From 3958e00caace6824783fd9ad33f4e7e160c8566e Mon Sep 17 00:00:00 2001 From: Devaris Date: Mon, 24 Aug 2026 17:01:12 -0700 Subject: [PATCH] test(lifecycle-poc): wait for the live status on the recovery path, don't instant-read it TestServiceLifecycle_Recovery_TransientErrorRecovers failed on a DOCS-ONLY push to main (run 32785888309, sha 7486372; the next docs commit c073543 passed). The failure was: service_test.go:778: Running != Recovering panic.go:615: persister did not drain within 10s (see #2746) controller.go:97: missing call(s) to *mock.SourcePlugin.Teardown controller.go:97: missing call(s) to *mock.DestinationPlugin.Teardown x2 This is a test-synchronization defect, not a recovery bug. Evidence: at the failing assertion the recorded status sequence is exactly [Running Recovering Running] - ONE recovery, not two - while the live instance still reads Recovering. Both dispensers are .Times(2) and neither reported an over-call, and the three missing Teardowns are precisely the recovered run's own plugins, i.e. run 2 was alive and healthy when the assertion aborted the test. Everything after line 778 (the persister drain panic, the "run didn't finish" lines, the missing Teardowns) is fallout from is.Equal calling FailNow while the pipeline was still running. Root cause: neither watermark the test waits on is ordered after the status write it then asserts. - statusRecorder.UpdateStatus appends the status and runs its hook BEFORE delegating to pipeline.Service, which is what calls Instance.SetStatus. So waitForRecovered's "Running after a Recovering" fires strictly before pl.GetStatus() returns Running, with a whole UpdateStatus call (Get, two metrics updates, store write) still to go. - runPipeline closes `registered` - releasing every worker goroutine - and only then calls UpdateStatus(StatusRunning). The recovered run can therefore read, write and ack its entire record set while that status write is still in flight, so waitForRecordsAcked returns immediately and adds no delay either. Unlike the initial start (ls.Start returns only after runPipeline's status write completes, which is why the same instant assertion is sound there and in pkg/lifecycle), a recovery restart runs on the tomb's cleanup goroutine: nothing in the test goroutine is synchronized with it. The assertion was passing on the ~microsecond margin between the recorder entry and SetStatus, which a 4-core CI runner under `-race -shuffle=on` with docker sidecars can lose to a single goroutine preemption. Fix (test-only, no production change): waitForRecovered now also polls the live pl.GetStatus() until it reports Running, so it means what its name says. A genuine second recovery is still caught - the helper does not latch past it, and the AC-7 exact-sequence assertion at the end of each test would fail on [Running Recovering Running Recovering ...]. Four call sites carried the same unsound pattern, all fixed: - TestServiceLifecycle_Recovery_TransientErrorRecovers - TestServiceLifecycle_NSource_TransientErrorOneSource_Recovers - TestServiceLifecycle_NxM_TransientErrorOneSource_RecoversAllSourcesAndDestinations - TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus The last one has no waitForRecovered call - it freezes the window with the onUpdate hook - and is the MOST exposed of the four, because close(release) only lets the status write proceed while the records it then waits on were already acked during the freeze. It gets waitForStatus instead. It was added by 34bf97d ("...and one async assertion behind three flaky tests"), which shipped a new assertion of the same class it was fixing; waitForRecovered itself dates to a61d4bc (#2718). This is polling a watermark, not a widened timeout, a retry or a sleep: the deadline (5s, unchanged from the existing helpers) only bounds the failure message. Perturbation proof. The flake could NOT be reproduced naturally on an M-series 16-core box: 3840 runs of the race-enabled binary (24x40 at -cpu=1,4 and 64x30 at GOMAXPROCS=1) were all green, so the CI signature was reproduced by injecting the scheduler preemption directly - a sleep in statusRecorder between recording the status and delegating to the wrapped service. With 200ms injected, across all four tests at -race -count=10: before fix: 40/40 FAIL, byte-identical to the CI output after fix: 40/40 PASS At 20ms the same 80-run matrix went 46/80 FAIL -> 0/80 PASS. Package green at `go test ./pkg/lifecycle-poc/... -race -count=3 -shuffle=on`; golangci-lint clean. The injection is not committed; it is reproducible by setting rec.onUpdate to sleep on the nth==2 StatusRunning, the seam that already exists for TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus. Tier 3 (test-only). No existing issue tracks this test; #2534 is the umbrella flaky-suite issue. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016xE861dwb3MLgqEdWRECLY --- pkg/lifecycle-poc/service_nxm_test.go | 2 +- pkg/lifecycle-poc/service_test.go | 65 +++++++++++++++++++++------ 2 files changed, 52 insertions(+), 15 deletions(-) diff --git a/pkg/lifecycle-poc/service_nxm_test.go b/pkg/lifecycle-poc/service_nxm_test.go index b547a7ccb..dae851717 100644 --- a/pkg/lifecycle-poc/service_nxm_test.go +++ b/pkg/lifecycle-poc/service_nxm_test.go @@ -514,7 +514,7 @@ func TestServiceLifecycle_NxM_TransientErrorOneSource_RecoversAllSourcesAndDesti is.NoErr(err) // Must have passed through Recovering and be Running again. - waitForRecovered(t, rec) + waitForRecovered(t, rec, pl) // Post-recovery records flow cleanly through the REBUILT shared // destinations - proof the poison flag (which never clears in place) is diff --git a/pkg/lifecycle-poc/service_test.go b/pkg/lifecycle-poc/service_test.go index 7606f9fad..ac474a781 100644 --- a/pkg/lifecycle-poc/service_test.go +++ b/pkg/lifecycle-poc/service_test.go @@ -770,7 +770,7 @@ func TestServiceLifecycle_Recovery_TransientErrorRecovers(t *testing.T) { // Wait until the pipeline has recovered: it must have passed through // Recovering and be Running again. (The initial run also briefly reports // Running, so "Running after a Recovering" is what distinguishes recovery.) - waitForRecovered(t, rec) + waitForRecovered(t, rec, pl) // The recovered run delivers its records end-to-end; wait for the source to // ack them so the graceful stop below has a deterministic last position. @@ -905,7 +905,12 @@ func TestServiceLifecycle_Recovery_LiveEntryPublishedBeforeRunningStatus(t *test // pre-recovery error resurfacing from a dead tomb, and no orphaned run // left behind (which is what stopAndWaitPersister above would hang on). waitForRecordsAcked(t, source, healthyRecords) - is.Equal(pipeline.StatusRunning, pl.GetStatus()) + // Poll rather than read instantly: close(release) only lets the post-recovery + // UpdateStatus(StatusRunning) call PROCEED, it does not wait for it to land on + // the instance, and the records this test just waited on were already acked + // while the window was frozen — so an instant read here races the status write + // with no intervening delay at all. See waitForRecovered for the full ordering. + waitForStatus(t, pl, pipeline.StatusRunning) is.NoErr(ls.Stop(ctx, pl.ID, false)) is.NoErr(ls.WaitPipeline(pl.ID)) @@ -1754,7 +1759,7 @@ func TestServiceLifecycle_NSource_TransientErrorOneSource_Recovers(t *testing.T) // Wait until the pipeline has recovered: it must have passed through // Recovering and be Running again. - waitForRecovered(t, rec) + waitForRecovered(t, rec, pl) waitForRecordsAcked(t, sourceA, healthyRecords) is.Equal(pipeline.StatusRunning, pl.GetStatus()) @@ -1925,24 +1930,56 @@ func (r *statusRecorder) snapshot() []pipeline.Status { return out } -// waitForRecovered blocks until the recorded status sequence shows a recovery: -// a Recovering entry followed by a later Running. Fails the test on timeout. -func waitForRecovered(t *testing.T, rec *statusRecorder) { +// waitForRecovered blocks until the pipeline has recovered: the recorded status +// sequence shows a Recovering entry followed by a later Running, AND the live +// instance actually reports StatusRunning. Fails the test on timeout. +// +// Both halves are load-bearing, and the second one is why this takes pl. +// +// The recorder is only a LEADING indicator. statusRecorder.UpdateStatus appends +// the status and runs its hook BEFORE delegating to the wrapped +// pipeline.Service, which is what actually calls Instance.SetStatus. So the +// recorded post-recovery Running appears strictly before pl.GetStatus() returns +// Running, and the gap is a full pipeline.Service.UpdateStatus call (a Get, two +// metrics updates, and a store write). +// +// Record flow is not a substitute watermark either: runPipeline releases every +// worker goroutine (close(registered)) BEFORE it calls +// UpdateStatus(StatusRunning), so the recovered run can read, write and ack its +// entire record set while that status write is still in flight. That is exactly +// how TestServiceLifecycle_Recovery_TransientErrorRecovers used to flake in CI +// — waitForRecovered followed by waitForRecordsAcked followed by an INSTANT +// is.Equal(StatusRunning, pl.GetStatus()) that read Recovering, because neither +// wait is ordered after the status write. Unlike the initial start (ls.Start +// returns only after runPipeline's status write completes), a recovery restart +// runs on the tomb's cleanup goroutine, so nothing in the test goroutine is +// synchronized with it. Polling the live value here is the ordering the test +// actually needs; it is a watermark, not a widened timeout. +func waitForRecovered(t *testing.T, rec *statusRecorder, pl *pipeline.Instance) { t.Helper() deadline := time.Now().Add(5 * time.Second) + recovered := false for { statuses := rec.snapshot() - seenRecovering := false - for _, s := range statuses { - switch { - case s == pipeline.StatusRecovering: - seenRecovering = true - case s == pipeline.StatusRunning && seenRecovering: - return // Running after a Recovering == recovered + if !recovered { + seenRecovering := false + for _, s := range statuses { + switch { + case s == pipeline.StatusRecovering: + seenRecovering = true + case s == pipeline.StatusRunning && seenRecovering: + recovered = true // Running after a Recovering == recovered + } } } + if recovered && pl.GetStatus() == pipeline.StatusRunning { + return + } if time.Now().After(deadline) { - t.Fatalf("timed out waiting for pipeline to recover (statuses: %v)", statuses) + t.Fatalf( + "timed out waiting for pipeline to recover (recorded: %v, live status: %s)", + statuses, pl.GetStatus(), + ) } time.Sleep(time.Millisecond) }