From d838a4ba7a3f151434b55125e48e8dfe1c417820 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 2 Nov 2026 11:09:22 -0800 Subject: [PATCH 1/2] fix: Park a synchronizer that reports Off or closes its results --- internal/datasystem/fdv2_datasystem.go | 13 +++++++--- internal/datasystem/fdv2_fallback_test.go | 31 +++++++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go index 4a7059ac..fe4828d3 100644 --- a/internal/datasystem/fdv2_datasystem.go +++ b/internal/datasystem/fdv2_datasystem.go @@ -566,11 +566,12 @@ func (f *FDv2) consumeSynchronizerResults( f.synchronizers.unparkAll() keyChanged = true case result, ok := <-resultChan: - // The status channel being closed means that we won't be receiving - // any more information from that synchronizer and we should - // probably fall back. if !ok { - return syncFallback, nil + // The synchronizer closed its result channel on its own. Treat it as an unexpected failure. + if keyChanged { + return syncRestart, nil + } + return syncPark, nil } if result.EnvironmentID.IsDefined() { @@ -599,6 +600,10 @@ func (f *FDv2) consumeSynchronizerResults( } case interfaces.DataSourceStateInterrupted: f.UpdateStatus(result.State, result.Error) + case interfaces.DataSourceStateOff: + // A synchronizer must not report Off. Treat it as an unexpected failure. + f.UpdateStatus(interfaces.DataSourceStateInterrupted, result.Error) + result.UnexpectedFailure = true } // FallbackToFDv1 may ride along on a Valid or Interrupted result too -- e.g. a diff --git a/internal/datasystem/fdv2_fallback_test.go b/internal/datasystem/fdv2_fallback_test.go index 8f7fe20d..d822e4db 100644 --- a/internal/datasystem/fdv2_fallback_test.go +++ b/internal/datasystem/fdv2_fallback_test.go @@ -399,3 +399,34 @@ func TestFDv2ReturnsToFDv2AfterFallbackTTLWhileFDv1FallbackRuns(t *testing.T) { require.FailNow(t, "timed out waiting for the primary synchronizer to start again") } } + +func TestFDv2ParksSynchronizerThatReportsOff(t *testing.T) { + primary := newScriptedSynchronizer("primary") + system := makeSDKKeyTestFDv2(t, primary) + system.Start(make(chan struct{})) + + // The only synchronizer reports Off with an error. + primary.requireRun(t) <- subsystems.DataSynchronizerResult{ + State: interfaces.DataSourceStateOff, + Error: interfaces.DataSourceErrorInfo{Kind: interfaces.DataSourceErrorKindUnknown, Message: "bad data"}, + } + + // The data system is interrupted with the error, and the synchronizer is parked. + require.Eventually(t, func() bool { + return system.DataSourceStatusProvider().GetStatus().LastError.Message == "bad data" + }, time.Second, time.Millisecond) + assert.Equal(t, interfaces.DataSourceStateInterrupted, system.DataSourceStatusProvider().GetStatus().State) + primary.assertNoRun(t) +} + +func TestFDv2ParksSynchronizerThatClosesItsResults(t *testing.T) { + primary := newScriptedSynchronizer("primary") + system := makeSDKKeyTestFDv2(t, primary) + system.Start(make(chan struct{})) + + // The only synchronizer closes its result channel on its own. + close(primary.requireRun(t)) + + // The synchronizer is parked, so it does not start again. + primary.assertNoRun(t) +} From 59492c2ea95cf23553279c2620d07950b9ce52f2 Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 2 Nov 2026 13:26:37 -0800 Subject: [PATCH 2/2] fix: Report Interrupted when a synchronizer closes its results --- internal/datasystem/fdv2_datasystem.go | 5 +++++ internal/datasystem/fdv2_fallback_test.go | 13 ++++++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go index fe4828d3..2d10c99d 100644 --- a/internal/datasystem/fdv2_datasystem.go +++ b/internal/datasystem/fdv2_datasystem.go @@ -568,6 +568,11 @@ func (f *FDv2) consumeSynchronizerResults( case result, ok := <-resultChan: if !ok { // The synchronizer closed its result channel on its own. Treat it as an unexpected failure. + f.UpdateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{ + Kind: interfaces.DataSourceErrorKindUnknown, + Message: "synchronizer closed its result channel", + Time: time.Now(), + }) if keyChanged { return syncRestart, nil } diff --git a/internal/datasystem/fdv2_fallback_test.go b/internal/datasystem/fdv2_fallback_test.go index d822e4db..d1f0fc64 100644 --- a/internal/datasystem/fdv2_fallback_test.go +++ b/internal/datasystem/fdv2_fallback_test.go @@ -424,9 +424,16 @@ func TestFDv2ParksSynchronizerThatClosesItsResults(t *testing.T) { system := makeSDKKeyTestFDv2(t, primary) system.Start(make(chan struct{})) - // The only synchronizer closes its result channel on its own. - close(primary.requireRun(t)) + // The only synchronizer gets data, then closes its result channel on its own. + results := primary.requireRun(t) + results <- validResult + require.True(t, system.DataSourceStatusProvider().WaitFor(interfaces.DataSourceStateValid, time.Second)) + close(results) - // The synchronizer is parked, so it does not start again. + // The data system is interrupted, and the synchronizer is parked, so it does not start again. + require.Eventually(t, func() bool { + return system.DataSourceStatusProvider().GetStatus().State == interfaces.DataSourceStateInterrupted + }, time.Second, time.Millisecond) + assert.Equal(t, interfaces.DataSourceErrorKindUnknown, system.DataSourceStatusProvider().GetStatus().LastError.Kind) primary.assertNoRun(t) }