test(lifecycle-poc): fix CI flake in TestServiceLifecycle_PipelineError/PipelineStop

[devarismeroxa]

Summary

Fixes the CI flake noted on #2534 that failed the test job on PRs #2576 and
#2577 (TestServiceLifecycle_PipelineError, TestServiceLifecycle_PipelineStop
in pkg/lifecycle-poc/service_test.go).

Risk tier: Tier 1 (pipeline lifecycle / status-reporting correctness in the
data-path-adjacent pkg/lifecycle-poc package — the arch-v2 successor to
pkg/lifecycle). Human sign-off required before merge per the standing policy.
Not merging this myself — opening for review as requested.

What was actually wrong

The starting hypothesis (a raw time.Sleep(100ms) used as synchronization) was
correct but incomplete. Digging into why the flake also produced a bare status
mismatch (SystemStopped != Running) turned up four distinct, real
concurrency bugs in product code, each found and fixed via -race -shuffle=on repetition — not just a test-timing issue.

1. Test-side: sleep-based sync (the originally-suspected cause)

TestServiceLifecycle_PipelineStop (and sibling PipelineSuccess) slept 100ms
to let 10 records "finish flowing" before calling Stop/StopAll. Under CI
load 100ms isn't always enough, so the stop call races the data path,
under-delivers records, and fails the source/destination mocks' record-count
assertions in t.Cleanup with an unrelated-looking error.

Fix: replaced both sleeps with waitForRecordsAcked, which polls the
source connector's exported, lock-guarded State field (set by Source.Ack)
until every record has actually been acked, instead of guessing a duration.

2. WaitPipeline lookup-after-delete race (service.go)

