fix: Park a synchronizer that reports Off or closes its results

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

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)
+}