fix(wfengine): close DrainWorker TOCTOU (bookshelf-81hq) #1388

Merged

zombor merged 3 commits from bd-bookshelf-81hq into main 2026-08-08 18:07:03 +00:00

Conversation 6 Commits 3 Files changed 5 +589 -17

zombor commented 2026-08-07 13:32:56 +00:00

Owner

Copy link

Summary

DrainWorker() unlocked to drain outside the mutex, then unconditionally wrote
workerStarted=false. If a newer worker generation started while the old one
was still draining (e.g. two overlapping DrainWorker calls racing a
StartWorker in between), the second DrainWorker call to finish would clobber
the newer worker's workerStarted=true, silently orphaning it.

Fix: add a workerGen counter incremented on every successful StartWorker.
DrainWorker snapshots the generation before releasing the lock to drain, and
only clears workerStarted afterward if the generation is unchanged — so a
newer worker's flag is never clobbered.

Not a data race (fields already mutex-guarded, -race clean before and after)
— a purely logical TOCTOU. Follow-up from PR #1029 / bookshelf-ke3w review.

Test plan

• New black-box test in internal/wfengine/engine_concurrent_test.go
  (package wfengine_test) deterministically drives the race via controllable
  stub channels: G1 begins draining and blocks, G2 drains and completes first
  (clearing workerStarted), a new StartWorker runs, then G1 finishes and must
  not clobber the new worker's flag.
• Verified RED against the pre-fix code (test fails), GREEN with the fix.
• go test -race ./internal/wfengine/... green, 100% coverage on the package.
• golangci-lint run ./internal/wfengine/... clean.

Docs: N/A because internal engine correctness, no user-facing surface.

Closes bead bookshelf-81hq on merge.

## Summary DrainWorker() unlocked to drain outside the mutex, then unconditionally wrote workerStarted=false. If a newer worker generation started while the old one was still draining (e.g. two overlapping DrainWorker calls racing a StartWorker in between), the second DrainWorker call to finish would clobber the newer worker's `workerStarted=true`, silently orphaning it. Fix: add a `workerGen` counter incremented on every successful StartWorker. DrainWorker snapshots the generation before releasing the lock to drain, and only clears workerStarted afterward if the generation is unchanged — so a newer worker's flag is never clobbered. Not a data race (fields already mutex-guarded, -race clean before and after) — a purely logical TOCTOU. Follow-up from PR #1029 / bookshelf-ke3w review. ## Test plan - New black-box test in `internal/wfengine/engine_concurrent_test.go` (`package wfengine_test`) deterministically drives the race via controllable stub channels: G1 begins draining and blocks, G2 drains and completes first (clearing workerStarted), a new StartWorker runs, then G1 finishes and must not clobber the new worker's flag. - Verified RED against the pre-fix code (test fails), GREEN with the fix. - `go test -race ./internal/wfengine/...` green, 100% coverage on the package. - `golangci-lint run ./internal/wfengine/...` clean. Docs: N/A because internal engine correctness, no user-facing surface. Closes bead bookshelf-81hq on merge.

zombor added 1 commit 2026-08-07 13:32:57 +00:00

fix(wfengine): close TOCTOU in DrainWorker that could orphan a concurrently-started worker ... 11caacd896

DrainWorker used to unconditionally write workerStarted=false after draining,
even if a newer StartWorker had run while the drain was in flight (e.g. two
overlapping DrainWorker calls racing a StartWorker in between). Add a
workerGen counter incremented on every successful StartWorker; DrainWorker
snapshots it before releasing the lock to drain and only clears
workerStarted if the generation is unchanged afterward, so a newer worker's
started flag is never clobbered.

Covered by a deterministic black-box test (engine_concurrent_test.go) that
drives two overlapping DrainWorker calls and an interleaved StartWorker via
controllable stub channels — fails against the old unconditional-write code
(verified RED), passes with the fix (GREEN), -race clean.

Closes bead bookshelf-81hq on merge.

zombor commented 2026-08-07 13:44:17 +00:00

Author

Owner

Copy link

