fix(wfengine): guard stopped/workerStarted with mutex (bookshelf-ke3w) #1029

Merged

zombor merged 1 commit from bd-bookshelf-ke3w into main 2026-07-08 06:15:20 +00:00

Conversation 1 Commits 1 Files changed 4 +129 -2

zombor commented 2026-07-08 05:52:33 +00:00

Owner

Copy link

Summary

• Add sync.Mutex to Engine struct protecting the stopped and workerStarted bool fields
• StartWorker holds the lock across check + start + set — idempotent under concurrent calls
• Stop locks for the guard check+set, drains and closes pools outside the lock
• DrainWorker locks for the guard check, drains outside, locks again to reset workerStarted
• WorkerStarted() in export_test.go reads under the lock
• New engine_concurrent_test.go: black-box race tests asserting 10 concurrent StartWorker calls invoke startWorker exactly once, and 10 concurrent Stop calls invoke drain exactly once
• Fix misleading SetDrainTimeoutForTest(-1) comment in worker_lifecycle_test.go

Test plan

• go test -race ./internal/wfengine/... passes (no race detector reports)
• New engine_concurrent_test.go specs: StartWorker idempotency + Stop idempotency under 10 concurrent goroutines — pass
• All 1329 existing wfengine specs pass
• make test passes across all internal packages

Closes bead bookshelf-ke3w on merge.

## Summary - Add `sync.Mutex` to `Engine` struct protecting the `stopped` and `workerStarted` bool fields - `StartWorker` holds the lock across check + start + set — idempotent under concurrent calls - `Stop` locks for the guard check+set, drains and closes pools outside the lock - `DrainWorker` locks for the guard check, drains outside, locks again to reset `workerStarted` - `WorkerStarted()` in `export_test.go` reads under the lock - New `engine_concurrent_test.go`: black-box race tests asserting 10 concurrent `StartWorker` calls invoke `startWorker` exactly once, and 10 concurrent `Stop` calls invoke drain exactly once - Fix misleading `SetDrainTimeoutForTest(-1)` comment in `worker_lifecycle_test.go` ## Test plan - [ ] `go test -race ./internal/wfengine/...` passes (no race detector reports) - [ ] New `engine_concurrent_test.go` specs: StartWorker idempotency + Stop idempotency under 10 concurrent goroutines — pass - [ ] All 1329 existing wfengine specs pass - [ ] `make test` passes across all internal packages Closes bead bookshelf-ke3w on merge.

zombor added 1 commit 2026-07-08 05:52:34 +00:00

fix(wfengine): guard stopped/workerStarted with mutex (bookshelf-ke3w) ... 2a443499e3

Stop(), DrainWorker(), and StartWorker() read/wrote the stopped and
workerStarted bool fields without synchronisation. The race detector
would flag any concurrent caller.

- Add sync.Mutex (mu) to Engine, guarding both bool fields.
- StartWorker: holds the lock throughout check + start + set, making
  it idempotent under concurrent calls (second caller returns nil).
- Stop: locks for check + set + capture, drains and closes pools outside.
- DrainWorker: locks for guard check, drains outside, locks again to
  reset workerStarted.
- WorkerStarted (export_test.go): reads under lock.
- Add engine_concurrent_test.go: black-box race tests asserting that
  10 concurrent StartWorker calls invoke startWorker exactly once, and
  10 concurrent Stop calls invoke drain exactly once (verifiable with
  -race).
- Clarify the SetDrainTimeoutForTest(-1) comment in
  worker_lifecycle_test.go: the 30s fallback timeout is irrelevant
  because the stub stopWorker returns immediately, not because of
  a 30s default.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

zombor commented 2026-07-08 06:04:41 +00:00

Author

Owner

Copy link

CODE REVIEW: APPROVED

Reviewed diff only (no test re-runs; CI is the behavioral source of truth).

---

Phase 0: DEMO verification

No DEMO block required for this bead type (concurrency guard / internal refactor with no user-facing surface). Proceeding directly to spec compliance and code quality.

---

Phase 1: Spec compliance

All four stated goals are met: sync.Mutex added to Engine; StartWorker holds the lock for check+start+set; Stop and DrainWorker lock for the guard check/set but drain outside the lock.

---

Phase 2: Code quality

Concurrency correctness — detailed findings:

NO DEADLOCK — PASS

startWorker and stopWorker are func fields (engine.go:189–190), not Engine methods. They cannot re-acquire e.mu. drainWorkerBounded (engine.go:1133–1155) also holds no lock. So StartWorker holding e.mu via defer e.mu.Unlock() while calling e.startWorker(ctx) is safe — there is no re-entrant lock path.

Lock/unlock pairing is clean on every path:

