From e48830afc3f8823d9ae6b717ae90ccfdea90b249 Mon Sep 17 00:00:00 2001 From: Devaris Date: Fri, 21 Aug 2026 07:20:57 -0700 Subject: [PATCH 1/2] fix(lifecycle): publish the live run before announcing StatusRunning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #2806. Between UpdateStatus(StatusRunning) and Start's runningPipelines.Set — which happened only after runPipeline returned — the map still pointed at the PREVIOUS run. On a recovery restart the old entry is deliberately left in place until that swap, so the window is real and reachable. Every public entry point resolves through that map: Stop, StopAll, WaitPipeline, StopAndWait (and so provisioning.ApplyPlanLive), plus StartWithBackoff's pointer guard. During the window WaitPipeline joined the dead tomb and returned the pre-recovery error, and Stop stopped the dead run and reported success while the recovered run kept going — connectors never torn down, persister never quiesced. A drain reported complete that never happened: invariant 7, silently. The fix publishes rp inside runPipeline, immediately before the status write. What it deliberately does NOT do is reorder the cleanup-goroutine registration to match pkg/lifecycle-poc's shape. Adversarial review of the plan found that doing so reintroduces the same bug class deterministically: in a recovery chain the cleanup goroutine runs the nested recoverPipeline -> StartWithBackoff -> Start synchronously, so a transient status-write failure in the NESTED run propagates back into the OUTER cleanup, which then deletes by key — erasing a newer run whose nodes are alive. Registering cleanup earlier would also let a fast-failing pipeline's terminal status be stomped by the still-pending Running write, since UpdateStatus is last-writer-wins with no versioning. Today's ordering is load-bearing and is kept. So instead: roll back the publication explicitly on the UpdateStatus error path, and make every delete a compare-and-delete (deleteRunningPipelineIfCurrent) so a superseded run can never erase a newer run's entry. csync.Map has no CAS primitive, so that helper is a Get-compare-Delete and its doc says plainly that it narrows the window rather than closing it — it is not atomic and must not be read as such. Tests hold the window open via a PipelineService wrapper that blocks inside the StatusRunning update, rather than racing it: a pre-fix -count=60 sweep of the equivalent v2 package passed while that bug was live, so repeat runs are not evidence. Verified against the exact pre-fix service.go from main: LiveEntryPublishedBeforeRunningStatus and StopDuringWindowTargetsLiveRun both FAIL without the change and pass with it. StopAll_Recovering, which binds an old run before the failure and legitimately expects the pre-recovery error, still passes — that behavior is by design and is untouched. Out of scope, filed separately as #2809: UpdateStatus mutates the shared in-memory pipeline status before the store write and never rolls it back, so a failed status write leaves the instance reporting Running and Start's own precondition rejects the retry. That is why the failed-status test asserts only map state and not that a later Start succeeds. --- pkg/lifecycle/service.go | 86 +++++- pkg/lifecycle/service_test.go | 503 ++++++++++++++++++++++++++++++++++ 2 files changed, 585 insertions(+), 4 deletions(-) diff --git a/pkg/lifecycle/service.go b/pkg/lifecycle/service.go index 86d683512..e8c0e2985 100644 --- a/pkg/lifecycle/service.go +++ b/pkg/lifecycle/service.go @@ -222,13 +222,15 @@ func (s *Service) Start( s.terminalErrors.Delete(pipelineID) s.logger.Trace(ctx).Str(log.PipelineIDField, pl.ID).Msg("running nodes") + // runPipeline publishes rp into runningPipelines itself, at the point the + // run actually goes live — see the Set call there for why that ordering + // is load-bearing (#2806) and why this function must not do it after the + // fact. if err := s.runPipeline(ctx, rp); err != nil { return cerrors.Errorf("failed to run pipeline %s: %w", pl.ID, err) } s.logger.Info(ctx).Str(log.PipelineIDField, pl.ID).Msg("pipeline started") - s.runningPipelines.Set(pl.ID, rp) - return nil } @@ -857,8 +859,49 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { }) } + // Publish this run as THE live run for its pipeline ID, here and not in + // Start (#2806, same invariant as pkg/lifecycle-poc's #2746 fix — see + // that package's runPipeline for the mirrored comment). + // + // Invariant: whenever a caller can observe the pipeline as running, + // runningPipelines[id] is the run that is actually running. Every public + // entry point resolves a pipeline through this map — Stop, StopAll, + // WaitPipeline, StopAndWait (and thus provisioning.ApplyPlanLive) — and + // StartWithBackoff's "am I still the live pipeline" guard (:270) is a + // pointer comparison against it. A stale entry does not fail loudly: it + // makes all of them silently operate on the previous, already-dead run. + // + // Start used to Set this AFTER runPipeline returned, i.e. after the + // UpdateStatus below had already announced StatusRunning. On a recovery + // restart the old entry is deliberately left in place until the swap + // (see the recovery arm below), so in that window the map still pointed + // at the FAILED run: WaitPipeline joined the dead tomb and returned the + // pre-recovery error for a pipeline that had just recovered, and Stop + // stopped the dead run while the recovered one kept running — connectors + // never torn down, persister never quiesced, a drain reported complete + // that never happened (invariant 7). + // + // Unlike v2, this package's cleanup goroutine (below) is registered + // AFTER this UpdateStatus call, deliberately — see its comment. That + // means, unlike v2, there is a real window here where this entry is + // published but nothing yet owns cleaning it up if UpdateStatus fails. + // So: roll back explicitly on that error path instead of relying on a + // cleanup goroutine that does not exist yet. + s.runningPipelines.Set(rp.pipeline.ID, rp) + err := s.pipelines.UpdateStatus(ctx, rp.pipeline.ID, pipeline.StatusRunning, "") if err != nil { + // Roll back the publication above: this run never went live, so it + // must not be reachable via Stop/WaitPipeline/the recovery pointer + // guard. Compare-and-delete (see deleteRunningPipelineIfCurrent), + // not a blind Delete(id): if a different run for this same pipeline + // ID were published under this key between the Set above and this + // point (e.g. this Start is itself the nested call inside another + // run's cleanup goroutine — recoverPipeline -> StartWithBackoff -> + // Start — and something else raced a further publish in the + // meantime), a blind Delete(id) would remove that OTHER run instead + // of just undoing this one's own publication. + s.deleteRunningPipelineIfCurrent(rp.pipeline.ID, rp) return err } @@ -926,8 +969,19 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { // delete leaves no window where neither is observable). s.terminalErrors.Set(rp.pipeline.ID, err) - // confirmed that all nodes stopped, we can now remove the pipeline from the running pipelines - s.runningPipelines.Delete(rp.pipeline.ID) + // confirmed that all nodes stopped, we can now remove the pipeline + // from the running pipelines — but only if the entry under this ID + // is still THIS run (#2806). This goroutine can itself be the one + // running synchronously inside an OLDER run's cleanup: recoverPipeline + // -> StartWithBackoff -> Start runs a nested runPipeline on the + // calling tomb, not a fresh goroutine. If that nested run's own + // UpdateStatus above fails, the error propagates back into the + // OUTER run's cleanup, which falls through to this same terminal + // block. A blind Delete(rp.pipeline.ID) there would delete the + // INNER run's freshly-published, still-alive entry — orphaning it, + // unreachable via Stop/WaitPipeline, exactly the bug class this + // fix closes. See deleteRunningPipelineIfCurrent. + s.deleteRunningPipelineIfCurrent(rp.pipeline.ID, rp) s.notify(rp.pipeline.ID, err) return err @@ -935,6 +989,30 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { return nil } +// deleteRunningPipelineIfCurrent removes id's entry from runningPipelines +// only if it still holds exactly rp — a compare-and-delete rather than a +// delete-by-key (#2806). This is what stops a stale owner (an older run's +// publish-rollback or its cleanup goroutine) from erasing a newer run's +// published entry, which is the bug class #2806 fixes: a superseded run +// falling through to an unconditional Delete(id) and taking a live run down +// with it. +// +// This is NOT atomic. csync.Map exposes Get/Set/Delete/Len/Keys/Values/ +// Clear/Copy and no compare-and-swap primitive, so this is a plain +// Get-then-compare-then-Delete with no lock held across the three steps. It +// narrows the window relative to a blind Delete(id) — the common failure +// mode this fixes requires the stale owner to still be current at Get time, +// which a blind Delete doesn't even check — but it does not close it: a +// concurrent Set(id, newer) can still land in the gap between this Get +// returning a match and the Delete call that follows it, in which case this +// still deletes that newer entry. Do not read this as airtight CAS; it is a +// narrower TOCTOU, not the absence of one. +func (s *Service) deleteRunningPipelineIfCurrent(id string, rp *runnablePipeline) { + if current, ok := s.runningPipelines.Get(id); ok && current == rp { + s.runningPipelines.Delete(id) + } +} + // recoverPipeline attempts to recover a pipeline that has stopped running. func (s *Service) recoverPipeline(ctx context.Context, rp *runnablePipeline) error { s.logger.Trace(ctx).Str(log.PipelineIDField, rp.pipeline.ID).Msg("recovering pipeline") diff --git a/pkg/lifecycle/service_test.go b/pkg/lifecycle/service_test.go index e826318f2..0109b1433 100644 --- a/pkg/lifecycle/service_test.go +++ b/pkg/lifecycle/service_test.go @@ -20,6 +20,8 @@ import ( "reflect" "strconv" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -1203,3 +1205,504 @@ func stopAndWaitPersister(t *testing.T, killAll context.CancelFunc, p *connector "ConnectorStopped (see #2746)", persisterDrainTimeout) } } + +// statusRecorder wraps a PipelineService and gives tests a seam into +// UpdateStatus calls, for the #2806 regression tests below (mirrors +// pkg/lifecycle-poc's #2746 test helper of the same name and shape — v1's +// runPipeline orders its cleanup-goroutine registration differently, which is +// exactly why v1 needed its own fix and its own tests, but the "hold the +// window open instead of racing it" technique transfers directly). +type statusRecorder struct { + PipelineService + + mu sync.Mutex + statuses []pipeline.Status + + // onUpdate, if set, is called from inside UpdateStatus for every call, + // with the status being written and its 1-based occurrence count for + // that status (so a hook can distinguish e.g. the initial StatusRunning + // from a post-recovery one). It is the seam that lets a test hold the + // lifecycle inside a "run is going live" window and inspect it + // deterministically, instead of trying to catch that window by racing + // it — see TestServiceLifecycle_StopAll_Recovering's sibling tests + // below, none of which sleep-and-hope. + // + // If onUpdate returns a non-nil error, that error is returned in place + // of calling the wrapped PipelineService — simulating a transient + // status-store failure without going anywhere near the real store. In + // particular it does NOT mutate the real pipeline.Instance's in-memory + // status as a side effect, unlike the genuine store write path (see + // #2809: pipeline.Service.UpdateStatus mutates the shared instance's + // status before attempting the store write and never rolls that back on + // error). Tests below intentionally do not depend on #2809's behavior — + // it is a separate, already-filed bug. + onUpdate func(status pipeline.Status, nth int) error +} + +func newStatusRecorder(inner PipelineService) *statusRecorder { + return &statusRecorder{PipelineService: inner} +} + +func (r *statusRecorder) UpdateStatus(ctx context.Context, id string, status pipeline.Status, errMsg string) error { + r.mu.Lock() + r.statuses = append(r.statuses, status) + nth := 0 + for _, s := range r.statuses { + if s == status { + nth++ + } + } + hook := r.onUpdate + r.mu.Unlock() + + if hook != nil { + if err := hook(status, nth); err != nil { + return err + } + } + return r.PipelineService.UpdateStatus(ctx, id, status, errMsg) +} + +// TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus is the +// #2806 regression test (v1's counterpart of #2746, fixed in +// pkg/lifecycle-poc by #2807). +// +// The invariant: whenever a caller can observe the pipeline as running, +// runningPipelines[id] is the run that is ACTUALLY running. Everything +// public resolves a pipeline through that map — Stop, StopAll, WaitPipeline, +// StopAndWait (and so provisioning.ApplyPlanLive) — and StartWithBackoff's +// "am I still the live pipeline" guard is a pointer comparison against it. +// +// Start used to publish the new run only AFTER runPipeline returned, i.e. +// after runPipeline had already announced StatusRunning. On a recovery +// restart the previous entry is deliberately left in place until that swap, +// so in the gap the map pointed at the FAILED run: WaitPipeline joined the +// dead tomb and returned the pre-recovery error for a pipeline that had just +// recovered, and Stop stopped the dead run while the recovered one kept +// running (invariant 7, silently). +// +// This does not try to catch that gap by racing it, which is what made the +// original bug intermittent (a pre-fix -count=60 sweep of this package +// passed while it was live). It HOLDS the lifecycle inside the gap: +// statusRecorder.onUpdate blocks in the middle of the post-recovery +// StatusRunning write — the exact instant an observer first learns the +// pipeline is running again — and the assertions run there. Pre-fix that is +// a deterministic failure, not a probabilistic one. +func TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus(t *testing.T) { + is := is.New(t) + ctx, killAll := context.WithCancel(context.Background()) + defer killAll() + logger := log.New(zerolog.Nop()) + db := &inmemory.DB{} + persister := connector.NewPersister(logger, db, time.Second, 3) + defer stopAndWaitPersister(t, killAll, persister) + wantErr := cerrors.New("lost connection to database") + + ps := pipeline.NewService(logger, db) + pl, err := ps.Create(ctx, uuid.NewString(), pipeline.Config{Name: "test pipeline"}, pipeline.ProvisionTypeAPI) + is.NoErr(err) + + ctrl := gomock.NewController(t) + wantRecords := generateRecords(0) + source, srcDispenser := asserterSource(ctrl, persister, wantRecords, nil, true, 2) + destination, destDispenser := asserterDestination(ctrl, persister, wantRecords, 2) + dlq, dlqDispenser := asserterDestination(ctrl, persister, nil, 2) + pl.DLQ.Plugin = dlq.Plugin + + pl, err = ps.AddConnector(ctx, pl.ID, source.ID) + is.NoErr(err) + pl, err = ps.AddConnector(ctx, pl.ID, destination.ID) + is.NoErr(err) + + // inWindow closes when the recovered run is mid-announcement; release + // unblocks it once the assertions below have run. + inWindow := make(chan struct{}) + release := make(chan struct{}) + rec := newStatusRecorder(ps) + rec.onUpdate = func(status pipeline.Status, nth int) error { + // The 2nd StatusRunning is the recovery restart (the 1st is the + // initial run). Fire once: a later run must not re-block. + if status == pipeline.StatusRunning && nth == 2 { + close(inWindow) + <-release + } + return nil + } + + ls := NewService( + logger, + testErrRecoveryCfg(), + testConnectorService{ + source.ID: source, + destination.ID: destination, + testDLQID: dlq, + }, + testProcessorService{}, + testConnectorPluginService{ + source.Plugin: srcDispenser, + destination.Plugin: destDispenser, + dlq.Plugin: dlqDispenser, + }, + rec, + ) + + is.NoErr(ls.Start(ctx, pl.ID)) + + // deadRp is the pre-recovery run. Capturing it lets the assertion below + // tell "the map points at the run that just failed" apart from "the map + // points at the new one" by identity, not just by aliveness. + deadRp, ok := ls.runningPipelines.Get(pl.ID) + is.True(ok) + + // wait for the pipeline to be consuming, then force a recoverable error. + time.Sleep(100 * time.Millisecond) + ls.StopAll(ctx, wantErr) + + <-inWindow + + // The assertion, taken while the lifecycle is frozen in the window: the + // entry a caller would resolve right now must be the live run, not the + // one that just failed. Reading runningPipelines directly (rather than + // going through Stop/WaitPipeline, both of which legitimately BLOCK on a + // live tomb and so cannot distinguish "correct" from "hung" here) is + // what keeps this deterministic. Pre-fix, ok is true but the entry is + // still deadRp, whose tomb died from wantErr. + rp, ok := ls.runningPipelines.Get(pl.ID) + is.True(ok) // no live entry at all while the pipeline reports Running + if rp == deadRp { + t.Fatalf("runningPipelines[%s] still points at the pre-recovery run while the recovered run's StatusRunning is being announced (#2806)", pl.ID) + } + is.True(rp.t != nil) + if !rp.t.Alive() { + t.Fatalf("runningPipelines[%s] points at a dead tomb while StatusRunning is being announced — Stop would tear down the wrong run (#2806)", pl.ID) + } + + close(release) + + // Let the recovered run actually finish starting, then tear it down — + // both dispensed plugin instances (rp1's and rp2's) still need their + // Stop/Teardown expectations satisfied, and an un-stopped pipeline would + // otherwise race the deferred persister drain (see stopAndWaitPersister). + is.NoErr(ls.Stop(ctx, pl.ID, false)) + is.NoErr(ls.WaitPipeline(pl.ID)) + is.Equal(pipeline.StatusUserStopped, pl.GetStatus()) +} + +// TestServiceLifecycle_RunPipeline_UpdateStatusRunningFails_RollsBackPublication +// is AC 2: a failed UpdateStatus(StatusRunning) must leave no entry in +// runningPipelines. +// +// This is deliberately narrow: it asserts ONLY the map state, not "a later +// Start succeeds" — pipeline.Service.UpdateStatus mutates the shared +// *Instance's in-memory status before attempting the store write and never +// rolls that back on error (#2809), so a retried Start would be rejected by +// its OWN precondition regardless of what this fix does to the map. That is +// a real, separate bug; asserting around it here would make this test depend +// on #2809 being fixed too. +// +// Exercises runPipeline directly rather than through Start, since the +// behavior under test — publish, then roll back on UpdateStatus failure — is +// entirely inside runPipeline, and calling it directly with a node-less +// runnablePipeline avoids needing any connector mocks (there is nothing to +// tear down: no nodes were ever spawned, and this must hold regardless). +func TestServiceLifecycle_RunPipeline_UpdateStatusRunningFails_RollsBackPublication(t *testing.T) { + is := is.New(t) + logger := log.New(zerolog.Nop()) + injectedErr := cerrors.New("status store: write timeout") + + rec := newStatusRecorder(testPipelineService{}) + rec.onUpdate = func(status pipeline.Status, _ int) error { + if status == pipeline.StatusRunning { + return injectedErr + } + return nil + } + + ls := NewService( + logger, + testErrRecoveryCfg(), + testConnectorService{}, + testProcessorService{}, + testConnectorPluginService{}, + rec, + ) + + rp := &runnablePipeline{ + pipeline: &pipeline.Instance{ + ID: uuid.NewString(), + Config: pipeline.Config{Name: "test-pipeline"}, + }, + backoff: testErrRecoveryCfg().toBackoff(), + recoveryAttempts: &atomic.Int64{}, + } + + err := ls.runPipeline(context.Background(), rp) + is.True(cerrors.Is(err, injectedErr)) + + _, ok := ls.runningPipelines.Get(rp.pipeline.ID) + is.True(!ok) // a failed UpdateStatus must leave no entry in runningPipelines (#2806) +} + +// TestServiceLifecycle_Recovery_NestedStartFailureDoesNotCorruptRunningPipelines +// is AC 2b, "the most important new test": an older run's cleanup cannot +// delete a newer run's published entry. +// +// recoverPipeline -> StartWithBackoff -> Start runs synchronously on the +// FAILED run's own cleanup goroutine — it is not a fresh goroutine. So when +// the recovered run (rp2) publishes itself and then its own +// UpdateStatus(StatusRunning) fails, the resulting error unwinds back into +// the ORIGINAL run's (rp1's) cleanup, which falls through to the same +// terminal block that removes a pipeline from runningPipelines. Naively +// deleting rp1's ID there — after rp2 already rolled itself back, or worse, +// before it does — is exactly the bug class #2806 fixes, just one recovery +// attempt deeper: an unconditional Delete(id) does not know or care that a +// DIFFERENT run's entry might now be under that key. +// +// Like the AC1 test above, this holds the window open on rp2's failing write +// rather than racing it, so the mid-write assertion (rp2 is already +// published, and alive, even though its own write is about to fail) fails +// deterministically pre-fix for the same reason AC1's does: pre-fix, Set +// happens only after runPipeline returns successfully, so at this exact +// moment the map still points at the dead rp1. +func TestServiceLifecycle_Recovery_NestedStartFailureDoesNotCorruptRunningPipelines(t *testing.T) { + is := is.New(t) + ctx, killAll := context.WithCancel(context.Background()) + defer killAll() + logger := log.New(zerolog.Nop()) + db := &inmemory.DB{} + persister := connector.NewPersister(logger, db, time.Second, 3) + defer stopAndWaitPersister(t, killAll, persister) + transientErr := cerrors.New("lost connection to source") + injectedStoreErr := cerrors.New("status store: write timeout") + + ps := pipeline.NewService(logger, db) + pl, err := ps.Create(ctx, uuid.NewString(), pipeline.Config{Name: "test pipeline"}, pipeline.ProvisionTypeAPI) + is.NoErr(err) + + ctrl := gomock.NewController(t) + noRecords := generateRecords(0) + // Both dispensed source instances fail on their own (no external Stop + // call needed): rp1's own failure is what triggers recovery in the + // first place; rp2's failure just lets its (now orphaned, since its + // UpdateStatus is about to fail and no cleanup goroutine will ever be + // registered for it) nodes wind down on their own instead of leaking + // for the rest of the test process. + source, srcDispenser := asserterSource(ctrl, persister, noRecords, transientErr, false, 2) + destination, destDispenser := asserterDestination(ctrl, persister, noRecords, 2) + dlq, dlqDispenser := asserterDestination(ctrl, persister, nil, 2) + pl.DLQ.Plugin = dlq.Plugin + + pl, err = ps.AddConnector(ctx, pl.ID, source.ID) + is.NoErr(err) + pl, err = ps.AddConnector(ctx, pl.ID, destination.ID) + is.NoErr(err) + + inWindow := make(chan struct{}) + release := make(chan struct{}) + rec := newStatusRecorder(ps) + rec.onUpdate = func(status pipeline.Status, nth int) error { + if status == pipeline.StatusRunning && nth == 2 { + close(inWindow) + <-release + return injectedStoreErr + } + return nil + } + + done := make(chan struct{}) + ls := NewService( + logger, + testErrRecoveryCfg(), + testConnectorService{ + source.ID: source, + destination.ID: destination, + testDLQID: dlq, + }, + testProcessorService{}, + testConnectorPluginService{ + source.Plugin: srcDispenser, + destination.Plugin: destDispenser, + dlq.Plugin: dlqDispenser, + }, + rec, + ) + ls.OnFailure(func(FailureEvent) { close(done) }) + + is.NoErr(ls.Start(ctx, pl.ID)) + + <-inWindow + // Mid-write for the RECOVERED run's (rp2's) announcement, moments + // before that write fails: the entry must already be rp2, alive, even + // though rp2 itself is about to be rolled back. + rp2, ok := ls.runningPipelines.Get(pl.ID) + is.True(ok) // no live entry at all while rp2's StatusRunning is in flight + is.True(rp2.t != nil) + if !rp2.t.Alive() { + t.Fatalf("runningPipelines[%s] points at a dead tomb while the recovered run's StatusRunning is being announced (#2806)", pl.ID) + } + close(release) + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("recovery chain did not complete within 5s") + } + + // Final state: rp2 rolled itself back (its own UpdateStatus failed), and + // rp1's cleanup — running on the SAME goroutine that just unwound from + // rp2's failure — must not have resurrected or corrupted the entry on + // its way through its own terminal block. Never left pointing at the + // dead rp1. + _, ok = ls.runningPipelines.Get(pl.ID) + is.True(!ok) + + // rp2 is orphaned (its own rollback means nothing will ever call + // Stop/WaitPipeline on it), but its nodes are still winding down on + // their own in the background, since its source fails on its own just + // like rp1's did. Wait for that tomb to fully finish before this test + // returns and the deferred persister drain runs — otherwise rp2's + // still-in-flight connector Open() calls race Persister.Wait(), the + // same shape of hazard #2746 documented for a live pipeline whose Stop + // was never called (see stopAndWaitPersister). + tombDone := make(chan struct{}) + go func() { + defer close(tombDone) + _ = rp2.t.Wait() + }() + select { + case <-tombDone: + case <-time.After(5 * time.Second): + t.Fatal("orphaned recovered run's nodes did not finish tearing down within 5s") + } +} + +// TestServiceLifecycle_StartWithBackoff_SupersededRunDoesNotRestart is AC 3: +// the recovery pointer guard in StartWithBackoff must still detect that it +// has been superseded and decline to restart. +// +// This is existing behavior (unchanged by #2806) that the fix must not +// regress: StartWithBackoff compares the runnablePipeline it was given +// against whatever is currently published under the same ID, and returns +// nil without restarting if they differ. +func TestServiceLifecycle_StartWithBackoff_SupersededRunDoesNotRestart(t *testing.T) { + is := is.New(t) + logger := log.New(zerolog.Nop()) + + ls := NewService( + logger, + testErrRecoveryCfg(), + testConnectorService{}, + testProcessorService{}, + testConnectorPluginService{}, + testPipelineService{}, + ) + + pipelineID := uuid.NewString() + staleRp := &runnablePipeline{ + pipeline: &pipeline.Instance{ID: pipelineID, Config: pipeline.Config{Name: "test"}}, + backoff: testErrRecoveryCfg().toBackoff(), + recoveryAttempts: &atomic.Int64{}, + } + freshRp := &runnablePipeline{ + pipeline: &pipeline.Instance{ID: pipelineID, Config: pipeline.Config{Name: "test"}}, + recoveryAttempts: &atomic.Int64{}, + } + + // freshRp supersedes staleRp in the map before staleRp's backoff elapses + // — e.g. the pipeline was stopped and restarted by a user while a + // recovery attempt for it was still waiting out its backoff. + ls.runningPipelines.Set(pipelineID, freshRp) + + err := ls.StartWithBackoff(context.Background(), staleRp) + is.NoErr(err) // the pointer guard must return nil, not attempt to restart a superseded run + + got, ok := ls.runningPipelines.Get(pipelineID) + is.True(ok) + is.True(got == freshRp) // the superseded run's early return must not touch the entry that superseded it +} + +// TestServiceLifecycle_Recovery_StopDuringWindowTargetsLiveRun is AC 4: Stop +// called during the announcement window must act on the LIVE (recovered) +// run, not the dead pre-recovery one. +// +// This is the concrete failure mode #2806 describes: Stop resolving to a +// dead tomb during the window either errors against nodes that already +// finished, or silently no-ops, while the actually-live run keeps going +// forever — connectors never torn down, a drain reported complete (or +// simply never attempted) when it never happened (invariant 7). The +// assertion that matters is the outcome: the live run must actually stop, +// promptly, and reach StatusUserStopped. Stop's own immediate return value +// during the window is logged but not asserted on, since it can legitimately +// differ pre-fix; that its target never stops is the real bug. +func TestServiceLifecycle_Recovery_StopDuringWindowTargetsLiveRun(t *testing.T) { + is := is.New(t) + ctx, killAll := context.WithCancel(context.Background()) + defer killAll() + logger := log.New(zerolog.Nop()) + db := &inmemory.DB{} + persister := connector.NewPersister(logger, db, time.Second, 3) + defer stopAndWaitPersister(t, killAll, persister) + wantErr := cerrors.New("lost connection to database") + + ps := pipeline.NewService(logger, db) + pl, err := ps.Create(ctx, uuid.NewString(), pipeline.Config{Name: "test pipeline"}, pipeline.ProvisionTypeAPI) + is.NoErr(err) + + ctrl := gomock.NewController(t) + wantRecords := generateRecords(0) + source, srcDispenser := asserterSource(ctrl, persister, wantRecords, nil, true, 2) + destination, destDispenser := asserterDestination(ctrl, persister, wantRecords, 2) + dlq, dlqDispenser := asserterDestination(ctrl, persister, nil, 2) + pl.DLQ.Plugin = dlq.Plugin + + pl, err = ps.AddConnector(ctx, pl.ID, source.ID) + is.NoErr(err) + pl, err = ps.AddConnector(ctx, pl.ID, destination.ID) + is.NoErr(err) + + inWindow := make(chan struct{}) + release := make(chan struct{}) + rec := newStatusRecorder(ps) + rec.onUpdate = func(status pipeline.Status, nth int) error { + if status == pipeline.StatusRunning && nth == 2 { + close(inWindow) + <-release + } + return nil + } + + ls := NewService( + logger, + testErrRecoveryCfg(), + testConnectorService{ + source.ID: source, + destination.ID: destination, + testDLQID: dlq, + }, + testProcessorService{}, + testConnectorPluginService{ + source.Plugin: srcDispenser, + destination.Plugin: destDispenser, + dlq.Plugin: dlqDispenser, + }, + rec, + ) + + is.NoErr(ls.Start(ctx, pl.ID)) + time.Sleep(100 * time.Millisecond) + ls.StopAll(ctx, wantErr) + + <-inWindow + // Stop while the recovered run's own StatusRunning announcement is + // still in flight — the exact window #2806 describes. + stopErr := ls.Stop(ctx, pl.ID, false) + close(release) + t.Logf("Stop() returned during the announcement window: %v", stopErr) + + c := make(cchan.Chan[error]) + go func() { c <- ls.WaitPipeline(pl.ID) }() + waitErr, _, ctxErr := c.RecvTimeout(ctx, 5*time.Second) + is.NoErr(ctxErr) // the LIVE run must actually stop, not run forever because Stop targeted a dead tomb + is.NoErr(waitErr) + is.Equal(pipeline.StatusUserStopped, pl.GetStatus()) +} From d61fa303b5c1fa6a6e7bd27d6ff7fdb16cc70eb6 Mon Sep 17 00:00:00 2001 From: Devaris Date: Fri, 21 Aug 2026 07:42:08 -0700 Subject: [PATCH 2/2] fix(lifecycle): serialize map writers, and correct two claims the review disproved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of e48830a returned "merge with changes". Three of them were merge gates. F1 — the compare-and-delete was closeable and I left it leaky. The helper's doc admitted a residual TOCTOU as if it were unavoidable; it is not. csync.Map has no CAS, but Service has only one other writer, so publishMu now serializes publication against compare-and-delete and the read-compare-delete becomes atomic with respect to it. The review demonstrated the gap rather than arguing it: 4 erasures of a live entry in 200,000 races, dropping to 0 with the lock. And it is reachable without any recovery chain — an operator Stop leaves the status UserStopped, which admits a concurrent Start whose Set can land inside a departing cleanup's window. F2 — my failure-mode analysis was wrong about how the bug presents. I wrote that Stop "reported success" against the dead run. It does not: SourceNode.Stop returns "source node is not running" deterministically (a dead run's source is already stopped), so StopAndWait surfaces an error to ApplyPlanLive and the drain is reported FAILED, not complete. The invariant-7 violation is real but arrives via StopAll, which swallows that error into a log warning, after which ls.Wait resolves instantly off the dead tomb and shutdown quiesces the persister and closes the DB while the recovered run is still live. Corrected here, in the commit message of e48830a's description, and in issue #2806. F3 — the invariant comment overclaimed. It asserted that the map always tracks the live run; this change only establishes that at the publication window. The recovery-backoff window holds a dead run for 1s..10m while Stop admits StatusRecovering, which is a far larger hole and entirely pre-existing. The comment is now scoped, and that window is filed as #2810 rather than being implied covered. F5 — the rollback test was one-sided. Asserting only that no entry remains would pass equally against code that never published at all, so it would not have noticed #2806 being reintroduced wholesale. It now asserts, from inside the status hook, that the entry IS published before the status write, and gone after it fails. Verified: removing the publication makes it fail. Also filed, not fixed here: #2811, pkg/lifecycle-poc still deletes by key in its recovery-failed arm, which is the delete-side half of this same bug in v2. F6 (StopAll holds the map's read lock across stopGraceful) and the named test coverage gaps are follow-ups, not merge gates. --- pkg/lifecycle/service.go | 71 ++++++++++++++++++++++++++--------- pkg/lifecycle/service_test.go | 27 ++++++++++++- 2 files changed, 79 insertions(+), 19 deletions(-) diff --git a/pkg/lifecycle/service.go b/pkg/lifecycle/service.go index e8c0e2985..1f686ddba 100644 --- a/pkg/lifecycle/service.go +++ b/pkg/lifecycle/service.go @@ -82,6 +82,24 @@ type Service struct { handlers []FailureHandler runningPipelines *csync.Map[string, *runnablePipeline] + // publishMu serializes WRITERS to runningPipelines — the publication in + // runPipeline and every compare-and-delete — so the read-compare-delete + // in deleteRunningPipelineIfCurrent is atomic with respect to a + // concurrent publication. csync.Map has no compare-and-swap primitive, so + // without this a stale owner can still erase a newer run's entry by + // landing its Delete after another goroutine's Set: measured at 4 in + // 200,000 races on the unserialized version, which is rare but is exactly + // the bug class #2806 exists to close, so "rare" is not good enough. + // + // Readers (Get/All) deliberately do NOT take this: csync.Map has its own + // RWMutex for memory safety, and a reader that observes a slightly stale + // pointer is the pre-existing, acceptable case. This lock exists only to + // make write-write ordering deterministic. + // + // It is never held across I/O or a node operation, so it cannot deadlock + // against the stop path. + publishMu sync.Mutex + // terminalErrors holds the terminal error of a pipeline after it has stopped // and been removed from runningPipelines, so WaitPipeline can still report it // to a caller that races the pipeline's own cleanup. Written before the @@ -863,8 +881,17 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { // Start (#2806, same invariant as pkg/lifecycle-poc's #2746 fix — see // that package's runPipeline for the mirrored comment). // - // Invariant: whenever a caller can observe the pipeline as running, - // runningPipelines[id] is the run that is actually running. Every public + // Invariant established here: at the publication window — from the moment + // StatusRunning is observable — runningPipelines[id] is the run that is + // actually running. + // + // Deliberately scoped. It is NOT a general claim that the map always + // tracks the live run: during StartWithBackoff's sleep the map holds the + // dead pre-recovery run on purpose, for MinDelay..MaxDelay (1s..10m), and + // Stop admits StatusRecovering (:299). That window is orders of magnitude + // larger than this one and is a separate, pre-existing bug — see the + // issue filed alongside this change. Do not read this comment as saying + // that one is covered. Every public // entry point resolves a pipeline through this map — Stop, StopAll, // WaitPipeline, StopAndWait (and thus provisioning.ApplyPlanLive) — and // StartWithBackoff's "am I still the live pipeline" guard (:270) is a @@ -875,11 +902,18 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { // UpdateStatus below had already announced StatusRunning. On a recovery // restart the old entry is deliberately left in place until the swap // (see the recovery arm below), so in that window the map still pointed - // at the FAILED run: WaitPipeline joined the dead tomb and returned the - // pre-recovery error for a pipeline that had just recovered, and Stop - // stopped the dead run while the recovered one kept running — connectors - // never torn down, persister never quiesced, a drain reported complete - // that never happened (invariant 7). + // at the FAILED run. WaitPipeline joined the dead tomb and returned the + // pre-recovery error for a pipeline that had just recovered. + // + // Stop, precisely: it resolves the dead run and returns an error from + // SourceNode.Stop ("source node is not running", stream/source.go:193-200, + // since a dead run's source is already stopped) — it does NOT silently + // report success, and StopAndWait therefore surfaces that error to + // provisioning.ApplyPlanLive rather than proceeding. The invariant-7 + // violation arrives through StopAll instead: it swallows that error into a + // log warning (:366-372), runtime then calls ls.Wait(exitTimeout), which + // resolves instantly off the dead tomb, and shutdown proceeds to quiesce + // the persister and close the DB while the recovered run is still live. // // Unlike v2, this package's cleanup goroutine (below) is registered // AFTER this UpdateStatus call, deliberately — see its comment. That @@ -887,7 +921,9 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { // published but nothing yet owns cleaning it up if UpdateStatus fails. // So: roll back explicitly on that error path instead of relying on a // cleanup goroutine that does not exist yet. + s.publishMu.Lock() s.runningPipelines.Set(rp.pipeline.ID, rp) + s.publishMu.Unlock() err := s.pipelines.UpdateStatus(ctx, rp.pipeline.ID, pipeline.StatusRunning, "") if err != nil { @@ -997,17 +1033,18 @@ func (s *Service) runPipeline(ctx context.Context, rp *runnablePipeline) error { // falling through to an unconditional Delete(id) and taking a live run down // with it. // -// This is NOT atomic. csync.Map exposes Get/Set/Delete/Len/Keys/Values/ -// Clear/Copy and no compare-and-swap primitive, so this is a plain -// Get-then-compare-then-Delete with no lock held across the three steps. It -// narrows the window relative to a blind Delete(id) — the common failure -// mode this fixes requires the stale owner to still be current at Get time, -// which a blind Delete doesn't even check — but it does not close it: a -// concurrent Set(id, newer) can still land in the gap between this Get -// returning a match and the Delete call that follows it, in which case this -// still deletes that newer entry. Do not read this as airtight CAS; it is a -// narrower TOCTOU, not the absence of one. +// csync.Map exposes no compare-and-swap primitive, so the read-compare-delete +// is made atomic the only way available: publishMu serializes it against the +// publication in runPipeline, which is the only other writer. Without that +// lock this is a genuine TOCTOU — a concurrent Set(id, newer) landing between +// the Get and the Delete makes a stale owner erase a live run — measured at 4 +// occurrences in 200,000 races during review, and it is reachable without any +// recovery chain: an operator Stop leaves the status UserStopped, which admits +// a concurrent Start, whose Set can land inside a departing cleanup's window. func (s *Service) deleteRunningPipelineIfCurrent(id string, rp *runnablePipeline) { + s.publishMu.Lock() + defer s.publishMu.Unlock() + if current, ok := s.runningPipelines.Get(id); ok && current == rp { s.runningPipelines.Delete(id) } diff --git a/pkg/lifecycle/service_test.go b/pkg/lifecycle/service_test.go index 0109b1433..013f515fc 100644 --- a/pkg/lifecycle/service_test.go +++ b/pkg/lifecycle/service_test.go @@ -1410,6 +1410,14 @@ func TestServiceLifecycle_RunPipeline_UpdateStatusRunningFails_RollsBackPublicat logger := log.New(zerolog.Nop()) injectedErr := cerrors.New("status store: write timeout") + // Captured inside the hook, which fires BEFORE the wrapped UpdateStatus. + // Asserting only the absence afterwards would pass just as well against + // code that never published at all — i.e. it would keep passing if + // someone deleted the Set and reintroduced #2806 wholesale. Two-sided: + // published before the status write, gone after the failure. + var publishedDuringStatusWrite bool + var rp *runnablePipeline + rec := newStatusRecorder(testPipelineService{}) rec.onUpdate = func(status pipeline.Status, _ int) error { if status == pipeline.StatusRunning { @@ -1427,7 +1435,7 @@ func TestServiceLifecycle_RunPipeline_UpdateStatusRunningFails_RollsBackPublicat rec, ) - rp := &runnablePipeline{ + rp = &runnablePipeline{ pipeline: &pipeline.Instance{ ID: uuid.NewString(), Config: pipeline.Config{Name: "test-pipeline"}, @@ -1436,11 +1444,26 @@ func TestServiceLifecycle_RunPipeline_UpdateStatusRunningFails_RollsBackPublicat recoveryAttempts: &atomic.Int64{}, } + // Re-point the hook now that rp exists, so it can observe the map at the + // instant the status write is attempted. + rec.onUpdate = func(status pipeline.Status, _ int) error { + if status == pipeline.StatusRunning { + got, ok := ls.runningPipelines.Get(rp.pipeline.ID) + publishedDuringStatusWrite = ok && got == rp + return injectedErr + } + return nil + } + err := ls.runPipeline(context.Background(), rp) is.True(cerrors.Is(err, injectedErr)) + // The publication must have happened BEFORE the status write ... + is.True(publishedDuringStatusWrite) + + // ... and must have been rolled back after it failed (#2806). _, ok := ls.runningPipelines.Get(rp.pipeline.ID) - is.True(!ok) // a failed UpdateStatus must leave no entry in runningPipelines (#2806) + is.True(!ok) } // TestServiceLifecycle_Recovery_NestedStartFailureDoesNotCorruptRunningPipelines