diff --git a/weed/s3api/s3lifecycle/DESIGN.md b/weed/s3api/s3lifecycle/DESIGN.md index c0cd9e434..315efed79 100644 --- a/weed/s3api/s3lifecycle/DESIGN.md +++ b/weed/s3api/s3lifecycle/DESIGN.md @@ -19,13 +19,16 @@ dailyrun.Run(ctx, cfg): rsh = engine.ReplayContentHash(snap) maxTTL = engine.MaxEffectiveTTL(snap) + cursors = load every shard's persisted cursor + walks = planWalks(cursors) → at most 2 shared walks (recovery, steady) + if rsh != [32]byte{}: // replay-eligible rules present globalStartTsNs = min over cfg.Shards of (persisted cursor or runNow - maxTTL) reader = subscribeMeta(ShardPredicate ∈ cfg.Shards, StartTsNs = globalStartTsNs) fanOut(reader.Events → shardEvents[shardID]) until ev.TsNs > runNow spawn one goroutine per shard: - runShard(ctx, cfg, snap, runNow, shardID, shardEvents[shardID]) + runShard(ctx, cfg, snap, runNow, shardID, shardEvents[shardID], walks) wait all teardown reader + fan-out @@ -37,7 +40,7 @@ runShard(ctx, cfg, snap, runNow, shardID, events): if rsh == [32]byte{}: // pure walker bucket if walkerDue and Walker: - cfg.Walker(walkView) // RulesForShard.walk + walkForShard(walkView) // shared walk, per-shard fallback lastWalkedNs = runNow.UnixNano() save cursor (TsNs=0, rsh, promoted, lastWalkedNs) return @@ -46,7 +49,7 @@ runShard(ctx, cfg, snap, runNow, shardID, events): mustWalkColdStart = !found if mustWalkRecovery or mustWalkColdStart: - cfg.Walker(engine.RecoveryView(snap)) // every rule, force-active + walkForShard(RecoveryView(snap)) // every rule, force-active walkedThisPass = true lastWalkedNs = runNow.UnixNano() if mustWalkRecovery: @@ -54,7 +57,7 @@ runShard(ctx, cfg, snap, runNow, shardID, events): return if Walker and !walkedThisPass and walkerDue: - cfg.Walker(walkView) // steady-state walker + walkForShard(walkView) // steady-state walker lastWalkedNs = runNow.UnixNano() startTsNs = !found ? runNow - maxTTL : persisted.TsNs // steady state honors cursor @@ -62,7 +65,7 @@ runShard(ctx, cfg, snap, runNow, shardID, events): save cursor (TsNs = lastOK, rsh, promoted, lastWalkedNs) ``` -Walker invocations — three call sites, distinct conditions: +Walker invocations — three call sites, distinct conditions. `cfg.Walker` takes a shard set, not a single shard: `planWalks` predicts each shard's branch from the pre-loaded cursors and runs one shared walk per group, so a pass lists each bucket at most twice (recovery group + steady group) instead of once per shard. `runShard` waits on its group's walk; a shard whose own freshly-loaded cursor disagrees with the plan falls back to a per-shard walk, so the plan can't under-cover a shard after a cursor is rewritten mid-pass. | Branch | View | Trigger | Throttle | |---|---|---|---| @@ -230,9 +233,10 @@ The lifecycle evaluator uses `ev.TsNs` for current-version rules and `entry.Nonc | `evaluate.go`, `due_at.go`, `rule_hash.go`, `tags.go` | Engine-side rule evaluation | | `reader/` | Meta-log subscribe; one subscription per dailyrun.Run pass | | `router/router.go` | Per-event rule evaluation | -| `bootstrap/walker.go` | Bucket walker with `RunForShard(view, shardID)` filter | +| `bootstrap/walker.go` | Bucket walker; shard-set entry filter | | `dispatcher/filer_persister.go` | Filer-backed cursor I/O | | `dailyrun/run.go` | Main pass orchestrator: subscription, fan-out, per-shard `runShard` | +| `dailyrun/walk_plan.go` | Groups shards into shared recovery/steady walks (one listing per group) | | `dailyrun/cursor.go` | Cursor type + filer JSON serializer | | `dailyrun/walker_dispatcher.go` | Adapter from walker to `LifecycleDelete` RPC | @@ -265,6 +269,6 @@ Worker config: Tracked as optimizations rather than blockers: 1. **Long-lived subscription across passes.** Today the subscription is rebuilt per `Run()`. Keeping it alive across passes would eliminate the 7s ctx-timeout per pass and the start/teardown overhead. Requires per-shard pending heap (events whose `DueTime > runNow` would be parked in-memory instead of replayed) and a hot-swappable snapshot for mid-pass config changes. Multi-day refactor; current model works. -2. **Bucket-coordinated walker.** Phase 4 has each shard walk the full bucket and filter by `ShardID(bucket, key)` — simple but 16× the listing cost. A per-bucket coordinator (the worker owning shard 0 for that bucket lists once, routes matches to other shards) would cut listing cost. Worth doing if listing becomes the bottleneck for very large buckets. +2. ~~Bucket-coordinated walker~~. Done: `planWalks` shares one listing across the recovery group and one across the steady group (issue #11791). Remaining headroom: directories could be pruned to rule prefixes, and per-shard views could partition work rather than fanning every entry to its shard's dispatcher. 3. **Per-bucket dispatch lag metric.** Currently only per-shard lag is exposed. Per-bucket would require a per-bucket cursor or a derived metric from `s3_lifecycle_dispatch_total{bucket,kind,outcome}`. Punted on cardinality concerns; revisit when an operator asks for it. 4. **Meta-log retention plumbing.** If the filer adds GC for `/topics/.system/log`, the `PromotedHash` partition flip needs to consume the filer's actual retention horizon (currently dormant because retention is effectively infinite). diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go index aa251f809..0a24d5c51 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go +++ b/weed/s3api/s3lifecycle/dailyrun/filer_list_func.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -31,40 +32,64 @@ func init() { listPageSize.Store(1024) } // MPU init records at .uploads/ with ExtMultipartObjectKey set // are emitted whole with IsMPUInit=true and DestKey carrying the // user's intended path. -func FilerListFunc(client filer_pb.SeaweedFilerClient, bucketsPath string) bootstrap.ListFunc { +// +// view supplies per-bucket state: a bucket that never had versioning +// enabled skips the .versions/ pre-pass entirely, and directories no +// active rule prefix can reach are pruned rather than listed. A nil +// view assumes the worst (versioned, unpruned), preserving the +// two-pass behavior. +func FilerListFunc(client filer_pb.SeaweedFilerClient, bucketsPath string, view *engine.Snapshot) bootstrap.ListFunc { return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error { if client == nil { return fmt.Errorf("FilerListFunc: nil client") } root := strings.TrimSuffix(bucketsPath, "/") + "/" + bucket - return walkBucketTree(ctx, client, root, root, start, cb) + versioned := view == nil || view.BucketVersioned(bucket) + var prefixes []string + if view != nil { + prefixes = view.BucketPrefixes(bucket) + if len(prefixes) == 0 { + return nil // no active rule can match anything in this bucket + } + } + return walkBucketTree(ctx, client, root, root, start, versioned, prefixes, cb) } } // walkBucketTree recurses through dir in two passes per level. Pass 1 // expands `.versions/` dirs (populating skipBare with the bare null- -// version keys that pass 2 must suppress). Pass 2 emits regular files -// and recurses into non-special subdirectories. +// version keys that pass 2 must suppress) and only runs when the bucket +// is versioned — on a never-versioned bucket it would re-list every +// directory and find nothing. Pass 2 emits regular files and recurses +// into non-special subdirectories. // -// The two-pass shape mirrors scheduler/bootstrap.go's walkBucketDir -// (see that file's comment for why `.versions/` has to be processed -// before its bare sibling). -func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot, start string, cb func(*bootstrap.Entry) error) error { +// `.versions/` dirs must be processed before their bare sibling so the +// walk sees whether the bare entry is the null version of the same +// logical object; hence the two passes on versioned buckets. +func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, bucketRoot, start string, versioned bool, prefixes []string, cb func(*bootstrap.Entry) error) error { skipBare := map[string]bool{} // Pass 1: .versions/ dirs only. - if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error { - if e == nil || e.Attributes == nil { - return nil + if versioned { + if err := listAll(ctx, client, dir, func(e *filer_pb.Entry) error { + if e == nil || e.Attributes == nil { + return nil + } + if !e.IsDirectory || !isVersionsDir(e) { + return nil + } + full := dir + "/" + e.Name + key := strings.TrimPrefix(full, bucketRoot+"/") + // The emitted logical key ("foo") is a string-prefix of the + // dir key ("foo.versions"), so dirInRuleScope is conservative + // here too. + if !dirInRuleScope(prefixes, key) { + return nil + } + return expandVersionsDir(ctx, client, bucketRoot, key, e, start, skipBare, versioned, prefixes, cb) + }); err != nil { + return err } - if !e.IsDirectory || !isVersionsDir(e) { - return nil - } - full := dir + "/" + e.Name - key := strings.TrimPrefix(full, bucketRoot+"/") - return expandVersionsDir(ctx, client, bucketRoot, key, e, start, skipBare, cb) - }); err != nil { - return err } // Pass 2: everything else. @@ -72,11 +97,19 @@ func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir if e == nil || e.Attributes == nil { return nil } - if e.IsDirectory && isVersionsDir(e) { - return nil - } full := dir + "/" + e.Name key := strings.TrimPrefix(full, bucketRoot+"/") + if e.IsDirectory && isVersionsDir(e) { + if versioned { + return nil // handled in pass 1 + } + // No versioning on this bucket: a *-suffixed user folder + // holds ordinary objects, not version files. + if !dirInRuleScope(prefixes, key) { + return nil + } + return walkBucketTree(ctx, client, full, bucketRoot, start, false, prefixes, cb) + } if e.IsDirectory { if isMPUInitDir(key, e) { if start != "" && key <= start { @@ -91,7 +124,12 @@ func walkBucketTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dir Size: int64(e.Attributes.FileSize), }) } - return walkBucketTree(ctx, client, full, bucketRoot, start, cb) + // .uploads is exempt: an AbortMPU rule's prefix applies to + // DestKey (the user's path), never to ".uploads/". + if key != s3_constants.MultipartUploadsFolder && !dirInRuleScope(prefixes, key) { + return nil + } + return walkBucketTree(ctx, client, full, bucketRoot, start, versioned, prefixes, cb) } if skipBare[key] { return nil @@ -135,7 +173,7 @@ type versionItem struct { // resume after a mid-expansion failure rewalks the whole sibling // group. Acceptable today because Phase 4b doesn't persist a // Checkpoint between runs (start is always "" via runShard). -func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, versionsKey string, versionsEntry *filer_pb.Entry, start string, skipBare map[string]bool, cb func(*bootstrap.Entry) error) error { +func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketRoot, versionsKey string, versionsEntry *filer_pb.Entry, start string, skipBare map[string]bool, versioned bool, prefixes []string, cb func(*bootstrap.Entry) error) error { logical := strings.TrimSuffix(versionsKey, s3_constants.VersionsFolder) if logical == "" { return nil @@ -160,7 +198,7 @@ func expandVersionsDir(ctx context.Context, client filer_pb.SeaweedFilerClient, // Coincidentally-named user folder, or an empty `.versions` // container. Treat it as a regular subdirectory so user-named // files inside still surface. - return walkBucketTree(ctx, client, versionsDir, bucketRoot, start, cb) + return walkBucketTree(ctx, client, versionsDir, bucketRoot, start, versioned, prefixes, cb) } if nullEntry, nullKey, explicit, ok := lookupNullVersion(ctx, client, bucketRoot, logical); ok { items = append(items, versionItem{ @@ -304,6 +342,22 @@ func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string } } +// dirInRuleScope reports whether any active rule prefix could match an +// object under dir key. Objects under dir/ carry keys starting "dir/", +// so a prefix p can match iff p is a prefix of "dir/" or "dir/" is a +// prefix of p. An empty prefixes list means "no constraint" (nil view). +func dirInRuleScope(prefixes []string, key string) bool { + if len(prefixes) == 0 { + return true + } + for _, p := range prefixes { + if p == "" || strings.HasPrefix(key+"/", p) || strings.HasPrefix(p, key+"/") { + return true + } + } + return false +} + func isVersionsDir(entry *filer_pb.Entry) bool { return entry.IsDirectory && strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) } diff --git a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go index 2e3083c12..aa9402435 100644 --- a/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/filer_list_func_test.go @@ -10,7 +10,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" @@ -55,8 +57,9 @@ func (s *fakeFilerStream) RecvMsg(any) error { return nil } type fakeFiler struct { filer_pb.SeaweedFilerClient - mu sync.Mutex - tree map[string][]*filer_pb.Entry + mu sync.Mutex + tree map[string][]*filer_pb.Entry + listCalls int } func (c *fakeFiler) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { @@ -73,6 +76,7 @@ func (c *fakeFiler) LookupDirectoryEntry(_ context.Context, in *filer_pb.LookupD func (c *fakeFiler) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { c.mu.Lock() defer c.mu.Unlock() + c.listCalls++ src := c.tree[in.Directory] // Mirror the filer: sort by name, honor StartFromFileName exclusive, // cap at Limit. listAll's pagination loop depends on these. @@ -117,7 +121,7 @@ func TestFilerListFunc_EmitsFlatFiles(t *testing.T) { client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ "/buckets/bkt": {file("a.txt", mtime, 10), file("b.txt", mtime, 20)}, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []string require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e.Path) @@ -135,7 +139,7 @@ func TestFilerListFunc_PropagatesTagsOnFlatFiles(t *testing.T) { client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ "/buckets/bkt": {tagged}, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got *bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = e @@ -152,7 +156,7 @@ func TestFilerListFunc_RecursesIntoSubdirs(t *testing.T) { "/buckets/bkt/logs": {dir("2026"), file("a.log", mtime, 5)}, "/buckets/bkt/logs/2026": {file("b.log", mtime, 7)}, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var paths []string require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { paths = append(paths, e.Path) @@ -179,7 +183,7 @@ func TestFilerListFunc_MPUInitEmitsWithDestKey(t *testing.T) { mpuDir, }, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -212,7 +216,7 @@ func TestFilerListFunc_MPUInitWithoutDestKeyIsSkipped(t *testing.T) { mpuDir, }, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var paths []string require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { paths = append(paths, e.Path) @@ -254,7 +258,7 @@ func TestFilerListFunc_VersionedExpansionMarksLatestByPointer(t *testing.T) { "/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v2")}, "/buckets/bkt/foo" + s3_constants.VersionsFolder: versions, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -291,7 +295,7 @@ func TestFilerListFunc_VersionedExpansionStalePointerFallsBackToNewestSibling(t "/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v_GHOST")}, "/buckets/bkt/foo" + s3_constants.VersionsFolder: versions, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -319,7 +323,7 @@ func TestFilerListFunc_VersionedExpansionNoPointerNewestSiblingWins(t *testing.T "/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "")}, "/buckets/bkt/foo" + s3_constants.VersionsFolder: versions, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -349,7 +353,7 @@ func TestFilerListFunc_VersionedExpansionExplicitNullIsLatestWhenPointerMissing( fileWithExt("v1", t1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}), }, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -396,7 +400,7 @@ func TestFilerListFunc_VersionedExpansionExplicitNullBeatsNewerSiblingWhenPointe fileWithExt("v1", tV1, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}), }, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []*bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = append(got, e) @@ -424,7 +428,7 @@ func TestFilerListFunc_VersionsDirWithoutMarkersRecursesAsRegular(t *testing.T) file("inner.txt", mtime, 1), }, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var paths []string require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { paths = append(paths, e.Path) @@ -445,7 +449,7 @@ func TestFilerListFunc_VersionedDeleteMarkerPropagates(t *testing.T) { "/buckets/bkt": {versionsDir("foo"+s3_constants.VersionsFolder, "v1")}, "/buckets/bkt/foo" + s3_constants.VersionsFolder: versions, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got *bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = e @@ -461,7 +465,7 @@ func TestFilerListFunc_HonorsStart(t *testing.T) { client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ "/buckets/bkt": {file("a", mtime, 1), file("b", mtime, 1), file("c", mtime, 1)}, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got []string require.NoError(t, listFn(context.Background(), "bkt", "a", func(e *bootstrap.Entry) error { got = append(got, e.Path) @@ -471,16 +475,88 @@ func TestFilerListFunc_HonorsStart(t *testing.T) { } func TestFilerListFunc_NilClient(t *testing.T) { - listFn := FilerListFunc(nil, "/buckets") + listFn := FilerListFunc(nil, "/buckets", nil) require.Error(t, listFn(context.Background(), "bkt", "", func(*bootstrap.Entry) error { return nil })) } +func snapshotForBucketRuleVersioned(t *testing.T, bucket string, days int, versioned bool) *engine.Snapshot { + return snapshotForBucketRules(t, bucket, versioned, + &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days}) +} + +func snapshotForBucketRules(t *testing.T, bucket string, versioned bool, rules ...*s3lifecycle.Rule) *engine.Snapshot { + t.Helper() + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: bucket, Versioned: versioned, Rules: rules}, + }, engine.CompileOptions{}) + snap := e.Snapshot() + for _, a := range snap.AllActions() { + snap.MarkActive(a.Key) + } + return snap +} + +func TestFilerListFunc_SkipsVersionsPrePassOnNonVersionedBucket(t *testing.T) { + // The .versions/ pre-pass costs one listing per directory. On a + // never-versioned bucket it can only find user folders, so the + // whole pass is wasted; per directory the walk should list once. + mtime := time.Now() + newFiler := func() *fakeFiler { + return &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": {dir("logs"), file("root.txt", mtime, 1)}, + "/buckets/bkt/logs": {file("a.log", mtime, 5)}, + }} + } + + unversioned := newFiler() + listFn := FilerListFunc(unversioned, "/buckets", snapshotForBucketRuleVersioned(t, "bkt", 7, false)) + var paths []string + require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + sort.Strings(paths) + assert.Equal(t, []string{"logs/a.log", "root.txt"}, paths) + assert.Equal(t, 2, unversioned.listCalls, "non-versioned bucket must list each dir once") + + versioned := newFiler() + listFn = FilerListFunc(versioned, "/buckets", snapshotForBucketRuleVersioned(t, "bkt", 7, true)) + paths = nil + require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + sort.Strings(paths) + assert.Equal(t, []string{"logs/a.log", "root.txt"}, paths) + assert.Equal(t, 4, versioned.listCalls, "versioned bucket still pays the .versions/ pre-pass per dir") +} + +func TestFilerListFunc_VersionsNamedUserDirOnNonVersionedBucket(t *testing.T) { + // On a never-versioned bucket a *-suffixed .versions folder is a + // coincidence: real user objects inside must still surface. + mtime := time.Now() + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": {versionsDir("user"+s3_constants.VersionsFolder, "")}, + "/buckets/bkt/user" + s3_constants.VersionsFolder: { + file("inner.txt", mtime, 1), + }, + }} + listFn := FilerListFunc(client, "/buckets", snapshotForBucketRuleVersioned(t, "bkt", 7, false)) + var paths []string + require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + assert.Equal(t, []string{"user" + s3_constants.VersionsFolder + "/inner.txt"}, paths) +} + func TestFilerListFunc_AttributesPropagate(t *testing.T) { mtime := time.Date(2026, 5, 11, 12, 0, 0, 1234, time.UTC) client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ "/buckets/bkt": {file("obj", mtime, 4096)}, }} - listFn := FilerListFunc(client, "/buckets") + listFn := FilerListFunc(client, "/buckets", nil) var got *bootstrap.Entry require.NoError(t, listFn(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { got = e @@ -493,3 +569,127 @@ func TestFilerListFunc_AttributesPropagate(t *testing.T) { assert.Equal(t, int64(4096), got.Size) assert.True(t, got.IsLatest) } + +func TestDirInRuleScope(t *testing.T) { + cases := []struct { + prefixes []string + key string + want bool + }{ + {nil, "any", true}, // nil view: no constraint + {[]string{""}, "any", true}, // "" prefix matches all + {[]string{"logs/"}, "logs", true}, // dir holds the prefix + {[]string{"logs"}, "logs", true}, // exact + {[]string{"logs"}, "logs2", true}, // "logs" can match "logs2/x" + {[]string{"logs/"}, "logs2", false}, // "logs2/x" can't match "logs/" + {[]string{"tmp/deep/"}, "tmp", true}, // prefix descends into dir + {[]string{"tmp/deep/"}, "tmp/deep", true}, + {[]string{"tmp/deep/"}, "tmp/dee", false}, + {[]string{"tmp/deep/"}, "tmp/shallow", false}, + {[]string{"a/", "b/"}, "b", true}, + {[]string{"a/", "b/"}, "c", false}, + } + for _, tc := range cases { + assert.Equal(t, tc.want, dirInRuleScope(tc.prefixes, tc.key), "prefixes=%v key=%q", tc.prefixes, tc.key) + } +} + +func TestFilerListFunc_PrunesDirsOutsideRulePrefixes(t *testing.T) { + // Dirs no active rule prefix can reach are never listed. Files are + // always emitted (prefix matching is walkEntry's job, downstream). + mtime := time.Now() + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": { + dir("logs"), dir("tmp"), dir("other"), file("root.txt", mtime, 1), + }, + "/buckets/bkt/logs": {file("a.log", mtime, 1)}, + "/buckets/bkt/tmp": {dir("deep"), dir("shallow")}, + "/buckets/bkt/tmp/deep": {file("d.txt", mtime, 1)}, + "/buckets/bkt/other": {file("o.txt", mtime, 1)}, + }} + view := snapshotForBucketRules(t, "bkt", false, + &s3lifecycle.Rule{ID: "r1", Status: s3lifecycle.StatusEnabled, Prefix: "logs/", ExpirationDays: 1}, + &s3lifecycle.Rule{ID: "r2", Status: s3lifecycle.StatusEnabled, Prefix: "tmp/deep/", ExpirationDays: 1}, + ) + var paths []string + require.NoError(t, FilerListFunc(client, "/buckets", view)(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + sort.Strings(paths) + assert.Equal(t, []string{"logs/a.log", "root.txt", "tmp/deep/d.txt"}, paths) + assert.Equal(t, 4, client.listCalls, "bkt+logs+tmp+tmp/deep listed; other and tmp/shallow pruned") +} + +func TestFilerListFunc_NoActiveActionsSkipsWalk(t *testing.T) { + // Every action inactive (e.g. bootstrap never completed): MatchPath + // could match nothing, so the whole bucket listing is wasted work. + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": {file("a", time.Now(), 1)}, + }} + e := engine.New() + e.Compile([]engine.CompileInput{{Bucket: "bkt", Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}, + }}}, engine.CompileOptions{}) + var paths []string + require.NoError(t, FilerListFunc(client, "/buckets", e.Snapshot())(context.Background(), "bkt", "", func(en *bootstrap.Entry) error { + paths = append(paths, en.Path) + return nil + })) + assert.Empty(t, paths) + assert.Equal(t, 0, client.listCalls) +} + +func TestFilerListFunc_UploadsDirExemptFromPruning(t *testing.T) { + // AbortMPU rules match DestKey (the user's key), never + // ".uploads/", so the uploads container stays in scope even + // when all rule prefixes point elsewhere. + mtime := time.Now() + mpuDir := dir("up-1") + mpuDir.Extended = map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte("logs/big.bin"), + } + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": {dir(s3_constants.MultipartUploadsFolder), dir("other")}, + "/buckets/bkt/" + s3_constants.MultipartUploadsFolder: {mpuDir}, + "/buckets/bkt/other": {file("o.txt", mtime, 1)}, + }} + view := snapshotForBucketRules(t, "bkt", false, + &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, Prefix: "logs/", AbortMPUDaysAfterInitiation: 3}) + var paths []string + require.NoError(t, FilerListFunc(client, "/buckets", view)(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + assert.Equal(t, []string{s3_constants.MultipartUploadsFolder + "/up-1"}, paths) +} + +func TestFilerListFunc_PrunesVersionsDirsOutsidePrefixes(t *testing.T) { + // Versioned bucket: stray.versions is out of scope for prefix + // "data/" — its logical key "stray" can't match — so it is never + // listed. data.versions still expands to logical "data". + mtime := time.Now() + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": { + versionsDir("data"+s3_constants.VersionsFolder, "v1"), + versionsDir("stray"+s3_constants.VersionsFolder, "v1"), + dir("other"), + }, + "/buckets/bkt/data" + s3_constants.VersionsFolder: { + fileWithExt("v1", mtime, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}), + }, + "/buckets/bkt/stray" + s3_constants.VersionsFolder: { + fileWithExt("v1", mtime, 1, map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v1")}), + }, + "/buckets/bkt/other": {file("x.txt", mtime, 1)}, + }} + view := snapshotForBucketRules(t, "bkt", true, + &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, Prefix: "data", ExpirationDays: 1}) + var paths []string + require.NoError(t, FilerListFunc(client, "/buckets", view)(context.Background(), "bkt", "", func(e *bootstrap.Entry) error { + paths = append(paths, e.Path) + return nil + })) + assert.Equal(t, []string{"data"}, paths) + assert.Equal(t, 3, client.listCalls, "bkt twice + data.versions; stray.versions and other pruned") +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index f459b1bdd..1488db6d5 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -27,11 +27,10 @@ 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 +// WalkerFunc runs one bucket walk covering every shard in shards under +// the given engine view, so a whole pass costs at most a couple of +// bucket listings rather than one listing per shard. +type WalkerFunc func(ctx context.Context, view *engine.Snapshot, shards []int) error type Config struct { Shards []int @@ -58,9 +57,9 @@ type Config struct { // 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. + // are walked, and the shard set the walk must cover. nil disables + // walker invocation entirely — the cursor still rewinds, matching + // Phase 4a behavior. Walker WalkerFunc // WalkerInterval throttles the steady-state and empty-replay walker @@ -132,6 +131,21 @@ func Run(ctx context.Context, cfg Config) error { // compatibility but otherwise inert. _ = cfg.Workers + retentionWindow := cfg.RetentionWindow + if retentionWindow <= 0 { + retentionWindow = maxTTL + } + + // One cursor scan feeds both the subscription floor and the walk + // plan. runShard re-loads its own cursor anyway, so a cursor + // rewritten between the two reads can't confuse it. + var cursors map[int]cursorLoad + if rsh != [32]byte{} || cfg.Walker != nil { + cursors = loadShardCursors(ctx, cfg) + } + promoted := engine.PromotedHash(snap, retentionWindow) + walks := planWalks(ctx, cfg, snap, runNow, cursors, rsh, promoted, retentionWindow) + // rsh==[32]byte{} means no replay-eligible rules — runShard's // walker-only branch fires and saves; no subscription needed. var ( @@ -151,7 +165,7 @@ func Run(ctx context.Context, cfg Config) error { // events on shards whose cursor is older than the floor — // exactly the case where a rule's TTL equals maxTTL and an // older event still has DueTime <= runNow. - globalStartTsNs = computeGlobalStartTsNs(ctx, cfg, runNow, maxTTL) + globalStartTsNs = computeGlobalStartTsNs(cursors, cfg, runNow, maxTTL) shardEvents, readerDone, fanoutDone, cancelRead = startSharedSubscription(ctx, cfg, runNow, globalStartTsNs, &stoppedOnPurpose) defer cancelRead() } @@ -167,7 +181,7 @@ func Run(ctx context.Context, cfg Config) error { if shardEvents != nil { ch = shardEvents[sh] } - if err := runShard(ctx, cfg, snap, runNow, sh, ch); err != nil { + if err := runShard(ctx, cfg, snap, runNow, sh, ch, walks); err != nil { errCh <- err } // A halted dispatch returns runShard mid-stream. Keep @@ -182,6 +196,18 @@ func Run(ctx context.Context, cfg Config) error { wg.Wait() close(errCh) + // Join every planned walk. A shard that exits early (cursor load + // failure, ctx cancel) never awaits pw.done, but planWalks already + // started the goroutine — Run must not return while it still lists + // or dispatches against connections the caller is about to close. + seenWalks := map[*pendingWalk]bool{} + for _, pw := range walks { + if !seenWalks[pw] { + seenWalks[pw] = true + <-pw.done + } + } + // Tear down the shared subscription. cancelRead unblocks both the // reader's gRPC stream and the fan-out's send loop; we wait on both // so their goroutines don't outlive Run. @@ -272,20 +298,20 @@ func summarizeShardCursorLag(ctx context.Context, cfg Config, runNow time.Time) return lagStr + " " + ageStr } -// computeGlobalStartTsNs scans every shard's persisted cursor and -// returns the minimum startTsNs the shared subscription must cover. -// For each shard the start point is the persisted cursor (steady state) -// or runNow - maxTTL (cold start, when no cursor exists). Load errors -// downgrade to the cold-start floor for that shard — failing closed -// would be worse than re-scanning maxTTL of events. -func computeGlobalStartTsNs(ctx context.Context, cfg Config, runNow time.Time, maxTTL time.Duration) int64 { +// computeGlobalStartTsNs returns the minimum startTsNs the shared +// subscription must cover. For each shard the start point is the +// persisted cursor (steady state) or runNow - maxTTL (cold start, when +// no cursor exists). Load errors downgrade to the cold-start floor for +// that shard — failing closed would be worse than re-scanning maxTTL +// of events. +func computeGlobalStartTsNs(cursors map[int]cursorLoad, cfg Config, runNow time.Time, maxTTL time.Duration) int64 { floor := runNow.Add(-maxTTL).UnixNano() min := int64(0) first := true for _, sh := range cfg.Shards { var start int64 - if persisted, found, err := cfg.Persister.Load(ctx, sh); err == nil && found { - start = persisted.TsNs + if cl := cursors[sh]; cl.err == nil && cl.found { + start = cl.cursor.TsNs } else { start = floor } @@ -452,7 +478,9 @@ func validate(cfg Config) error { } // runShard executes one daily-replay pass; see DESIGN.md for algorithm. -// Three walker invocation paths under cfg.Walker (when set): +// Three walker invocation paths under cfg.Walker (when set); each waits +// on the pass's shared walk via walkForShard rather than listing the +// bucket itself: // - cold start / recovery: RecoveryView. Unconditional — these are // bounded events (first run for a shard, or rule-content edit / partition // flip) and the walker must run once when the trigger condition is met. @@ -462,7 +490,7 @@ func validate(cfg Config) error { // - empty replay: same throttling — buckets with only walker-bound rules // use the walk view, and a too-fast Run() driver would otherwise burn // the filer with a full subtree scan per tick. -func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int, events <-chan *reader.Event) error { +func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int, events <-chan *reader.Event, walks map[int]*pendingWalk) error { shardLabel := strconv.Itoa(shardID) shardStart := time.Now() defer func() { @@ -522,7 +550,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim // the bug TestLifecycleExpirationDateInThePast caught. if cfg.Walker != nil && walkerDue(lastWalkedNs, runNow, cfg.WalkerInterval) { if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { - if werr := cfg.Walker(ctx, walkView, shardID); werr != nil { + if werr := walkForShard(ctx, cfg.Walker, walks, shardID, false, walkView); werr != nil { return fmt.Errorf("shard=%d: steady walk (empty replay): %w", shardID, werr) } lastWalkedNs = runNow.UnixNano() @@ -551,7 +579,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim mustWalkColdStart := !found if mustWalkRecovery || mustWalkColdStart { if cfg.Walker != nil { - if werr := cfg.Walker(ctx, engine.RecoveryView(snap), shardID); werr != nil { + if werr := walkForShard(ctx, cfg.Walker, walks, shardID, true, engine.RecoveryView(snap)); werr != nil { return fmt.Errorf("shard=%d: recovery walk: %w", shardID, werr) } lastWalkedNs = runNow.UnixNano() @@ -586,7 +614,7 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim // where this is a normal pass that didn't trigger cold-start. if cfg.Walker != nil && !walkedThisPass && walkerDue(lastWalkedNs, runNow, cfg.WalkerInterval) { if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { - if werr := cfg.Walker(ctx, walkView, shardID); werr != nil { + if werr := walkForShard(ctx, cfg.Walker, walks, shardID, false, walkView); werr != nil { return fmt.Errorf("shard=%d: steady walk: %w", shardID, werr) } lastWalkedNs = runNow.UnixNano() diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go index bfd63afae..2a0eac358 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go @@ -12,14 +12,14 @@ import ( ) // WalkBuckets runs bootstrap.Walk for each bucket using a per-bucket -// ListFunc wrapped to drop entries whose ShardID doesn't belong to -// shardID. Returns the first bucket error; remaining buckets log and -// continue so one bucket's filer error doesn't kill the whole walk. +// ListFunc wrapped to drop entries whose ShardID isn't in shards. +// Returns the first bucket error; remaining buckets log and continue +// so one bucket's filer error doesn't kill the whole walk. // // Composable shape: callers (the worker handler) supply a real // filer-backed ListFunc and a real Dispatcher (WalkerDispatcher). // Tests pass bootstrap.EntryCallback and a stub Dispatcher. -func WalkBuckets(ctx context.Context, view *engine.Snapshot, shardID int, buckets []string, list bootstrap.ListFunc, dispatch bootstrap.Dispatcher) error { +func WalkBuckets(ctx context.Context, view *engine.Snapshot, shards []int, buckets []string, list bootstrap.ListFunc, dispatch bootstrap.Dispatcher) error { if view == nil { return errors.New("WalkBuckets: nil view") } @@ -29,7 +29,11 @@ func WalkBuckets(ctx context.Context, view *engine.Snapshot, shardID int, bucket if dispatch == nil { return errors.New("WalkBuckets: nil dispatch") } - filtered := perShardListFunc(list, shardID) + shardSet := make(map[int]bool, len(shards)) + for _, sh := range shards { + shardSet[sh] = true + } + filtered := perShardListFunc(list, shardSet) var firstErr error for _, b := range buckets { if err := ctx.Err(); err != nil { @@ -47,19 +51,19 @@ func WalkBuckets(ctx context.Context, view *engine.Snapshot, shardID int, bucket } // perShardListFunc wraps an inner ListFunc so only entries whose -// logical-key shard matches shardID reach the callback. The walker +// logical-key shard is in shardSet reach the callback. The walker // emits one entry per logical object — versioned siblings share the // same logical key so they're either all in-shard or all out — and // MPU init records carry the user's destination in DestKey, so the // shard predicate must use DestKey there to match what the dispatcher // will send. -func perShardListFunc(inner bootstrap.ListFunc, shardID int) bootstrap.ListFunc { +func perShardListFunc(inner bootstrap.ListFunc, shardSet map[int]bool) bootstrap.ListFunc { return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error { return inner(ctx, bucket, start, func(e *bootstrap.Entry) error { if e == nil { return nil } - if entryShardID(bucket, e) != shardID { + if !shardSet[entryShardID(bucket, e)] { return nil } return cb(e) diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go index 4f3f5b75c..b311b4f5f 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go @@ -104,7 +104,7 @@ func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) { snap := snapshotForBucketRule(t, bucket, 7) // 30-day-old objects expire d := &recordingDispatcher{} - err := WalkBuckets(context.Background(), snap, targetShard, []string{bucket}, + err := WalkBuckets(context.Background(), snap, []int{targetShard}, []string{bucket}, bootstrap.EntryCallback(all), d) require.NoError(t, err) @@ -124,9 +124,9 @@ func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) { } func TestWalkBuckets_NilGuards(t *testing.T) { - require.Error(t, WalkBuckets(context.Background(), nil, 0, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{})) - require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, nil, &recordingDispatcher{})) - require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), 0, nil, bootstrap.EntryCallback(nil), nil)) + require.Error(t, WalkBuckets(context.Background(), nil, []int{0}, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{})) + require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, nil, &recordingDispatcher{})) + require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, bootstrap.EntryCallback(nil), nil)) } func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) { @@ -141,7 +141,7 @@ func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) { } snap := snapshotForBucketRule(t, "good", 7) - err := WalkBuckets(context.Background(), snap, 0, []string{"bad", "good"}, list, &recordingDispatcher{}) + err := WalkBuckets(context.Background(), snap, []int{0}, []string{"bad", "good"}, list, &recordingDispatcher{}) require.Error(t, err) assert.ErrorIs(t, err, listErr) } @@ -152,7 +152,7 @@ func TestWalkBuckets_HonorsContextCancellationBetweenBuckets(t *testing.T) { // short-circuit. ctx, cancel := context.WithCancel(context.Background()) cancel() - err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), 0, []string{"b"}, + err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), []int{0}, []string{"b"}, bootstrap.EntryCallback(nil), &recordingDispatcher{}) assert.ErrorIs(t, err, context.Canceled) } diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_plan.go b/weed/s3api/s3lifecycle/dailyrun/walk_plan.go new file mode 100644 index 000000000..dc51adfd9 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walk_plan.go @@ -0,0 +1,105 @@ +package dailyrun + +import ( + "context" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" +) + +// cursorLoad is the outcome of one shard's Persister.Load. +type cursorLoad struct { + cursor Cursor + found bool + err error +} + +func loadShardCursors(ctx context.Context, cfg Config) map[int]cursorLoad { + out := make(map[int]cursorLoad, len(cfg.Shards)) + for _, sh := range cfg.Shards { + c, found, err := cfg.Persister.Load(ctx, sh) + out[sh] = cursorLoad{cursor: c, found: found, err: err} + } + return out +} + +// pendingWalk is one shared bucket walk covering a group of shards. +// runShard goroutines wait on done instead of running their own listing. +type pendingWalk struct { + done chan struct{} + err error + recovery bool +} + +// planWalks groups the shards runShard will send through the recovery +// and steady-state walk branches — same cursor data, same hash and +// walkerDue checks — and starts one shared walk per group, so a pass +// lists each bucket at most twice instead of once per shard. +// +// The plan is only a prediction: runShard re-loads its own cursor and +// keeps the authoritative decision. A shard whose fresh read disagrees +// (a cursor rewritten by a concurrent pass between the two loads) +// falls back to a per-shard walk in walkForShard. +func planWalks(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, cursors map[int]cursorLoad, rsh, promoted [32]byte, retentionWindow time.Duration) map[int]*pendingWalk { + if cfg.Walker == nil || len(cfg.Shards) == 0 { + return nil + } + // RulesForShard ignores shardID today, so one view serves every + // shard. Revisit if it ever partitions rules per shard. + _, walkView := snap.RulesForShard(cfg.Shards[0], retentionWindow) + walkViewEmpty := walkView == nil || len(walkView.AllActions()) == 0 + + var recovery, steady []int + for _, sh := range cfg.Shards { + cl := cursors[sh] + if cl.err != nil { + continue + } + if rsh != [32]byte{} && (!cl.found || cl.cursor.RuleSetHash != rsh || cl.cursor.PromotedHash != promoted) { + recovery = append(recovery, sh) + continue + } + var lastWalkedNs int64 + if cl.found { + lastWalkedNs = cl.cursor.LastWalkedNs + } + if !walkViewEmpty && walkerDue(lastWalkedNs, runNow, cfg.WalkerInterval) { + steady = append(steady, sh) + } + } + + walks := make(map[int]*pendingWalk, len(recovery)+len(steady)) + start := func(shards []int, recovery bool, view *engine.Snapshot) { + pw := &pendingWalk{done: make(chan struct{}), recovery: recovery} + for _, sh := range shards { + walks[sh] = pw + } + go func() { + defer close(pw.done) + pw.err = cfg.Walker(ctx, view, shards) + }() + } + if len(recovery) > 0 { + start(recovery, true, engine.RecoveryView(snap)) + } + if len(steady) > 0 { + start(steady, false, walkView) + } + return walks +} + +// walkForShard waits on the shard's planned shared walk when one covers +// this branch (same recovery/steady kind), or runs a per-shard walk when +// the plan missed — a cursor drift between the plan load and runShard's +// own load — or no plan exists (direct runShard calls, tests). +func walkForShard(ctx context.Context, walker WalkerFunc, walks map[int]*pendingWalk, shardID int, recovery bool, view *engine.Snapshot) error { + if pw := walks[shardID]; pw != nil && pw.recovery == recovery { + select { + case <-pw.done: + return pw.err + case <-ctx.Done(): + return ctx.Err() + } + } + return walker(ctx, view, []int{shardID}) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_plan_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_plan_test.go new file mode 100644 index 000000000..7b3ffdb98 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walk_plan_test.go @@ -0,0 +1,232 @@ +package dailyrun + +import ( + "context" + "errors" + "sort" + "sync" + "sync/atomic" + "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" +) + +// Issue #11791: a pass used to invoke the walker once per shard, so the +// whole bucket tree was listed ShardCount times. The walker now gets +// the full shard set in at most two calls per pass — one for the +// recovery group, one for the steady group. +func TestRun_SharesBucketWalkAcrossShards(t *testing.T) { + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: "b1", Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, + ExpirationDate: time.Now().Add(24 * time.Hour)}, + }}, + }, engine.CompileOptions{}) + + shards := make([]int, 0, s3lifecycle.ShardCount) + for i := 0; i < s3lifecycle.ShardCount; i++ { + shards = append(shards, i) + } + + var mu sync.Mutex + var calls [][]int + p := newMemPersister() + cfg := Config{ + Shards: shards, + BucketsPath: "/buckets", + Engine: e, + FilerClient: stubFilerClient{}, + Client: stubLifecycleClient{}, + Persister: p, + Lister: stubSiblingLister{}, + Walker: func(_ context.Context, _ *engine.Snapshot, shards []int) error { + mu.Lock() + defer mu.Unlock() + calls = append(calls, append([]int{}, shards...)) + return nil + }, + } + runNow := time.Unix(1_700_000_000, 0).UTC() + cfg.Now = func() time.Time { return runNow } + require.NoError(t, Run(context.Background(), cfg)) + + require.LessOrEqual(t, len(calls), 2, "a pass must not list the bucket once per shard") + covered := map[int]bool{} + for _, c := range calls { + for _, sh := range c { + covered[sh] = true + } + } + for _, sh := range shards { + assert.True(t, covered[sh], "shard %d missed the shared walk", sh) + c, found, err := p.Load(context.Background(), sh) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, runNow.UnixNano(), c.LastWalkedNs, + "shard %d must credit the shared walk to its own cursor", sh) + } +} + +// A recovery group and a steady group in the same pass share one walk +// each: two walker calls total, each covering its own shard set and +// its own view. +func TestRun_RecoveryAndSteadyGroupsWalkSeparately(t *testing.T) { + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: "b1", Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, + ExpirationDays: 30, + ExpirationDate: time.Now().Add(24 * time.Hour)}, + }}, + }, engine.CompileOptions{}) + snap := e.Snapshot() + for _, a := range snap.AllActions() { + snap.MarkActive(a.Key) + } + + p := newMemPersister() + var stale [32]byte + for i := range stale { + stale[i] = 0xAA + } + // Shard 0's stale rule hash sends it through the recovery walk. + // Shards 1 and 2 have matching hashes and a zero walk anchor, so + // the steady-state branch takes them. + require.NoError(t, p.Save(context.Background(), 0, Cursor{TsNs: 42, RuleSetHash: stale})) + matching := Cursor{ + TsNs: 1234, + RuleSetHash: engine.ReplayContentHash(snap), + PromotedHash: engine.PromotedHash(snap, engine.MaxEffectiveTTL(snap)), + } + for _, sh := range []int{1, 2} { + require.NoError(t, p.Save(context.Background(), sh, matching)) + } + + var mu sync.Mutex + var calls [][]int + var actions []int + cfg := Config{ + Shards: []int{0, 1, 2}, + BucketsPath: "/buckets", + Engine: e, + FilerClient: &idleSubscribeClient{}, + Client: stubLifecycleClient{}, + Persister: p, + Lister: stubSiblingLister{}, + Walker: func(_ context.Context, view *engine.Snapshot, shards []int) error { + mu.Lock() + defer mu.Unlock() + calls = append(calls, append([]int{}, shards...)) + actions = append(actions, len(view.AllActions())) + return nil + }, + } + require.NoError(t, Run(context.Background(), cfg)) + + require.Len(t, calls, 2, "one shared walk per group: recovery + steady") + for i, c := range calls { + sort.Ints(c) + switch { + case len(c) == 1 && c[0] == 0: + assert.Equal(t, 2, actions[i], "recovery view carries every non-disabled action") + case len(c) == 2 && c[0] == 1 && c[1] == 2: + assert.Equal(t, 1, actions[i], "walk view carries only the walker-bound action") + default: + t.Fatalf("unexpected walk group %v", c) + } + } +} + +// A shared walk error must surface on every shard it covered, not just +// the one that happened to run the listing. +func TestRun_SharedWalkErrorFailsCoveredShards(t *testing.T) { + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: "b1", Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, + ExpirationDate: time.Now().Add(24 * time.Hour)}, + }}, + }, engine.CompileOptions{}) + + cfg := Config{ + Shards: []int{0, 1}, + BucketsPath: "/buckets", + Engine: e, + FilerClient: stubFilerClient{}, + Client: stubLifecycleClient{}, + Persister: newMemPersister(), + Lister: stubSiblingLister{}, + Walker: func(_ context.Context, _ *engine.Snapshot, _ []int) error { + return errors.New("walk boom") + }, + } + err := Run(context.Background(), cfg) + require.Error(t, err) + assert.Contains(t, err.Error(), "walk boom") +} + +// countingPersister fails Load after the first failAfter calls: call 1 +// feeds planWalks, the failing call 2 is runShard's authoritative read. +type countingPersister struct { + inner CursorPersister + calls atomic.Int32 + failAfter int32 +} + +func (p *countingPersister) Load(ctx context.Context, shardID int) (Cursor, bool, error) { + if p.calls.Add(1) > p.failAfter { + return Cursor{}, false, errors.New("transient load failure") + } + return p.inner.Load(ctx, shardID) +} + +func (p *countingPersister) Save(ctx context.Context, shardID int, c Cursor) error { + return p.inner.Save(ctx, shardID, c) +} + +// A shard whose second cursor load fails exits runShard without +// awaiting the planned walk it was grouped into. The walk goroutine is +// already running — Run must still join it, or the handler tears down +// the filer connection while the walk is mid-listing. +func TestRun_JoinsPlannedWalkWhenShardExitsEarly(t *testing.T) { + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: "b1", Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, + ExpirationDate: time.Now().Add(24 * time.Hour)}, + }}, + }, engine.CompileOptions{}) + + walkStarted := make(chan struct{}) + walkRelease := make(chan struct{}) + cfg := Config{ + Shards: []int{0}, + BucketsPath: "/buckets", + Engine: e, + FilerClient: stubFilerClient{}, + Client: stubLifecycleClient{}, + Persister: &countingPersister{inner: newMemPersister(), failAfter: 1}, + Lister: stubSiblingLister{}, + Walker: func(context.Context, *engine.Snapshot, []int) error { + close(walkStarted) + <-walkRelease + return nil + }, + } + runDone := make(chan error, 1) + go func() { runDone <- Run(context.Background(), cfg) }() + + <-walkStarted + select { + case <-runDone: + t.Fatal("Run returned while the planned walk was still running") + case <-time.After(200 * time.Millisecond): + } + close(walkRelease) + require.Error(t, <-runDone) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go index e6af3d875..27f9230d8 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go @@ -84,7 +84,7 @@ func TestRunShard_WalkerThrottle(t *testing.T) { calls := 0 cfg := Config{ Persister: p, - Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + Walker: func(_ context.Context, _ *engine.Snapshot, _ []int) error { calls++ return nil }, @@ -127,7 +127,7 @@ func TestRunShard_WalkerThrottle(t *testing.T) { close(closedEvents) // Pass 1. - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents, nil)) afterPass1, _, _ := p.Load(context.Background(), 0) require.Equal(t, 1, calls, "pass 1 must fire the walker (cold-start anchor)") require.Equal(t, runNow.UnixNano(), afterPass1.LastWalkedNs) @@ -136,7 +136,7 @@ func TestRunShard_WalkerThrottle(t *testing.T) { runNow2 := runNow.Add(tc.secondPassAfter) closedEvents2 := make(chan *readerEventAlias) close(closedEvents2) - require.NoError(t, runShard(context.Background(), cfg, snap, runNow2, 0, closedEvents2)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow2, 0, closedEvents2, nil)) afterPass2, _, _ := p.Load(context.Background(), 0) assert.Equal(t, tc.wantTotalCalls, calls, "walker calls after 2 passes") if tc.wantSecondAdvNs { @@ -221,7 +221,7 @@ func TestRunShard_ColdStartDoesNotDoubleWalk(t *testing.T) { calls := 0 cfg := Config{ Persister: p, - Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + Walker: func(_ context.Context, _ *engine.Snapshot, _ []int) error { calls++ return nil }, @@ -231,7 +231,7 @@ func TestRunShard_ColdStartDoesNotDoubleWalk(t *testing.T) { closedEvents := make(chan *readerEventAlias) close(closedEvents) - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, closedEvents, nil)) assert.Equal(t, 1, calls, "cold-start walker must fire exactly once per pass even with interval=0") } @@ -256,14 +256,14 @@ func TestRunShard_RecoveryWalkerSetsLastWalkedAnchor(t *testing.T) { calls := 0 cfg := Config{ Persister: p, - Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + Walker: func(_ context.Context, _ *engine.Snapshot, _ []int) error { calls++ return nil }, WalkerInterval: time.Hour, } runNow := time.Unix(1_700_000_000, 0).UTC() - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 5, nil)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 5, nil, nil)) require.Equal(t, 1, calls, "recovery branch fires the walker unconditionally") got, ok, _ := p.Load(context.Background(), 5) diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go index 1dbe068f1..245221cdf 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_recovery_test.go @@ -66,23 +66,23 @@ func TestRunShard_WalkerInvokedOnRecoveryBranch(t *testing.T) { require.NoError(t, p.Save(context.Background(), 3, Cursor{TsNs: 1234, RuleSetHash: stale})) var gotView *engine.Snapshot - var gotShard int + var gotShards []int calls := 0 cfg := Config{ Persister: p, - Walker: func(_ context.Context, view *engine.Snapshot, shardID int) error { + Walker: func(_ context.Context, view *engine.Snapshot, shards []int) error { calls++ gotView = view - gotShard = shardID + gotShards = shards return nil }, } runNow := time.Unix(1_700_000_000, 0).UTC() - require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3, nil)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 3, nil, nil)) 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) + assert.Equal(t, []int{3}, gotShards) // Cursor rewound to runNow - maxTTL with the new hashes persisted. got, ok, err := p.Load(context.Background(), 3) @@ -106,7 +106,7 @@ func TestRunShard_NilWalkerOnRecoveryIsNoop(t *testing.T) { 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, nil)) + require.NoError(t, runShard(context.Background(), cfg, snap, runNow, 0, nil, nil)) got, ok, err := p.Load(context.Background(), 0) require.NoError(t, err) @@ -128,12 +128,12 @@ func TestRunShard_WalkerErrorPropagates(t *testing.T) { cfg := Config{ Persister: p, - Walker: func(_ context.Context, _ *engine.Snapshot, _ int) error { + 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, nil) + err := runShard(context.Background(), cfg, snap, runNow, 7, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "walker boom") diff --git a/weed/s3api/s3lifecycle/engine/engine.go b/weed/s3api/s3lifecycle/engine/engine.go index 0c6b40f04..d7ad7e7c5 100644 --- a/weed/s3api/s3lifecycle/engine/engine.go +++ b/weed/s3api/s3lifecycle/engine/engine.go @@ -119,6 +119,30 @@ func (s *Snapshot) BucketVersioned(bucket string) bool { return false } +// BucketPrefixes returns the distinct rule prefixes of the bucket's +// active actions in this view. An empty result means no active action +// can match any entry in the bucket; a "" element means some rule +// matches every key. +func (s *Snapshot) BucketPrefixes(bucket string) []string { + bi := s.buckets[bucket] + if bi == nil { + return nil + } + seen := map[string]bool{} + var out []string + for _, k := range bi.actionKeys { + a := s.actions[k] + if a == nil || !a.IsActive() || a.Mode == ModeDisabled || a.Rule == nil { + continue + } + if !seen[a.Rule.Prefix] { + seen[a.Rule.Prefix] = true + out = append(out, a.Rule.Prefix) + } + } + return out +} + // BucketActionKeys returns a defensive copy of the action keys for bucket, // or nil if the bucket has no compiled actions in this snapshot. func (s *Snapshot) BucketActionKeys(bucket string) []s3lifecycle.ActionKey { diff --git a/weed/s3api/s3lifecycle/scheduler/configload.go b/weed/s3api/s3lifecycle/scheduler/configload.go index 9da85e537..90c6e13ab 100644 --- a/weed/s3api/s3lifecycle/scheduler/configload.go +++ b/weed/s3api/s3lifecycle/scheduler/configload.go @@ -82,14 +82,17 @@ func LoadCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, } // IsBucketVersioned returns true when the bucket's filer entry has the -// versioning extended attribute set to Enabled or Suspended. +// versioning extended attribute set to Enabled or Suspended, or when +// Object Lock is on: lock requires versioning, so the gateway writes +// version siblings even if the versioning attribute is absent. func IsBucketVersioned(entry *filer_pb.Entry) bool { - v, ok := entry.Extended[s3_constants.ExtVersioningKey] - if !ok { - return false + if v, ok := entry.Extended[s3_constants.ExtVersioningKey]; ok { + s := strings.ToLower(strings.TrimSpace(string(v))) + if s == "enabled" || s == "suspended" { + return true + } } - s := strings.ToLower(strings.TrimSpace(string(v))) - return s == "enabled" || s == "suspended" + return string(entry.Extended[s3_constants.ExtObjectLockEnabledKey]) == s3_constants.ObjectLockEnabled } // AllActivePriorStates seeds every compiled action as bootstrap-complete + diff --git a/weed/s3api/s3lifecycle/scheduler/configload_test.go b/weed/s3api/s3lifecycle/scheduler/configload_test.go index 5929ea2ba..0512adb1a 100644 --- a/weed/s3api/s3lifecycle/scheduler/configload_test.go +++ b/weed/s3api/s3lifecycle/scheduler/configload_test.go @@ -63,6 +63,18 @@ func TestIsBucketVersioned_RejectsOtherValues(t *testing.T) { } } +func TestIsBucketVersioned_ObjectLockImpliesVersioned(t *testing.T) { + // Object Lock requires versioning; every writer sets both attrs, + // but a lock flag alone must still classify as versioned so a + // hand-mutated or legacy entry can't hide its version siblings. + e := &filer_pb.Entry{Extended: map[string][]byte{ + s3_constants.ExtObjectLockEnabledKey: []byte(s3_constants.ObjectLockEnabled), + }} + assert.True(t, IsBucketVersioned(e)) + e.Extended[s3_constants.ExtObjectLockEnabledKey] = []byte("false") + assert.False(t, IsBucketVersioned(e)) +} + func TestAllActivePriorStates_EmptyInputs(t *testing.T) { assert.Empty(t, AllActivePriorStates(nil)) assert.Empty(t, AllActivePriorStates([]engine.CompileInput{})) diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index 1b2f45bab..47025bf8c 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -117,7 +117,6 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i defer cancel() } client := &lifecycleClientCallable{c: rpcClient} - listFn := dailyrun.FilerListFunc(filerClient, bucketsPath) walkerDispatch := &dailyrun.WalkerDispatcher{Client: client} fmt.Fprintf(writer, "running shards %s (event budget=%d, runtime=%s, refresh=%s)…\n", @@ -166,8 +165,9 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i buckets = append(buckets, in.Bucket) } } - walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error { - return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, listFn, walkerDispatch) + walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shards []int) error { + return dailyrun.WalkBuckets(walkCtx, view, shards, buckets, + dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch) }) return dailyrun.Run(passCtx, dailyrun.Config{ Shards: shards, diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 173725009..50d9d4a02 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -313,12 +313,12 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe buckets = append(buckets, in.Bucket) } } - walkerListFn := dailyrun.FilerListFunc(filerClient, bucketsPath) // Share the limiter with processMatches so walker + replay can't // combine to burst past the cluster cap. walkerDispatch := &dailyrun.WalkerDispatcher{Client: client, Limiter: limiter} - walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error { - return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, walkerListFn, walkerDispatch) + walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shards []int) error { + return dailyrun.WalkBuckets(walkCtx, view, shards, buckets, + dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch) }) _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{