diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability.go b/weed/s3api/s3lifecycle/dailyrun/replayability.go deleted file mode 100644 index 5f72b5962..000000000 --- a/weed/s3api/s3lifecycle/dailyrun/replayability.go +++ /dev/null @@ -1,69 +0,0 @@ -package dailyrun - -import ( - "errors" - "fmt" - - "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" - "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" -) - -// 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 - Reason string -} - -func (e *UnsupportedRuleError) Error() string { - return fmt.Sprintf("daily_replay: unsupported action kind %s on bucket %q: %s", e.Kind, e.Bucket, e.Reason) -} - -func IsUnsupportedRule(err error) bool { - var u *UnsupportedRuleError - return errors.As(err, &u) -} - -func isReplayEligibleKind(k s3lifecycle.ActionKind) bool { - switch k { - case s3lifecycle.ActionKindExpirationDays, - s3lifecycle.ActionKindNoncurrentDays, - s3lifecycle.ActionKindAbortMPU: - return true - } - return false -} - -// 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 - } - for _, a := range snap.AllActions() { - if a == nil || !a.IsActive() { - continue - } - if !isReplayEligibleKind(a.Key.ActionKind) { - return &UnsupportedRuleError{ - Bucket: a.Bucket, - Kind: a.Key.ActionKind, - Reason: "Phase 2 only routes ExpirationDays / NoncurrentDays / AbortMPU; ExpirationDate, ExpiredDeleteMarker, NewerNoncurrent land in Phase 4", - } - } - if a.Mode != engine.ModeEventDriven { - return &UnsupportedRuleError{ - Bucket: a.Bucket, - Kind: a.Key.ActionKind, - Reason: fmt.Sprintf("action is in Mode=%v (router.Route only dispatches ModeEventDriven); scan_only promotions land in Phase 4", a.Mode), - } - } - } - return nil -} diff --git a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go b/weed/s3api/s3lifecycle/dailyrun/replayability_test.go deleted file mode 100644 index b79ad840a..000000000 --- a/weed/s3api/s3lifecycle/dailyrun/replayability_test.go +++ /dev/null @@ -1,117 +0,0 @@ -package dailyrun - -import ( - "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" -) - -func newSnapshotWith(t *testing.T, inputs []engine.CompileInput) *engine.Snapshot { - t.Helper() - e := engine.New() - e.Compile(inputs, engine.CompileOptions{}) - snap := e.Snapshot() - for _, a := range snap.AllActions() { - snap.MarkActive(a.Key) - } - return snap -} - -func ruleExpirationDays(days int) *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days} -} - -func ruleExpirationDate(t time.Time) *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r-date", Status: s3lifecycle.StatusEnabled, ExpirationDate: t} -} - -func ruleNoncurrentDays(days int) *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r-nc", Status: s3lifecycle.StatusEnabled, NoncurrentVersionExpirationDays: days} -} - -func ruleAbortMPU(days int) *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r-mpu", Status: s3lifecycle.StatusEnabled, AbortMPUDaysAfterInitiation: days} -} - -func ruleNewerNoncurrent(n int) *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r-newer", Status: s3lifecycle.StatusEnabled, NewerNoncurrentVersions: n} -} - -func ruleExpiredDeleteMarker() *s3lifecycle.Rule { - return &s3lifecycle.Rule{ID: "r-edm", Status: s3lifecycle.StatusEnabled, ExpiredObjectDeleteMarker: true} -} - -func TestCheckSnapshotForUnsupported_AllReplayKindsAccepted(t *testing.T) { - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ - ruleExpirationDays(30), - ruleNoncurrentDays(7), - ruleAbortMPU(7), - }}, - }) - require.Nil(t, checkSnapshotForUnsupported(snap)) -} - -func TestCheckSnapshotForUnsupported_ExpirationDateRejected(t *testing.T) { - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDate(time.Now().Add(48 * time.Hour))}}, - }) - err := checkSnapshotForUnsupported(snap) - require.NotNil(t, err) - assert.Equal(t, s3lifecycle.ActionKindExpirationDate, err.Kind) - assert.Equal(t, "b1", err.Bucket) -} - -func TestCheckSnapshotForUnsupported_NewerNoncurrentRejected(t *testing.T) { - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleNewerNoncurrent(2)}}, - }) - err := checkSnapshotForUnsupported(snap) - require.NotNil(t, err) - assert.Equal(t, s3lifecycle.ActionKindNewerNoncurrent, err.Kind) -} - -func TestCheckSnapshotForUnsupported_ExpiredDeleteMarkerRejected(t *testing.T) { - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpiredDeleteMarker()}}, - }) - err := checkSnapshotForUnsupported(snap) - require.NotNil(t, err) - assert.Equal(t, s3lifecycle.ActionKindExpiredDeleteMarker, err.Kind) -} - -func TestCheckSnapshotForUnsupported_NonEventDrivenModeRejected(t *testing.T) { - // router.Route silently drops non-ModeEventDriven actions; gate - // must reject loudly. - snap := newSnapshotWith(t, []engine.CompileInput{ - {Bucket: "b1", Rules: []*s3lifecycle.Rule{ruleExpirationDays(30)}}, - }) - for _, a := range snap.AllActions() { - if a.Key.ActionKind == s3lifecycle.ActionKindExpirationDays { - a.Mode = engine.ModeScanOnly - } - } - err := checkSnapshotForUnsupported(snap) - require.NotNil(t, err) - assert.Equal(t, s3lifecycle.ActionKindExpirationDays, err.Kind) - assert.Contains(t, err.Reason, "ModeEventDriven") -} - -func TestIsUnsupportedRule_TypeCheck(t *testing.T) { - var u error = &UnsupportedRuleError{Bucket: "b", Kind: s3lifecycle.ActionKindExpirationDate, Reason: "x"} - assert.True(t, IsUnsupportedRule(u)) - assert.False(t, IsUnsupportedRule(nil)) - assert.False(t, IsUnsupportedRule(assertNonNilError())) -} - -func assertNonNilError() error { return errPlain } - -type plainErr struct{} - -func (plainErr) Error() string { return "plain" } - -var errPlain = plainErr{} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index 14d6a8ba3..56417be2e 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -89,9 +89,6 @@ func Run(ctx context.Context, cfg Config) error { // Capture once so a mid-run Compile can't make shards disagree. snap := cfg.Engine.Snapshot() - if unsupported := checkSnapshotForUnsupported(snap); unsupported != nil { - return unsupported - } workers := cfg.Workers if workers <= 0 { @@ -157,10 +154,12 @@ func validate(cfg Config) error { } // 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. +// Two walker invocations under cfg.Walker (when set): +// - recovery branch: RecoveryView, so already-due objects across the +// rewritten rule set fire before the cursor rewinds. +// - steady state: RulesForShard's walk view, so walker-bound and +// scan_only-promoted rules fire every day even when replay rules +// are unchanged. 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 { @@ -210,6 +209,20 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim return cfg.Persister.Save(ctx, shardID, next) } + // Steady-state walker for walker-bound and scan_only-promoted rules. + // RulesForShard splits the snapshot using the same retentionWindow + // PromotedHash used, so the walk view is exactly the partition the + // hash already accounted for. Empty walk view (no rules need walking + // today) skips the call so non-versioned, replay-only deployments + // don't pay an O(N) bucket-walk per run. + if cfg.Walker != nil { + if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { + if werr := cfg.Walker(ctx, walkView, shardID); werr != nil { + return fmt.Errorf("shard=%d: steady walk: %w", shardID, werr) + } + } + } + // Cold start: scan from now-maxTTL so already-due objects within // meta-log retention still expire. startTsNs := persisted.TsNs diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 7bd75707f..7e927e9db 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -357,12 +357,6 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe Walker: walker, ClientName: "worker-s3-lifecycle-daily", }) - if dailyrun.IsUnsupportedRule(runErr) { - // Surface the typed error verbatim so admin marks the run as - // failed with the user-facing reason in the activity log. - glog.Warningf("daily_replay: %v", runErr) - return runErr - } if runErr != nil { glog.Warningf("daily_replay: %v", runErr) return runErr