Skip to content

test: stabilize flaky TestTableTimerStore #69921

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
flaky-claw wants to merge 1 commit into
base: master
Choose a base branch
from
+42 −3
Open

test: stabilize flaky TestTableTimerStore #69921

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 42 additions & 3 deletions pkg/timer/store_intergartion_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ func TestMemTimerStore(t *testing.T) {

store = api.NewMemoryTimerStore()
defer store.Close()
runTimerStoreWatchTest(t, store)
runTimerStoreWatchTest(t, store, nil)
}

// createTimerTableSQL returns a SQL to create timer table
Expand Down Expand Up @@ -106,7 +106,43 @@ func TestTableTimerStore(t *testing.T) {
tk.MustExec(createTimerTableSQL(dbName, tblName))
timerStore = tablestore.NewTableTimerStore(1, pool, dbName, tblName, cli)
defer timerStore.Close()
runTimerStoreWatchTest(t, timerStore)
// Watch starts asynchronously in the etcd notifier, so synchronize before producing table events.
runTimerStoreWatchTest(t, timerStore, func(ch api.WatchTimerChan) {
readyDeadline := time.NewTimer(time.Minute)
defer readyDeadline.Stop()
retryReady:
for {
readyID := "watch-ready-" + uuid.NewString()
readyKey := fmt.Sprintf("/tidb/timer/cluster/%d/notify/%s", 1, readyID)
putCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
_, err := cli.Put(putCtx, readyKey, fmt.Sprintf(
`{"events":[{"tp":"create","timer_id":%q,"timestamp":%d}]}`,
readyID,
time.Now().Unix(),
))
cancel()
require.NoError(t, err)

attemptTimer := time.NewTimer(100 * time.Millisecond)
for {
select {
case resp, ok := <-ch:
require.True(t, ok)
for _, event := range resp.Events {
if event.Tp == api.WatchTimerEventCreate && event.TimerID == readyID {
attemptTimer.Stop()
return
}
}
case <-attemptTimer.C:
continue retryReady
case <-readyDeadline.C:
attemptTimer.Stop()
require.FailNow(t, "watch ready timeout")
}
}
}
})
Comment on lines +109 to +145

@coderabbitai coderabbitai Bot Jul 19, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Prevent tail-chasing race condition on slow CI environments.

The fixed 100ms attemptTimer can cause this test to flake and eventually timeout if the etcd watch event delivery latency consistently exceeds 100ms.

Because event.TimerID is strictly matched against the latest readyID, receiving an older event (e.g., delayed by >100ms) causes it to be ignored. The inner loop then hits the 100ms timeout before the newly expected event can arrive, triggering a new Put and resetting the 100ms timer. This creates a continuous tail-chasing scenario where the timer always expires before the correctly matched event arrives.

Use an exponential backoff for the retry interval. This ensures the timeout eventually exceeds the delivery latency, allowing the matching event to be received without leaving stray events in the channel for subsequent assertions.

🐛 Proposed fix to add exponential backoff 
-	retryReady:
+	retryInterval := 100 * time.Millisecond
+retryReady:
 		for {
 			readyID := "watch-ready-" + uuid.NewString()
 			readyKey := fmt.Sprintf("/tidb/timer/cluster/%d/notify/%s", 1, readyID)
 			putCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 			_, err := cli.Put(putCtx, readyKey, fmt.Sprintf(
 				`{"events":[{"tp":"create","timer_id":%q,"timestamp":%d}]}`,
 				readyID,
 				time.Now().Unix(),
 			))
 			cancel()
 			require.NoError(t, err)
 
-			attemptTimer := time.NewTimer(100 * time.Millisecond)
+			attemptTimer := time.NewTimer(retryInterval)
 			for {
 				select {
 				case resp, ok := <-ch:
 					require.True(t, ok)
 					for _, event := range resp.Events {
 						if event.Tp == api.WatchTimerEventCreate && event.TimerID == readyID {
 							attemptTimer.Stop()
 							return
 						}
 					}
 				case <-attemptTimer.C:
+					if retryInterval < 2*time.Second {
+						retryInterval *= 2
+					}
 					continue retryReady
 				case <-readyDeadline.C:
 					attemptTimer.Stop()
 					require.FailNow(t, "watch ready timeout")
 				}
 			}
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// Watch starts asynchronously in the etcd notifier, so synchronize before producing table events.
runTimerStoreWatchTest(t, timerStore, func(ch api.WatchTimerChan) {
readyDeadline := time.NewTimer(time.Minute)
defer readyDeadline.Stop()
retryReady:
for {
readyID := "watch-ready-" + uuid.NewString()
readyKey := fmt.Sprintf("/tidb/timer/cluster/%d/notify/%s", 1, readyID)
putCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
_, err := cli.Put(putCtx, readyKey, fmt.Sprintf(
`{"events":[{"tp":"create","timer_id":%q,"timestamp":%d}]}`,
readyID,
time.Now().Unix(),
))
cancel()
require.NoError(t, err)
attemptTimer := time.NewTimer(100 * time.Millisecond)
for {
select {
case resp, ok := <-ch:
require.True(t, ok)
for _, event := range resp.Events {
if event.Tp == api.WatchTimerEventCreate && event.TimerID == readyID {
attemptTimer.Stop()
return
}
}
case <-attemptTimer.C:
continue retryReady
case <-readyDeadline.C:
attemptTimer.Stop()
require.FailNow(t, "watch ready timeout")
}
}
}
})
// Watch starts asynchronously in the etcd notifier, so synchronize before producing table events.
runTimerStoreWatchTest(t, timerStore, func(ch api.WatchTimerChan) {
readyDeadline := time.NewTimer(time.Minute)
defer readyDeadline.Stop()
retryInterval := 100 * time.Millisecond
retryReady:
for {
readyID := "watch-ready-" + uuid.NewString()
readyKey := fmt.Sprintf("/tidb/timer/cluster/%d/notify/%s", 1, readyID)
putCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
_, err := cli.Put(putCtx, readyKey, fmt.Sprintf(
`{"events":[{"tp":"create","timer_id":%q,"timestamp":%d}]}`,
readyID,
time.Now().Unix(),
))
cancel()
require.NoError(t, err)
attemptTimer := time.NewTimer(retryInterval)
for {
select {
case resp, ok := <-ch:
require.True(t, ok)
for _, event := range resp.Events {
if event.Tp == api.WatchTimerEventCreate && event.TimerID == readyID {
attemptTimer.Stop()
return
}
}
case <-attemptTimer.C:
if retryInterval < 2*time.Second {
retryInterval *= 2
}
continue retryReady
case <-readyDeadline.C:
attemptTimer.Stop()
require.FailNow(t, "watch ready timeout")
}
}
}
})
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/timer/store_intergartion_test.go` around lines 109 - 145, Update the
readiness retry loop in runTimerStoreWatchTest to use an exponential backoff for
attemptTimer instead of a fixed 100ms interval. Increase the retry interval
after each timeout, while preserving the latest readyID matching and the overall
readyDeadline timeout, so slow event delivery can be observed without repeatedly
enqueueing readiness events.


// check pool
require.False(t, pool.inuse.Load())
Expand Down Expand Up @@ -577,7 +613,7 @@ func runTimerStoreInsertAndList(ctx context.Context, t *testing.T, store *api.Ti
checkList([]*api.TimerRecord{&recordTpl1}, timers)
}

func runTimerStoreWatchTest(t *testing.T, store *api.TimerStore) {
func runTimerStoreWatchTest(t *testing.T, store *api.TimerStore, prepareWatch func(api.WatchTimerChan)) {
require.True(t, store.WatchSupported())
ctx, cancel := context.WithCancel(context.Background())
defer func() {
Expand All @@ -595,6 +631,9 @@ func runTimerStoreWatchTest(t *testing.T, store *api.TimerStore) {
}

ch := store.Watch(ctx)
if prepareWatch != nil {
prepareWatch(ch)
}
assertWatchEvent := func(tp api.WatchTimerEventType, id string) {
timeout := time.NewTimer(time.Minute)
defer timeout.Stop()
Expand Down
Loading