From 443ae1cd838237bd8abb866292a277e62bc786e1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 17 Oct 2026 15:21:53 -0700 Subject: [PATCH] s3 lifecycle: share daily bucket walk and prune listings (#11797) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * s3 lifecycle: share the daily bucket walk across shards The daily pass invoked the walker once per shard, and every shard's walk listed the full bucket tree while keeping only its own 1/16 of the entries — 16 full listings per pass by default (#11791). Run now groups the shards by the walk branch they'll take (recovery vs steady, using the same cursor data and hash checks runShard applies) and starts one shared walk per group, so a pass lists each bucket at most twice. WalkerFunc and WalkBuckets take a shard set instead of a single shard. runShard keeps its own decision and waits on its group's walk; a shard whose freshly-loaded cursor disagrees with the plan falls back to a per-shard walk, so mid-pass cursor rewrites can't under-cover a shard. * s3 lifecycle: skip the .versions/ pre-pass on non-versioned buckets walkBucketTree listed every directory twice — once looking for .versions/ dirs, once for regular entries. On a bucket that never had versioning enabled the first pass can only find coincidentally-named user folders, so the extra listing is pure cost (#11791). FilerListFunc now takes the walk view and consults Snapshot.BucketVersioned; non-versioned buckets skip pass 1 entirely. A *-suffixed .versions user folder on such a bucket is recursed into as a regular directory so real objects inside still surface; a nil view keeps the two-pass behavior. * s3 lifecycle: prune walk directories to active rule prefixes The daily walk listed every directory under a bucket even when no rule prefix could reach it. Collect the active rule prefixes of each bucket from the walk view (new Snapshot.BucketPrefixes) and skip recursing into directories whose keys can never match; a bucket with no active actions is skipped entirely, saving the listing. .uploads stays in scope regardless of prefixes because AbortMPU rules match DestKey, and .versions expansion is pruned only conservatively (the logical key is a string-prefix of the dir key). * s3 lifecycle: join planned walks before Run returns planWalks starts a shared walk goroutine per group, but a shard whose authoritative cursor load then fails exits runShard without waiting on it. Run returned while the walk still listed or dispatched deletes against connections the caller closes on return. Join every planned walk after the shard waitgroup. * s3 lifecycle: treat object-locked buckets as versioned in the walk Object Lock requires versioning and every writer sets both extended attributes, but IsBucketVersioned only read ExtVersioningKey. A locked bucket missing that attribute would have its .versions dirs walked as ordinary folders, hiding version siblings from lifecycle. Classify ExtObjectLockEnabledKey as versioned too. --- weed/s3api/s3lifecycle/DESIGN.md | 18 +- .../s3lifecycle/dailyrun/filer_list_func.go | 104 ++++++-- .../dailyrun/filer_list_func_test.go | 234 ++++++++++++++++-- weed/s3api/s3lifecycle/dailyrun/run.go | 76 ++++-- .../s3lifecycle/dailyrun/walk_buckets.go | 20 +- .../s3lifecycle/dailyrun/walk_buckets_test.go | 12 +- weed/s3api/s3lifecycle/dailyrun/walk_plan.go | 105 ++++++++ .../s3lifecycle/dailyrun/walk_plan_test.go | 232 +++++++++++++++++ .../dailyrun/walker_interval_test.go | 14 +- .../dailyrun/walker_recovery_test.go | 16 +- weed/s3api/s3lifecycle/engine/engine.go | 24 ++ .../s3api/s3lifecycle/scheduler/configload.go | 15 +- .../s3lifecycle/scheduler/configload_test.go | 12 + weed/shell/command_s3_lifecycle_run_shard.go | 6 +- weed/worker/tasks/s3_lifecycle/handler.go | 6 +- 15 files changed, 780 insertions(+), 114 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/walk_plan.go create mode 100644 weed/s3api/s3lifecycle/dailyrun/walk_plan_test.go 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{