fix(lifecycle-poc): publish the live run before releasing its workers

pkg/lifecycle-poc/service.go

@@ -84,6 +84,28 @@ type Service struct {
 	// docs/design-documents/20260706-forceful-stop-test-determinism.md.
 	terminalErrors *csync.Map[string, error]
 
+	// testWorkersReleased, if set, is called synchronously from runPipeline
+	// immediately after this run's worker goroutines are released (i.e.
+	// right after close(registered)) — the earliest instant a worker could
+	// possibly do observable work (read/process/ack a record). It exists
+	// solely so a test can hold that instant open deterministically, instead
+	// of racing it, to prove the #2833 regression: that runningPipelines
+	// already points at THIS run before any worker can be released, even on
+	// a recovery restart where a dead run's entry would otherwise still be
+	// in the map. Mirrors the same "wrap/hook a collaborator instead of
+	// sleeping" seam shape as statusRecorder.onUpdate in the test file,
+	// applied here because the window under test closes before the
+	// PipelineService is ever called, so hooking UpdateStatus cannot observe
+	// it.
+	//
+	// Nil in production — NewService never sets it — so the call site below
+	// is a single unlocked nil-check with no synchronization and no
+	// observable effect when unset: zero behavior change outside tests.
+	// Exported to the package's test files only by being unexported (they
+	// share this package) and set directly on a *Service under test; there
+	// is no production code path that can set or read it.
+	testWorkersReleased func(rp *runnablePipeline)
+
 	isGracefulShutdown atomic.Bool
 	metricsDisabled    bool
 }
@@ -1661,25 +1683,22 @@ func (s *Service) runPipeline(rp *runnablePipeline) error {
 		return err
 	})
 
-	// All N+1 goroutines (every worker plus the cleanup goroutine) are now
-	// registered on the tomb, so release the workers: none of them can any
-	// longer drive tomb.alive to 0 before the cleanup goroutine exists. This
-	// must happen BEFORE the UpdateStatus call below, which is potentially
-	// slow — the workers only need the registration barrier, not the status
-	// write (the cleanup goroutine is the one that waits for that, via
-	// startupDone).
-	close(registered)
-
 	// Publish this run as THE live run for its pipeline ID, here and not in
-	// Start (#2746).
+	// Start (#2746), and — critically — BEFORE close(registered) below
+	// releases the worker goroutines (#2833).
 	//
-	// 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 is a pointer
-	// comparison against it. A stale entry does not fail loudly; it makes all
-	// of them operate on the previous, already-dead run.
+	// Invariant 7: whenever a caller can observe the pipeline as running (and,
+	// as of #2833, whenever any worker can be observed to be doing work at
+	// all), 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 is a pointer comparison against it. A stale entry
+	// does not fail loudly; it makes all of them operate on the previous,
+	// already-dead run, which is a graceful-shutdown/Stop correctness bug
+	// (invariant 7): a Stop that lands in the stale window stops the DEAD
+	// run and returns success while the actually-running one keeps consuming
+	// and acking records, unsupervised.
 	//
 	// Start used to Set this AFTER runPipeline returned, i.e. after the
 	// UpdateStatus below had already announced StatusRunning. On a recovery
