feat(lifecycle,provisioning): StopAndWait + ApplyPlanLive (PR1 of #2588)

[devarismeroxa]

Summary

Risk tier: Tier 1 (data path). Needs human sign-off — do not merge on this review alone.

PR1 of the #2588 arc (live-server apply to a running pipeline). Scope is the engine only: no API RPCs, no CLI/MCP wiring — that's PR2. Implements the corrected design from docs/design-documents/20260708-live-server-deploy-apply.md, specifically the "Review outcome & required rework" section (the Tier-1 design review found the original §2 raced an in-flight drain).

• lifecycle.Service.StopAndWait(ctx, id) error — Stop(ctx,id,false) → WaitPipeline(id) (all node goroutines exited, i.e. genuinely drained — not just "stop signal injected") → connectors.WaitPersisted() (blocks until the connector position/state flush that draining triggered is durably committed). Only after all three does it return.
• provisioning.Service.ApplyPlanLive(ctx, desired, hash) (Diff, error) — like ApplyPlan (recompute+verify hash → plan_stale; empty diff → no-op) but for a running pipeline: StopAndWait → transactionalImport (the #2595 transactional path, factored out so ApplyPlan and ApplyPlanLive share it) → Start. Any error after StopAndWait leaves the pipeline stopped — never auto-restarted into a half-applied state. For a stopped pipeline, it's identical to ApplyPlan's tail. The standalone ApplyPlan is unchanged and still refuses running pipelines outright (it has no lifecycle to drive).
• pkg/lifecycle-poc (the Preview.PipelineArchV2 / funnel arch) gets a StopAndWait that always refuses with a coded error (lifecycle_v2.stop_and_wait_unsupported). Its Stop/drain semantics haven't had the audit pkg/lifecycle's got in the design review — silently building on an unaudited stop path would reintroduce the exact class of bug the review found. ApplyPlanLive needs no poc-specific branch: whatever StopAndWait returns is surfaced as-is.

How StopAndWait awaits durability

connector.Persister.Wait() (existing) blocks on connWg (every connector across the whole process reaching ConnectorStopped) and flushWg. Using it from a single pipeline's stop-and-drain path would deadlock for as long as any other pipeline stays running, since the persister batches writes across all pipelines.

New Persister.WaitPendingWrites() blocks only on flushWg — the flush(es) already triggered. A connector's own Teardown → ConnectorStopped → triggerFlush increments flushWg synchronously before returning, and that happens strictly before SourceNode.Run() returns (Teardown is called from Run's deferred cleanup, after openMsgTracker.Wait() — i.e. after every in-flight message has been acked end-to-end), which is strictly before WaitPipeline unblocks. So by the time StopAndWait calls WaitPersisted, the flush for this pipeline's connectors has already been triggered, and flushWg.Wait() is guaranteed to observe it complete (sync.WaitGroup can't miss an Add that happened-before the Wait call). connector.Service.WaitPersisted() exposes this via the connector service already held by the lifecycle service — no persister handle is passed to provisioning, per the design doc.

Test results

Drain-no-loss (AC-3) — TestServiceLifecycle_StopAndWait_DrainsAndPersists (pkg/lifecycle): runs a real pipeline (mocked plugins, real node goroutines, real connector.Persister, real inmemory.DB wrapped in a delayingDB that adds a 200ms delay to every NewTransaction, so a StopAndWait that returned early would deterministically read stale/missing state, not just flake). Asserts StopAndWait blocks for at least the flush delay and that the source's position, read directly back from the store, matches the last record. Verified this test actually catches the regression: temporarily removed the WaitPersisted() call from StopAndWait — the test failed immediately (not flakily); restored — it passes. -race -count=20: clean, no flakes.

Kill-mid-apply (AC-6) / restart-failure / rollback — proxied at the provisioning control-flow layer (real SIGKILL chaos infra isn't live yet per CLAUDE.md's Process maturity table — "Chaos suite nightly — by Phase 2"), since the actual durability guarantee is proven at the lifecycle layer above and #2595's transaction already proves the store is all-or-nothing:

• TestApplyPlanLive_ImportFails_RunningPipeline_LeftStopped — import fails after StopAndWait succeeds → Start is never called (no mock expectation set, so gomock fails the test if it is) → error surfaced.
• TestApplyPlanLive_RestartFails_LeftStoppedWithNewConfig — import commits, Start fails → error surfaced, pipeline left stopped with the new (valid) config, matching the design doc's "Restart failure" mode.
• TestApplyPlanLive_StopAndWaitFails_NoMutation — StopAndWait itself fails → no Create/Update/Delete/Start call happens at all (this is also the generic mechanism the lifecycle-poc guard relies on).