• StartWorker (engine.go:1029–1041): Lock() + defer Unlock() — all three exit paths (idempotent return, error return, success return) unlock correctly.
• DrainWorker (engine.go:1066–1080): early-return path does explicit Unlock before return nil; normal path does Unlock, drain, re-Lock, Unlock — both paths fully paired. On drain error, the function returns without the re-lock, leaving workerStarted=true (correct: drain failed so the worker may still be live, keeping the state true enables a retry).
• Stop (engine.go:1093–1115): early-return path does explicit Unlock before return nil; normal path sets stopped=true, captures workerWasStarted, then Unlock before the drain. Fully paired on all exit paths.

NO REMAINING RACE — PASS

Every read and write of stopped and workerStarted is under e.mu:

• StartWorker: read workerStarted (line 1032) and write workerStarted=true (line 1038) — both under defer Unlock.
• DrainWorker: read e.stopped || !e.workerStarted (line 1068) under lock; write workerStarted=false (line 1078) under re-acquired lock.
• Stop: read e.stopped (line 1095) and write e.stopped=true (line 1097) under lock; read e.workerStarted into workerWasStarted (line 1100) under same lock before unlocking.
• WorkerStarted() in export_test.go (line 368–371): reads e.workerStarted under Lock() + defer Unlock(). ✓

No unsynchronized access remains. -race will find nothing.

StartWorker idempotency — PASS

Check (workerStarted), call (startWorker), and set (workerStarted=true) are all within the scope of defer e.mu.Unlock(). Concurrent goroutines queue on the lock; whichever gets it first will either see workerStarted=true and early-return, or start and set it — the next waiter will then see workerStarted=true and return. Exactly one worker is ever started.

Stop-after-Start ordering — PASS

Stop sets stopped=true and reads workerWasStarted atomically under the lock before unlocking. Subsequent concurrent Stop calls see stopped=true and return immediately. The drain happens exactly once.

Captured-state pattern — PASS

Stop captures workerWasStarted := e.workerStarted while holding the lock (engine.go:1100), then unlocks (line 1101), then branches on the captured value. No shared mutable state is read after the unlock. Correct.

DrainWorker resetting workerStarted — PASS (with one MINOR)

The re-lock at engine.go:1077 correctly gates the workerStarted=false write. However:

---

[MINOR] internal/wfengine/engine.go:1072–1079 — DrainWorker re-lock creates a logical TOCTOU window for workerStarted

