From 0e987c20aef3b714dc31804c2e921bee166d3f91 Mon Sep 17 00:00:00 2001 From: "dkoosis@gmail.com" Date: Wed, 12 Aug 2026 19:04:32 -0400 Subject: [PATCH 1/2] fix(st-k6z): serialize the counts refresh read-modify-write across processes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit atomicfile stopped torn writes; it did not stop lost updates. The launchd --all run, a manual `strand counts`, and the server's per-write `strand counts ` spawns each read counts.json + counts-mtimes, compute, and replace both files — last writer wins wholesale. Take an advisory flock on the cache dir for the whole refresh, opened before the first read. Non-blocking poll so ctx cancellation and a 2-minute wait cap stay honored: a wedged holder fails the run loudly instead of piling up blocked refreshers behind launchd's every-minute fire. --- internal/counts/lock.go | 77 +++++++++++++++++++++ internal/counts/lock_test.go | 126 +++++++++++++++++++++++++++++++++++ internal/counts/refresh.go | 34 +++++----- 3 files changed, 221 insertions(+), 16 deletions(-) create mode 100644 internal/counts/lock.go create mode 100644 internal/counts/lock_test.go diff --git a/internal/counts/lock.go b/internal/counts/lock.go new file mode 100644 index 0000000..fab0505 --- /dev/null +++ b/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 ` 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): + } + } +} diff --git a/internal/counts/lock_test.go b/internal/counts/lock_test.go new file mode 100644 index 0000000..1d2402c --- /dev/null +++ b/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") + } +} diff --git a/internal/counts/refresh.go b/internal/counts/refresh.go index 928fa49..6162987 100644 --- a/internal/counts/refresh.go +++ b/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 ` 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) From da1af901d7a17410d349854c335714d89462ab26 Mon Sep 17 00:00:00 2001 From: "dkoosis@gmail.com" Date: Wed, 12 Aug 2026 19:11:29 -0400 Subject: [PATCH 2/2] fix(st-k6z): cancel the lock wait with the caller's context Codex review: defaultRefreshCounts discarded goBackground's context and counts.Run used context.Background(), so a slow lock holder could keep every queued post-write goroutine alive past its 10s timeout and pin Server.Stop's bgWG wait behind lockWait. Add counts.RunContext for in-process callers that own a lifetime; Run keeps context.Background() as the CLI root. Cover the lockWait and cancellation paths with named-error assertions (CodeRabbit). --- internal/counts/lock.go | 5 ++-- internal/counts/lock_test.go | 53 +++++++++++++++++++++++++++++------- internal/counts/refresh.go | 10 ++++++- internal/server/server.go | 9 ++++-- 4 files changed, 62 insertions(+), 15 deletions(-) diff --git a/internal/counts/lock.go b/internal/counts/lock.go index fab0505..3196797 100644 --- a/internal/counts/lock.go +++ b/internal/counts/lock.go @@ -18,8 +18,9 @@ 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 +// give up loudly instead of hanging. A var, not a const, so the timeout test can shrink +// it rather than sleep for two minutes. +var lockWait = 2 * time.Minute // lockPoll is the retry interval while another process holds the lock. const lockPoll = 50 * time.Millisecond diff --git a/internal/counts/lock_test.go b/internal/counts/lock_test.go index 1d2402c..ae07a01 100644 --- a/internal/counts/lock_test.go +++ b/internal/counts/lock_test.go @@ -2,6 +2,7 @@ package counts import ( "context" + "errors" "os" "path/filepath" "strings" @@ -97,30 +98,62 @@ func TestRefreshSerializesConcurrentRuns(t *testing.T) { } } -// 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() +// holdLock parks an exclusive holder on dir and returns once it is held; the cleanup +// releases it. +func holdLock(t *testing.T, dir string) { + t.Helper() held := make(chan struct{}) holding := make(chan struct{}) + done := make(chan struct{}) go func() { - _ = withLock(context.Background(), cache, func() error { + defer close(done) + _ = withLock(context.Background(), dir, func() error { close(holding) <-held return nil }) }() <-holding - defer close(held) + t.Cleanup(func() { + close(held) + <-done + }) +} + +// TestRefreshLockGivesUpOnWedgedHolder: a holder that never releases must not hang a +// refresh forever — past lockWait the run fails with errLockBusy, so launchd's next +// fire does not pile up blocked refreshers behind it. +func TestRefreshLockGivesUpOnWedgedHolder(t *testing.T) { + cache := t.TempDir() + holdLock(t, cache) + + orig := lockWait + lockWait = 150 * time.Millisecond // the real 2 minutes, without a 2-minute test + t.Cleanup(func() { lockWait = orig }) + + err := withLock(context.Background(), cache, func() error { + t.Error("fn ran while another holder had the lock") + return nil + }) + if !errors.Is(err, errLockBusy) { + t.Fatalf("withLock error = %v, want errLockBusy", err) + } +} + +// TestRefreshLockHonorsContextCancel: the wait for the lock ends when the caller's +// context does — the server's post-write refresh runs under goBackground's 10s +// timeout and Server.Stop, neither of which may be pinned behind lockWait. +func TestRefreshLockHonorsContextCancel(t *testing.T) { + cache := t.TempDir() + holdLock(t, cache) ctx, cancel := context.WithCancel(context.Background()) - cancel() // stands in for the lockWait deadline, without a 2-minute test + cancel() 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") + if !errors.Is(err, context.Canceled) { + t.Fatalf("withLock error = %v, want context.Canceled", err) } } diff --git a/internal/counts/refresh.go b/internal/counts/refresh.go index 6162987..941ea4b 100644 --- a/internal/counts/refresh.go +++ b/internal/counts/refresh.go @@ -48,6 +48,14 @@ type config struct { // strand counts --all # every discovered repo, unconditionally // strand counts ... # only the named repo roots func Run(args []string, version string) error { + return RunContext(context.Background(), args, version) //nolint:forbidigo // CLI subcommand root: `strand counts` is a one-shot command, not request-scoped +} + +// RunContext is Run for an in-process caller that owns a lifetime — the server's +// post-write refresh, which runs under goBackground's 10s timeout and dies with +// Server.Stop. Cancelling ctx aborts both the bd reads and the wait for the refresh +// lock, so a slow holder cannot pin a shutdown behind lockWait (st-k6z review). +func RunContext(ctx context.Context, args []string, version string) error { fs := flag.NewFlagSet("counts", flag.ContinueOnError) all := fs.Bool("all", false, "refresh every discovered repo unconditionally") bin := fs.String("bd", "bd", "path to the bd binary") @@ -70,7 +78,7 @@ func Run(args []string, version string) error { case *all: cfg.mode = modeAll } - return refresh(context.Background(), &cfg) //nolint:forbidigo // CLI subcommand root: `strand counts` is a one-shot command, not request-scoped + return refresh(ctx, &cfg) } // refresh visits the run's repos, recomputes each row, and writes counts.json. It is diff --git a/internal/server/server.go b/internal/server/server.go index 7c5e425..f60daaa 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -254,9 +254,14 @@ func (s *Server) goBackground(timeout time.Duration, fn func(context.Context)) { // seconds. A failure is logged, not surfaced: the write itself already // succeeded, and a stale count self-heals on the next poll or the next launchd // cycle either way. +// +// It runs on goBackground's context, not a detached one: a refresh waits on the +// cross-process counts lock (st-k6z), so without cancellation a slow holder would keep +// every queued post-write goroutine alive past the 10s timeout and pin Stop's bgWG +// wait behind it. func (s *Server) defaultRefreshCounts(repo registry.Repo) { - s.goBackground(10*time.Second, func(_ context.Context) { - if err := counts.Run([]string{repo.Path}, Version); err != nil { + s.goBackground(10*time.Second, func(ctx context.Context) { + if err := counts.RunContext(ctx, []string{repo.Path}, Version); err != nil { log.Printf("strand: post-write counts refresh for %s: %v", repo.Path, err) } })