Stop a deleted transcript from pausing the watcher

internal/client/watch/pressure.go

@@ -10,15 +10,65 @@ import (
 	"github.com/jssblck/akari/internal/client/upload"
 )
 
+// exhaustionErrnos are the kernel limits a later attempt can plausibly clear.
+// EAGAIN is a process table with no room to fork Git; the descriptor and memory
+// limits are the same story for a watcher tracking thousands of files. Backing
+// off gives the machine room, so these stay pressure.
+//
+// They are listed rather than inferred because the net.Error match below used to
+// classify them by accident, and an accident is not a contract.
+var exhaustionErrnos = [...]syscall.Errno{
+	syscall.EAGAIN,
+	syscall.EMFILE,
+	syscall.ENFILE,
+	syscall.ENOMEM,
+}
+
+// pressureFailure reports whether err is worth pausing the whole watcher for.
+// The worker re-marks a pressured file, so answering yes to a condition that
+// cannot change puts the watcher in a loop it never leaves.
 func pressureFailure(err error) bool {
 	if err == nil || errors.Is(err, context.Canceled) {
 		return false
 	}
-	if errors.Is(err, syscall.EAGAIN) || errors.Is(err, upload.ErrRetryableStatus) {
+	if errors.Is(err, upload.ErrRetryableStatus) {
 		return true
 	}
+	for _, errno := range exhaustionErrnos {
+		if errors.Is(err, errno) {
+			return true
+		}
+	}
+	return networkFailure(err)
+}
+
+// networkFailure reports whether err came from the transport rather than from
+// the filesystem.
+//
+// Matching the net.Error interface alone does not answer that. syscall.Errno
+// implements both Timeout() and Temporary(), so every errno satisfies net.Error
+// on its own, and errors.As walks straight past *fs.PathError — which has
+// Timeout() but no Temporary() — to land on the errno underneath. An ENOENT from
+// a transcript that was deleted between discovery and upload therefore matched
+// as firmly as a dropped connection.
+//
+// That is not a theoretical mismatch. On one macOS host it held the watcher in a
+// 30-second backoff loop over 938 session files from removed worktrees: each
+// attempt failed with ENOENT, was classified as pressure, was re-marked into the
+// dirty set, and slept. The set could never drain, and the loop wrote ~2.3MB a
+// day of identical errors into an unrotated log for eleven days.
+//
+// A genuine transport failure always presents a net package type ahead of any
+// errno — *url.Error from the upload client, with *net.OpError or *net.DNSError
+// beneath it — so a match that lands on a bare errno is the filesystem talking,
+// not the network.
+func networkFailure(err error) bool {
 	var networkError net.Error
-	return errors.As(err, &networkError)
+	if !errors.As(err, &networkError) {
+		return false
+	}
+	_, bareErrno := networkError.(syscall.Errno)
+	return !bareErrno
 }
 
 func waitForPressureBackoff(ctx context.Context, delay time.Duration) bool {

internal/client/watch/pressure_test.go

@@ -4,7 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io/fs"
 	"net"
+	"net/url"
+	"os"
+	"path/filepath"
 	"syscall"
 	"testing"
 	"time"
@@ -14,17 +18,44 @@ import (
 	"github.com/jssblck/akari/internal/client/upload"
 )
 
+// vanishedHeaderError is what the worker sees for a transcript deleted between
+// discovery and upload: resolve wraps the *fs.PathError that os.Lstat returned.
+func vanishedHeaderError(path string) error {
+	return fmt.Errorf("read session header: %w", &fs.PathError{
+		Op:   "lstat",
+		Path: path,
+		Err:  syscall.ENOENT,
+	})
+}
+
 func TestPressureFailureClassification(t *testing.T) {
 	tests := []struct {
 		name string
 		err  error
 		want bool
 	}{
 		{name: "dns", err: &net.DNSError{Err: "no such host", Name: "akari.example"}, want: true},
+		{name: "dial", err: &net.OpError{Op: "dial", Err: syscall.ECONNREFUSED}, want: true},
+		{
+			name: "upload transport",
+			err: &url.Error{
+				Op:  "Post",
+				URL: "https://akari.example/api/v1/ingest/session",
+				Err: &net.OpError{Op: "read", Err: syscall.ECONNRESET},
+			},
+			want: true,
+		},
 		{name: "process capacity", err: fmt.Errorf("start git: %w", syscall.EAGAIN), want: true},
+		{name: "descriptor capacity", err: fmt.Errorf("open: %w", syscall.EMFILE), want: true},
 		{name: "server status", err: fmt.Errorf("announce: %w", upload.ErrRetryableStatus), want: true},
 		{name: "canceled", err: context.Canceled, want: false},
-		{name: "file", err: errors.New("read session header"), want: false},
+		{name: "vanished session", err: vanishedHeaderError("/gone/session.jsonl"), want: false},
+		{
+			name: "unreadable session",
+			err:  fmt.Errorf("read session header: %w", &fs.PathError{Op: "open", Path: "/x", Err: syscall.EACCES}),
+			want: false,
+		},
+		{name: "opaque", err: errors.New("read session header"), want: false},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -35,6 +66,20 @@ func TestPressureFailureClassification(t *testing.T) {
 	}
 }
 
+// The synthetic *fs.PathError above only reproduces the bug for as long as the
+// standard library keeps its method set as it is. syscall.Errno satisfying
+// net.Error is what made a missing file look like a dropped connection, so take
+// the error from a real syscall too and let the kernel supply the shape.
+func TestPressureFailureOnRealMissingFile(t *testing.T) {
+	_, err := os.Lstat(filepath.Join(t.TempDir(), "never-written.jsonl"))
+	if err == nil {
+		t.Fatal("lstat of a missing path returned no error")
+	}
+	if pressureFailure(fmt.Errorf("read session header: %w", err)) {
+		t.Fatalf("a missing transcript was classified as resource pressure: %v", err)
+	}
+}
+
 func TestPressureBackoffDefault(t *testing.T) {
 	if got := (Options{}).withDefaults().PressureBackoff; got != 30*time.Second {
 		t.Fatalf("pressure backoff = %s, want 30s", got)
@@ -88,3 +133,53 @@ func TestWorkerBacksOffAndRetriesAfterPressure(t *testing.T) {
 		t.Fatal("worker did not stop")
 	}
 }
+
+// A deleted transcript is never coming back, so the worker must drop it. Keeping
+// it costs more than one wasted attempt: a re-marked file is retried forever, and
+// each retry pauses every other file behind the pressure backoff.
+func TestWorkerDropsVanishedFile(t *testing.T) {
+	const backoff = 10 * time.Second // long enough that a pause would fail the test
+	attempted := make(chan struct{}, 4)
+	file := discover.File{Agent: "claude", Path: "/gone/session.jsonl"}
+	w := &Watcher{
+		sync: func(context.Context, discover.File) syncer.Result {
+			attempted <- struct{}{}
+			return syncer.Result{File: file, Err: vanishedHeaderError(file.Path)}
+		},
+		opt: Options{PressureBackoff: backoff, Logf: func(string, ...any) {}},
+	}
+	rs := &runState{w: w, dirty: map[discover.File]struct{}{}, wake: make(chan struct{}, 1)}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		rs.worker(ctx)
+		close(done)
+	}()
+	rs.mark(file)
+
+	select {
+	case <-attempted:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not attempt the vanished file")
+	}
+	select {
+	case <-attempted:
+		t.Fatal("worker retried a vanished file instead of dropping it")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	rs.mu.Lock()
+	remaining := len(rs.dirty)
+	rs.mu.Unlock()
+	if remaining != 0 {
+		t.Fatalf("vanished file left %d entries in the dirty set, want 0", remaining)
+	}
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not stop")
+	}
+}