From 3db133f088559b54f5b34a4dce841782ed154fed Mon Sep 17 00:00:00 2001 From: Bee Klimt Date: Mon, 2 Nov 2026 12:58:11 -0800 Subject: [PATCH] fix: Leave StateSince alone when a synchronizer starts --- internal/datasystem/fdv2_datasystem.go | 16 +++++++-------- internal/datasystem/fdv2_fallback_test.go | 24 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/internal/datasystem/fdv2_datasystem.go b/internal/datasystem/fdv2_datasystem.go index 521b7207..b0488758 100644 --- a/internal/datasystem/fdv2_datasystem.go +++ b/internal/datasystem/fdv2_datasystem.go @@ -102,6 +102,9 @@ type FDv2 struct { fallbackCond func(status interfaces.DataSourceStatus) bool recoveryCond func(status interfaces.DataSourceStatus) bool + // The time the current synchronizer started. Only the goroutine that runs the synchronizers uses it. + syncStartedAt time.Time + // The default duration of a directed fallback to FDv1, before jitter. defaultFDv1FallbackTTL time.Duration @@ -437,7 +440,7 @@ func (f *FDv2) runSynchronizers(ctx context.Context, closeWhenReady chan struct{ } f.loggers.Infof("Synchronizer at index %d (%s) is starting", f.synchronizers.currentIndex(), sync.Name()) - f.restartStatusClock() + f.syncStartedAt = time.Now() resultChan := sync.Sync(f.store) action, err := f.consumeSynchronizerResults(ctx, resultChan, closeWhenReady) @@ -520,6 +523,10 @@ const ( // conditionAction returns the action the configured conditions call for, or syncNone. Recovery // is checked first, so it takes precedence when both conditions hold. func (f *FDv2) conditionAction(status interfaces.DataSourceStatus) syncAction { + // The conditions measure from the later of the last state change and the start of the current synchronizer. + if f.syncStartedAt.After(status.StateSince) { + status.StateSince = f.syncStartedAt + } if !f.synchronizers.isCurrentFirstEligible() && f.recoveryCond(status) { f.loggers.Debugf("Recovery condition met") return syncRecover @@ -772,13 +779,6 @@ func (f *FDv2) maybeUpdateStatus( return f.status, changed } -// restartStatusClock sets the time the current status has been held since to now. -func (f *FDv2) restartStatusClock() { - f.mu.Lock() - defer f.mu.Unlock() - f.status.StateSince = time.Now() -} - func (f *FDv2) getStatus() interfaces.DataSourceStatus { f.mu.Lock() defer f.mu.Unlock() diff --git a/internal/datasystem/fdv2_fallback_test.go b/internal/datasystem/fdv2_fallback_test.go index d1f0fc64..fe592a15 100644 --- a/internal/datasystem/fdv2_fallback_test.go +++ b/internal/datasystem/fdv2_fallback_test.go @@ -437,3 +437,27 @@ func TestFDv2ParksSynchronizerThatClosesItsResults(t *testing.T) { assert.Equal(t, interfaces.DataSourceErrorKindUnknown, system.DataSourceStatusProvider().GetStatus().LastError.Kind) primary.assertNoRun(t) } + +func TestFDv2StartingSynchronizerDoesNotChangeStateSince(t *testing.T) { + primary := newScriptedSynchronizer("primary") + secondary := newScriptedSynchronizer("secondary") + system := makeSDKKeyTestFDv2(t, primary, secondary) + system.Start(make(chan struct{})) + + // The primary synchronizer fails, so the data system is interrupted. + primary.requireRun(t) <- unauthorizedResult + require.Eventually(t, func() bool { + return system.DataSourceStatusProvider().GetStatus().State == interfaces.DataSourceStateInterrupted + }, time.Second, time.Millisecond) + stateSince := system.DataSourceStatusProvider().GetStatus().StateSince + + // The secondary synchronizer starts after the failover delay. + select { + case <-secondary.runs: + case <-time.After(5 * time.Second): + require.FailNow(t, "timed out waiting for the secondary synchronizer to start") + } + + // The state did not change, so neither did the time it changed. + assert.Equal(t, stateSince, system.DataSourceStatusProvider().GetStatus().StateSince) +}