@@ -1688,7 +1707,18 @@ func (s *Service) runPipeline(rp *runnablePipeline) error {
 	// 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 going — leaving a
-	// pipeline nobody could stop and a persister that never quiesced.
+	// pipeline nobody could stop and a persister that never quiesced. #2746/
+	// #2812 fixed that window (Set before UpdateStatus) but left a second,
+	// narrower one: close(registered) below hands worker goroutines their
+	// release signal, and a worker released from <-registered can reach
+	// w.Do — reading, processing and acking records — before this goroutine
+	// gets back around to the Set call, if Set runs after it. On a recovery
+	// restart that is exactly the same stale-map window as #2746, just
+	// shrunk to two adjacent statements instead of spanning the UpdateStatus
+	// call: a Stop landing between close(registered) and Set still resolves
+	// the DEAD run (#2833). Publishing here, before close(registered), closes
+	// it: by the time any worker can possibly be released to do anything
+	// observable, the map already points at THIS run.
 	//
 	// This is the correct publish point, and it needs no rollback on error:
 	//   - every earlier return in this function (sink Open, worker Open)
@@ -1701,6 +1731,24 @@ func (s *Service) runPipeline(rp *runnablePipeline) error {
 	//     outside the map.
 	s.runningPipelines.Set(rp.pipeline.ID, rp)
 
+	// All N+1 goroutines (every worker plus the cleanup goroutine) are now
+	// registered on the tomb, so release the workers: none of them can any
+	// longer drive tomb.alive to 0 before the cleanup goroutine exists. This
+	// must happen BEFORE the UpdateStatus call below, which is potentially
+	// slow — the workers only need the registration barrier and the Set
+	// above, not the status write (the cleanup goroutine is the one that
+	// waits for that, via startupDone) — and it must happen AFTER the Set
+	// above, per the invariant-7 comment there (#2833): a worker released
+	// any earlier could act while the map still pointed at a dead run.
+	close(registered)
+
+	// testWorkersReleased, if set, lets a test observe/hold this exact
+	// instant — workers released, run published, status not yet announced —
+	// deterministically. See its doc.
+	if s.testWorkersReleased != nil {
+		s.testWorkersReleased(rp)
+	}
+
 	// It's now safe to make the potentially slow UpdateStatus call and then
 	// release the cleanup goroutine to make its own. close(startupDone)
 	// unconditionally, including on error, so the cleanup goroutine (already

pkg/lifecycle-poc/service_test.go

@@ -919,6 +919,165 @@ func TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus(t *test
 	})
 }
 
