Fix: Self-heal corrupt session metadata and bound the lock wait

authbridge/authlib/observe/claude/harvest_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -222,27 +223,170 @@ func TestHarvest_IncrementalRequiresMerge(t *testing.T) {
 	}
 }
 
-// A corrupt existing file is refused under Merge, distinguishably, so the command layer
-// can name --merge=false as the way past.
-func TestHarvest_CorruptMetadataIsDistinguishable(t *testing.T) {
+// writeMetadataFile puts raw bytes at the metadata path, creating the directory.
+//
+// Its own helper because every corrupt-file test needs the same four lines, and the mode
+// matters: SaveMetadata writes 0o600, so a fixture that differs would test a file the
+// product never produces.
+func writeMetadataFile(t *testing.T, body string) string {
+	t.Helper()
+	path, err := SessionMetadataPath()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+// A file that does not parse is rebuilt from the transcripts rather than refused, so the
+// titles come back on their own instead of waiting for a hand-run --merge=false.
+func TestHarvest_CorruptMetadataIsRebuilt(t *testing.T) {
 	metadataHome(t)
 	cfg := filepath.Join(t.TempDir(), "claude")
 	writeSessionTranscript(t, filepath.Join(cfg, "projects", "-p"), "s1.jsonl",
 		`{"type":"ai-title","aiTitle":"t"}`)
+	path := writeMetadataFile(t, "{not json")
+
+	// Incremental too, which is how `abctl observe` calls it: the rebuild has to clear the
+	// baseline, or a skip test against an empty map is the only thing making this work.
+	res, err := Harvest(Options{ConfigDir: cfg, Merge: true, Incremental: true})
+	if err != nil {
+		t.Fatalf("Harvest: %v", err)
+	}
+	if !res.Rebuilt {
+		t.Error("Rebuilt = false, want true: the caller cannot tell a rebuild from a first run")
+	}
+	if got := res.Meta["s1"].Title; got != "t" {
+		t.Errorf("Meta[s1].Title = %q, want the harvested title", got)
+	}
+	if got := readMetadataFile(t, path)["s1"].Title; got != "t" {
+		t.Errorf("on disk Title = %q, want the file replaced with the rebuild", got)
+	}
+}
+
+// An UNREADABLE file is still refused, and left alone. This is the destructive case the
+// rebuild must not reach: a permission or I/O failure says nothing about the contents, so
+// replacing the file there would drop entries that are very likely intact.
+func TestHarvest_UnreadableMetadataIsRefusedNotRebuilt(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root ignores the mode bits this test relies on")
+	}
+	metadataHome(t)
+	cfg := filepath.Join(t.TempDir(), "claude")
+	writeSessionTranscript(t, filepath.Join(cfg, "projects", "-p"), "s1.jsonl",
+		`{"type":"ai-title","aiTitle":"t"}`)
+	// VALID JSON, so the only thing making this unreadable is the mode. A corrupt body here
+	// would let the test pass for the wrong reason if the classification were inverted.
+	path := writeMetadataFile(t, `{"old":{"title":"keep me"}}`)
+	if err := os.Chmod(path, 0o000); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.Chmod(path, 0o600) })
+
+	res, err := Harvest(Options{ConfigDir: cfg, Merge: true})
+	if !errors.Is(err, ErrCorruptMetadata) {
+		t.Fatalf("err = %v, want it to wrap ErrCorruptMetadata", err)
+	}
+	if res.Rebuilt {
+		t.Error("Rebuilt = true: an unreadable file must not be rebuilt over")
+	}
+	if err := os.Chmod(path, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if got := readMetadataFile(t, path)["old"].Title; got != "keep me" {
+		t.Errorf("Title = %q, want the untouched entry: the file was replaced", got)
+	}
+}
+
+// An oversized but VALID file is refused, not rebuilt over. The read cap makes such a file
+// fail json.Unmarshal, which would classify it as a parse failure and destroy it: measured at
+// 18.5MB/17000 entries replaced by 412 bytes/1 entry.
+func TestHarvest_OversizedValidMetadataIsRefusedNotRebuilt(t *testing.T) {
+	metadataHome(t)
 	path, err := SessionMetadataPath()
 	if err != nil {
 		t.Fatal(err)
 	}
 	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
 		t.Fatal(err)
 	}
