feat(s3/lifecycle): swap daily_run to engine hash APIs (Phase 4a) (#9457)

* feat(s3/lifecycle): swap daily_run to engine hash APIs (Phase 4a)

Replace the local replay-content-hash / max-effective-TTL helpers in
dailyrun with the engine package's canonical versions (ReplayContentHash,
MaxEffectiveTTL, PromotedHash) that landed with the Phase 4 view surface.

Adds PromotedHash to the cursor's recovery triggers: a partition flip
(rule moving between replay and walk because retention shifted) now
fires the rule-change branch alongside RuleSetHash mismatch. The
retentionWindow is set to MaxEffectiveTTL today, which keeps the
promoted set empty and the trigger dormant; Phase 4b will plumb the
real meta-log retention boundary so true scan_only promotions are
detected.

Cursor schema is unchanged — PromotedHash was already persisted as
the zero hash in Phase 2.

* docs(s3/lifecycle): note the one-time cursor rewind on hash format change

gemini-code-assist flagged that swapping localReplayContentHash for
engine.ReplayContentHash changes the persisted RuleSetHash byte layout
(sort order + tagged-field encoding). Phase-2 cursors mismatch on first
post-upgrade run and drop into the rule-change branch.

Going with option 3 (document the intentional one-time rewind). The
rewind is bounded to runNow - maxTTL (not time-zero), self-healing on
the next save, and daily_replay is off by default so the affected
population is limited to early adopters of the algorithm flag. A
migration shim or a hash-compat layer would carry the legacy encoder
forever for one bounded re-scan; not worth it.

Comment in runShard makes the trade explicit so a future reader doesn't
hunt for the "why does my cursor rewind once after upgrade" mystery.

* chore(s3/lifecycle): trim verbose comments in dailyrun

Cut multi-paragraph headers and narration that just described what the
code does. Kept the small WHY notes (per-match skip vs per-rule, the
one-time post-upgrade cursor rewind, scan_only rejection rationale).
Same behavior, ~150 fewer lines of comment.

* fix(s3/lifecycle): persist PromotedHash on the successful runShard save

The comment-trim pass dropped the field alongside a "stays empty in
Phase 2" comment. Harmless today (promoted is always zero), but Phase 4b
turns promoted into a real value — and a save that writes zero would
make the next run falsely detect drift and rewind. Spotted by
gemini-code-assist on PR 9457.

Other save paths (recovery, drain-error) already persisted it; the
success path is the only one that was missing it. Now consistent.

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
-}

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))
-}

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
 }