feat(s3/lifecycle): wire Walker hook into runShard's recovery branch

Adds a Config.Walker callback that fires on rule-content edit /
partition flip BEFORE the cursor rewinds, so already-due objects across
the rewritten rule set get caught instead of waiting on meta-log
replay alone. The callback receives engine.RecoveryView(snap) and the
per-shard ID; nil disables it (Phase 4a behavior preserved).

Decoupling the wiring from the implementation: the handler-side
WalkerFunc that drives bootstrap.Walk via the filer is the follow-up
commit, and tests can stub the callback without standing up the full
filer/client/lister harness.

Tests pin: walker fires exactly once on hash mismatch, walker error
propagates and leaves the cursor unchanged, nil Walker is a no-op.

weed/s3api/s3lifecycle/dailyrun/run.go

@@ -25,6 +25,12 @@ type LifecycleClient interface {
 	LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
 }
 
+// WalkerFunc handles the per-shard bucket walk for a given engine view.
+// Phase 4b uses it on the recovery branch (rule-content edit / partition
+// flip) so already-due objects across the rewritten rule set get caught
+// before the cursor rewinds.
+type WalkerFunc func(ctx context.Context, view *engine.Snapshot, shardID int) error
+
 type Config struct {
 	Shards      []int
 	BucketsPath string
@@ -47,6 +53,14 @@ type Config struct {
 	// trigger dormant.
 	RetentionWindow time.Duration
 
+	// Walker is invoked on the recovery branch (rule-content edit or
+	// partition flip) before the cursor rewind. It receives the
+	// engine.RecoveryView so only the rules that need bulk re-evaluation
+	// are walked, and the per-shard ID so the implementation can filter
+	// entries. nil disables walker invocation entirely — the cursor
+	// still rewinds, matching Phase 4a behavior.
+	Walker WalkerFunc
+
 	ClientName string
 	// 0 -> randomized per-run.
 	ClientID int32
@@ -179,10 +193,15 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim
 	}
 
 	// 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.
+	// flip (PromotedHash mismatch). Walk the rewritten rule set so
+	// already-due objects fire before the cursor rewinds; then rewind
+	// and let the sliding meta-log replay catch up steady state.
 	if found && (persisted.RuleSetHash != rsh || persisted.PromotedHash != promoted) {
+		if cfg.Walker != nil {
+			if werr := cfg.Walker(ctx, engine.RecoveryView(snap), shardID); werr != nil {
+				return fmt.Errorf("shard=%d: recovery walk: %w", shardID, werr)
+			}
+		}
 		next := Cursor{
 			TsNs:         runNow.Add(-maxTTL).UnixNano(),
 			RuleSetHash:  rsh,

weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go

@@ -0,0 +1,143 @@
+package dailyrun
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// memPersister is a minimal in-memory CursorPersister. Phase 4b tests
+// drive runShard directly; the recovery-branch path returns before
+// drainShardEvents so the heavier filer/client/lister fakes aren't
+// needed here.
+type memPersister struct {
+	store map[int]Cursor
+}
+
+func newMemPersister() *memPersister { return &memPersister{store: map[int]Cursor{}} }
+
+func (p *memPersister) Load(_ context.Context, shardID int) (Cursor, bool, error) {
+	c, ok := p.store[shardID]
+	return c, ok, nil
+}
+
+func (p *memPersister) Save(_ context.Context, shardID int, c Cursor) error {
+	p.store[shardID] = c
+	return nil
+}
+
+func snapshotWithRule(t *testing.T, days int) *engine.Snapshot {
+	t.Helper()
+	e := engine.New()
+	e.Compile([]engine.CompileInput{
+		{Bucket: "b1", Rules: []*s3lifecycle.Rule{
+			{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days},
+		}},
+	}, engine.CompileOptions{})
+	snap := e.Snapshot()
+	for _, a := range snap.AllActions() {
+		snap.MarkActive(a.Key)
+	}
+	return snap
+}
+
+func TestRunShard_WalkerInvokedOnRecoveryBranch(t *testing.T) {
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	// Seed a persisted cursor whose RuleSetHash differs from snap's, so
+	// the rule-change branch fires.
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xAA
+	}
+	require.NoError(t, p.Save(context.Background(), 3, Cursor{TsNs: 1234, RuleSetHash: stale}))
+
+	var gotView *engine.Snapshot
+	var gotShard int
+	calls := 0
+	cfg := Config{
+		Persister: p,
+		Walker: func(_ context.Context, view *engine.Snapshot, shardID int) error {
+			calls++
+			gotView = view
+			gotShard = shardID
+			return nil
+		},
+	}
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3))
+
+	assert.Equal(t, 1, calls, "walker must fire exactly once on recovery")
+	require.NotNil(t, gotView, "walker received the RecoveryView")
+	assert.Equal(t, 3, gotShard)
+
+	// Cursor rewound to runNow - maxTTL with the new hashes persisted.
+	got, ok, err := p.Load(context.Background(), 3)
+	require.NoError(t, err)
+	require.True(t, ok)
+	expectedFloor := runNow.Add(-engine.MaxEffectiveTTL(snap)).UnixNano()
+	assert.Equal(t, expectedFloor, got.TsNs, "cursor must rewind to runNow - maxTTL after recovery walk")
+	assert.NotEqual(t, stale, got.RuleSetHash, "stale hash must be replaced")
+}
+
+func TestRunShard_NilWalkerOnRecoveryIsNoop(t *testing.T) {
+	// Phase 4a behavior: a nil Walker must not crash and must still
+	// rewind the cursor.
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xBB
+	}
+	require.NoError(t, p.Save(context.Background(), 0, Cursor{TsNs: 9999, RuleSetHash: stale}))
+
+	cfg := Config{Persister: p} // Walker nil
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0))
+
+	got, ok, err := p.Load(context.Background(), 0)
+	require.NoError(t, err)
+	require.True(t, ok)
+	assert.Equal(t, runNow.Add(-engine.MaxEffectiveTTL(snap)).UnixNano(), got.TsNs)
+}
+
+func TestRunShard_WalkerErrorPropagates(t *testing.T) {
+	// A walker failure must surface as the runShard error so the daily
+	// run treats the shard as interrupted, and must NOT advance the
+	// cursor (the seeded cursor stays).
+	snap := snapshotWithRule(t, 30)
+	p := newMemPersister()
+	var stale [32]byte
+	for i := range stale {
+		stale[i] = 0xCC
+	}
+	require.NoError(t, p.Save(context.Background(), 7, Cursor{TsNs: 42, RuleSetHash: stale}))
+
+	cfg := Config{
+		Persister: p,
+		Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error {
+			return errors.New("walker boom")
+		},
+	}
+	runNow := time.Unix(1_700_000_000, 0).UTC()
+	err := runShard(context.Background(), cfg, snap, runNow, 7)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "walker boom")
+
+	// Cursor untouched.
+	got, _, _ := p.Load(context.Background(), 7)
+	assert.Equal(t, int64(42), got.TsNs, "walker failure must leave cursor unchanged")
+	assert.Equal(t, stale, got.RuleSetHash)
+}
+
+// The "matching cursor doesn't invoke walker" case is implicit: the
+// walker call is inside the rule-change / partition-flip `if` and
+// can't be reached otherwise. Exercising it end-to-end requires the
+// full filer + lister + client harness; covered by the integration
+// tests once Phase 4b is wired into the handler.