-	if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
+	// Built as text rather than by marshalling a map: cheaper, and it keeps the fixture's size
+	// the point of the test rather than a side effect of the struct.
+	var b []byte
+	b = append(b, '{')
+	pad := strings.Repeat("x", 4<<10)
+	for i := 0; len(b) <= 16<<20; i++ {
+		if i > 0 {
+			b = append(b, ',')
+		}
+		b = append(b, fmt.Sprintf("%q:{\"title\":%q}", fmt.Sprintf("sess-%06d", i), pad)...)
+	}
+	b = append(b, '}')
+	if !json.Valid(b) {
+		t.Fatal("fixture is not valid JSON, so this would not test the rebuild path")
+	}
+	if err := os.WriteFile(path, b, 0o600); err != nil {
 		t.Fatal(err)
 	}
 
-	_, err = Harvest(Options{ConfigDir: cfg, Merge: true})
+	cfg := filepath.Join(t.TempDir(), "claude")
+	writeSessionTranscript(t, filepath.Join(cfg, "projects", "-p"), "fresh.jsonl",
+		`{"type":"ai-title","aiTitle":"new"}`)
+
+	res, err := Harvest(Options{ConfigDir: cfg, Merge: true})
+	if err == nil {
+		t.Fatalf("Harvest succeeded (Rebuilt=%v); want it refused rather than rebuilding", res.Rebuilt)
+	}
 	if !errors.Is(err, ErrCorruptMetadata) {
-		t.Fatalf("err = %v, want it to wrap ErrCorruptMetadata", err)
+		t.Errorf("err = %v, want it to wrap ErrCorruptMetadata", err)
+	}
+	// Distinguishable through both wraps, so a caller can give the right remedy: this file's
+	// permissions are fine and "fix the permissions" would be wrong advice.
+	if !errors.Is(err, ErrMetadataTooLarge) {
+		t.Errorf("err = %v, want it to wrap ErrMetadataTooLarge", err)
+	}
+	if res.Rebuilt {
+		t.Error("Rebuilt is true for a file that was merely too large to read")
+	}
+	// The assertion that matters: the bytes are still there.
+	st, serr := os.Stat(path)
+	if serr != nil {
+		t.Fatal(serr)
+	}
+	if got := st.Size(); got != int64(len(b)) {
+		t.Errorf("file is %d bytes, was %d — it was rewritten", got, len(b))
+	}
+}
+
+// A rebuild is still a merge otherwise: it replaces the unparseable file, and the next
+// harvest keeps what it wrote.
+func TestHarvest_RebuildThenMergeKeepsEntries(t *testing.T) {
+	metadataHome(t)
+	cfg := filepath.Join(t.TempDir(), "claude")
+	dir := filepath.Join(cfg, "projects", "-p")
+	writeSessionTranscript(t, dir, "s1.jsonl", `{"type":"ai-title","aiTitle":"one"}`)
+	writeMetadataFile(t, "{not json")
+
+	if _, err := Harvest(Options{ConfigDir: cfg, Merge: true}); err != nil {
+		t.Fatalf("rebuild: %v", err)
+	}
+	writeSessionTranscript(t, dir, "s2.jsonl", `{"type":"ai-title","aiTitle":"two"}`)
+	res, err := Harvest(Options{ConfigDir: cfg, Merge: true})
+	if err != nil {
+		t.Fatalf("second harvest: %v", err)
+	}
+	if res.Rebuilt {
+		t.Error("Rebuilt = true on a file this package just wrote")
+	}
+	for id, want := range map[string]string{"s1": "one", "s2": "two"} {
+		if got := res.Meta[id].Title; got != want {
+			t.Errorf("Meta[%s].Title = %q, want %q", id, got, want)
+		}
 	}
 }
 
@@ -2709,3 +2853,42 @@ func TestTitleFromTranscript_CwdIsBounded(t *testing.T) {
 			got[max(0, len(got)-20):], "/the-leaf")
 	}
 }
