diff --git a/pkg/lifecycle/service.go b/pkg/lifecycle/service.go index 86d683512..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 @@ -222,13 +240,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 +877,67 @@ 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 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 + // 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. + // + // 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 + // 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.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 { + // 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 +1005,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 +1025,31 @@ 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. +// +// 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) + } +} + // 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..013f515fc 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,527 @@ 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") + + // 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 { + 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{}, + } + + // 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) +} + +// 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()) +}