WaitPipeline looked up the pipeline in runningPipelines and returned a bare
nil if not found — but the cleanup goroutine deletes the entry after
finishing, so a caller racing that cleanup could get a false nil instead of
the pipeline's terminal error. This is the identical bug already diagnosed
and fixed in the sibling pkg/lifecycle package (see
docs/design-documents/20260706-forceful-stop-test-determinism.md, filed as
#2521) — it just hadn't been ported to pkg/lifecycle-poc. Ported the same
terminalErrors-map fix here.

3. tomb.Kill ordering race on a fatal worker error (service.go)

The worker goroutine's error only got recorded on the tomb via tomb.v2's own
post-return bookkeeping (t.run, after f() returns) — which races the
cleanup goroutine waking from workersWg.Wait() and reading rp.t.Err().
Losing that race made a fatal stop get reported as a clean graceful stop
(reproduced locally as Degraded != UserStopped). Fixed by calling
rp.t.Kill(err) synchronously before returning from the worker goroutine.

4. UpdateStatus concurrency on the same *pipeline.Instance — the actual root cause of the CI-observed "SystemStopped != Running"

runPipeline's own initial UpdateStatus(StatusRunning) call and the cleanup
goroutine's terminal UpdateStatus call are not safe to run concurrently
on the same instance: SetStatus is lock-guarded, but the Error field write
and the store's JSON-encode of the whole struct for persistence are not. With
mocks providing zero I/O latency, the cleanup goroutine could reach its own
UpdateStatus call while the initial one was still mid-flight (both goroutines
were already registered via tomb.Go before the initial call ran) and win
the race — clobbering a correct terminal status back to "running". This is
confirmed by an actual -race DATA RACE report (not just a logic bug), and
it's the mechanism that best explains the exact CI symptom.

Fix: gate the cleanup goroutine's UpdateStatus call on a channel closed only
after the initial one fully completes. Both tomb.Go registrations stay
adjacent (nothing slow between them) so tomb.alive reaches 2 before either
goroutine can finish — a naive first attempt that moved the initial
UpdateStatus call between the two tomb.Go calls introduced a different
bug (tomb.Go called after all goroutines terminated panic) when the worker
finished before that call returned.

5. Worker.Stop teardown-before-stop-flag race (funnel/worker.go)

Stop tore down the source (which nils the plugin) before setting
w.stop, so a concurrent Read() racing the stop could observe
plugin.ErrPluginNotRunning before doTask's graceful-stop check
(w.stop.Load()) saw stop=true, misreporting a graceful stop as a real
failure (reproduced as "failed to read from source: plugin is not running").
Fixed by setting w.stop before tearing down.

6. Missing persister.Wait() (test hygiene)

PipelineError and PipelineStop were missing the defer persister.Wait()
that PipelineSuccess already had. Without it, the persister's background
flush goroutine can log via log.Test(t) after the test function returns,
which panics (Log in goroutine after ... has completed) and can corrupt
whatever test happens to be running concurrently in the same process — this
alone could plausibly explain some of the wider "unrelated" CI flakiness.
Confirmed by an actual panic + race report during repro.

Verification (exact commands, all pkg/lifecycle-poc/)

go test ./pkg/lifecycle-poc/ -run 'TestServiceLifecycle' -race -count=500
→ 0 failures (ok, 9-11s)

for i in $(seq 1 3); do yes >/dev/null & done
go test ./pkg/lifecycle-poc/ -run 'TestServiceLifecycle' -count=300
→ 0 failures (ok), load generators killed after

go test ./pkg/lifecycle-poc/... -race -shuffle=on -count=50
→ 0 failures (ok, both pkg/lifecycle-poc and pkg/lifecycle-poc/funnel)

Each of the four product-side bugs was independently reproduced before its
fix (bare status mismatches, a real data race report, and a real panic — logs
available on request) and confirmed gone after. All three commands above were
also run a second time for confidence, still clean.

go build ./...            → clean
golangci-lint run ./pkg/lifecycle-poc/... → 0 issues
go test ./pkg/conduit/...  → ok (sole external consumer of lifecycle-poc.Service)

Failure-mode analysis (Tier 1 requirement)

• What could this break: all four product fixes are in pkg/lifecycle-poc,
  which is gated behind a feature flag and not yet the default engine
  (pkg/lifecycle is still primary). Blast radius is limited to that package
  and its direct consumer in pkg/conduit/runtime.go.
• Which metric/alert would show a regression: pipeline status
  (StatusRunning/StatusDegraded/etc.) reported via the API/CLI would be
  wrong, or WaitPipeline callers (including graceful shutdown) would hang or
  return stale results. No production traffic depends on pkg/lifecycle-poc
  yet, so live-system risk is effectively zero right now; the value is
  fixing arch-v2 before it becomes the default.
• Rollback: revert this commit; no serialized format or public contract
  changes.

Not fixed here (flagged, out of scope for a flaky-test PR)

• pkg/lifecycle-poc's force-stop path kills the tomb with a bare
  pipeline.ErrForceStop instead of cerrors.FatalError(...) like
  pkg/lifecycle does — meaning a force-stopped pipeline in lifecycle-poc
  hits the "recovery not implemented" no-op branch below instead of being
  reported Degraded.
• The "recovery not implemented" branch in the cleanup goroutine's switch is a
  complete no-op (all commented out, TODO) — a pipeline that stops with a
  non-fatal, non-recognized-as-graceful error is left at status Running
  forever, with no terminal status ever recorded. This is a real correctness
  gap (an operator would see a dead pipeline reported as running) but fixing
  it means either implementing recovery or deciding what the interim
  behavior should be — a design decision beyond this PR's scope.
• Under much heavier artificial stress than requested (6x yes loaders +
  -shuffle=on -count=1500, well beyond the specified bar) I intermittently
  saw a mock-helper assertion (pkg/plugin/connector/mock/source.go:116,
  done.Load()) fail — a background mock goroutine's last few non-blocking
  instructions hadn't been scheduled by the time t.Cleanup ran. This did
  not reproduce at the requested stress levels (all three verification
  commands above, run twice, were clean) and touches shared mock
  infrastructure used well beyond this package, so I left it alone rather
  than risk a wider, rushed change.

Self-review

Re-read the diff hunting for error paths, concurrency, and resource cleanup:

• close(startupDone) in runPipeline is unconditional (both the success and
  UpdateStatus-error paths reach it), so the cleanup goroutine — already
  blocked on <-startupDone — can never hang waiting for it.
• Both tomb.Go registrations for the worker and cleanup goroutines stay
  textually adjacent with nothing slow between them, so tomb.alive can't
  reach 0 (and panic on the next Go call) before the cleanup goroutine is
  registered.
• rp.t.Kill(err) called manually from inside the worker goroutine before
  returning is idempotent with tomb.v2's own subsequent kill(err) call for
  the same goroutine (verified against the vendored tomb.v2 source: a second
  kill call on an already-non-ErrStillAlive reason is a no-op).
• Worker.Stop's reordering (w.stop.Store(true) before tearDownSource)
  only tightens an existing signal; nothing else in the codepath depended on
  the old order (checked all w.stop.Load() call sites).

🤖 Generated with Claude Code

https://claude.ai/code/session_01Tapg9dLWXKMZJoL65R24vd