From 6a74b64c7a5f4dea4115f62d5e097c552744e201 Mon Sep 17 00:00:00 2001 From: Steve Dudenhoeffer Date: Sat, 18 Jul 2026 12:47:36 -0400 Subject: [PATCH] =?UTF-8?q?fix(concurrency):=20address=20gadfly=20review?= =?UTF-8?q?=20=E2=80=94=20fail-open=20pool=20+=20doc=20drift?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gadfly's own review of #27 surfaced a real robustness cluster (5 models, error-handling) plus stale comments I missed. Robustness — the flock permit pool could hang forever: - tryAcquire swallowed os.OpenFile errors and treated every flock error as "busy", so a broken/missing pool dir (or a filesystem without flock) would spin-poll indefinitely; the fanout context is uncancellable and the per-lens timeout only starts AFTER acquire returns. Can't trigger in the deploy (entrypoint mkdir -p's the dir) but fixed defensively. - tryAcquire now returns a structural error, distinguished from a healthy-full pool (EWOULDBLOCK = busy → keep polling). acquire FAILS OPEN on a structural error: logs once and runs the lens unthrottled rather than hanging the review. - acquire uses time.NewTimer + Stop() (no per-poll timer leak on cancellation). - activeLensSem warns on stderr when GADFLY_LENS_SEM_DIR is set but the size is invalid (was a silent degrade to unthrottled). - New test: a broken pool dir fails open promptly. Doc drift (stale references to the removed model cap): - main.go defaultLensConcurrency + runSpecialists doc, entrypoint.sh status pre-seed + lane-launch comments, and the pre-existing lens_concurrency_test.go header all updated to the provider-wide-budget wording. Accepted (graded real, not changed): all-models-start-at-once startup burst (intended tradeoff) and index-0 sweep bias (cosmetic). One false positive (one model using the whole budget is the intended lone-model behavior). Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/gadfly/lens_concurrency_test.go | 2 +- cmd/gadfly/lenssem.go | 57 ++++++++++++++++++++++------- cmd/gadfly/lenssem_test.go | 48 ++++++++++++++++++------ cmd/gadfly/main.go | 29 ++++++++------- entrypoint.sh | 10 ++--- 5 files changed, 101 insertions(+), 45 deletions(-) diff --git a/cmd/gadfly/lens_concurrency_test.go b/cmd/gadfly/lens_concurrency_test.go index 4b67a6a..1ea5e81 100644 --- a/cmd/gadfly/lens_concurrency_test.go +++ b/cmd/gadfly/lens_concurrency_test.go @@ -156,7 +156,7 @@ func TestRunSpecialists_PerProviderFanOut(t *testing.T) { // TestLensConcurrency covers the resolution matrix: scalar default, scalar // override, and per-provider override keyed by the model's resolved lane (same -// lane rule entrypoint.sh uses for GADFLY_PROVIDER_CONCURRENCY). +// lane rule entrypoint.sh uses for GADFLY_PROVIDER_LENS_CONCURRENCY). func TestLensConcurrency(t *testing.T) { tests := []struct { name string diff --git a/cmd/gadfly/lenssem.go b/cmd/gadfly/lenssem.go index ad806a4..50fd187 100644 --- a/cmd/gadfly/lenssem.go +++ b/cmd/gadfly/lenssem.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "os" "path/filepath" @@ -43,44 +44,72 @@ func activeLensSem() *lensSem { } n := envInt("GADFLY_LENS_SEM_SIZE", 0) if n < 1 { + // A dir was requested but the size is missing/blank/invalid. Rather than + // silently run unthrottled, say so on stderr — it's almost certainly a + // misconfiguration in entrypoint.sh (the two are set together). + fmt.Fprintf(os.Stderr, "gadfly: GADFLY_LENS_SEM_DIR set but GADFLY_LENS_SEM_SIZE=%q invalid; lenses run unthrottled\n", os.Getenv("GADFLY_LENS_SEM_SIZE")) return nil } return &lensSem{dir: dir, size: n} } // acquire blocks until a permit is free (or ctx is done) and returns a release -// func. On cancellation it returns a no-op release plus ctx.Err(), so callers can -// surface the lens as "did not run" rather than leaking a permit. +// func. It FAILS OPEN: if the pool directory is structurally unusable (missing, +// unwritable, or on a filesystem without flock) it logs once and returns a no-op +// release with a nil error, so the lens runs UNTHROTTLED rather than hanging the +// whole review forever — this is an advisory reviewer, a slightly oversubscribed +// backend beats a stuck one. Only a full-but-healthy pool actually blocks; on ctx +// cancellation it returns a no-op release plus ctx.Err() so the caller can surface +// the lens as "did not run" instead of leaking a permit. func (s *lensSem) acquire(ctx context.Context) (func(), error) { for { - if release, ok := s.tryAcquire(); ok { + release, ok, err := s.tryAcquire() + if ok { return release, nil } + if err != nil { + fmt.Fprintf(os.Stderr, "gadfly: lens permit pool %q unusable (%v); proceeding without a permit\n", s.dir, err) + return func() {}, nil + } + timer := time.NewTimer(lensSemPollInterval) select { case <-ctx.Done(): + timer.Stop() // don't leak the timer when we bail on cancellation return func() {}, ctx.Err() - case <-time.After(lensSemPollInterval): + case <-timer.C: } } } -// tryAcquire makes one non-blocking sweep over the permit files, returning a -// release func for the first one it locks. The permit files are created lazily -// (entrypoint.sh only guarantees the directory exists), so a fresh pool needs no -// seeding step beyond mkdir. Closing the *os.File in the returned func drops the -// flock (the lock lives on the open file description). -func (s *lensSem) tryAcquire() (func(), bool) { +// tryAcquire makes one non-blocking sweep over the permit files. It returns a +// release func for the first one it locks; otherwise it distinguishes a healthy +// FULL pool (some permit was openable but flock-busy → ok=false, err=nil → keep +// polling) from a structurally BROKEN pool (no permit was even acquirable and none +// was merely busy → err set → caller fails open). Permit files are created lazily +// (entrypoint.sh only guarantees the directory exists). Closing the *os.File in +// the returned func drops the flock (the lock lives on the open file description). +func (s *lensSem) tryAcquire() (func(), bool, error) { + sawBusy := false + var structural error for i := 0; i < s.size; i++ { path := filepath.Join(s.dir, fmt.Sprintf("permit.%d", i)) f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600) if err != nil { + structural = err // e.g. the pool dir is gone or unwritable continue } - if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err == nil { + return func() { f.Close() }, true, nil + } else if errors.Is(err, syscall.EWOULDBLOCK) { + sawBusy = true // this permit is held by someone else — pool has capacity + f.Close() + } else { + structural = err // flock unsupported / other → not a transient "busy" f.Close() - continue } - return func() { f.Close() }, true } - return nil, false + if sawBusy { + return nil, false, nil // full but healthy: another lens will free a permit + } + return nil, false, structural } diff --git a/cmd/gadfly/lenssem_test.go b/cmd/gadfly/lenssem_test.go index f42472f..52cbc69 100644 --- a/cmd/gadfly/lenssem_test.go +++ b/cmd/gadfly/lenssem_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "path/filepath" "sync" "sync/atomic" "testing" @@ -29,27 +30,52 @@ func TestActiveLensSemUnset(t *testing.T) { func TestLensSemExhaustionAndRelease(t *testing.T) { s := &lensSem{dir: t.TempDir(), size: 2} - r1, ok := s.tryAcquire() - if !ok { - t.Fatal("first acquire should succeed") + r1, ok, err := s.tryAcquire() + if !ok || err != nil { + t.Fatalf("first acquire should succeed: ok=%v err=%v", ok, err) } - r2, ok := s.tryAcquire() - if !ok { - t.Fatal("second acquire should succeed") + r2, ok, err := s.tryAcquire() + if !ok || err != nil { + t.Fatalf("second acquire should succeed: ok=%v err=%v", ok, err) } - if _, ok := s.tryAcquire(); ok { - t.Fatal("third acquire should fail: pool of 2 is exhausted") + // Pool full but healthy: ok=false with NO error (keep polling), not a + // structural failure. + if _, ok, err := s.tryAcquire(); ok || err != nil { + t.Fatalf("third acquire should be full-but-healthy: ok=%v err=%v", ok, err) } r1() // free one permit - r3, ok := s.tryAcquire() - if !ok { - t.Fatal("acquire should succeed after a release") + r3, ok, err := s.tryAcquire() + if !ok || err != nil { + t.Fatalf("acquire should succeed after a release: ok=%v err=%v", ok, err) } r2() r3() } +// A structurally broken pool (dir missing/unwritable) must FAIL OPEN — tryAcquire +// surfaces the error and acquire returns promptly with a no-op release and no +// error, so a lens runs unthrottled instead of spinning forever. +func TestLensSemBrokenPoolFailsOpen(t *testing.T) { + s := &lensSem{dir: filepath.Join(t.TempDir(), "does", "not", "exist"), size: 2} + + if _, ok, err := s.tryAcquire(); ok || err == nil { + t.Fatalf("tryAcquire on a broken pool should report a structural error: ok=%v err=%v", ok, err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + start := time.Now() + release, err := s.acquire(ctx) + if err != nil { + t.Fatalf("acquire should fail open (nil err), got %v", err) + } + if waited := time.Since(start); waited > time.Second { + t.Fatalf("acquire on a broken pool should return promptly, waited %v", waited) + } + release() // must be a safe no-op +} + func TestLensSemAcquireBlocksThenCancels(t *testing.T) { s := &lensSem{dir: t.TempDir(), size: 1} diff --git a/cmd/gadfly/main.go b/cmd/gadfly/main.go index 9aa19fb..006bba9 100644 --- a/cmd/gadfly/main.go +++ b/cmd/gadfly/main.go @@ -97,9 +97,11 @@ const ( // calls and then hard-failing with "max steps reached without a final // answer" — it always has a few steps left to wrap up. defaultWrapUpReserve = 4 - // defaultLensConcurrency is how many specialist lenses run at once within a - // single model. 1 keeps the suite sequential (the historical behavior); - // higher values overlap the independent per-lens passes. See runSpecialists. + // defaultLensConcurrency is the fallback lens budget when neither + // GADFLY_PROVIDER_LENS_CONCURRENCY nor GADFLY_LENS_CONCURRENCY is set. 1 keeps + // the suite sequential (the historical behavior); higher values overlap the + // independent per-lens passes. Under entrypoint.sh the resolved value is the + // provider-wide budget shared across the provider's models. See runSpecialists. defaultLensConcurrency = 1 ) @@ -245,17 +247,15 @@ func run() error { // runSpecialists reviews the diff through each lens and returns the results in // the SAME order as specialists, regardless of finish order. It uses executus's -// fanout primitive: up to GADFLY_LENS_CONCURRENCY lenses run concurrently (the -// default of 1 keeps the suite sequential, exactly as before), and fanout.Run -// returns one result per lens in input order. Each lens already runs under its -// own per-lens timeout (reviewWithSpecialist) and the lenses only read the -// immutable repoFS, so concurrency simply overlaps independent passes. +// fanout primitive to overlap independent lens passes (fanout.Run returns one +// result per lens in input order); each lens runs under its own per-lens timeout +// (reviewWithSpecialist) and the lenses only read the immutable repoFS. // -// Provider-wide throttling: when entrypoint.sh runs several of a provider's -// models at once it seeds a shared lens-permit pool (activeLensSem) that every -// model's lenses draw from, so the real cap is total lens passes in flight per -// provider — not (models at once) × (lenses at once). Absent that pool (local -// runs, tests) this falls back to the in-process lensConcurrency() limit alone. +// Throttling: when entrypoint.sh runs several of a provider's models at once it +// seeds a shared lens-permit pool (activeLensSem) that every model's lenses draw +// from, so the real cap is total lens passes in flight per PROVIDER — not +// (models at once) × (lenses at once). Absent that pool (local runs, tests) the +// in-process GADFLY_LENS_CONCURRENCY limit applies alone (default 1 = sequential). func runSpecialists(eng reviewEngine, base string, specialists []Specialist, task, diff string) []specialistResult { // Optional live status board: publishes this model's per-lens progress to a // file the entrypoint board renders. Inert (no-op) unless GADFLY_STATUS_FILE @@ -265,10 +265,11 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas // The cross-process pool (if any) is the real ceiling; size the in-process // fanout to it so a lone model in its lane can use the whole provider budget, // while extra goroutines simply block in sem.acquire until a permit frees. + // Absent a pool, fall back to the in-process lens limit. sem := activeLensSem() maxConcurrent := lensConcurrency() if sem != nil { - maxConcurrent = sem.size + maxConcurrent = sem.size // the shared pool is the real ceiling } fanResults := fanout.Run(context.Background(), specialists, fanout.Options[Specialist]{ diff --git a/entrypoint.sh b/entrypoint.sh index f6dedf4..43147de 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -354,8 +354,8 @@ BOARD_PID="" if [ "${GADFLY_STATUS_BOARD:-1}" != "0" ]; then rm -rf "$STATUS_DIR"; mkdir -p "$STATUS_DIR" # Pre-seed every model as queued so the board shows the full swarm from t=0, - # even models still waiting on their provider lane's concurrency cap. Each - # binary overwrites its own file with real per-lens detail once it starts. + # even models whose lenses are still waiting on their provider's lens budget. + # Each binary overwrites its own file with real per-lens detail once it starts. for m in "${MODEL_LIST[@]}"; do jq -n --arg model "$m" --arg provider "$(provider_of "$m")" \ '{model:$model, provider:$provider, started:0, updated:0, done:false, lenses:[]}' \ @@ -394,9 +394,9 @@ if [ "${GADFLY_PR_BUDGET_SECS:-0}" -gt 0 ] 2>/dev/null; then fi log "providers: ${PROVIDERS:-none}" -# Each provider lane runs in parallel; cap is enforced within each lane. Track -# the lane PIDs so we wait ONLY for the review work — not the status board, -# which intentionally runs until we signal it below. +# Each provider lane runs in parallel; the shared lens budget throttles within +# each lane. Track the lane PIDs so we wait ONLY for the review work — not the +# status board, which intentionally runs until we signal it below. LANE_PIDS=() for p in $PROVIDERS; do run_lane "$p" &