fix(s3/lifecycle): persist cursor with fresh ctx after passCtx timeout

drainShardEvents only exits via ctx cancellation for an idle subscription
— that's the steady-state when all replayed events are already past.
Saving the cursor with the canceled passCtx silently drops every
advance, so the next pass re-subscribes from the same floor and
re-replays the same events. Symptom in s3tests: status=error shards=16
errors=16 on every pass, and 1/6 expire3/* dispatches lost to a race
between concurrent shard drains all retrying the same events.

Use a 5s timeout derived from context.Background for the save, and
treat passCtx Deadline/Canceled from drain as a clean end-of-pass —
not a shard-level error to log.
This commit is contained in:
Chris Lu
2026-05-12 23:39:09 -07:00
parent 13c731ee6c
commit bfea01e113
+15 -2
View File
@@ -280,12 +280,25 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
}
lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs)
// Cursor save uses a fresh ctx because the steady-state drain exits
// via passCtx cancellation (the only signal the filer subscription
// gets when no new events arrive). Saving with the canceled passCtx
// would silently drop the cursor and the next pass would re-replay
// from the same floor — defeating advancement entirely.
saveCtx, saveCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer saveCancel()
if drainErr != nil {
_ = cfg.Persister.Save(ctx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted})
_ = cfg.Persister.Save(saveCtx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted})
// passCtx timeout is the expected end-of-pass for an idle
// subscription; not a real error. Other drain errors still
// propagate.
if errors.Is(drainErr, context.DeadlineExceeded) || errors.Is(drainErr, context.Canceled) {
return nil
}
return fmt.Errorf("shard=%d: drain: %w", shardID, drainErr)
}
return cfg.Persister.Save(ctx, shardID, Cursor{
return cfg.Persister.Save(saveCtx, shardID, Cursor{
TsNs: lastOK,
RuleSetHash: rsh,
PromotedHash: promoted,