From d2b9f53edb7bedc8e102bd91905504833d811d23 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 11 May 2026 21:58:40 -0700 Subject: [PATCH] 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 ++- .../dailyrun/walker_recovery_test.go | 143 ++++++++++++++++++ 2 files changed, 165 insertions(+), 3 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 11badc600..14d6a8ba3 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/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, diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go new file mode 100644 index 000000000..aba8ed877 --- /dev/null +++ b/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.