From bfea01e1136c6055e80fb4232e16c32adc2cea25 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 12 May 2026 23:39:09 -0700 Subject: [PATCH] fix(s3/lifecycle): persist cursor with fresh ctx after passCtx timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- weed/s3api/s3lifecycle/dailyrun/run.go | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index e1380f5ae..ebf676279 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -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,