+
+// A short write must not be renamed over the good file. This is the seam writeAll exists for:
+// the failure is a write error that Close and Rename both survive, which no real filesystem
+// produces, and the bug it guards — `err :=` shadowing inside SaveMetadata — is invisible to
+// every test that writes to a working disk.
+func TestSaveMetadata_AFailedWriteKeepsTheOldFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "session-metadata.json")
+	const keep = `{"old":{"title":"keep me"}}`
+	if err := os.WriteFile(path, []byte(keep), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	was := writeAll
+	writeAll = func(io.Writer, []byte) (int, error) { return 0, errors.New("disk on fire") }
+	t.Cleanup(func() { writeAll = was })
+
+	if err := SaveMetadata(path, map[string]SessionMetadata{"new": {Title: "t"}}); err == nil {
+		t.Fatal("SaveMetadata returned nil after the write failed")
+	}
+	// The point of the test: the old file, not a truncated new one.
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(b) != keep {
+		t.Errorf("file = %q, want the original %q", b, keep)
+	}
+	// And no temp file left behind to accumulate one per failure.
+	ents, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range ents {
+		if strings.HasSuffix(e.Name(), ".tmp") {
+			t.Errorf("temp file %s was left behind", e.Name())
+		}
+	}
+}

authbridge/authlib/observe/claude/lock_unix.go

@@ -3,11 +3,52 @@
 package claude
 
 import (
+	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
 	"syscall"
+	"time"
 )
 
+// lockTimeout bounds how long a harvest waits for the metadata lock before proceeding
+// without it.
+//
+// THE TWO COSTS ARE NOT SYMMETRIC, which is what sets the size. Too long only delays a
+// harvest that was going to be wrong anyway — the wedged case this bound exists for, where
+// no wait of any length succeeds. Too short corrupts a HEALTHY run: expiry means proceeding
+// unlocked, and an unlocked harvest can have its whole contribution erased by another run's
+// rename. So the deadline has to sit far above the slowest legitimate wait, and the penalty
+// for overshooting is measured in seconds of a background goroutine nobody is watching.
+//
+// SIZED AGAINST A QUEUE, not a single scan, because the wait is the queue ahead of you and
+// not the holder alone. That was the sizing error in the first version of this: 2s was chosen
+// as "many times one scan" from a ~3ms one-file measurement, and it fell over as soon as
+// several harvests contended on a slow machine. Six concurrent harvests of a padded tree took
+// ~190ms serialized on a developer laptop and blew straight past 2s on a shared CI runner —
+// the package's own concurrency test failed, with five of six children losing every entry.
+// A runner is not an exotic environment; it is the slowest machine this code routinely runs on
+// and therefore the one that sets the number.
+//
+// A VAR RATHER THAN A CONST only so the tests can shrink it: waiting the real deadline twice
+// would add a minute to the package for no extra coverage, and a test that asserts the logic at
+// 40ms asserts exactly the same logic. Nothing outside the tests assigns it.
+//
+// 30s is deliberately far past any queue this file can produce. `abctl observe` harvests one
+// tree per tick, `read-claude-sessions` is one process, and the realistic worst case is a
+// handful of viewers plus a manual run — a queue of seconds, not minutes, even derated for a
+// loaded runner. What 30s buys is that reaching it means no wait would have worked.
+var lockTimeout = 30 * time.Second
+
+// lockPoll is how often acquisition is retried inside lockTimeout.
+//
+// Polled rather than blocking because the two are exclusive in this API: syscall.Flock
+// either blocks forever (LOCK_EX) or returns at once (LOCK_NB), and there is no
+// deadline variant. 20ms adds at most 20ms to an uncontended handoff, and costs 1500
+// cheap syscalls across a full 30s timeout — paid only by a harvest that is already
+// losing, since a lock that frees up is acquired on the next tick.
+const lockPoll = 20 * time.Millisecond
+
 // lockMetadata takes an exclusive advisory lock covering the read-modify-write of the metadata
 // file, and returns the release.
 //