Between the first Unlock (line 1072, after the guard check) and the re-Lock (line 1077, after drain), a concurrent StartWorker call can acquire and release the lock, set workerStarted=true, and start a new worker. DrainWorker then re-locks and unconditionally writes workerStarted=false, silently orphaning the new worker (it runs but is no longer tracked; future DrainWorker/Stop calls will skip it). This is not a data race (the -race detector won't catch it) but a logical race requiring the caller to overlap StartWorker and DrainWorker. The DrainWorker docstring says "StartWorker may be called again after DrainWorker returns" — not during — so this is an edge case requiring unusual API misuse. This issue is architecturally pre-existing (the PR did not introduce the unlock-drain-relock structure; it only added the lock). Suggested fix for a follow-up: after drain, re-check !e.stopped under the re-lock before writing workerStarted=false, or document the "do not call StartWorker while DrainWorker is in flight" constraint in the function's doc comment.

---

Tests — PASS

engine_concurrent_test.go:

• package wfengine_test — black-box ✓
• Genuine goroutine fan-out (10 goroutines + sync.WaitGroup) ✓
• sync/atomic.Int32 for the shared counter — no data race on the counter itself ✓
• StartWorker idempotency: asserts startCount == 1 and WorkerStarted() == true after 10 concurrent calls ✓
• Stop idempotency: asserts stopCount == 1 after 10 concurrent calls; the stub wired as stopWorker is what drainWorkerBounded calls ✓
• WorkerStarted() shim reads under lock ✓

No white-box tests. No coverage exclusions added. No new .golangci.yml exclusions observed in the diff.

---

REVIEW VERDICT: 0 blocker, 0 major, 1 minor

## CODE REVIEW: APPROVED Reviewed diff only (no test re-runs; CI is the behavioral source of truth). --- ### Phase 0: DEMO verification No DEMO block required for this bead type (concurrency guard / internal refactor with no user-facing surface). Proceeding directly to spec compliance and code quality. --- ### Phase 1: Spec compliance All four stated goals are met: `sync.Mutex` added to `Engine`; `StartWorker` holds the lock for check+start+set; `Stop` and `DrainWorker` lock for the guard check/set but drain outside the lock. --- ### Phase 2: Code quality **Concurrency correctness — detailed findings:** **NO DEADLOCK — PASS** `startWorker` and `stopWorker` are `func` fields (engine.go:189–190), not `Engine` methods. They cannot re-acquire `e.mu`. `drainWorkerBounded` (engine.go:1133–1155) also holds no lock. So `StartWorker` holding `e.mu` via `defer e.mu.Unlock()` while calling `e.startWorker(ctx)` is safe — there is no re-entrant lock path. Lock/unlock pairing is clean on every path: - `StartWorker` (engine.go:1029–1041): `Lock()` + `defer Unlock()` — all three exit paths (idempotent return, error return, success return) unlock correctly. - `DrainWorker` (engine.go:1066–1080): early-return path does explicit `Unlock` before `return nil`; normal path does `Unlock`, drain, re-`Lock`, `Unlock` — both paths fully paired. On drain error, the function returns without the re-lock, leaving `workerStarted=true` (correct: drain failed so the worker may still be live, keeping the state true enables a retry). - `Stop` (engine.go:1093–1115): early-return path does explicit `Unlock` before `return nil`; normal path sets `stopped=true`, captures `workerWasStarted`, then `Unlock` before the drain. Fully paired on all exit paths. **NO REMAINING RACE — PASS** Every read and write of `stopped` and `workerStarted` is under `e.mu`: - `StartWorker`: read `workerStarted` (line 1032) and write `workerStarted=true` (line 1038) — both under `defer Unlock`. - `DrainWorker`: read `e.stopped || !e.workerStarted` (line 1068) under lock; write `workerStarted=false` (line 1078) under re-acquired lock. - `Stop`: read `e.stopped` (line 1095) and write `e.stopped=true` (line 1097) under lock; read `e.workerStarted` into `workerWasStarted` (line 1100) under same lock before unlocking. - `WorkerStarted()` in export_test.go (line 368–371): reads `e.workerStarted` under `Lock()` + `defer Unlock()`. ✓ No unsynchronized access remains. `-race` will find nothing. **StartWorker idempotency — PASS** Check (`workerStarted`), call (`startWorker`), and set (`workerStarted=true`) are all within the scope of `defer e.mu.Unlock()`. Concurrent goroutines queue on the lock; whichever gets it first will either see `workerStarted=true` and early-return, or start and set it — the next waiter will then see `workerStarted=true` and return. Exactly one worker is ever started. **Stop-after-Start ordering — PASS** `Stop` sets `stopped=true` and reads `workerWasStarted` atomically under the lock before unlocking. Subsequent concurrent `Stop` calls see `stopped=true` and return immediately. The drain happens exactly once. **Captured-state pattern — PASS** `Stop` captures `workerWasStarted := e.workerStarted` while holding the lock (engine.go:1100), then unlocks (line 1101), then branches on the captured value. No shared mutable state is read after the unlock. Correct. **DrainWorker resetting workerStarted — PASS (with one MINOR)** The re-lock at engine.go:1077 correctly gates the `workerStarted=false` write. However: --- [MINOR] internal/wfengine/engine.go:1072–1079 — DrainWorker re-lock creates a logical TOCTOU window for workerStarted Between the first `Unlock` (line 1072, after the guard check) and the re-`Lock` (line 1077, after drain), a concurrent `StartWorker` call can acquire and release the lock, set `workerStarted=true`, and start a new worker. DrainWorker then re-locks and unconditionally writes `workerStarted=false`, silently orphaning the new worker (it runs but is no longer tracked; future `DrainWorker`/`Stop` calls will skip it). This is not a data race (the `-race` detector won't catch it) but a logical race requiring the caller to overlap `StartWorker` and `DrainWorker`. The DrainWorker docstring says "StartWorker may be called again *after* DrainWorker returns" — not during — so this is an edge case requiring unusual API misuse. This issue is architecturally pre-existing (the PR did not introduce the unlock-drain-relock structure; it only added the lock). Suggested fix for a follow-up: after drain, re-check `!e.stopped` under the re-lock before writing `workerStarted=false`, or document the "do not call StartWorker while DrainWorker is in flight" constraint in the function's doc comment. --- **Tests — PASS** `engine_concurrent_test.go`: - `package wfengine_test` — black-box ✓ - Genuine goroutine fan-out (10 goroutines + `sync.WaitGroup`) ✓ - `sync/atomic.Int32` for the shared counter — no data race on the counter itself ✓ - `StartWorker` idempotency: asserts `startCount == 1` and `WorkerStarted() == true` after 10 concurrent calls ✓ - `Stop` idempotency: asserts `stopCount == 1` after 10 concurrent calls; the stub wired as `stopWorker` is what `drainWorkerBounded` calls ✓ - `WorkerStarted()` shim reads under lock ✓ No white-box tests. No coverage exclusions added. No new `.golangci.yml` exclusions observed in the diff. --- REVIEW VERDICT: 0 blocker, 0 major, 1 minor

zombor force-pushed bd-bookshelf-ke3w from 2a443499e3 to d41b732160 2026-07-08 06:09:15 +00:00 Compare

zombor merged commit c02c44da95 into main 2026-07-08 06:15:20 +00:00

zombor referenced this pull request from a commit 2026-07-08 06:15:25 +00:00

Merge pull request 'fix(wfengine): guard stopped/workerStarted with mutex (bookshelf-ke3w)' (#1029) from bd-bookshelf-ke3w into main

zombor referenced this pull request 2026-08-07 13:32:56 +00:00

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

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!1029

No description provided.