Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 13:30:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(s3/lifecycle): steady-state walker + drop unsupported-rule gate

Browse Source 
Adds the second walker invocation in runShard. After the recovery
check passes, runShard derives the walk view via snap.RulesForShard
(using the same retentionWindow PromotedHash used, so the partition
is consistent) and runs the walker over it. The view holds
walker-bound action kinds (ExpirationDate / ExpiredDeleteMarker /
NewerNoncurrent) plus any replay-eligible rules promoted to walk by
retention shortage; an empty view skips the call so non-versioned,
replay-only deployments don't pay an O(N) bucket walk per run.

With the walker now servicing every rule kind, checkSnapshotForUnsupported
and its UnsupportedRuleError type are obsolete. router.Route gates
replay on Mode == ModeEventDriven, so walker-bound and scan_only
rules are silently dropped by replay and picked up by the walker
instead — no double-dispatch. Drop the gate, delete replayability.go
+ replayability_test.go, and remove the handler's redundant
IsUnsupportedRule branch.
This commit is contained in:
Chris Lu
2026-05-11 23:42:36 -07:00
parent 20f2b71a4a
commit 01d0de6ede
4 changed files with 20 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3lifecycle/dailyrun/replayability.go
-69
View File
@@ -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
}
weed/s3api/s3lifecycle/dailyrun/replayability_test.go
-117
View File
@@ -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{}
weed/s3api/s3lifecycle/dailyrun/run.go
+20 -7
View File
@@ -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
weed/worker/tasks/s3_lifecycle/handler.go
-6
View File
@@ -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