fix(st-k6z): serialize the counts refresh read-modify-write across processes

internal/counts/lock.go

@@ -0,0 +1,77 @@
+package counts
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+	"time"
+)
+
+// lockName is the advisory lock file inside the counts cache dir. It carries no
+// content — flock's kernel-side state is the lock, so a crashed holder releases it
+// automatically when its fd closes (unlike a hand-rolled O_EXCL marker, which would
+// strand the next run behind a stale file).
+const lockName = ".refresh.lock"
+
+// lockWait bounds how long a refresh waits for the holder. A wedged process must not
+// let launchd's every-minute fire pile up blocked refreshers forever; past the wait we
+// give up loudly instead of hanging.
+const lockWait = 2 * time.Minute
+
+// lockPoll is the retry interval while another process holds the lock.
+const lockPoll = 50 * time.Millisecond
+
+// errLockBusy is returned when the wait elapses with another refresh still holding the
+// lock — a wedged or very slow holder, not a transient overlap.
+var errLockBusy = errors.New("counts: refresh already running")
+
+// withLock runs fn while holding an exclusive cross-process lock on dir, so the whole
+// read-compute-write of counts.json + counts-mtimes is serialized (st-k6z). Atomic
+// writes alone are not enough: two refreshes (launchd's `--all` run, a manual
+// `strand counts`, and the server's per-write `strand counts <repo>` spawns) each read
+// the same base, compute, and replace the file — the last writer wins wholesale and
+// silently drops the other's rows. The lock closes that window; the atomic write still
+// protects concurrent *readers* mid-rename.
+//
+// The lock is advisory and per-fd, so it serializes goroutines in one process as well
+// as separate processes. It is taken non-blocking on a poll loop rather than with a
+// blocking LOCK_EX so ctx cancellation and lockWait both stay honored.
+func withLock(ctx context.Context, dir string, fn func() error) error {
+	f, err := os.OpenFile(filepath.Join(dir, lockName), os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return fmt.Errorf("counts: open lock: %w", err)
+	}
+	defer f.Close()
+
+	if err := acquire(ctx, f); err != nil {
+		return err
+	}
+	defer syscall.Flock(int(f.Fd()), syscall.LOCK_UN) //nolint:errcheck // closing the fd releases the lock regardless
+
+	return fn()
+}
+
+// acquire polls for the exclusive lock until it lands, ctx ends, or lockWait elapses.
+func acquire(ctx context.Context, f *os.File) error {
+	deadline := time.Now().Add(lockWait)
+	for {
+		err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+		if err == nil {
+			return nil
+		}
+		if !errors.Is(err, syscall.EWOULDBLOCK) {
+			return fmt.Errorf("counts: lock: %w", err)
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("%w (waited %s for %s)", errLockBusy, lockWait, f.Name())
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("counts: waiting for refresh lock: %w", ctx.Err())
+		case <-time.After(lockPoll):
+		}
+	}
+}

internal/counts/lock_test.go

@@ -0,0 +1,126 @@
+package counts
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dkoosis/strand/internal/bd"
+	"github.com/dkoosis/strand/internal/bdcounts"
+)
+
+// gatedSource is a source whose first read parks until released, so a test can hold
+// one refresh mid-compute while a second one starts against the same cache dir.
+type gatedSource struct {
+	fakeSource
+	entered chan struct{} // closed when the read starts
+	release chan struct{} // the read returns once this is closed
+	once    sync.Once
+}
+
+func (g *gatedSource) List(ctx context.Context, opts bd.ListOpts) ([]bd.Issue, error) {
+	g.once.Do(func() {
+		close(g.entered)
+		<-g.release
+	})
+	return g.fakeSource.List(ctx, opts)
+}
+
+// TestRefreshSerializesConcurrentRuns is the st-k6z regression: two refreshes over
+// disjoint repos, overlapping in time, must both land their rows in counts.json and
+// their gate entries in counts-mtimes. Atomic writes alone do not give this — before
+// the cache-dir lock, the second run read the base while the first was still computing
+// and then replaced the file, silently dropping the first run's repo.
+func TestRefreshSerializesConcurrentRuns(t *testing.T) {
+	projects := t.TempDir()
+	cache := t.TempDir()
+	a := mkRepo(t, projects, "repo-a")
+	b := mkRepo(t, projects, "repo-b")
+
+	gate := &gatedSource{
+		fakeSource: fakeSource{issues: []bd.Issue{{ID: "x", Status: bd.StatusOpen}}},
+		entered:    make(chan struct{}),
+		release:    make(chan struct{}),
+	}
+	slow := config{
+		cacheDir: cache, projects: projects, mode: modeExplicit, targets: []string{a},
+		newSource: func(string) source { return gate },
+	}
+	fast := config{
+		cacheDir: cache, projects: projects, mode: modeExplicit, targets: []string{b},
+		newSource: func(string) source { return oneOpenBead() },
+	}
+
+	errs := make(chan error, 2)
+	go func() { errs <- refresh(context.Background(), &slow) }()
+
+	select {
+	case <-gate.entered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("slow refresh never reached its first read")
+	}
+	go func() { errs <- refresh(context.Background(), &fast) }()
+	// Give the second run time to reach the lock (and, unlocked, to read the base and
+	// race ahead) before the first is allowed to finish.
+	time.Sleep(100 * time.Millisecond)
+	close(gate.release)
+
+	for range 2 {
+		select {
+		case err := <-errs:
+			if err != nil {
+				t.Fatalf("refresh: %v", err)
+			}
+		case <-time.After(10 * time.Second):
+			t.Fatal("refresh did not return — lock not released?")
+		}
+	}
+
+	r := bdcounts.NewReaderAt(filepath.Join(cache, "counts.json"))
+	for _, root := range []string{a, b} {
+		if _, ok := r.Lookup(root); !ok {
+			t.Errorf("no counts.json row for %s — a concurrent refresh dropped it", root)
+		}
+	}
+	state, err := os.ReadFile(filepath.Join(cache, "counts-mtimes"))
+	if err != nil {
+		t.Fatalf("read state: %v", err)
+	}
+	for _, root := range []string{a, b} {
+		if !strings.Contains(string(state), root+"\t") {
+			t.Errorf("no counts-mtimes entry for %s — a concurrent refresh dropped it", root)
+		}
+	}
+}
+
+// TestRefreshLockTimesOutOnWedgedHolder: a holder that never releases must not hang a
+// refresh forever — past the wait the run fails loudly so launchd's next fire does not
+// pile up blocked refreshers behind it.
+func TestRefreshLockTimesOutOnWedgedHolder(t *testing.T) {
+	cache := t.TempDir()
+	held := make(chan struct{})
+	holding := make(chan struct{})
+	go func() {
+		_ = withLock(context.Background(), cache, func() error {
+			close(holding)
+			<-held
+			return nil
+		})
+	}()
+	<-holding
+	defer close(held)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // stands in for the lockWait deadline, without a 2-minute test
+	err := withLock(ctx, cache, func() error {
+		t.Error("fn ran while another holder had the lock")
+		return nil
+	})
+	if err == nil {
+		t.Fatal("withLock returned nil while the lock was held")
+	}
+}

