diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability.go b/weed/s3api/s3lifecycle/dailyrun/replayability.go index 139a61dcb..5f72b5962 100644 --- a/weed/s3api/s3lifecycle/dailyrun/replayability.go +++ b/weed/s3api/s3lifecycle/dailyrun/replayability.go @@ -1,22 +1,17 @@ package dailyrun import ( - "crypto/sha256" - "encoding/binary" "errors" "fmt" - "sort" - "time" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" ) -// UnsupportedRuleError is returned by Run when the bucket's compiled -// rules include any action kind Phase 2 cannot service. The handler -// surfaces this so admin marks the run failed in the activity log; -// flipping the algorithm flag to daily_replay on a bucket with these -// rules is a loud failure rather than a silent dropped rule. +// UnsupportedRuleError fails the run loudly when the snapshot contains +// a rule the Phase 2 replay path can't service. Surfaced verbatim to +// the activity log so flipping algorithm=daily_replay on an +// incompatible bucket isn't a silent dropped rule. type UnsupportedRuleError struct { Bucket string Kind s3lifecycle.ActionKind @@ -27,16 +22,11 @@ func (e *UnsupportedRuleError) Error() string { return fmt.Sprintf("daily_replay: unsupported action kind %s on bucket %q: %s", e.Kind, e.Bucket, e.Reason) } -// IsUnsupportedRule reports whether err is or wraps an -// UnsupportedRuleError. Callers (the worker handler) use this to -// classify the run outcome. func IsUnsupportedRule(err error) bool { var u *UnsupportedRuleError return errors.As(err, &u) } -// isReplayEligibleKind reports whether the engine's daily-replay path -// can service action kind k. Mirror this list when adding new kinds. func isReplayEligibleKind(k s3lifecycle.ActionKind) bool { switch k { case s3lifecycle.ActionKindExpirationDays, @@ -47,15 +37,11 @@ func isReplayEligibleKind(k s3lifecycle.ActionKind) bool { return false } -// checkSnapshotForUnsupported walks every active CompiledAction in snap -// and returns the first kind that isn't replay-eligible OR is in a Mode -// other than ModeEventDriven. router.Route gates dispatch on -// `Mode == ModeEventDriven` (see weed/s3api/s3lifecycle/router/router.go), -// so a replay-kind action that's been promoted to ModeScanOnly would -// silently get no matches at all — the daily run must reject it loudly -// so admin sees the failure in the activity log. Phase 4 partitions -// these into walk-bound actions and runs them through the walker, -// removing the gate. +// checkSnapshotForUnsupported rejects (a) walker-bound action kinds and +// (b) replay-kind actions in any Mode other than ModeEventDriven. +// router.Route silently drops non-ModeEventDriven actions; rejecting +// them here turns the silent drop into a loud failure. Phase 4 +// partitions these into walk-bound actions and removes the gate. func checkSnapshotForUnsupported(snap *engine.Snapshot) *UnsupportedRuleError { if snap == nil { return nil @@ -81,116 +67,3 @@ func checkSnapshotForUnsupported(snap *engine.Snapshot) *UnsupportedRuleError { } return nil } - -// localReplayContentHash hashes the rule definitions of replay-eligible -// actions in snap. Stable across reorderings — actions are sorted by -// (bucket, rule_hash, action_kind) before mixing. Phase 4 replaces -// this with engine.ReplayContentHash; both must produce the same value -// on a snapshot that's already fully replay-eligible (which Phase 2 -// enforces via checkSnapshotForUnsupported). -// -// Includes the effective TTL in the hash so a TTL change (e.g. 30 → 60 -// days) is detected as a rule-content change, even though the rule's -// RuleHash also captures it — defense in depth against any future -// RuleHash collision and an explicit dependency for the cursor -// rewind decision. -func localReplayContentHash(snap *engine.Snapshot) [32]byte { - if snap == nil { - return [32]byte{} - } - type rec struct { - bucket string - ruleHash [8]byte - actionKind s3lifecycle.ActionKind - ttlNs int64 - } - var recs []rec - for _, a := range snap.AllActions() { - if a == nil || !a.IsActive() { - continue - } - if !isReplayEligibleKind(a.Key.ActionKind) { - continue - } - recs = append(recs, rec{ - bucket: a.Bucket, - ruleHash: a.Key.RuleHash, - actionKind: a.Key.ActionKind, - ttlNs: int64(effectiveTTL(a)), - }) - } - if len(recs) == 0 { - return [32]byte{} - } - sort.Slice(recs, func(i, j int) bool { - if recs[i].bucket != recs[j].bucket { - return recs[i].bucket < recs[j].bucket - } - if recs[i].ruleHash != recs[j].ruleHash { - for k := 0; k < len(recs[i].ruleHash); k++ { - if recs[i].ruleHash[k] != recs[j].ruleHash[k] { - return recs[i].ruleHash[k] < recs[j].ruleHash[k] - } - } - } - return int(recs[i].actionKind) < int(recs[j].actionKind) - }) - h := sha256.New() - var scratch [16]byte - for _, r := range recs { - h.Write([]byte(r.bucket)) - h.Write([]byte{0}) - h.Write(r.ruleHash[:]) - binary.LittleEndian.PutUint64(scratch[:8], uint64(r.actionKind)) - binary.LittleEndian.PutUint64(scratch[8:], uint64(r.ttlNs)) - h.Write(scratch[:]) - } - var out [32]byte - copy(out[:], h.Sum(nil)) - return out -} - -// effectiveTTL returns the per-action TTL the lifecycle engine clocks -// against. Walker-bound kinds (ExpirationDate, ExpiredDeleteMarker, -// NewerNoncurrent) return 0 — they don't participate in the replay -// sliding window so they don't contribute to MaxEffectiveTTL or the -// content hash. Phase 2 guarantees those kinds never reach here via -// checkSnapshotForUnsupported, but the switch stays exhaustive so a -// future kind addition is a compile-time prompt to decide. -func effectiveTTL(a *engine.CompiledAction) time.Duration { - if a == nil || a.Rule == nil { - return 0 - } - switch a.Key.ActionKind { - case s3lifecycle.ActionKindExpirationDays: - return s3lifecycle.DaysToDuration(a.Rule.ExpirationDays) - case s3lifecycle.ActionKindNoncurrentDays: - return s3lifecycle.DaysToDuration(a.Rule.NoncurrentVersionExpirationDays) - case s3lifecycle.ActionKindAbortMPU: - return s3lifecycle.DaysToDuration(a.Rule.AbortMPUDaysAfterInitiation) - } - return 0 -} - -// localMaxEffectiveTTL returns the largest effective TTL across active -// replay-eligible actions. Returns zero when snap is empty/nil — caller -// is responsible for routing through the empty-replay branch in that -// case. Phase 4 replaces this with engine.MaxEffectiveTTL. -func localMaxEffectiveTTL(snap *engine.Snapshot) time.Duration { - if snap == nil { - return 0 - } - var max time.Duration - for _, a := range snap.AllActions() { - if a == nil || !a.IsActive() { - continue - } - if !isReplayEligibleKind(a.Key.ActionKind) { - continue - } - if d := effectiveTTL(a); d > max { - max = d - } - } - return max -} diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go b/weed/s3api/s3lifecycle/dailyrun/replayability_test.go index daeaf70d3..b79ad840a 100644 --- a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/replayability_test.go @@ -15,10 +15,6 @@ func newSnapshotWith(t *testing.T, inputs []engine.CompileInput) *engine.Snapsho e := engine.New() e.Compile(inputs, engine.CompileOptions{}) snap := e.Snapshot() - // Mark every action active so isActive checks fire as expected; the - // production compile path also marks them active for replay-eligible - // kinds when prior.BootstrapComplete is true. Tests just need them - // visible to AllActions(). for _, a := range snap.AllActions() { snap.MarkActive(a.Key) } @@ -89,26 +85,18 @@ func TestCheckSnapshotForUnsupported_ExpiredDeleteMarkerRejected(t *testing.T) { } func TestCheckSnapshotForUnsupported_NonEventDrivenModeRejected(t *testing.T) { - // A replay-eligible action whose Mode isn't ModeEventDriven (e.g. - // promoted to ModeScanOnly by retention checks) is silently - // ignored by router.Route. Phase 2 must catch this loudly. Build a - // snapshot with one ExpirationDays action and mutate its Mode to - // ModeScanOnly directly — there's no production path that produces - // this without retention plumbing we don't have here, but the gate - // must still reject it. + // router.Route silently drops non-ModeEventDriven actions; gate + // must reject loudly. snap := newSnapshotWith(t, []engine.CompileInput{ {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDays(30)}}, }) - // Mutate the compiled action to ModeScanOnly. AllActions returns - // the live snapshot's actions; this is safe in a test where no - // concurrent worker is reading. for _, a := range snap.AllActions() { if a.Key.ActionKind == s3lifecycle.ActionKindExpirationDays { a.Mode = engine.ModeScanOnly } } err := checkSnapshotForUnsupported(snap) - require.NotNil(t, err, "ModeScanOnly on a replay-kind action must be rejected") + require.NotNil(t, err) assert.Equal(t, s3lifecycle.ActionKindExpirationDays, err.Kind) assert.Contains(t, err.Reason, "ModeEventDriven") } @@ -120,8 +108,6 @@ func TestIsUnsupportedRule_TypeCheck(t *testing.T) { assert.False(t, IsUnsupportedRule(assertNonNilError())) } -// assertNonNilError returns a plain non-nil error of a different type -// so IsUnsupportedRule's errors.As check has a negative case to refuse. func assertNonNilError() error { return errPlain } type plainErr struct{} @@ -129,48 +115,3 @@ type plainErr struct{} func (plainErr) Error() string { return "plain" } var errPlain = plainErr{} - -func TestLocalReplayContentHash_StableAcrossReorderings(t *testing.T) { - // Compile the same rules in two snapshots and verify the hash is - // identical — sort order in AllActions() is implementation detail - // and must not affect the cursor's content hash. - rules := []*s3lifecycle.Rule{ruleExpirationDays(30), ruleNoncurrentDays(7), ruleAbortMPU(7)} - snapA := newSnapshotWith(t, []engine.CompileInput{{Bucket: "b1", Rules: rules}}) - // Build snapB with the same rules in reverse order; hash must match. - rev := []*s3lifecycle.Rule{rules[2], rules[1], rules[0]} - snapB := newSnapshotWith(t, []engine.CompileInput{{Bucket: "b1", Rules: rev}}) - assert.Equal(t, localReplayContentHash(snapA), localReplayContentHash(snapB)) -} - -func TestLocalReplayContentHash_ChangesOnTTLEdit(t *testing.T) { - snap30 := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDays(30)}}, - }) - snap60 := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDays(60)}}, - }) - assert.NotEqual(t, localReplayContentHash(snap30), localReplayContentHash(snap60)) -} - -func TestLocalReplayContentHash_EmptyIsZero(t *testing.T) { - snap := newSnapshotWith(t, nil) - var zero [32]byte - assert.Equal(t, zero, localReplayContentHash(snap)) -} - -func TestLocalMaxEffectiveTTL_PicksLargest(t *testing.T) { - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ - ruleExpirationDays(7), - ruleNoncurrentDays(30), - ruleAbortMPU(14), - }}, - }) - got := localMaxEffectiveTTL(snap) - assert.Equal(t, s3lifecycle.DaysToDuration(30), got) -} - -func TestLocalMaxEffectiveTTL_EmptyReturnsZero(t *testing.T) { - snap := newSnapshotWith(t, nil) - assert.Equal(t, time.Duration(0), localMaxEffectiveTTL(snap)) -} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 879936ada..b684d7fba 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -19,17 +19,12 @@ import ( "golang.org/x/time/rate" ) -// LifecycleClient is the same RPC contract the streaming dispatcher -// uses. Replicated here to avoid an import cycle with dispatcher; both -// shapes target s3_lifecycle_pb.SeaweedS3LifecycleInternalClient and -// neither owns the protobuf interface. +// LifecycleClient mirrors dispatcher's contract; duplicated to avoid an +// import cycle. type LifecycleClient interface { LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) } -// Config bundles everything daily_run needs to make one shard-fan-out -// pass over the meta-log. Fields with zero values use documented -// defaults; required fields are checked at the top of Run. type Config struct { Shards []int BucketsPath string @@ -39,43 +34,27 @@ type Config struct { Persister CursorPersister Lister router.SiblingLister - // Workers caps how many shards run in parallel. Each shard owns its - // own meta-log subscription and rate limiter is shared across all - // of them, so the cap is about filer-side concurrency rather than - // throughput. Zero or negative → 1 (serial, the legacy default). + // Workers <= 0 -> 1 (serial). Workers int - // Limiter is the optional per-worker rate.Limiter shared across all - // shard goroutines on this worker. nil = no rate limit (the legacy - // "drain as fast as the filer can" behavior). Phase 3 wires the - // per-worker share from ClusterContext.Metadata into this field. + // nil -> no rate limit. Shared across all shard goroutines. Limiter *rate.Limiter - // ClientName / ClientID identify this worker on the meta-log - // subscription. ClientID 0 randomizes per-run. ClientName string - ClientID int32 + // 0 -> randomized per-run. + ClientID int32 - // Now overrides time.Now for tests. nil = production wall clock. + // nil -> time.Now. Now func() time.Time - // EventBudget caps how many meta-log events each shard's reader - // consumes before returning. Zero = unbounded (the run-until-now - // behavior — see "stop condition" below). Tests set a small value - // to bound deterministic runs. + // 0 -> unbounded. EventBudget int } // Run executes the daily replay for every shard in cfg.Shards -// concurrently. Returns the first non-nil error from any shard; other -// shards still run to completion so a transient filer error on one -// shard doesn't lose progress on the others. -// -// Phase 2 scope: replay only. A bucket whose compiled rules require -// walker-bound dispatch (ExpirationDate, ExpiredDeleteMarker, -// NewerNoncurrent) or any rule promoted to scan_only fails the whole -// run with an UnsupportedRuleError — the handler reports this so admin -// can either revert algorithm=streaming or wait for Phase 4. +// concurrently. Returns the first shard error; the rest log and run to +// completion so one shard's transient failure doesn't lose other shards' +// progress. func Run(ctx context.Context, cfg Config) error { if err := validate(cfg); err != nil { return err @@ -84,25 +63,15 @@ func Run(ctx context.Context, cfg Config) error { if now == nil { now = func() time.Time { return time.Now().UTC() } } - // Freeze "now" at the start of the run. Every shard, every match's - // DueTime comparison, and every cursor-floor calculation uses the - // same instant so a long-running shard doesn't see the boundary - // drift relative to a short-running peer. + // Freeze "now" so shards agree on the boundary. runNow := now() - // Refuse runs whose snapshot includes a walker-bound action kind - // or any rule the router won't route. Done once against an - // immutable snapshot — every shard reuses this exact value so a - // mid-execution Compile can't make shards disagree about the rule - // set or the hash. + // Capture once so a mid-run Compile can't make shards disagree. snap := cfg.Engine.Snapshot() if unsupported := checkSnapshotForUnsupported(snap); unsupported != nil { return unsupported } - // Concurrency cap. cfg.Workers controls how many shards run in - // parallel; the rate limiter (Phase 3) governs throughput. With - // Workers=1 (the legacy default) the 16 shards process serially. workers := cfg.Workers if workers <= 0 { workers = 1 @@ -128,8 +97,6 @@ func Run(ctx context.Context, cfg Config) error { } wg.Wait() close(errCh) - // Return the first error; the rest are logged at glog level so - // they're recoverable in the operational stream. var first error for err := range errCh { if first == nil { @@ -168,63 +135,50 @@ func validate(cfg Config) error { return nil } -// runShard executes the daily replay loop for a single shard. The -// algorithm is documented in DESIGN.md ("Algorithm" section). -// -// Phase 2 omissions vs. the full design: -// - No walker invocation on rule-change / cold-start / retention -// loss. The cursor is rewritten and the worker exits; Phase 4 -// wires the walker over engine.RecoveryView(snap) on these -// branches. -// - PromotedHash is always the empty hash. The partition-flip -// trigger is dormant until Phase 4. -// - All walker-bound action kinds and scan_only-promoted rules are -// refused at validate-time (see checkSnapshotForUnsupported), so -// replay only ever sees ExpirationDays / NoncurrentDays / AbortMPU. -// -// snap is the engine snapshot captured at the top of Run; reusing the -// same value across shards guarantees they observe identical rules and -// hashes. runNow is the frozen instant for due-time comparisons. +// runShard executes one daily-replay pass; see DESIGN.md for algorithm. +// Phase 2: no walker on rule-change / cold-start; PromotedHash trigger +// is dormant until Phase 4b wires real retention. +// checkSnapshotForUnsupported already rejected walker-bound and +// scan_only rules. func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int) error { persisted, found, err := cfg.Persister.Load(ctx, shardID) if err != nil { return fmt.Errorf("shard=%d: load cursor: %w", shardID, err) } - rsh := localReplayContentHash(snap) - maxTTL := localMaxEffectiveTTL(snap) + // engine.ReplayContentHash has a different byte layout than the + // Phase-2 local helper. First post-upgrade run mismatches every + // cursor and drops into the rule-change branch below — bounded + // one-time rewind to runNow - maxTTL, self-healing on save. + rsh := engine.ReplayContentHash(snap) + maxTTL := engine.MaxEffectiveTTL(snap) + // retentionWindow=maxTTL keeps promoted empty (no rule's TTL + // exceeds the max). Phase 4b plumbs real meta-log retention. + promoted := engine.PromotedHash(snap, maxTTL) - // Empty-replay sentinel: no replay-eligible active rules in the - // snapshot. We persist the hash so a future rule addition is - // detected as a content change. Use the hash rather than maxTTL - // here as the explicit "no replay state" signal — both fire on - // the same set in practice (action_kind.go only emits actions - // when their Days field is > 0), but the hash captures intent. if rsh == [32]byte{} { return cfg.Persister.Save(ctx, shardID, Cursor{ TsNs: 0, RuleSetHash: rsh, - PromotedHash: [32]byte{}, // Phase 2: always empty + PromotedHash: promoted, }) } - // Rule-change branch: hash mismatch (and a persisted cursor exists) - // rewinds to now - max_ttl. Phase 4 adds the walker call here. - if found && persisted.RuleSetHash != rsh { + // Recovery: rule-content edit (RuleSetHash mismatch) or partition + // flip (PromotedHash mismatch — dormant until real retention). + // Phase 4b adds the walker here; until then we rewind and let the + // sliding meta-log replay catch up. + if found && (persisted.RuleSetHash != rsh || persisted.PromotedHash != promoted) { next := Cursor{ - TsNs: runNow.Add(-maxTTL).UnixNano(), - RuleSetHash: rsh, - // PromotedHash stays empty in Phase 2. + TsNs: runNow.Add(-maxTTL).UnixNano(), + RuleSetHash: rsh, + PromotedHash: promoted, } return cfg.Persister.Save(ctx, shardID, next) } - // Cold start with no prior cursor: start scanning from now - max_ttl - // so a freshly-installed worker still expires already-due objects - // whose PUT events sit within meta-log retention. Phase 4 would - // invoke the walker here for the longer-than-retention case; Phase 2 - // trusts that retention >= max_ttl in the deployments this code is - // enabled on. + // Cold start: scan from now-maxTTL so already-due objects within + // meta-log retention still expire. startTsNs := persisted.TsNs floor := runNow.Add(-maxTTL).UnixNano() if startTsNs < floor { @@ -233,43 +187,22 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, snap, startTsNs) if drainErr != nil { - // Drain failed (transport, ctx cancel, or limiter shutdown): - // persist whatever progress we have so subsequent runs don't - // re-process the events we already dispatched, then propagate - // the error. Cursor save uses ctx — if ctx is already cancelled - // the save will fail too; that's fine, we still surface the - // underlying drain error. - _ = cfg.Persister.Save(ctx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh}) + _ = cfg.Persister.Save(ctx, shardID, Cursor{TsNs: lastOK, RuleSetHash: rsh, PromotedHash: promoted}) return fmt.Errorf("shard=%d: drain: %w", shardID, drainErr) } - // drainShardEvents stops advancing lastOK at the first event with a - // skipped (not-yet-due) match. Persisting that value means the next - // run resumes from before the skipped event and re-scans it — - // without that, future-DueTime matches would be lost. halted=true - // adds nothing here because lastOK already reflects the - // stuck-at-failure boundary. return cfg.Persister.Save(ctx, shardID, Cursor{ - TsNs: lastOK, - RuleSetHash: rsh, - // PromotedHash stays empty in Phase 2. + TsNs: lastOK, + RuleSetHash: rsh, + PromotedHash: promoted, }) } -// drainShardEvents subscribes to the meta-log starting at startTsNs, -// routes every event through router.Route, and dispatches each Match -// whose due_time is past runNow. Returns (cursorAdvanceTo, halted, error): -// - cursorAdvanceTo: the highest TsNs the persisted cursor may safely -// advance to. Equals the TsNs of the last event whose matches were -// ALL dispatched (DONE/NOOP_RESOLVED/SKIPPED_OBJECT_LOCK) AND every -// prior event was likewise fully processed. Once an event with a -// not-yet-due match is encountered, cursorAdvanceTo stops growing — -// so subsequent runs re-scan that event and any after it. -// - halted: true when an unresolved dispatch outcome (RETRY_LATER / -// BLOCKED / transport error after in-run retries) stopped the loop. -// - error: stream/setup failure or context cancellation; caller -// persists cursorAdvanceTo and propagates the error so the run is -// marked as interrupted rather than successful. +// drainShardEvents subscribes to the meta-log from startTsNs and +// dispatches matches whose due_time is past runNow. cursorAdvanceTo +// stops growing at the first event with a not-yet-due match so that +// event is re-scanned in a later run. halted=true marks an unresolved +// dispatch outcome. func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID int, snap *engine.Snapshot, startTsNs int64) (int64, bool, error) { clientName := cfg.ClientName if clientName == "" { @@ -281,7 +214,6 @@ func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID } runUpTo := runNow.UnixNano() if startTsNs >= runUpTo { - // Nothing to do — cursor already at or past the run boundary. return startTsNs, false, nil } @@ -303,10 +235,6 @@ func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID }() cursorAdvanceTo := startTsNs - // stuck flips to true at the first event with a skipped (not-yet-due) - // match. From that event onward, cursorAdvanceTo no longer rises — - // future runs must re-scan everything from cursorAdvanceTo + 1 onward - // so the future-due matches get re-evaluated when they age in. stuck := false halted := false @@ -314,9 +242,6 @@ drain: for { select { case <-ctx.Done(): - // Parent context cancellation (worker shutdown, MaxRuntime). - // Return whatever progress was made so far and propagate the - // error so the caller marks the run as interrupted. cancelReader() <-readerDone return cursorAdvanceTo, true, ctx.Err() @@ -328,19 +253,12 @@ drain: continue } if ev.TsNs > runUpTo { - // Reached the run boundary: events past now belong to - // tomorrow's pass. Cancel the reader so it doesn't keep - // pulling live events, then exit. cancelReader() break drain } matches := router.Route(ctx, snap, ev, runNow, cfg.Lister) eventSkipped, eventHalted, eventErr := processMatches(ctx, cfg, runNow, ev, matches) if eventErr != nil { - // A non-dispatch error (e.g. limiter wait cancelled by - // shutdown). Cursor stays at the last fully-processed - // event; surface the error so the caller treats it as - // an interrupt. cancelReader() <-readerDone return cursorAdvanceTo, true, eventErr @@ -350,11 +268,6 @@ drain: break drain } if eventSkipped { - // First event with a not-yet-due match. Don't advance - // the cursor past it; keep processing later events to - // dispatch any due ones, but the persisted cursor stays - // at the previous cursorAdvanceTo so this event is - // re-scanned tomorrow. stuck = true continue } @@ -364,7 +277,6 @@ drain: } } - // Wait for reader to drain so its goroutine doesn't outlive us. cancelReader() if rerr := <-readerDone; rerr != nil && !errors.Is(rerr, context.Canceled) { glog.V(2).Infof("daily_run shard=%d: reader returned: %v", shardID, rerr) @@ -372,29 +284,11 @@ drain: return cursorAdvanceTo, halted, nil } -// processMatches dispatches every match emitted from one event. Returns -// (skippedAny, halted, error): -// - skippedAny=true if at least one match had a DueTime past runNow. -// The caller must NOT advance the persisted cursor past this event -// so the skipped match gets re-scanned in a later run. -// - halted=true on an unresolved outcome (RETRY_LATER / BLOCKED / -// transport error after in-run retries). Caller exits the drain -// loop. -// - error: propagated from limiter.Wait when ctx is cancelled. -// -// A future-DueTime match is silently skipped — it does NOT terminate -// the loop or get cached as "done for this rule." A single event can -// produce multiple matches for the same ActionKey against different -// objects (routePointerTransitionExpand emits one match per noncurrent -// sibling, each with its own SuccessorModTime derived from a different -// demoting event), so a not-yet-due sibling must never gate a sibling -// that's already past its DueTime. Without per-object state the only -// safe behavior is per-match independence; the cursor-advance gate -// upstream handles re-scanning the skipped events tomorrow. -// -// runNow is the frozen run-start instant. Using it (rather than a -// fresh now() per call) keeps the boundary stable across all matches -// in this run. +// processMatches dispatches matches from one event. A future-DueTime +// match is silently skipped (per-match, not per-rule): one event can +// emit multiple matches for the same ActionKey across noncurrent +// siblings, so a not-yet-due sibling must not gate a due one. The +// cursor-advance gate upstream re-scans skipped events tomorrow. func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reader.Event, matches []router.Match) (skippedAny, halted bool, err error) { for _, m := range matches { if m.DueTime.After(runNow) { @@ -410,8 +304,6 @@ func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reade } outcome, dispatchErr := dispatchWithRetry(ctx, cfg.Client, m) if dispatchErr != nil { - // Exhausted in-run transport retries: halt; tomorrow retries - // from the same cursor. glog.V(1).Infof("daily_run: transport error on %s/%s %s: %v", m.Bucket, m.ObjectKey, m.Key.ActionKind, dispatchErr) return skippedAny, true, nil @@ -420,9 +312,6 @@ func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reade case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK: - // Cursor advances. The dispatch's own metric (in the RPC - // server) records the outcome; the daily-run side stays - // metric-quiet to avoid double-counting. case s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED: glog.V(1).Infof("daily_run: %s on %s/%s %s", @@ -432,7 +321,7 @@ func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reade glog.V(1).Infof("daily_run: unknown outcome %v on %s/%s", outcome, m.Bucket, m.ObjectKey) return skippedAny, true, nil } - _ = ev // ev kept available for future per-event logging + _ = ev } return skippedAny, false, nil }