Stale-hash / rollback / running-left-stopped — TestApplyPlanLive_StaleHash_RefusedNoMutation_RunningPipeline (AC-4), TestApplyPlanLive_RunningPipeline_StopsAppliesRestarts (AC-3 control-flow: asserts call order stop → import → start), TestApplyPlanLive_StoppedPipeline_NoLifecycleCalls / TestApplyPlanLive_Idempotent_NoOp (confirms parity with ApplyPlan and that no lifecycle calls happen when there's nothing live to disrupt).

Lifecycle-poc parity — TestServiceLifecycle_StopAndWait_Unsupported pins the refusal + its error code.

Standalone ApplyPlan still refuses running pipelines — unchanged, TestApplyPlan_RunningPipeline_Refused still passes.

-count=20 -race on both the pkg/lifecycle StopAndWait tests and the pkg/provisioning ApplyPlanLive tests: clean, no flakes. Full targeted suite (./pkg/lifecycle/... ./pkg/lifecycle-poc/... ./pkg/provisioning/... ./pkg/connector/...) green under -race. Full repo go test ./... green. golangci-lint run ./pkg/lifecycle/... ./pkg/lifecycle-poc/... ./pkg/provisioning/... ./pkg/connector/... ./pkg/conduit/...: 0 issues. go generate ./...: clean (only the expected pkg/provisioning/mock/provisioning.go diff, +StopAndWait).

Adversarial self-review findings

1. Does StopAndWait actually achieve quiescence+durability, and is there a window where importPipeline runs against a not-fully-stopped pipeline? Traced the full happens-before chain: Stop injects → SourceNode.Run's openMsgTracker.Wait() blocks until every in-flight message is acked end-to-end (pre-existing drain machinery) → deferred Source.Teardown() runs (still inside Run, before it returns) → Teardown calls persister.ConnectorStopped() → triggerFlush synchronously increments flushWg before Run() can return → node's nodesWg.Done() → WaitPipeline's tomb.Wait() unblocks only after this → WaitPersisted/flushWg.Wait() is guaranteed to observe the already-incremented counter. No window found where transactionalImport can run before the pipeline is both drained and its position writes are durable.
2. Doc-comment bug — caught a stray blank line that would have detached ApplyPlanLive's doc comment from the function in go doc. Fixed; verified with go doc.
3. Regression-test validity — didn't just trust the durability test; deliberately broke StopAndWait (removed the WaitPersisted call) and confirmed the test fails hard, then confirmed it passes again after restoring.
4. Persister.WaitPendingWrites has no timeout/ctx — inherited from the existing Persister.Wait()'s design (also no ctx, used unconditionally at process shutdown), not a new gap. A stuck store write for an unrelated pipeline could in principle block another pipeline's StopAndWait indefinitely, since the persister batches across all pipelines. Flagged, not fixed — would be scope creep for PR1 and is a pre-existing characteristic of the shared persister.
5. No per-pipeline provisioning lock — the design doc's Failure modes list "Concurrent applies / apply-during-manual-stop" as needing one; ApplyPlanLive doesn't take one. Between StopAndWait returning and transactionalImport/Start, a concurrent caller could independently Start the now-stopped pipeline, causing a config-drift (not data-loss — invariants 1-3 hold) condition. Not reachable in PR1 — no API/CLI/MCP surface calls ApplyPlanLive yet. Flagged in code comments and here; belongs in or before whichever PR (likely PR2) first exposes this to concurrent external callers.
6. Pre-existing isRunningStatus/StatusDegraded dead end — a Degraded pipeline has already removed itself from lifecycle.Service.runningPipelines by the time its status is visible (see runPipeline's cleanup goroutine), so isRunningStatus correctly reports it as "needs an explicit decision" but StopAndWait/Stop then fails with ErrPipelineNotRunning against it — a dead end. This is identical, pre-existing behavior in ApplyPlan (not introduced here); flagged in a code comment for an explicit decision rather than left as silent archaeology.

Test plan

• go build ./...
• go test ./pkg/lifecycle/... ./pkg/lifecycle-poc/... ./pkg/provisioning/... ./pkg/connector/... -race -count=1 — green
• go test ./pkg/lifecycle/... -run TestServiceLifecycle_StopAndWait -race -count=20 — green, no flakes
• go test ./pkg/provisioning/... -run TestApplyPlanLive -race -count=20 — green, no flakes
• go test ./... -count=1 (full repo) — green
• golangci-lint run on all touched packages — 0 issues (36 pre-existing issues elsewhere in the repo, verified unrelated via git stash -u)
• go generate ./... — clean, only the expected mock regen
• Manually verified the new durability test fails without the fix and passes with it

Roadmap

Phase 1/2 execution plan item: live-server deploy/apply (issue #2588), engine primitives half of the arc. PR2 will add the PlanPipeline/ApplyPipeline API RPCs, CLI/MCP live-server-preferred client logic, and the enforced operator-authorization gate for data-path applies to running pipelines (design doc §3-4).

🤖 Generated with Claude Code

https://claude.ai/code/session_01Tapg9dLWXKMZJoL65R24vd

---

Tier-1 review outcome (SOUND-WITH-CONCERNS) + follow-ups

An independent Tier-1 data-integrity review verified the core safety case in the code (not just the comments): every link of the StopAndWait happens-before chain checks out — WaitPipeline unblocks only after all node goroutines incl. source Teardown; openMsgTracker.Wait() blocks teardown until every in-flight record is acked end-to-end; the position Persist happens-before the tracker decrement (reverse-order status handlers); ConnectorStopped→triggerFlush→flushWg.Add(1) runs synchronously before Run returns; and WaitPersisted(flushWg) is the correct durability wait (the post-WaitPipeline call ordering eliminates the buffered-but-not-triggered window). No window found where a mutation races the drain (invariants 1/3 upheld). ApplyPlanLive failure handling (leave-stopped, no auto-restart, transactional), the poc guard, and the delayingDB durability test (fails without WaitPersisted) all CONFIRMED.

Concerns (documentation/process, addressed / tracked — not code):

1. AC-6 crash-recovery is deferred, not claimed — corrected the test comment: what's delivered is the txn atomicity (fix(provisioning): wrap ApplyPlan's importPipeline in a DB transaction #2595) + StopAndWait durable checkpoint + the leave-stopped control flow; a real SIGKILL test is Phase-2 chaos (per CLAUDE.md's maturity table).
2. Per-pipeline provisioning lock is a HARD GATE on PR2 — ApplyPlanLive has no external caller in PR1 (verified), so the concurrent-Start / !running TOCTOU is unreachable now; it must land in PR2 (the API/CLI/MCP surface) before external exposure.
3. Minor deferrable: the isRunningStatus/StatusDegraded dead-end (fails closed, safe, pre-existing shared with ApplyPlan).