Security Review (PR #1388 / bookshelf-81hq)

Scope: internal/wfengine/engine.go, internal/wfengine/engine_concurrent_test.go — workerGen counter closing the DrainWorker/StartWorker TOCTOU.

[MINOR] internal/wfengine/engine.go:731-748 — Doc comment note only, no fix needed
The new doc comment correctly calls out that a concurrent StartWorker racing an in-flight DrainWorker still allows two goroutines to invoke e.drainWorkerBounded() concurrently on the same underlying ws object (worth confirming go-workflows' WorkerServer.WaitForCompletion/Start tolerate concurrent invocation), and that StartWorker's early if e.workerStarted { return nil } no-op means a caller racing mid-drain can silently no-op instead of actually restarting the poller once drain finishes. Both behaviors predate this diff (the prior code had the identical race, just with a worse consequence — unconditionally clobbering the flag). This PR strictly narrows the blast radius (no longer orphans a newer worker's workerStarted flag) without introducing a new hang/deadlock: workerGen is a monotonically increasing counter compared for equality, so there is no path where workerStarted gets stuck false while a real worker is running, or vice versa, from this change alone. Flagging as MINOR only as a paper trail — no change requested.

Other angles checked, none flagged:

• Availability/DoS: Traced the new workerGen gate through StartWorker (increments only after a successful e.startWorker) and DrainWorker (snapshot-before-unlock, clear-only-if-unchanged-after). No path leaves workerStarted permanently desynced from reality or wedges the mutex — the lock is always released via the existing Unlock() calls and the new logic adds only a plain uint64 compare under the same lock.
• Background job / retention semantics: DrainWorker still does not close wfDB/diagBackend; the retention sweep and SSE monitor lifecycle (lifecycleStarted) are untouched by this diff. No stale-state or cleanup-worker impact.
• Auth/scoping/input boundary/secrets: None present — purely internal engine lifecycle state (booleans/counter guarded by e.mu), no request-supplied data, no logging of sensitive values.
• Test hygiene: engine_concurrent_test.go stays package wfengine_test (black-box), uses the existing exported NewTestEngine/WorkerStarted() test seams — no new white-box access. Deterministic interleaving via channels (g1Entered/g1Blocking) + atomic.Int32 call counter, no wall-clock timing assertions, no flake risk per the standard's flake-prevention checklist.
• Unrelated gofmt-only realignment in newWithFactory's struct literal and two closures (StartBulkCoversGenerateByIDsWorkflow, StartRecalcLibraryMatchScoresWorkflow) wrapped onto multiple lines — cosmetic, no behavior change.

REVIEW VERDICT: 0 blocker, 0 major, 1 minor

## Security Review (PR #1388 / bookshelf-81hq) Scope: `internal/wfengine/engine.go`, `internal/wfengine/engine_concurrent_test.go` — `workerGen` counter closing the DrainWorker/StartWorker TOCTOU. [MINOR] internal/wfengine/engine.go:731-748 — Doc comment note only, no fix needed The new doc comment correctly calls out that a concurrent `StartWorker` racing an in-flight `DrainWorker` still allows two goroutines to invoke `e.drainWorkerBounded()` concurrently on the same underlying `ws` object (worth confirming go-workflows' `WorkerServer.WaitForCompletion`/`Start` tolerate concurrent invocation), and that `StartWorker`'s early `if e.workerStarted { return nil }` no-op means a caller racing mid-drain can silently no-op instead of actually restarting the poller once drain finishes. Both behaviors predate this diff (the prior code had the identical race, just with a worse consequence — unconditionally clobbering the flag). This PR strictly narrows the blast radius (no longer orphans a newer worker's `workerStarted` flag) without introducing a new hang/deadlock: `workerGen` is a monotonically increasing counter compared for equality, so there is no path where `workerStarted` gets stuck false while a real worker is running, or vice versa, from this change alone. Flagging as MINOR only as a paper trail — no change requested. Other angles checked, none flagged: - **Availability/DoS:** Traced the new `workerGen` gate through `StartWorker` (increments only after a successful `e.startWorker`) and `DrainWorker` (snapshot-before-unlock, clear-only-if-unchanged-after). No path leaves `workerStarted` permanently desynced from reality or wedges the mutex — the lock is always released via the existing `Unlock()` calls and the new logic adds only a plain uint64 compare under the same lock. - **Background job / retention semantics:** DrainWorker still does not close `wfDB`/`diagBackend`; the retention sweep and SSE monitor lifecycle (`lifecycleStarted`) are untouched by this diff. No stale-state or cleanup-worker impact. - **Auth/scoping/input boundary/secrets:** None present — purely internal engine lifecycle state (booleans/counter guarded by `e.mu`), no request-supplied data, no logging of sensitive values. - **Test hygiene:** `engine_concurrent_test.go` stays `package wfengine_test` (black-box), uses the existing exported `NewTestEngine`/`WorkerStarted()` test seams — no new white-box access. Deterministic interleaving via channels (`g1Entered`/`g1Blocking`) + `atomic.Int32` call counter, no wall-clock timing assertions, no flake risk per the standard's flake-prevention checklist. - Unrelated gofmt-only realignment in `newWithFactory`'s struct literal and two closures (`StartBulkCoversGenerateByIDsWorkflow`, `StartRecalcLibraryMatchScoresWorkflow`) wrapped onto multiple lines — cosmetic, no behavior change. REVIEW VERDICT: 0 blocker, 0 major, 1 minor

zombor commented 2026-08-07 13:45:25 +00:00

Author

Owner

Copy link

Code Review — PR #1388 (bookshelf-81hq)

Reviewed the diff only (origin/main...origin/bd-bookshelf-81hq). CI green is
taken as behavioral truth; I did not re-run tests.

Correctness walkthrough of the workerGen fix

Traced the interleaving the fix targets: StartWorker only increments
workerGen/sets workerStarted=true while holding e.mu (engine.go:696-698),
and DrainWorker snapshots startGen under e.mu before unlocking
(engine.go:745-750) and compares under e.mu after re-locking
(engine.go:756-759). All reads/writes of workerGen are inside the lock on
both sides — no bare/unsynchronized access, consistent with the -race clean
CI run.

Walked the actual trigger scenario (confirmed it requires two overlapping
DrainWorker calls, not a lone StartWorker racing a single DrainWorker —
a solo StartWorker during a single drain always no-ops because workerStarted
stays true until the drain's own final Lock/Unlock): G1 and G2 both drain
concurrently with startGen=1; G2 finishes first and clears the flag
(gen==startGen); a real StartWorker then runs, bumping gen to 2; G1
finishes later, sees gen(2) != startGen(1), and correctly skips clearing
workerStarted. Verified there is no reverse case where the flag is left
true after a worker was actually drained (workerGen only advances on a
successful StartWorker, and StartWorker is a no-op whenever workerStarted
is still true). The bookkeeping fix is correct and complete for the
workerStarted/workerGen pair itself.

Findings

[MAJOR] internal/wfengine/engine.go:731-744 (DrainWorker doc comment) — fix closes the flag race but not the underlying workerSet race it now explicitly documents/tests
The new doc comment (and the new test) establish overlapping DrainWorker/StartWorker
calls on the same Engine as an expected scenario the code "handles." But the fix
only synchronizes the Engine's own workerStarted/workerGen bookkeeping — it does
not synchronize the underlying workerSet (internal/wfengine/worker_set.go,
unchanged by this diff) that drainWorkerBounded/e.stopWorker/e.startWorker
actually call. drainWorkerBounded runs e.stopWorker() (= ws.waitForCompletion,
worker_set.go:183-187) outside e.mu, and it mutates s.drained and reads
s.workers with no lock of its own (worker_set.go:163-187). Meanwhile StartWorker's
call into ws.start() (which also mutates s.workers/s.drained,
worker_set.go:163-177) runs under e.mu — but e.mu is exactly what the drain
path releases for the length of the real drain. So the same overlapping-drain scenario
this PR's own test constructs (two concurrent DrainWorker calls, one of which lets a
real StartWorker run in between) will, with the real (non-stub) waitForCompletion/
start, produce concurrent unsynchronized writes to workerSet.drained/workerSet.workers
from two goroutines — a genuine data race that the new test's mocked stop/start
closures (plain functions with no shared workerSet state) cannot exercise, so -race
stays clean without proving the real path is safe. This is pre-existing workerSet
design, but this PR is the one that newly documents and tests the overlap as a
supported/considered scenario without covering the layer where the actual race lives.
Suggested fix: either (a) add a dedicated drainMu sync.Mutex in Engine that
DrainWorker/StartWorker hold for the full duration of the real
stopWorker/startWorker call (serializing calls into workerSet without blocking
the flag bookkeeping under e.mu), or (b) make DrainWorker reject/wait rather than
silently proceed when a drain is already in flight (e.g. a draining bool guarded by
e.mu), so two DrainWorker calls never call into workerSet concurrently. The
weaker option is to at least narrow the new doc comment so it doesn't imply the overlap
is safe end-to-end — but a real fix is preferable since production App.StopWorker
(internal/app/app.go:1280-1297) already runs DrainWorker on a spawned goroutine,
making a doubled call plausible if StopWorker is ever invoked twice concurrently
(e.g. racing shutdown + test cleanup paths).

Other checks (no findings)

• Test file internal/wfengine/engine_concurrent_test.go is black-box (package wfengine_test), var-at-top, one Expect per It, and the new spec is fully
  deterministic — synchronization is via unbuffered/closed channels and
  sync.WaitGroup, no sleep/wall-clock timing, no asserting on async
  side-effects (assertions run only after wg.Wait()).
• No .golangci.yml changes; gofmt on the diff's engine.go is clean (the
  struct-literal realignment in newWithFactory is pure gofmt, not a
  behavior change).
• No new coverage exclusions; the normal (matching-generation) DrainWorker
  path was already covered by worker_lifecycle_test.go before this PR.

REVIEW VERDICT: 0 blocker, 1 major, 0 minor

## Code Review — PR #1388 (bookshelf-81hq) Reviewed the diff only (`origin/main...origin/bd-bookshelf-81hq`). CI green is taken as behavioral truth; I did not re-run tests. ### Correctness walkthrough of the `workerGen` fix Traced the interleaving the fix targets: `StartWorker` only increments `workerGen`/sets `workerStarted=true` while holding `e.mu` (`engine.go:696-698`), and `DrainWorker` snapshots `startGen` under `e.mu` before unlocking (`engine.go:745-750`) and compares under `e.mu` after re-locking (`engine.go:756-759`). All reads/writes of `workerGen` are inside the lock on both sides — no bare/unsynchronized access, consistent with the `-race` clean CI run. Walked the actual trigger scenario (confirmed it requires **two overlapping `DrainWorker` calls**, not a lone `StartWorker` racing a single `DrainWorker` — a solo `StartWorker` during a single drain always no-ops because `workerStarted` stays `true` until the drain's own final `Lock`/`Unlock`): G1 and G2 both drain concurrently with `startGen=1`; G2 finishes first and clears the flag (`gen==startGen`); a real `StartWorker` then runs, bumping `gen` to 2; G1 finishes later, sees `gen(2) != startGen(1)`, and correctly **skips** clearing `workerStarted`. Verified there is no reverse case where the flag is left `true` after a worker was actually drained (`workerGen` only advances on a successful `StartWorker`, and `StartWorker` is a no-op whenever `workerStarted` is still `true`). The bookkeeping fix is correct and complete for the `workerStarted`/`workerGen` pair itself. ### Findings [MAJOR] internal/wfengine/engine.go:731-744 (DrainWorker doc comment) — fix closes the flag race but not the underlying workerSet race it now explicitly documents/tests The new doc comment (and the new test) establish overlapping `DrainWorker`/`StartWorker` calls on the same `Engine` as an *expected* scenario the code "handles." But the fix only synchronizes the Engine's own `workerStarted`/`workerGen` bookkeeping — it does **not** synchronize the underlying `workerSet` (`internal/wfengine/worker_set.go`, unchanged by this diff) that `drainWorkerBounded`/`e.stopWorker`/`e.startWorker` actually call. `drainWorkerBounded` runs `e.stopWorker()` (= `ws.waitForCompletion`, `worker_set.go:183-187`) **outside `e.mu`**, and it mutates `s.drained` and reads `s.workers` with no lock of its own (`worker_set.go:163-187`). Meanwhile `StartWorker`'s call into `ws.start()` (which also mutates `s.workers`/`s.drained`, `worker_set.go:163-177`) runs *under* `e.mu` — but `e.mu` is exactly what the drain path releases for the length of the real drain. So the same overlapping-drain scenario this PR's own test constructs (two concurrent `DrainWorker` calls, one of which lets a real `StartWorker` run in between) will, with the *real* (non-stub) `waitForCompletion`/ `start`, produce concurrent unsynchronized writes to `workerSet.drained`/`workerSet.workers` from two goroutines — a genuine data race that the new test's mocked `stop`/`start` closures (plain functions with no shared workerSet state) cannot exercise, so `-race` stays clean without proving the real path is safe. This is pre-existing workerSet design, but this PR is the one that newly documents and tests the overlap as a supported/considered scenario without covering the layer where the actual race lives. Suggested fix: either (a) add a dedicated `drainMu sync.Mutex` in `Engine` that `DrainWorker`/`StartWorker` hold for the full duration of the real `stopWorker`/`startWorker` call (serializing calls into `workerSet` without blocking the flag bookkeeping under `e.mu`), or (b) make `DrainWorker` reject/wait rather than silently proceed when a drain is already in flight (e.g. a `draining bool` guarded by `e.mu`), so two `DrainWorker` calls never call into `workerSet` concurrently. The weaker option is to at least narrow the new doc comment so it doesn't imply the overlap is safe end-to-end — but a real fix is preferable since production `App.StopWorker` (`internal/app/app.go:1280-1297`) already runs `DrainWorker` on a spawned goroutine, making a doubled call plausible if `StopWorker` is ever invoked twice concurrently (e.g. racing shutdown + test cleanup paths). ### Other checks (no findings) - Test file `internal/wfengine/engine_concurrent_test.go` is black-box (`package wfengine_test`), var-at-top, one `Expect` per `It`, and the new spec is fully deterministic — synchronization is via unbuffered/closed channels and `sync.WaitGroup`, no `sleep`/wall-clock timing, no asserting on async side-effects (assertions run only after `wg.Wait()`). - No `.golangci.yml` changes; `gofmt` on the diff's `engine.go` is clean (the struct-literal realignment in `newWithFactory` is pure `gofmt`, not a behavior change). - No new coverage exclusions; the normal (matching-generation) `DrainWorker` path was already covered by `worker_lifecycle_test.go` before this PR. REVIEW VERDICT: 0 blocker, 1 major, 0 minor

zombor referenced this pull request from a commit 2026-08-07 14:50:11 +00:00

fix(review): serialize workerSet start/drain to close overlapping-drain race

zombor force-pushed bd-bookshelf-81hq from 11caacd896 to 7dc7d92c93 2026-08-07 14:50:12 +00:00 Compare

zombor commented 2026-08-07 15:02:51 +00:00

Author

Owner

Copy link

Code Review — PR #1388 (bookshelf-81hq) — RE-REVIEW after concurrency deep-fix

Walked every interleaving in worker_set.go's new leader/follower dedup +
eager-drained design, cross-checked against engine.go's workerGen TOCTOU
guard and both new test files.

Deadlock / hang freedom

waitForCompletion()'s leader path has exactly one exit: mu.Lock() → draining=false, drainErr=err, close(done) → mu.Unlock()
(internal/wfengine/worker_set.go:322-326). There is no early return between
draining=true (line 300) and close(done) that could skip the close, so
every leader invocation unconditionally closes drainDone — a follower
parked on <-done (line 305) cannot hang forever under normal (non-panic)
operation. Confirmed no leader path swallows the close.

Locking discipline

All reads/writes of draining, drainDone, drainErr, s.drained, and
s.workers happen under s.mu, including the follower's re-lock after
<-done to read drainErr (worker_set.go:302-307). The two genuinely
blocking calls — drainWorkers(fns) (leader) and <-done (follower) — both
run outside the lock, matching the documented intent. start()
(worker_set.go:255-283) never reads draining/drainDone at all, only the
non-blocking s.drained bool, so Engine.mu (held across StartWorker →
start()) is never held across a drain — confirmed by
engine_concurrent_test.go's StartWorker idempotency spec.

Panic prevention (double WaitForCompletion)

For the intended same-generation overlap (Stop||DrainWorker,
DrainWorker||DrainWorker), the leader/follower dedup correctly prevents a
second call into the underlying go-workflows Worker.WaitForCompletion —
confirmed by the worker_set_test.go "real Start/Drain overlap" spec, which
exercises real *goworker.Worker instances (not just the field-only
bookshelf-ti0zf regression test) and would panic/-race abort if either
double-drain or start-during-drain occurred.

[MINOR] worker_set.go:292-330 — drain leader/follower dedup is not generation-scoped

draining/drainDone/drainErr are single, ungenerationed fields. If
waitForCompletion()'s leader for generation K is still blocked past its
30s drainWorkerBounded timeout (orphaned goroutine, per the original
bookshelf-ti0zf race) and a concurrent StartWorker rebuilds generation
K+1 in the interim, a third, later call to DrainWorker/Stop targeting
K+1 can observe s.draining == true (still true from K's orphaned leader)
and become a follower of K's drain, not K+1's. It returns success
without ever calling WaitForCompletion on K+1's workers, and K+1's
pollers keep running untouched.

This is honestly disclosed by the PR itself — engine.go:731-740's
DrainWorker doc says outright "this guard only prevents the new worker
from being silently orphaned, it does not make the drain itself apply to
the new generation," and the worker_set_test.go "real Start/Drain
overlap" spec explicitly exercises and accepts this
either-dedup-or-fresh-drain nondeterminism as "safe" (no crash/panic). It
also matches the pre-existing DrainWorker/Stop guidance to avoid
overlapping calls where possible. Given it's a triple-stacked race
(timeout-induced orphan + concurrent Start + concurrent second Drain) and
explicitly tested/documented rather than a silent regression, I'm grading
this MINOR rather than MAJOR — but recommend a follow-up bead to key
draining/drainDone to a generation stamp (e.g. capture len(s.workers)-
independent counter alongside s.workers and compare) so a drain call
always targets the CURRENT generation instead of whichever one happens to
be draining at call time. Worth flagging because Engine.Stop() closes
wfDB/diagBackend right after its drainWorkerBounded() call
(engine.go Stop()), so in the (documented-as-discouraged) overlap case
a still-running generation's pollers would keep polling against soon-to-be-
closed DB pools — not introduced by this diff (pre-existing lack of an
e.stopped check in StartWorker), but worth linking in the follow-up.

Other checks

• Test hygiene: both new test files declare package wfengine_test
  (worker_set_test.go:1, engine_concurrent_test.go:1) — black-box,
  correct.
• export_test.go seams: StartForRaceTest (line ~1851) is a thin
  passthrough to production ws.start(); DrainingForRaceTest (line ~1859)
  reads the real w.ws.draining field under w.ws.mu — both drive real
  production code paths, consistent with internal/wfengine's existing
  *ForTest seam convention (not export-to-game-blackbox).
• No .golangci.yml/scripts/check-coverage.sh changes — confirmed via
  diff, no new exclusions.
• engine.go workerGen TOCTOU guard (DrainWorker snapshotting
  e.workerGen before releasing the lock, only clearing workerStarted if
  unchanged) is correctly exercised by the new
  engine_concurrent_test.go "DrainWorker TOCTOU: newer worker generation
  started mid-drain" spec with a deterministic channel-based interleaving —
  good regression coverage for the exact scenario it targets.
• Eager s.drained = true does not corrupt the normal single-caller flow: a
  drain that errors still leaves drained = true correctly (the real
  underlying Worker.WaitForCompletion still permanently closes each
  queue's channel regardless of per-queue error), and a subsequent
  already-drained call correctly short-circuits via the
  s.drained && !s.draining no-op branch (worker_set.go:308-311).

REVIEW VERDICT: 0 blocker, 0 major, 1 minor

## Code Review — PR #1388 (bookshelf-81hq) — RE-REVIEW after concurrency deep-fix Walked every interleaving in `worker_set.go`'s new leader/follower dedup + eager-drained design, cross-checked against `engine.go`'s `workerGen` TOCTOU guard and both new test files. ### Deadlock / hang freedom `waitForCompletion()`'s leader path has exactly one exit: `mu.Lock() → draining=false, drainErr=err, close(done) → mu.Unlock()` (`internal/wfengine/worker_set.go:322-326`). There is no early return between `draining=true` (line 300) and `close(done)` that could skip the close, so every leader invocation unconditionally closes `drainDone` — a follower parked on `<-done` (line 305) cannot hang forever under normal (non-panic) operation. Confirmed no leader path swallows the close. ### Locking discipline All reads/writes of `draining`, `drainDone`, `drainErr`, `s.drained`, and `s.workers` happen under `s.mu`, including the follower's re-lock after `<-done` to read `drainErr` (`worker_set.go:302-307`). The two genuinely blocking calls — `drainWorkers(fns)` (leader) and `<-done` (follower) — both run outside the lock, matching the documented intent. `start()` (`worker_set.go:255-283`) never reads `draining`/`drainDone` at all, only the non-blocking `s.drained` bool, so `Engine.mu` (held across `StartWorker` → `start()`) is never held across a drain — confirmed by `engine_concurrent_test.go`'s `StartWorker idempotency` spec. ### Panic prevention (double `WaitForCompletion`) For the intended same-generation overlap (`Stop`||`DrainWorker`, `DrainWorker`||`DrainWorker`), the leader/follower dedup correctly prevents a second call into the underlying go-workflows `Worker.WaitForCompletion` — confirmed by the `worker_set_test.go` "real Start/Drain overlap" spec, which exercises real `*goworker.Worker` instances (not just the field-only `bookshelf-ti0zf` regression test) and would panic/`-race` abort if either double-drain or start-during-drain occurred. ### [MINOR] worker_set.go:292-330 — drain leader/follower dedup is not generation-scoped `draining`/`drainDone`/`drainErr` are single, ungenerationed fields. If `waitForCompletion()`'s leader for generation K is still blocked past its 30s `drainWorkerBounded` timeout (orphaned goroutine, per the original bookshelf-ti0zf race) and a concurrent `StartWorker` rebuilds generation K+1 in the interim, a *third*, later call to `DrainWorker`/`Stop` targeting K+1 can observe `s.draining == true` (still true from K's orphaned leader) and become a **follower of K's drain**, not K+1's. It returns success without ever calling `WaitForCompletion` on K+1's workers, and K+1's pollers keep running untouched. This is honestly disclosed by the PR itself — `engine.go:731-740`'s `DrainWorker` doc says outright "this guard only prevents the new worker from being silently orphaned, it does not make the drain itself apply to the new generation," and the `worker_set_test.go` "real Start/Drain overlap" spec explicitly exercises and accepts this either-dedup-or-fresh-drain nondeterminism as "safe" (no crash/panic). It also matches the pre-existing `DrainWorker`/`Stop` guidance to avoid overlapping calls where possible. Given it's a triple-stacked race (timeout-induced orphan + concurrent Start + concurrent second Drain) and explicitly tested/documented rather than a silent regression, I'm grading this MINOR rather than MAJOR — but recommend a follow-up bead to key `draining`/`drainDone` to a generation stamp (e.g. capture `len(s.workers)`- independent counter alongside `s.workers` and compare) so a drain call always targets the CURRENT generation instead of whichever one happens to be `draining` at call time. Worth flagging because `Engine.Stop()` closes `wfDB`/`diagBackend` right after its `drainWorkerBounded()` call (`engine.go` `Stop()`), so in the (documented-as-discouraged) overlap case a still-running generation's pollers would keep polling against soon-to-be- closed DB pools — not introduced by this diff (pre-existing lack of an `e.stopped` check in `StartWorker`), but worth linking in the follow-up. ### Other checks - **Test hygiene:** both new test files declare `package wfengine_test` (`worker_set_test.go:1`, `engine_concurrent_test.go:1`) — black-box, correct. - **export_test.go seams:** `StartForRaceTest` (line ~1851) is a thin passthrough to production `ws.start()`; `DrainingForRaceTest` (line ~1859) reads the real `w.ws.draining` field under `w.ws.mu` — both drive real production code paths, consistent with `internal/wfengine`'s existing `*ForTest` seam convention (not export-to-game-blackbox). - **No `.golangci.yml`/`scripts/check-coverage.sh` changes** — confirmed via diff, no new exclusions. - **`engine.go` `workerGen` TOCTOU guard** (`DrainWorker` snapshotting `e.workerGen` before releasing the lock, only clearing `workerStarted` if unchanged) is correctly exercised by the new `engine_concurrent_test.go` "DrainWorker TOCTOU: newer worker generation started mid-drain" spec with a deterministic channel-based interleaving — good regression coverage for the exact scenario it targets. - Eager `s.drained = true` does not corrupt the normal single-caller flow: a drain that errors still leaves `drained = true` correctly (the real underlying `Worker.WaitForCompletion` still permanently closes each queue's channel regardless of per-queue error), and a subsequent already-drained call correctly short-circuits via the `s.drained && !s.draining` no-op branch (`worker_set.go:308-311`). REVIEW VERDICT: 0 blocker, 0 major, 1 minor

zombor commented 2026-08-07 15:03:08 +00:00

Author

Owner

Copy link

Security/Availability re-review — PR #1388 (bookshelf-81hq)

Scope: git diff origin/main...origin/bd-bookshelf-81hq (5 files, +330/-25), focused on the workerSet start/drain serialization fix in internal/wfengine/worker_set.go + engine.go.

Traced every leader-exit path in workerSet.waitForCompletion (worker_set.go:306-340) and how Engine.DrainWorker/Engine.Stop consume it.

[MAJOR] internal/wfengine/worker_set.go:306-340 — leader/follower drain dedup is generation-unaware; a follower can silently report "drained" for a stale generation while the true-current generation keeps running un-drained
waitForCompletion's dedup is keyed purely on the boolean s.draining, with no generation/epoch tag tying a follower's wait to the specific worker generation it thinks it's draining. Reachable sequence:

1. DrainWorker (or Stop) begins draining generation A; drainWorkerBounded (engine.go:826-843) wraps the real call in a 30s bound and orphans the goroutine if it fires first — this is an intentional, documented behavior for CI/DB-latency stalls, not a hypothetical.
2. Because the bound fired, DrainWorker clears workerStarted (generation-matched via workerGen) and a subsequent StartWorker succeeds, calling ws.start() which — correctly, per this PR's fix — rebuilds fresh Worker instances for generation B while A's orphaned drain is still in flight at the ws level (s.draining still true).
3. A second drain call for generation B (a later DrainWorker, or Engine.Stop at true shutdown) arrives while s.draining is still true from A's orphaned goroutine. It becomes a follower on A's drainDone, not a new leader for B — worker_set.go:308-316. When A's orphaned drain eventually finishes, the follower gets A's drainErr and returns, believing generation B has been drained, when in fact generation B's pollers were never touched.
4. Engine.Stop (engine.go:786-811) has no generation check analogous to DrainWorker's workerGen guard — it unconditionally closes e.wfDB and e.diagBackend after drainWorkerBounded returns "success". If step 3's follower call was inside Stop, this closes the DB pools out from under generation B's still-live pollers, which will start failing every DB round-trip (in-flight covers/metadata/scan/enrich activities lost or erroring) instead of being drained cleanly — exactly the "background engine wedge / lost work" class this review was asked to hunt for.
   The new worker_set_test.go "real Start/Drain overlap" spec explicitly documents and accepts this ambiguity ("the second drain call... may either dedup against the still-in-flight generation-K leader OR end up performing the first-ever drain of generation K+1 — both are safe") — but "safe" there is scoped to "no panic/no -race failure", not "drains the generation the caller actually asked to drain". That's a real, if narrow, correctness gap the comment doesn't flag as a limitation anywhere a caller (Engine.Stop) would see it.
   Reachability in production is narrow — it requires the 30s drain-timeout to actually fire (documented as rare/CI-only) plus an overlapping StartWorker + second drain within that window; today's only multi-cycle Start/Stop caller is the e2e harness (App.StartWorker/StopWorker, explicitly for "per-journey" reuse), not steady-state production (which starts/stops the worker exactly once). That narrows it to MAJOR rather than BLOCKER, but it's the direct trade for eliminating the double-WaitForCompletion panic: the old code crashed loudly on this race, the new code can now succeed silently on the wrong generation.
   Fix: tag drainDone/drainErr (and draining) with a generation counter (mirror Engine.workerGen, or simply compare the follower's own s.workers snapshot identity against what the in-flight leader captured). A follower should only dedup against a leader draining the same generation it observes at entry; if the current generation differs from the in-flight drain's generation, it should become the leader for its own generation instead of following a stale one. Alternatively, thread Engine.workerGen through to ws.waitForCompletion/ws.start so the ws layer refuses to conflate generations at all.

[MINOR] internal/wfengine/worker_set.go:306-340 — no recover() around the leader's blocking drain; an unrelated panic inside drainWorkers/WaitForCompletion (not the double-call case this PR targets) leaves s.draining permanently true and any waiting follower blocked forever on <-done
In practice an unrecovered panic in this goroutine (spawned via drainWorkerBounded, engine.go:833) crashes the whole process (Go terminates the program on an unhandled goroutine panic), so this doesn't manifest as a silent forever-wedge in isolation — it's a full crash instead, and start() itself never blocks on drainDone so a fresh generation can still start after a crash-free restart. Still, worth a defensive recover() + close(done) in a defer so any future non-double-call panic degrades to a returned error on followers rather than a hard process crash. Low priority — not a regression introduced by this diff (the underlying call was never protected), just noting since the doc comment now makes strong claims about follower safety that assume the leader always reaches close(done).

Confirmed scope of the change is purely internal worker lifecycle plumbing (workerSet/Engine fields, generation counter, drain synchronization) — no authn/authz surface, no request-derived input, no new logging of secrets/PII, no SQL/HTTP boundary touched. Nothing here implicates multi-user scoping, SSRF, or CSP.

REVIEW VERDICT: 0 blocker, 1 major, 1 minor

## Security/Availability re-review — PR #1388 (bookshelf-81hq) Scope: `git diff origin/main...origin/bd-bookshelf-81hq` (5 files, +330/-25), focused on the workerSet start/drain serialization fix in `internal/wfengine/worker_set.go` + `engine.go`. Traced every leader-exit path in `workerSet.waitForCompletion` (worker_set.go:306-340) and how `Engine.DrainWorker`/`Engine.Stop` consume it. [MAJOR] internal/wfengine/worker_set.go:306-340 — leader/follower drain dedup is generation-unaware; a follower can silently report "drained" for a stale generation while the true-current generation keeps running un-drained `waitForCompletion`'s dedup is keyed purely on the boolean `s.draining`, with no generation/epoch tag tying a follower's wait to the specific worker generation it thinks it's draining. Reachable sequence: 1. `DrainWorker` (or `Stop`) begins draining generation A; `drainWorkerBounded` (engine.go:826-843) wraps the real call in a 30s bound and *orphans* the goroutine if it fires first — this is an intentional, documented behavior for CI/DB-latency stalls, not a hypothetical. 2. Because the bound fired, `DrainWorker` clears `workerStarted` (generation-matched via `workerGen`) and a subsequent `StartWorker` succeeds, calling `ws.start()` which — correctly, per this PR's fix — rebuilds fresh Worker instances for generation B while A's orphaned drain is still in flight at the ws level (`s.draining` still true). 3. A second drain call for generation B (a later `DrainWorker`, or `Engine.Stop` at true shutdown) arrives while `s.draining` is still true from A's orphaned goroutine. It becomes a **follower on A's `drainDone`**, not a new leader for B — worker_set.go:308-316. When A's orphaned drain eventually finishes, the follower gets A's `drainErr` and returns, believing generation B has been drained, when in fact generation B's pollers were never touched. 4. `Engine.Stop` (engine.go:786-811) has no generation check analogous to `DrainWorker`'s `workerGen` guard — it unconditionally closes `e.wfDB` and `e.diagBackend` after `drainWorkerBounded` returns "success". If step 3's follower call was inside `Stop`, this closes the DB pools out from under generation B's still-live pollers, which will start failing every DB round-trip (in-flight covers/metadata/scan/enrich activities lost or erroring) instead of being drained cleanly — exactly the "background engine wedge / lost work" class this review was asked to hunt for. The new `worker_set_test.go` "real Start/Drain overlap" spec explicitly documents and *accepts* this ambiguity ("the second drain call... may either dedup against the still-in-flight generation-K leader OR end up performing the first-ever drain of generation K+1 — both are safe") — but "safe" there is scoped to "no panic/no `-race` failure", not "drains the generation the caller actually asked to drain". That's a real, if narrow, correctness gap the comment doesn't flag as a limitation anywhere a caller (`Engine.Stop`) would see it. Reachability in production is narrow — it requires the 30s drain-timeout to actually fire (documented as rare/CI-only) plus an overlapping `StartWorker` + second drain within that window; today's only multi-cycle Start/Stop caller is the e2e harness (`App.StartWorker`/`StopWorker`, explicitly for "per-journey" reuse), not steady-state production (which starts/stops the worker exactly once). That narrows it to MAJOR rather than BLOCKER, but it's the direct trade for eliminating the double-`WaitForCompletion` panic: the old code crashed loudly on this race, the new code can now succeed silently on the wrong generation. Fix: tag `drainDone`/`drainErr` (and `draining`) with a generation counter (mirror `Engine.workerGen`, or simply compare the follower's own `s.workers` snapshot identity against what the in-flight leader captured). A follower should only dedup against a leader draining the *same* generation it observes at entry; if the current generation differs from the in-flight drain's generation, it should become the leader for its own generation instead of following a stale one. Alternatively, thread `Engine.workerGen` through to `ws.waitForCompletion`/`ws.start` so the ws layer refuses to conflate generations at all. [MINOR] internal/wfengine/worker_set.go:306-340 — no `recover()` around the leader's blocking drain; an unrelated panic inside `drainWorkers`/`WaitForCompletion` (not the double-call case this PR targets) leaves `s.draining` permanently `true` and any waiting follower blocked forever on `<-done` In practice an unrecovered panic in this goroutine (spawned via `drainWorkerBounded`, engine.go:833) crashes the whole process (Go terminates the program on an unhandled goroutine panic), so this doesn't manifest as a silent forever-wedge in isolation — it's a full crash instead, and `start()` itself never blocks on `drainDone` so a *fresh* generation can still start after a crash-free restart. Still, worth a defensive `recover()` + `close(done)` in a `defer` so any future non-double-call panic degrades to a returned error on followers rather than a hard process crash. Low priority — not a regression introduced by this diff (the underlying call was never protected), just noting since the doc comment now makes strong claims about follower safety that assume the leader always reaches `close(done)`. Confirmed scope of the change is purely internal worker lifecycle plumbing (`workerSet`/`Engine` fields, generation counter, drain synchronization) — no authn/authz surface, no request-derived input, no new logging of secrets/PII, no SQL/HTTP boundary touched. Nothing here implicates multi-user scoping, SSRF, or CSP. REVIEW VERDICT: 0 blocker, 1 major, 1 minor

zombor referenced this pull request from a commit 2026-08-08 01:25:35 +00:00

fix(review): serialize workerSet start/drain to close overlapping-drain race

zombor force-pushed bd-bookshelf-81hq from 7dc7d92c93 to ebc0a6954d 2026-08-08 01:25:35 +00:00 Compare

zombor referenced this pull request from a commit 2026-08-08 16:20:43 +00:00

fix(review): serialize workerSet start/drain to close overlapping-drain race

zombor force-pushed bd-bookshelf-81hq from ebc0a6954d to b45cb2ce8b 2026-08-08 16:20:43 +00:00 Compare

zombor commented 2026-08-08 16:30:18 +00:00

Author

Owner

Copy link

Security/Availability re-review (round 3) — PR #1388 / bookshelf-81hq

Scope: internal/wfengine/{engine.go,worker_set.go,worker_set_test.go,engine_concurrent_test.go,export_test.go} diff vs origin/main. Focus: can the generation-scoped leader/follower drain dedup leave the engine (all background workflow processing: covers/metadata/enrich/scan/bookdrop) permanently wedged, and does it actually close the previously-reported "Engine.Stop closes DB pools under a live newer-generation worker" hole.

Trace: every exit path of workerSet.waitForCompletion()

1. Follower branch (draining && drainGen == generation): unlocks, blocks only on <-d.done (never holds mu while waiting). The d pointer it captured is read atomically with the drainGen/generation check under the same lock acquisition, so it can never capture a stale d for a mismatched generation.
2. No-op branch (drained && !draining): returns immediately — correct, since a real leader already resolved this exact generation.
3. Leader branch: sets leaderGen := generation, draining = true, drainGen = leaderGen, drained = true, allocates a fresh *drainResult and unlocks before the blocking runDrain. On return it re-locks and clears draining only if drainGen == leaderGen (i.e., a newer generation's leader hasn't since taken over the shared flag) — then unconditionally sets d.err/close(d.done) on its own local d, outside the lock.

Because each leader owns a private, generation-scoped *drainResult (never a shared mutable field), an orphaned older-generation leader that finishes late can never corrupt or fail to resolve the channel a same-generation follower is waiting on — it only ever writes to the d it allocated for itself, regardless of what s.drain/s.drainGen have been overwritten to in the meantime by a newer leader. Verified this holds for the "stale orphan" scenario the PR is fixing (older leader still in flight past drainWorkerBounded's 30s bound, newer generation rebuilt via StartWorker, a third caller arrives): the third caller's generation check correctly fails to dedup against the stale drainGen, so it becomes the CURRENT generation's own leader and genuinely drains the live pollers — it does not fall through to any code path that would leave it unresolved.

Panic path: runDrain wraps drainWorkers(fns) in recover() and synthesizes an error, so a panicking per-queue WaitForCompletion still reaches the d.err = err; close(d.done) lines in the caller — a same-generation follower can't be stranded by a leader panic.

No path leaves draining stuck true for the CURRENT generation without a corresponding d.done close, and no path leaves a follower's captured d unresolved. I did not find a hang/DoS in the diff.

Does this close the prior finding, or just narrow it?

Closes it. On origin/main, workerSet had no generation concept at all — draining/drain were single un-scoped fields, so ANY caller with draining==true would dedup onto whatever drain was in flight regardless of which worker generation it belonged to. That's exactly the hole: an orphaned old-generation drain (bounded 30s caller returned, real goroutine still running) would cause a later caller — including Engine.Stop() — to falsely "succeed" without ever draining the new generation's live pollers, before Stop() closes wfDB/diagBackend out from under them. The new generation/drainGen pair makes the dedup conditional on drainGen == generation, so a caller for a rebuilt generation always leads its own real drain instead of silently no-op'ing via a stale dedup. internal/wfengine/worker_set_test.go's new "cross-generation drain" spec reproduces this ordering directly and is deliberately RED against the pre-fix dedup logic (Eventually would time out) — it's a real regression test, not just a plausibility check.

Note this does not change the separate, pre-existing, and intentionally-documented trade-off that drainWorkerBounded's 30s timeout can still let Engine.Stop() proceed and close pools if a genuinely slow (not orphaned/stale-generation) drain exceeds the bound — that's an existing, explicitly-commented production guard against a 9-minute CI hang, unrelated to and unmodified by this diff.

Test coverage

The four new specs in worker_set_test.go + the one in engine_concurrent_test.go directly exercise: (1) real Start/Drain overlap against actual go-workflows Worker instances (not just field pokes), (2) the cross-generation false-dedup scenario deterministically, (3) the same-generation follower dedup deterministically, and (4) leader panic recovery. This is good regression coverage that matches the hazards traced above rather than just asserting non-determinism.

Surface / scoping check

Purely internal worker-lifecycle synchronization state (workerGen, generation, drainGen, draining, drain *drainResult) — no request-derived input, no auth/authz surface, no per-user scoping concern, nothing newly logged (existing Warn on drain timeout is unchanged), no SQL/network I/O added. Not applicable to multi-user scoping, injection, or CSP rules.

Findings

No blockers, no majors.

[MINOR] internal/wfengine/engine.go:685-700 (StartWorker) — StartWorker does not check e.stopped before starting a new worker generation (only checks e.workerStarted), so a StartWorker call that races after Engine.Stop() has already set e.stopped = true and begun draining can still start a brand-new worker generation. This is pre-existing/unchanged behavior (not introduced by this diff), and it's workerGen/generation-adjacent enough to flag given this PR's exact focus on Start/Drain/Stop races — worth a follow-up bead to gate StartWorker on !e.stopped, but does not block this PR.

REVIEW VERDICT: 0 blocker, 0 major, 1 minor

## Security/Availability re-review (round 3) — PR #1388 / bookshelf-81hq Scope: `internal/wfengine/{engine.go,worker_set.go,worker_set_test.go,engine_concurrent_test.go,export_test.go}` diff vs `origin/main`. Focus: can the generation-scoped leader/follower drain dedup leave the engine (all background workflow processing: covers/metadata/enrich/scan/bookdrop) permanently wedged, and does it actually close the previously-reported "Engine.Stop closes DB pools under a live newer-generation worker" hole. ### Trace: every exit path of `workerSet.waitForCompletion()` 1. **Follower branch** (`draining && drainGen == generation`): unlocks, blocks only on `<-d.done` (never holds `mu` while waiting). The `d` pointer it captured is read atomically with the `drainGen`/`generation` check under the same lock acquisition, so it can never capture a stale `d` for a mismatched generation. 2. **No-op branch** (`drained && !draining`): returns immediately — correct, since a real leader already resolved this exact generation. 3. **Leader branch**: sets `leaderGen := generation`, `draining = true`, `drainGen = leaderGen`, `drained = true`, allocates a **fresh** `*drainResult` and unlocks before the blocking `runDrain`. On return it re-locks and clears `draining` **only if** `drainGen == leaderGen` (i.e., a newer generation's leader hasn't since taken over the shared flag) — then unconditionally sets `d.err`/`close(d.done)` on its own local `d`, outside the lock. Because each leader owns a **private, generation-scoped `*drainResult`** (never a shared mutable field), an orphaned older-generation leader that finishes late can never corrupt or fail to resolve the channel a same-generation follower is waiting on — it only ever writes to the `d` it allocated for itself, regardless of what `s.drain`/`s.drainGen` have been overwritten to in the meantime by a newer leader. Verified this holds for the "stale orphan" scenario the PR is fixing (older leader still in flight past `drainWorkerBounded`'s 30s bound, newer generation rebuilt via `StartWorker`, a third caller arrives): the third caller's generation check correctly fails to dedup against the stale `drainGen`, so it becomes the CURRENT generation's own leader and genuinely drains the live pollers — it does not fall through to any code path that would leave it unresolved. **Panic path**: `runDrain` wraps `drainWorkers(fns)` in `recover()` and synthesizes an error, so a panicking per-queue `WaitForCompletion` still reaches the `d.err = err; close(d.done)` lines in the caller — a same-generation follower can't be stranded by a leader panic. No path leaves `draining` stuck `true` for the CURRENT generation without a corresponding `d.done` close, and no path leaves a follower's captured `d` unresolved. I did not find a hang/DoS in the diff. ### Does this close the prior finding, or just narrow it? **Closes it.** On `origin/main`, `workerSet` had no generation concept at all — `draining`/`drain` were single un-scoped fields, so ANY caller with `draining==true` would dedup onto whatever drain was in flight regardless of which worker generation it belonged to. That's exactly the hole: an orphaned old-generation drain (bounded 30s caller returned, real goroutine still running) would cause a later caller — including `Engine.Stop()` — to falsely "succeed" without ever draining the new generation's live pollers, before `Stop()` closes `wfDB`/`diagBackend` out from under them. The new `generation`/`drainGen` pair makes the dedup conditional on `drainGen == generation`, so a caller for a rebuilt generation always leads its own real drain instead of silently no-op'ing via a stale dedup. `internal/wfengine/worker_set_test.go`'s new "cross-generation drain" spec reproduces this ordering directly and is deliberately RED against the pre-fix dedup logic (`Eventually` would time out) — it's a real regression test, not just a plausibility check. Note this does **not** change the separate, pre-existing, and intentionally-documented trade-off that `drainWorkerBounded`'s 30s timeout can still let `Engine.Stop()` proceed and close pools if a *genuinely slow* (not orphaned/stale-generation) drain exceeds the bound — that's an existing, explicitly-commented production guard against a 9-minute CI hang, unrelated to and unmodified by this diff. ### Test coverage The four new specs in `worker_set_test.go` + the one in `engine_concurrent_test.go` directly exercise: (1) real `Start`/`Drain` overlap against actual go-workflows `Worker` instances (not just field pokes), (2) the cross-generation false-dedup scenario deterministically, (3) the same-generation follower dedup deterministically, and (4) leader panic recovery. This is good regression coverage that matches the hazards traced above rather than just asserting non-determinism. ### Surface / scoping check Purely internal worker-lifecycle synchronization state (`workerGen`, `generation`, `drainGen`, `draining`, `drain *drainResult`) — no request-derived input, no auth/authz surface, no per-user scoping concern, nothing newly logged (existing `Warn` on drain timeout is unchanged), no SQL/network I/O added. Not applicable to multi-user scoping, injection, or CSP rules. ### Findings No blockers, no majors. [MINOR] internal/wfengine/engine.go:685-700 (StartWorker) — `StartWorker` does not check `e.stopped` before starting a new worker generation (only checks `e.workerStarted`), so a `StartWorker` call that races after `Engine.Stop()` has already set `e.stopped = true` and begun draining can still start a brand-new worker generation. This is pre-existing/unchanged behavior (not introduced by this diff), and it's `workerGen`/generation-adjacent enough to flag given this PR's exact focus on Start/Drain/Stop races — worth a follow-up bead to gate `StartWorker` on `!e.stopped`, but does not block this PR. REVIEW VERDICT: 0 blocker, 0 major, 1 minor

zombor commented 2026-08-08 16:31:36 +00:00

Author

Owner

Copy link

Round-3 Code Review — PR #1388 (bookshelf-81hq)

Reviewed git diff origin/main...origin/bd-bookshelf-81hq at head b45cb2ce: generation-scoped drain dedup (workerSet.generation/drainGen/*drainResult), Engine.workerGen-guarded DrainWorker, runDrain recover, and the new deterministic cross-generation spec.

Correctness walkthrough (verified against the actual diff, not just the doc comments)

• Generation predicate: waitForCompletion()'s leader/follower gate is exactly s.draining && s.drainGen == s.generation (worker_set.go). Traced the 3-generation interleaving (K orphaned in flight, K+1 leads its own drain, K+2 arrives later): each leader captures leaderGen := s.generation under the lock and only clears s.draining in its post-drain unlock block if s.drainGen == leaderGen. Because drainGen is only ever overwritten by a strictly newer leader (serialized through s.mu), an orphaned older leader can never clobber a newer leader's in-flight state, and a caller for a newer generation never dedups onto a stale *drainResult. This closes the exact hole the round-2 re-review found.
• Deadlock/hang freedom: every leader exit path sets d.err = err; close(d.done) unconditionally after the drain call returns (worker_set.go, end of waitForCompletion). runDrain's defer recover() guarantees s.runDrain(fns) always returns normally (never propagates a panic), so the only way to skip close(d.done) would be a panic in s.drains() while still holding s.mu before the first Unlock() — that risk is pre-existing (identical shape existed before this PR) and not introduced/worsened by this diff, so not scoring it against round 3.
• Locking discipline: generation, drainGen, draining, drain, drained, workers are read/written only under s.mu.Lock(), except the final d.err = err; close(d.done) pair, which is safe without the lock because each leader owns an exclusive *drainResult it alone writes, and followers only ever read d.err after receiving from the closed d.done channel — a valid happens-before edge via channel close/receive, not a data race (confirmed by -race per the completion comment).
• start() non-blocking: start() only inspects/writes s.drained/s.generation/s.workers under s.mu, copies the per-worker start closures, and unlocks before calling startWorkers(fns) outside the lock — it never blocks on drainGen/draining/drain.done. Engine.mu is never held across e.startWorker/e.stopWorker (both release before their respective blocking calls) — consistent with pre-existing structure.
• Engine-level workerGen guard (engine.go DrainWorker): correctly snapshots startGen before releasing e.mu for the (bounded) drain and only clears workerStarted if e.workerGen == startGen on return — mirrors the ws-level pattern one layer up.

Tests

• The new cross-generation spec (worker_set_test.go, "workerSet cross-generation drain") drives the real ws.start()/ws.waitForCompletion() via the StartForRaceTest/WaitForCompletionForRaceTest/DrainingForRaceTest seams (thin passthroughs to production code, consistent with wfengine's standing export_test allowlist) and uses a genuinely deterministic proof (generation K is never cancelled/released for the spec's lifetime, so the only way the third call can return within the bounded Eventually window is by leading its own drain of K+1) — this is a real RED/GREEN regression test, not a flaky timing assertion.
• runDrain's recover path is exercised directly via RunDrainPanicForTest (thin passthrough), and the same-generation follower branch is now covered deterministically (separate from the nondeterministic "real Start/Drain overlap" spec). All test files declare package wfengine_test (black-box).
• No .golangci.yml or scripts/check-coverage.sh changes in this diff — no new lint/coverage exclusions.

Findings

[MINOR] internal/wfengine/engine.go workerGen field vs internal/wfengine/worker_set.go generation field — two independent, uncoordinated generation counters now exist (Engine.workerGen at the engine layer, workerSet.generation at the worker-set layer), tracking overlapping but not identical concepts (Engine tracks "has StartWorker run since the last completed drain"; workerSet tracks "which batch of live Worker instances is current"). They happen to stay in lockstep today because Engine.StartWorker/DrainWorker are the only callers of ws.start/ws.waitForCompletion, but the duplication is a latent maintenance trap if a third caller is ever added at either layer. Not a bug in this diff — flagging for awareness; no fix required now, worth a doc note tying the two together explicitly (e.g. "Engine.workerGen and workerSet.generation are deliberately parallel, not shared, because ...").

[MINOR] internal/wfengine/engine.go Stop() / internal/wfengine/worker_set.go "GENERATION SCOPING" doc — generation-scoping fixes the false-dedup hole, but a residual, pre-existing limitation remains undocumented at the Engine.Stop() call site: if drainWorkerBounded's 30s timeout previously orphaned an older generation's real drain goroutine that is still running when Stop() later closes wfDB/diagBackend, that orphaned goroutine's in-flight go-workflows calls will hit closed connection pools (this is unchanged by round 3 — round 3 only guarantees Stop() itself now drains the current generation correctly, not that all older orphaned generations have finished). This was already implicitly accepted (the new cross-generation spec explicitly asserts "leaves the orphaned older generation's drain still running") but Engine.Stop()'s doc comment doesn't call this out the way DrainWorker()'s does. Suggest a one-line addition to Stop()'s doc, or a follow-up bead, to keep this documented pre-existing limitation from being lost. Does not block this PR — pre-existing, not introduced/worsened here.

No blockers or majors found. The generation-scoping fix correctly closes the cross-generation dedup hole identified in the round-2 re-review, done is closed on every leader exit path (including the new recover()), all shared fields are lock-protected except the intentionally lock-free channel-synchronized d.err/close(d.done) pair, start() remains non-blocking, and the new tests exercise real production code paths deterministically.

REVIEW VERDICT: 0 blocker, 0 major, 2 minor

## Round-3 Code Review — PR #1388 (bookshelf-81hq) Reviewed `git diff origin/main...origin/bd-bookshelf-81hq` at head b45cb2ce: generation-scoped drain dedup (`workerSet.generation`/`drainGen`/`*drainResult`), `Engine.workerGen`-guarded `DrainWorker`, `runDrain` recover, and the new deterministic cross-generation spec. ### Correctness walkthrough (verified against the actual diff, not just the doc comments) - **Generation predicate**: `waitForCompletion()`'s leader/follower gate is exactly `s.draining && s.drainGen == s.generation` (worker_set.go). Traced the 3-generation interleaving (K orphaned in flight, K+1 leads its own drain, K+2 arrives later): each leader captures `leaderGen := s.generation` under the lock and only clears `s.draining` in its post-drain unlock block `if s.drainGen == leaderGen`. Because `drainGen` is only ever overwritten by a strictly newer leader (serialized through `s.mu`), an orphaned older leader can never clobber a newer leader's in-flight state, and a caller for a newer generation never dedups onto a stale `*drainResult`. This closes the exact hole the round-2 re-review found. - **Deadlock/hang freedom**: every leader exit path sets `d.err = err; close(d.done)` unconditionally after the drain call returns (worker_set.go, end of `waitForCompletion`). `runDrain`'s `defer recover()` guarantees `s.runDrain(fns)` always returns normally (never propagates a panic), so the only way to skip `close(d.done)` would be a panic in `s.drains()` while still holding `s.mu` before the first `Unlock()` — that risk is pre-existing (identical shape existed before this PR) and not introduced/worsened by this diff, so not scoring it against round 3. - **Locking discipline**: `generation`, `drainGen`, `draining`, `drain`, `drained`, `workers` are read/written only under `s.mu.Lock()`, *except* the final `d.err = err; close(d.done)` pair, which is safe without the lock because each leader owns an exclusive `*drainResult` it alone writes, and followers only ever read `d.err` after receiving from the closed `d.done` channel — a valid happens-before edge via channel close/receive, not a data race (confirmed by -race per the completion comment). - **start() non-blocking**: `start()` only inspects/writes `s.drained`/`s.generation`/`s.workers` under `s.mu`, copies the per-worker start closures, and unlocks before calling `startWorkers(fns)` outside the lock — it never blocks on `drainGen`/`draining`/`drain.done`. `Engine.mu` is never held across `e.startWorker`/`e.stopWorker` (both release before their respective blocking calls) — consistent with pre-existing structure. - **Engine-level `workerGen` guard** (engine.go `DrainWorker`): correctly snapshots `startGen` before releasing `e.mu` for the (bounded) drain and only clears `workerStarted` if `e.workerGen == startGen` on return — mirrors the ws-level pattern one layer up. ### Tests - The new cross-generation spec (`worker_set_test.go`, "workerSet cross-generation drain") drives the real `ws.start()`/`ws.waitForCompletion()` via the `StartForRaceTest`/`WaitForCompletionForRaceTest`/`DrainingForRaceTest` seams (thin passthroughs to production code, consistent with wfengine's standing export_test allowlist) and uses a genuinely deterministic proof (generation K is *never* cancelled/released for the spec's lifetime, so the only way the third call can return within the bounded `Eventually` window is by leading its own drain of K+1) — this is a real RED/GREEN regression test, not a flaky timing assertion. - `runDrain`'s recover path is exercised directly via `RunDrainPanicForTest` (thin passthrough), and the same-generation follower branch is now covered deterministically (separate from the nondeterministic "real Start/Drain overlap" spec). All test files declare `package wfengine_test` (black-box). - No `.golangci.yml` or `scripts/check-coverage.sh` changes in this diff — no new lint/coverage exclusions. ### Findings **[MINOR]** internal/wfengine/engine.go workerGen field vs internal/wfengine/worker_set.go generation field — two independent, uncoordinated generation counters now exist (`Engine.workerGen` at the engine layer, `workerSet.generation` at the worker-set layer), tracking overlapping but not identical concepts (Engine tracks "has StartWorker run since the last completed drain"; workerSet tracks "which batch of live Worker instances is current"). They happen to stay in lockstep today because `Engine.StartWorker`/`DrainWorker` are the only callers of `ws.start`/`ws.waitForCompletion`, but the duplication is a latent maintenance trap if a third caller is ever added at either layer. Not a bug in this diff — flagging for awareness; no fix required now, worth a doc note tying the two together explicitly (e.g. "Engine.workerGen and workerSet.generation are deliberately parallel, not shared, because ..."). **[MINOR]** internal/wfengine/engine.go Stop() / internal/wfengine/worker_set.go "GENERATION SCOPING" doc — generation-scoping fixes the *false-dedup* hole, but a residual, pre-existing limitation remains undocumented at the `Engine.Stop()` call site: if `drainWorkerBounded`'s 30s timeout previously orphaned an older generation's real drain goroutine that is *still* running when `Stop()` later closes `wfDB`/`diagBackend`, that orphaned goroutine's in-flight go-workflows calls will hit closed connection pools (this is unchanged by round 3 — round 3 only guarantees `Stop()` itself now drains the *current* generation correctly, not that all older orphaned generations have finished). This was already implicitly accepted (the new cross-generation spec explicitly asserts "leaves the orphaned older generation's drain still running") but `Engine.Stop()`'s doc comment doesn't call this out the way `DrainWorker()`'s does. Suggest a one-line addition to `Stop()`'s doc, or a follow-up bead, to keep this documented pre-existing limitation from being lost. Does not block this PR — pre-existing, not introduced/worsened here. No blockers or majors found. The generation-scoping fix correctly closes the cross-generation dedup hole identified in the round-2 re-review, `done` is closed on every leader exit path (including the new recover()), all shared fields are lock-protected except the intentionally lock-free channel-synchronized `d.err`/`close(d.done)` pair, `start()` remains non-blocking, and the new tests exercise real production code paths deterministically. REVIEW VERDICT: 0 blocker, 0 major, 2 minor

zombor merged commit c833c0e234 into main 2026-08-08 18:07:03 +00:00

zombor referenced this pull request from a commit 2026-08-08 18:07:05 +00:00

Merge pull request 'fix(wfengine): close DrainWorker TOCTOU (bookshelf-81hq)' (#1388) from bd-bookshelf-81hq into main

zombor referenced this pull request 2026-08-28 17:30:18 +00:00

fix(wfengine): worker-lifecycle hardening — e.stopped guard + generation/Stop docs (bookshelf-289dd) #1458

Sign in to join this conversation.

Reviewers

No reviewers

Labels

Clear labels

No items

No labels

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

1 participant

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference

zombor/pergamum!1388

No description provided.