From 9007d8a108211355db76fa67208aea10180bdbfa Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sat, 10 Oct 2026 19:24:31 +0800 Subject: [PATCH] Fix/syncer timer generation guard (#20725) --- internal/syncer/scheduler.go | 54 ++++++-- internal/syncer/scheduler_timer_test.go | 177 ++++++++++++++++++++++++ 2 files changed, 220 insertions(+), 11 deletions(-) create mode 100644 internal/syncer/scheduler_timer_test.go diff --git a/internal/syncer/scheduler.go b/internal/syncer/scheduler.go index e729bd58e1..97078f3790 100644 --- a/internal/syncer/scheduler.go +++ b/internal/syncer/scheduler.go @@ -172,22 +172,54 @@ func (s *Scheduler) ScheduleTaskAfter(ctx context.Context, taskID string, delay return s.publish(ctx, taskID, false) } s.timerMu.Lock() + defer s.timerMu.Unlock() + s.scheduleTimerLocked(ctx, taskID, delay) + return nil +} + +// scheduleTimerLocked replaces a timer while holding timerMu. +func (s *Scheduler) scheduleTimerLocked(ctx context.Context, taskID string, delay time.Duration) { if existing := s.timers[taskID]; existing != nil { existing.Stop() } - timer := time.AfterFunc(delay, func() { - if err := s.publish(ctx, taskID, true); err != nil && ctx.Err() == nil { - common.Warn("syncer scheduler timer publish failed", zap.String("task_id", taskID), zap.Error(err)) - _ = s.ScheduleTaskAfter(ctx, taskID, 3*time.Second) - return - } - s.timerMu.Lock() - delete(s.timers, taskID) - s.timerMu.Unlock() + var timer *time.Timer + timer = time.AfterFunc(delay, func() { + s.handleTimerWakeup(ctx, taskID, &timer) }) s.timers[taskID] = timer - s.timerMu.Unlock() - return nil +} + +// handleTimerWakeup is also exercised directly by callback-path unit tests. +// The timer pointer is only read while holding timerMu, which also protects +// the assignment made by scheduleTimerLocked. +func (s *Scheduler) handleTimerWakeup(ctx context.Context, taskID string, timer **time.Timer) { + if err := s.publish(ctx, taskID, true); err != nil && ctx.Err() == nil { + common.Warn("syncer scheduler timer publish failed", zap.String("task_id", taskID), zap.Error(err)) + s.retryTimerIfCurrent(ctx, taskID, timer) + return + } + s.retireTimer(taskID, timer) +} + +// retryTimerIfCurrent prevents an outdated callback's failure from replacing +// a newer schedule. The identity check and replacement are one critical section. +func (s *Scheduler) retryTimerIfCurrent(ctx context.Context, taskID string, timer **time.Timer) { + s.timerMu.Lock() + defer s.timerMu.Unlock() + if ctx.Err() != nil || s.timers[taskID] != *timer { + return + } + s.scheduleTimerLocked(ctx, taskID, 3*time.Second) +} + +// retireTimer removes only the timer whose callback finished. A newer +// schedule for the same task must remain registered for replacement/shutdown. +func (s *Scheduler) retireTimer(taskID string, timer **time.Timer) { + s.timerMu.Lock() + defer s.timerMu.Unlock() + if s.timers[taskID] == *timer { + delete(s.timers, taskID) + } } func (s *Scheduler) publish(ctx context.Context, taskID string, wakeup bool) error { diff --git a/internal/syncer/scheduler_timer_test.go b/internal/syncer/scheduler_timer_test.go new file mode 100644 index 0000000000..7946489d25 --- /dev/null +++ b/internal/syncer/scheduler_timer_test.go @@ -0,0 +1,177 @@ +package syncer + +import ( + "context" + "errors" + "testing" + "time" +) + +// TestSchedulerExpiredCallbackKeepsReplacementTimer verifies the stale +// callback cleanup cannot remove a newer timer for the same task. +func TestSchedulerExpiredCallbackKeepsReplacementTimer(t *testing.T) { + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), nil, &fakeSyncTaskBroker{}) + defer scheduler.stopTimers() + + if err := scheduler.ScheduleTaskAfter(t.Context(), "task-1", time.Hour); err != nil { + t.Fatalf("schedule first timer: %v", err) + } + scheduler.timerMu.Lock() + original := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if original == nil { + t.Fatal("original timer not registered") + } + + if err := scheduler.ScheduleTaskAfter(t.Context(), "task-1", 2*time.Hour); err != nil { + t.Fatalf("schedule replacement timer: %v", err) + } + scheduler.timerMu.Lock() + replacement := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if replacement == nil || replacement == original { + t.Fatalf("replacement timer = %p, original = %p", replacement, original) + } + defer replacement.Stop() + + // Simulate the original callback reaching its cleanup after replacement. + // Invoke the exact same method used by the real timer callback: unlike a + // sleep-based assertion, this cannot pass before cleanup has executed. + scheduler.retireTimer("task-1", &original) + + scheduler.timerMu.Lock() + got := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if got != replacement { + t.Fatalf("stale callback removed replacement: got %p, want %p", got, replacement) + } + + // The current callback must still be allowed to retire its own timer. + replacement.Stop() + scheduler.retireTimer("task-1", &replacement) + scheduler.timerMu.Lock() + _, exists := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if exists { + t.Fatal("current callback did not remove its timer") + } +} + +// TestSchedulerStaleFailureKeepsReplacementTimer verifies that the retry +// path cannot overwrite a newer timer when an old callback fails. +func TestSchedulerStaleFailureKeepsReplacementTimer(t *testing.T) { + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), nil, &fakeSyncTaskBroker{}) + defer scheduler.stopTimers() + + if err := scheduler.ScheduleTaskAfter(t.Context(), "task-1", time.Hour); err != nil { + t.Fatalf("schedule first timer: %v", err) + } + scheduler.timerMu.Lock() + original := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + + if err := scheduler.ScheduleTaskAfter(t.Context(), "task-1", 2*time.Hour); err != nil { + t.Fatalf("schedule replacement timer: %v", err) + } + scheduler.timerMu.Lock() + replacement := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if original == nil || replacement == nil || original == replacement { + t.Fatalf("invalid timer replacement: old %p, new %p", original, replacement) + } + defer replacement.Stop() + + // Simulate the stale callback's retry, using the same path as the + // production callback. The newer schedule must not be overwritten. + scheduler.retryTimerIfCurrent(t.Context(), "task-1", &original) + scheduler.timerMu.Lock() + got := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if got != replacement { + t.Fatalf("stale retry replaced newer timer: got %p, want %p", got, replacement) + } + + // A genuinely current timer must still be retried on publish failure. + scheduler.retryTimerIfCurrent(t.Context(), "task-1", &replacement) + scheduler.timerMu.Lock() + retry := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if retry == nil || retry == replacement { + t.Fatalf("current timer not retried: got %p, previous %p", retry, replacement) + } +} + +// TestSchedulerCallbackPreservesReplacement exercises the actual callback +// path through publication and cleanup, rather than just the cleanup helper. +func TestSchedulerCallbackPreservesReplacement(t *testing.T) { + ctx := t.Context() + var scheduler *Scheduler + broker := &fakeSyncTaskBroker{} + broker.onPublish = func(taskID string) { + if err := scheduler.ScheduleTaskAfter(ctx, taskID, 2*time.Hour); err != nil { + t.Errorf("schedule replacement: %v", err) + } + } + scheduler = NewNATSScheduler(make(chan TaskEnvelope, 1), nil, broker) + defer scheduler.stopTimers() + + if err := scheduler.ScheduleTaskAfter(ctx, "task-1", time.Hour); err != nil { + t.Fatal(err) + } + scheduler.timerMu.Lock() + original := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + original.Stop() + + // The old callback's publish synchronously schedules the replacement. + // The handler cannot return until the stale cleanup has also completed. + scheduler.handleTimerWakeup(ctx, "task-1", &original) + scheduler.timerMu.Lock() + replacement := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if replacement == nil || replacement == original { + t.Fatalf("callback removed the replacement: old=%p new=%p", original, replacement) + } +} + +type failingTimerWakeupBroker struct { + *fakeSyncTaskBroker + onWakeup func(string) +} + +func (b *failingTimerWakeupBroker) PublishSyncerTaskWakeup(taskID string) error { + if b.onWakeup != nil { + b.onWakeup(taskID) + } + return errors.New("temporary publish failure") +} + +// TestSchedulerCallbackFailurePreservesReplacement covers the case in which +// a stale publish failure must not overwrite an already replaced timer. +func TestSchedulerCallbackFailurePreservesReplacement(t *testing.T) { + ctx := context.Background() + var scheduler *Scheduler + broker := &failingTimerWakeupBroker{fakeSyncTaskBroker: &fakeSyncTaskBroker{}} + broker.onWakeup = func(taskID string) { + if err := scheduler.ScheduleTaskAfter(ctx, taskID, 2*time.Hour); err != nil { + t.Errorf("schedule replacement: %v", err) + } + } + scheduler = NewNATSScheduler(make(chan TaskEnvelope, 1), nil, broker) + defer scheduler.stopTimers() + if err := scheduler.ScheduleTaskAfter(ctx, "task-1", time.Hour); err != nil { + t.Fatal(err) + } + scheduler.timerMu.Lock() + original := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + original.Stop() + + scheduler.handleTimerWakeup(ctx, "task-1", &original) + scheduler.timerMu.Lock() + replacement := scheduler.timers["task-1"] + scheduler.timerMu.Unlock() + if replacement == nil || replacement == original { + t.Fatalf("stale failure replaced the new schedule: old=%p new=%p", original, replacement) + } +}