internal/counts/refresh.go

@@ -89,7 +89,20 @@ func Run(args []string, version string) error {
 // this cycle always schedules exactly one guaranteed follow-up derive next cycle
 // (regardless of whether the mtime moves again), by which time bd's commit has
 // settled. See repoState.next for the exact carry-forward rule.
+// The whole sequence runs under one cross-process lock (st-k6z): the reads below are
+// the R of a read-modify-write, so the lock opens before them, not just around the
+// writes. Without it, the launchd `--all` run, a manual `strand counts`, and the
+// server's per-write `strand counts <repo>` spawns each read the same base and replace
+// the file — last writer wins wholesale, dropping the others' rows.
 func refresh(ctx context.Context, cfg *config) error {
+	if err := os.MkdirAll(cfg.cacheDir, 0o755); err != nil {
+		return fmt.Errorf("counts: cache dir: %w", err)
+	}
+	return withLock(ctx, cfg.cacheDir, func() error { return refreshLocked(ctx, cfg) })
+}
+
+// refreshLocked is refresh's body, run with the cache-dir lock held.
+func refreshLocked(ctx context.Context, cfg *config) error {
 	targets := cfg.targets
 	if cfg.mode != modeExplicit {
 		targets = discover(cfg.projects)
@@ -120,9 +133,6 @@ func refresh(ctx context.Context, cfg *config) error {
 		nextState[root] = repoState{mtime: cur, pending: nextPending(cfg.mode, mtimeChanged, err)}
 	}
 
-	if err := os.MkdirAll(cfg.cacheDir, 0o755); err != nil {
-		return fmt.Errorf("counts: cache dir: %w", err)
-	}
 	// counts.json is rewritten every run, even when no row changed: the write stamps a
 	// fresh liveness meta (last-run time + binary version) so a dead or wedged refresher
 	// shows a stale flag in the masthead instead of freezing the file indistinguishably
@@ -210,12 +220,9 @@ func readRows(path string) map[string]Row {
 // under bdcounts.MetaKey — a reserved key no repo path collides with — so keyed readers
 // (bdcounts.Reader.Lookup, the status line's `.[$repo]`) are untouched by its presence.
 //
-// NOTE (RMW race, not fixed by atomic write): two concurrent refreshes (the
-// launchd --all run and a manual `strand counts`) each read-compute-write the
-// whole rows set independently. atomicfile.WriteFile stops either write from
-// being torn, but it does not serialize the two writers — whichever finishes
-// last wins wholesale, silently dropping the other's rows. Follow-up: a lock
-// around the refresh, or a merge-on-write like writeState's.
+// The atomic write protects concurrent *readers* from a half-written file; concurrent
+// *writers* are handled a level up, by refresh's cache-dir lock (st-k6z) — this
+// function assumes the caller holds it.
 func writeRowsAtomic(path string, rows map[string]Row, meta bdcounts.Meta) error {
 	out := make(map[string]any, len(rows)+1)
 	for k, v := range rows {
@@ -278,13 +285,8 @@ func readState(path string) map[string]repoState {
 // a plain overwrite would truncate every other repo's gate entry — the next launchd
 // changed-mode run would then find no prior mtime for those repos and cold-recompute
 // them all (st-dd9). Reading the prior state and overlaying keeps the untouched repos'
-// gate entries intact.
-//
-// NOTE (RMW race, not fixed by atomic write): the read-merge-write above is not
-// serialized against a concurrent writer — two refreshes racing this function can
-// each read the same prior state, merge their own entries on top, and whichever
-// atomicfile.WriteFile finishes last wins wholesale, silently dropping the other's
-// merged entries. Same follow-up as writeRowsAtomic.
+// gate entries intact. The merge is a within-run concern; cross-run serialization of
+// this read-merge-write is refresh's cache-dir lock (st-k6z), which the caller holds.
 func writeState(path string, state map[string]repoState) error {
 	merged := readState(path)
 	maps.Copy(merged, state)