@@ -24,13 +65,33 @@ import (
 // replaces the metadata file itself. Locking the metadata file would lock an inode the rename is
 // about to detach, protecting nothing.
 //
-// Blocking, with no timeout. A harvest holds this for the length of one scan, and the failure mode
-// of a timeout here is the lost update this exists to prevent; a caller that cannot wait should not
-// be harvesting. A crashed holder releases on process exit, since the kernel owns the lock.
+// BOUNDED BY lockTimeout, then it gives up and lets the caller proceed unlocked. This used to
+// block indefinitely, on the reasoning that a caller who cannot wait should not be harvesting.
+// The case that reasoning did not cover is a holder that never releases — not a crash, which the
+// kernel cleans up on process exit, but a process still alive and stuck. `abctl observe` harvests
+// on a timer, so every later attempt queued behind the same lock and the viewer showed no titles
+// at all, indefinitely, with nothing on screen to say why.
+//
+// What the bound costs, and why the deadline is generous rather than tight: past it two harvests
+// interleave, and the loser's entries can be erased wholesale by the winner's rename. That is the
+// very lost update this lock exists to prevent, so expiry must mean "no wait would have worked"
+// and never "this machine is slow today". recoverConcurrentEntries is a weaker backstop than it
+// looks — a SINGLE re-read after saving, which by its own documentation cannot converge when more
+// than one rename lands inside its window, i.e. exactly the many-writer case a short deadline
+// creates. It does not cover for a deadline that fires under ordinary contention.
+//
+// So: unlocked is strictly better than wedged, and strictly worse than locked — which makes the
+// bound worth having and worth sizing so that only the wedged case ever reaches it. See
+// lockTimeout for the measurements that set the number.
 //
 // Errors are returned rather than swallowed so the caller can proceed UNLOCKED instead of refusing
 // to harvest: a filesystem that cannot flock (some network mounts) should still get titles, on the
-// same best-effort footing as before this existed.
+// same best-effort footing as before this existed. A timeout joins that path, reported as
+// ErrLockTimeout so the two are distinguishable.
+//
+// The release func is nil on every error return, which the caller relies on to decide whether to
+// defer it — see Harvest's `if lerr == nil` gate. Returning a no-op func alongside an error would
+// make that gate look optional, and it is not.
 func lockMetadata(path string) (func(), error) {
 	lockPath := filepath.Join(filepath.Dir(path), "."+filepath.Base(path)+".lock")
 	if err := os.MkdirAll(filepath.Dir(lockPath), 0o700); err != nil {
@@ -40,9 +101,33 @@ func lockMetadata(path string) (func(), error) {
 	if err != nil {
 		return nil, err
 	}
-	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
-		_ = f.Close()
-		return nil, err
+	// LOCK_NB polled to a deadline. EWOULDBLOCK (EAGAIN on Linux, where they are the same errno)
+	// is the one retryable answer: it means held, not broken. Every other errno is a real failure
+	// and returns at once, exactly as the blocking call used to.
+	//
+	// UNTESTED BRANCH, said out loud rather than left as a silent hole: no test distinguishes this
+	// discrimination from "retry on every errno", because provoking a non-EWOULDBLOCK flock error
+	// needs a filesystem that refuses flock outright (some network mounts) — not something a unit
+	// test can conjure. A mutation that retries every errno passes the whole suite. What it would
+	// cost in production is the timeout spent sleeping against an error that will never change,
+	// then reported as a lock timeout rather than as the ENOLCK it was.
+	deadline := time.Now().Add(lockTimeout)
+	for {
+		err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+		if err == nil {
+			break
+		}
+		if !errors.Is(err, syscall.EWOULDBLOCK) {
+			_ = f.Close()
+			return nil, err
+		}
+		if !time.Now().Before(deadline) {
+			// Closed here too. The caller proceeds unlocked and never sees this descriptor, so
+			// leaking it would leak one per harvest — and `abctl observe` harvests on a timer.
+			_ = f.Close()
+			return nil, fmt.Errorf("%w after %s", ErrLockTimeout, lockTimeout)
+		}
+		time.Sleep(lockPoll)
 	}
 	return func() {
 		// Unlock before close, though closing the descriptor would release it anyway: being