+// TestServiceLifecycle_Recovery_ForceStopDuringWorkerReleaseWindow is the
+// #2833 regression test.
+//
+// runPipeline releases this run's worker goroutines (close(registered)) and
+// publishes it as the live entry in runningPipelines (Set) as two adjacent
+// statements with nothing between them. Before this fix, close(registered)
+// ran FIRST: a worker released at that instant is already free to do
+// observable work (read, process, ack), while runningPipelines[id] still
+// points at the PREVIOUS run. On a recovery restart that previous entry is
+// deliberately left in place until the swap (see recoverPipeline), so the
+// window is real and reachable — this is #2746/#2806's failure mode again,
+// just shrunk to two adjacent statements with no I/O between them, which is
+// why #2812 (which fixed the *2746* window, one call further out — ordering
+// Set before UpdateStatus) did not close this one: it never touched the
+// close(registered)-vs-Set order.
+//
+// A Stop landing in that window resolves the DEAD run: Kill lands on an
+// already-dead tomb (a documented no-op — tomb.v2 keeps only the first
+// death reason, see gopkg.in/tomb.v2's kill()), and the actually-running
+// pipeline is left completely unsupervised. Invariant 7 (shutdown is
+// graceful by default), silently.
+//
+// This does not try to catch the window by racing it — a pre-fix repeat
+// sweep of the equivalent window in #2746 passed far more often than it
+// failed, so a green run there proved nothing. It HOLDS runPipeline paused
+// at the exact instant under test via testWorkersReleased (see that
+// field's doc): the same "wrap/hook a collaborator instead of sleeping"
+// seam shape as statusRecorder.onUpdate below, applied one call earlier,
+// because THIS window closes before UpdateStatus — and therefore
+// PipelineService — is ever invoked, so hooking UpdateStatus (as the
+// sibling #2746 test above does) cannot observe it.
+//
+// force=true (not a graceful Stop) is deliberate: stopRunnablePipeline's
+// force branch is a single unconditional rp.t.Kill call with no worker
+// interaction, so the assertion below reduces to a synchronous, racy-drain
+// -free tomb.Alive() check. See TestServiceLifecycle_PipelineForceStop's
+// doc for the same zero-records/force-stop rationale, reused here.
+func TestServiceLifecycle_Recovery_ForceStopDuringWorkerReleaseWindow(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)
+
+	ps := pipeline.NewService(logger, db)
+	pl, err := ps.Create(ctx, uuid.NewString(), pipeline.Config{Name: "test pipeline"}, pipeline.ProvisionTypeAPI)
+	is.NoErr(err)
+
+	transientErr := cerrors.New("lost connection to source")
+	// Zero records for the recovered run too: avoids any dependency on ack
+	// timing (see TestServiceLifecycle_PipelineForceStop's doc) — this test
+	// only needs the recovered run's worker to be released and then sit
+	// blocked on its mocked stream, exactly like that test's "Run blocks"
+	// case.
+	noRecords := generateRecords(0)
+
+	ctrl := gomock.NewController(t)
+	source, srcDispenser := sourceRecoversAfterTransientError(ctrl, persister, noRecords, transientErr)
+	destination, destDispenser := destinationRecovers(ctrl, persister, noRecords)
+	dlq, dlqDispenser := dlqDispenserTimes(ctrl, persister, 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)
+
+	ls := NewService(
+		logger,
+		testErrRecoveryCfg(),
+		testConnectorService{
+			source.ID:      source,
+			destination.ID: destination,
+			testDLQID:      dlq,
+		},
+		testProcessorService{},
+		testConnectorPluginService{
+			source.Plugin:      srcDispenser,
+			destination.Plugin: destDispenser,
+			dlq.Plugin:         dlqDispenser,
+		},
+		ps,
+		false,
+	)
+
+	// inWindow closes once the RECOVERY restart's runPipeline has released
+	// its workers; release unblocks it once the assertions below have run.
+	inWindow := make(chan struct{})
+	release := make(chan struct{})
+	var runCount atomic.Int64
+	var liveRp *runnablePipeline
+	ls.testWorkersReleased = func(rp *runnablePipeline) {
+		// The 1st call is the initial run: it has no prior runningPipelines
+		// entry to race against, so #2833's window doesn't apply to it (the
+		// issue is explicit: "the initial-start path does not have this
+		// window"). Only intercept the 2nd call, the recovery restart.
+		if runCount.Add(1) != 2 {
+			return
+		}
+		liveRp = rp
+		close(inWindow)
+		<-release
+	}
+
+	is.NoErr(ls.Start(ctx, pl.ID))
+
+	// deadRp is the pre-recovery run, captured immediately (before the
+	// transient failure) so identity — not just aliveness — distinguishes
+	// "the dead run" from "the live run" below.
+	deadRp, ok := ls.runningPipelines.Get(pl.ID)
+	is.True(ok)
+
+	<-inWindow
+	is.True(liveRp != nil)
+	is.True(liveRp != deadRp) // sanity: genuinely a different run
+
+	// The regression: a Stop landing right now — runPipeline paused exactly
+	// after releasing the recovered run's workers — must reach the run that
+	// is actually running, not the one that already failed.
+	stopErr := ls.Stop(ctx, pl.ID, true)
+	is.NoErr(stopErr) // force-stop always returns nil, whichever run it resolved
+
+	liveWasKilled := !liveRp.t.Alive()
+
+	// Safety net, unconditional and run regardless of the assertion below:
+	// guarantee the live run's tomb dies so its worker — parked on a mocked
+	// stream Recv with nothing else to wake it (see the zero-records
+	// rationale above) — unblocks via ctx cancellation and cleanup
+	// completes deterministically. Without this, a pre-fix failure here
+	// would otherwise depend on stopAndWaitPersister's 10s timeout to
+	// notice the recovered pipeline never stopped. A no-op if the Stop call
+	// above already reached this tomb (Kill keeps only the first reason).
+	liveRp.t.Kill(cerrors.FatalError(pipeline.ErrForceStop))
+
+	close(release)
+
+	if !liveWasKilled {
+		t.Fatalf(
+			"Stop(force=true), called the instant runPipeline had released the recovered " +
+				"run's workers, did not kill the live run (tomb still alive) - it landed on " +
+				"the dead pre-recovery run instead, leaving the actually-running pipeline " +
+				"completely unsupervised (#2833)",
+		)
+	}
+
+	err = ls.WaitPipeline(pl.ID)
+	is.True(err != nil)
+	is.True(cerrors.IsFatalError(err))
+	is.True(cerrors.Is(err, pipeline.ErrForceStop))
+	is.Equal(pipeline.StatusDegraded, pl.GetStatus())
+
+	// The pre-recovery run's own terminal classification (the transient
+	// error that triggered recovery) must be untouched by our stray Kill
+	// call above - pin that explicitly rather than assuming a no-op.
+	is.True(cerrors.Is(deadRp.t.Err(), transientErr))
+}
+
 // TestServiceLifecycle_Recovery_MaxRetriesExhausted proves the bounded-retry
 // path (design-doc path: running → recovering → degraded). With a finite
 // MaxRetries and a source that fails on every run, the pipeline attempts exactly