diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go index 01c63c49..521b7207 100644 --- a/internal/datasystem/fdv2_datasystem.go +++ b/internal/datasystem/fdv2_datasystem.go @@ -563,11 +563,17 @@ 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. + f.UpdateStatus(interfaces.DataSourceStateInterrupted, interfaces.DataSourceErrorInfo{ + Kind: interfaces.DataSourceErrorKindUnknown, + Message: "synchronizer closed its result channel", + Time: time.Now(), + }) + if keyChanged { + return syncRestart, nil + } + return syncPark, nil } if result.EnvironmentID.IsDefined() { @@ -596,6 +602,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..d1f0fc64 100644 --- a/internal/datasystem/fdv2_fallback_test.go +++ b/internal/datasystem/fdv2_fallback_test.go @@ -399,3 +399,41 @@ 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 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 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) +}