Skip to content

Commit d00ae43

Browse files
wesmwesm
authored
fix(sync): preserve retries across repeated stale paths (#1310)
Live activity hints can resolve through several stale indexed paths while a rollout is moving. #1308 kept the initial missing-path lookup retryable, but a later retry that resolved to another missing path consumed that state and could leave the session stale until another prompt arrived. This follow-up retains the bounded retry until the candidate path is successfully statted and records retry and refresh lookups in the poll-wide attempted set. Cursor resets can replay older hints, so carried retry state is restored before merging the replay and only genuinely newer activity can restart the retry window. Each session still incurs at most one lookup per poll, while the existing retry TTL and global state bounds remain unchanged. Co-authored-by: Wes McKinney <wesm@users.noreply.github.com>
1 parent d322d64 commit d00ae43

2 files changed

Lines changed: 134 additions & 13 deletions

File tree

‎internal/sync/live_activity.go‎

Lines changed: 28 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -55,11 +55,12 @@ type LiveActivityPollStats struct {
5555
}
5656

5757
type liveActivityHotEntry struct {
58-
target int
59-
source LiveActivitySource
60-
lastActivity time.Time
61-
pending bool
62-
refreshRetry *liveActivityRetryEntry
58+
target int
59+
source LiveActivitySource
60+
lastActivity time.Time
61+
pending bool
62+
refreshRetry *liveActivityRetryEntry
63+
retryPendingStat bool
6364
}
6465

6566
type liveActivityRetryEntry struct {
@@ -278,6 +279,7 @@ func (p *LiveActivityPoller) PollOnce(
278279
if _, ok := attempted[fullID]; ok {
279280
continue
280281
}
282+
attempted[fullID] = struct{}{}
281283
stats.SessionLookups++
282284
source, found, err := p.lookup(ctx, fullID)
283285
if ctxErr := ctx.Err(); ctxErr != nil {
@@ -308,6 +310,7 @@ func (p *LiveActivityPoller) PollOnce(
308310
if _, ok := attempted[fullID]; ok {
309311
continue
310312
}
313+
attempted[fullID] = struct{}{}
311314
stats.SessionLookups++
312315
source, found, err := p.lookup(ctx, fullID)
313316
if ctxErr := ctx.Err(); ctxErr != nil {
@@ -352,12 +355,13 @@ func (p *LiveActivityPoller) PollOnce(
352355
}
353356
if errors.Is(err, os.ErrNotExist) {
354357
delete(p.hot, fullID)
358+
if entry.refreshRetry != nil {
359+
p.retries[fullID] = entry.refreshRetry
360+
}
355361
if hint, ok := hinted[fullID]; ok {
356362
p.addRetry(
357363
fullID, hint.target, now, hint.lastHint,
358364
)
359-
} else if entry.refreshRetry != nil {
360-
p.retries[fullID] = entry.refreshRetry
361365
}
362366
continue
363367
}
@@ -366,6 +370,10 @@ func (p *LiveActivityPoller) PollOnce(
366370
fmt.Errorf("stat live activity source %q: %w", entry.source.Path, err))
367371
continue
368372
}
373+
if entry.retryPendingStat {
374+
entry.refreshRetry = nil
375+
entry.retryPendingStat = false
376+
}
369377
inode, device := getFileIdentity(entry.source.Path, info)
370378
if entry.source.HasStoredStat &&
371379
entry.source.StoredSize == info.Size() &&
@@ -495,6 +503,7 @@ func (p *LiveActivityPoller) addHotRefreshRetry(
495503
retry.target = target
496504
retry.lastHint = lastHint
497505
}
506+
entry.retryPendingStat = false
498507
}
499508

500509
func (p *LiveActivityPoller) setHot(
@@ -504,6 +513,7 @@ func (p *LiveActivityPoller) setHot(
504513
lastActivity time.Time,
505514
) {
506515
source.Path = filepath.Clean(source.Path)
516+
var refreshRetry *liveActivityRetryEntry
507517
if entry := p.hot[fullID]; entry != nil {
508518
if entry.lastActivity.After(lastActivity) {
509519
lastActivity = entry.lastActivity
@@ -512,15 +522,20 @@ func (p *LiveActivityPoller) setHot(
512522
retry.lastHint.After(lastActivity) {
513523
lastActivity = retry.lastHint
514524
}
525+
refreshRetry = entry.refreshRetry
515526
}
516-
if retry := p.retries[fullID]; retry != nil &&
517-
retry.lastHint.After(lastActivity) {
518-
lastActivity = retry.lastHint
527+
if retry := p.retries[fullID]; retry != nil {
528+
if retry.lastHint.After(lastActivity) {
529+
lastActivity = retry.lastHint
530+
}
531+
refreshRetry = retry
519532
}
520533
p.hot[fullID] = &liveActivityHotEntry{
521-
target: target,
522-
source: source,
523-
lastActivity: lastActivity,
534+
target: target,
535+
source: source,
536+
lastActivity: lastActivity,
537+
refreshRetry: refreshRetry,
538+
retryPendingStat: refreshRetry != nil,
524539
}
525540
delete(p.retries, fullID)
526541
}

‎internal/sync/live_activity_test.go‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -537,6 +537,112 @@ func TestLiveActivityRetriesHintWhoseIndexedPathIsMissing(t *testing.T) {
537537
assert.Equal(t, []string{canonical}, synced)
538538
}
539539

540+
func TestLiveActivityRetriesRepeatedMissingIndexedPaths(t *testing.T) {
541+
now := time.Unix(1_800_000_000, 0).UTC()
542+
dir := t.TempDir()
543+
history := filepath.Join(dir, "history.jsonl")
544+
firstMissing := filepath.Join(dir, "first-missing.jsonl")
545+
secondMissing := filepath.Join(dir, "second-missing.jsonl")
546+
canonical := filepath.Join(dir, "canonical.jsonl")
547+
require.NoError(t, os.WriteFile(
548+
history, []byte(hintRecord("move", now)), 0o644,
549+
))
550+
require.NoError(t, os.WriteFile(canonical, []byte("active\n"), 0o644))
551+
provider := newLiveActivityTestProvider(history)
552+
lookupPaths := []string{firstMissing, secondMissing, canonical}
553+
lookups := 0
554+
var synced []string
555+
poller := NewLiveActivityPoller([]LiveActivityTarget{{
556+
Provider: provider,
557+
Hints: provider,
558+
Sources: []parser.ActivityHintSource{{Path: history}},
559+
}}, func(_ context.Context, id string) (LiveActivitySource, bool, error) {
560+
assert.Equal(t, "codex:move", id)
561+
require.Less(t, lookups, len(lookupPaths))
562+
path := lookupPaths[lookups]
563+
lookups++
564+
return LiveActivitySource{Path: path}, true, nil
565+
}, func(_ context.Context, paths []string) error {
566+
synced = append(synced, paths...)
567+
return nil
568+
}, nil)
569+
570+
_, err := poller.PollOnce(t.Context(), now)
571+
require.NoError(t, err)
572+
require.Contains(t, poller.retries, "codex:move")
573+
574+
_, err = poller.PollOnce(t.Context(), now.Add(time.Second))
575+
require.NoError(t, err)
576+
require.Contains(t, poller.retries, "codex:move",
577+
"a second stale indexed path must not consume the retry")
578+
579+
_, err = poller.PollOnce(t.Context(), now.Add(2*time.Second))
580+
require.NoError(t, err)
581+
assert.Equal(t, 3, lookups)
582+
require.Contains(t, poller.hot, "codex:move")
583+
assert.Equal(t, canonical, poller.hot["codex:move"].source.Path)
584+
assert.NotContains(t, poller.retries, "codex:move")
585+
assert.Equal(t, []string{canonical}, synced)
586+
}
587+
588+
func TestLiveActivityOlderReplayPreservesRetryAcrossRepeatedMissingPaths(
589+
t *testing.T,
590+
) {
591+
now := time.Unix(1_800_000_000, 0).UTC()
592+
dir := t.TempDir()
593+
history := filepath.Join(dir, "history.jsonl")
594+
firstMissing := filepath.Join(dir, "first-missing.jsonl")
595+
secondMissing := filepath.Join(dir, "second-missing.jsonl")
596+
canonical := filepath.Join(dir, "canonical.jsonl")
597+
require.NoError(t, os.WriteFile(
598+
history, []byte(hintRecord("move", now)), 0o644,
599+
))
600+
require.NoError(t, os.WriteFile(canonical, []byte("active\n"), 0o644))
601+
provider := newLiveActivityTestProvider(history)
602+
lookupPaths := []string{firstMissing, secondMissing, canonical}
603+
lookups := 0
604+
var synced []string
605+
poller := NewLiveActivityPoller([]LiveActivityTarget{{
606+
Provider: provider,
607+
Hints: provider,
608+
Sources: []parser.ActivityHintSource{{Path: history}},
609+
}}, func(_ context.Context, id string) (LiveActivitySource, bool, error) {
610+
assert.Equal(t, "codex:move", id)
611+
require.Less(t, lookups, len(lookupPaths))
612+
path := lookupPaths[lookups]
613+
lookups++
614+
return LiveActivitySource{Path: path}, true, nil
615+
}, func(_ context.Context, paths []string) error {
616+
synced = append(synced, paths...)
617+
return nil
618+
}, nil)
619+
620+
_, err := poller.PollOnce(t.Context(), now)
621+
require.NoError(t, err)
622+
require.Contains(t, poller.retries, "codex:move")
623+
624+
replacement := history + ".older"
625+
require.NoError(t, os.WriteFile(
626+
replacement,
627+
[]byte(hintRecord("move", now.Add(-time.Hour))),
628+
0o644,
629+
))
630+
require.NoError(t, os.Rename(replacement, history))
631+
_, err = poller.PollOnce(t.Context(), now.Add(time.Minute))
632+
require.NoError(t, err)
633+
require.Contains(t, poller.retries, "codex:move")
634+
assert.Equal(t, now, poller.retries["codex:move"].firstSeen)
635+
assert.Equal(t, now, poller.retries["codex:move"].lastHint)
636+
637+
_, err = poller.PollOnce(t.Context(), now.Add(time.Minute+time.Second))
638+
require.NoError(t, err)
639+
assert.Equal(t, 3, lookups)
640+
require.Contains(t, poller.hot, "codex:move")
641+
assert.Equal(t, canonical, poller.hot["codex:move"].source.Path)
642+
assert.NotContains(t, poller.retries, "codex:move")
643+
assert.Equal(t, []string{canonical}, synced)
644+
}
645+
540646
func TestLiveActivityPreservesRefreshRetryAcrossHotExpiration(t *testing.T) {
541647
now := time.Unix(1_800_000_000, 0).UTC()
542648
dir := t.TempDir()

0 commit comments

Comments
 (0)