diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go new file mode 100644 index 000000000..bfd63afae --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go @@ -0,0 +1,76 @@ +package dailyrun + +import ( + "context" + "errors" + "fmt" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" +) + +// 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. +// +// 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 { + if view == nil { + return errors.New("WalkBuckets: nil view") + } + if list == nil { + return errors.New("WalkBuckets: nil list") + } + if dispatch == nil { + return errors.New("WalkBuckets: nil dispatch") + } + filtered := perShardListFunc(list, shardID) + var firstErr error + for _, b := range buckets { + if err := ctx.Err(); err != nil { + return err + } + if _, err := bootstrap.Walk(ctx, view, b, filtered, dispatch, bootstrap.WalkOptions{}); err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("walk %s: %w", b, err) + } else { + glog.V(1).Infof("walker: additional bucket %s: %v", b, err) + } + } + } + return firstErr +} + +// perShardListFunc wraps an inner ListFunc so only entries whose +// logical-key shard matches shardID 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 { + 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 { + return nil + } + return cb(e) + }) + } +} + +func entryShardID(bucket string, e *bootstrap.Entry) int { + key := e.Path + if e.IsMPUInit && e.DestKey != "" { + key = e.DestKey + } + return s3lifecycle.ShardID(bucket, key) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go new file mode 100644 index 000000000..4f3f5b75c --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go @@ -0,0 +1,175 @@ +package dailyrun + +import ( + "context" + "errors" + "sort" + "testing" + "time" + + "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" +) + +// recordingDispatcher captures every (action, entry) pair the walker +// emits so tests can assert dispatch order and shard filtering. +type recordingDispatcher struct { + calls []recordedDispatch + err error +} + +type recordedDispatch struct { + bucket string + path string +} + +func (d *recordingDispatcher) Delete(_ context.Context, action *engine.CompiledAction, entry *bootstrap.Entry) error { + d.calls = append(d.calls, recordedDispatch{bucket: action.Bucket, path: entry.Path}) + return d.err +} + +// findShardForPath returns the shard ID for an entry with given path +// in given bucket. Helper for tests that want to force entries onto +// a known shard. +func findShardForPath(bucket, path string) int { + return s3lifecycle.ShardID(bucket, path) +} + +// fixedShardEntries returns a slice of bootstrap.Entries whose paths +// all live in the same shard. Used so a per-shard filter test has +// deterministic expectations. +func fixedShardEntries(t *testing.T, bucket string, shardID int, count int) []*bootstrap.Entry { + t.Helper() + var out []*bootstrap.Entry + for i := 0; tries(i, count); i++ { + path := "obj-" + intToStr(i) + if findShardForPath(bucket, path) == shardID { + out = append(out, &bootstrap.Entry{ + Path: path, + // Old enough to expire under a 7-day ExpirationDays rule. + ModTime: time.Now().Add(-90 * 24 * time.Hour), + Size: 1, + IsLatest: true, + }) + if len(out) >= count { + return out + } + } + } + t.Fatalf("could not find %d entries in shard %d for bucket %s after 4096 attempts", count, shardID, bucket) + return nil +} + +func intToStr(i int) string { + const hex = "0123456789abcdef" + if i == 0 { + return "0" + } + var out []byte + for n := i; n > 0; n /= 16 { + out = append([]byte{hex[n%16]}, out...) + } + return string(out) +} + +func tries(i, count int) bool { return i < count*256+256 } + +func snapshotForBucketRule(t *testing.T, bucket string, days int) *engine.Snapshot { + t.Helper() + e := engine.New() + e.Compile([]engine.CompileInput{ + {Bucket: bucket, Rules: []*s3lifecycle.Rule{ + {ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: days}, + }}, + }, engine.CompileOptions{}) + snap := e.Snapshot() + for _, a := range snap.AllActions() { + snap.MarkActive(a.Key) + } + return snap +} + +func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) { + bucket := "b1" + const targetShard = 0 + // Build entries: half in shard 0, half elsewhere. The walker's + // per-shard filter must drop the non-matching ones. + inShard := fixedShardEntries(t, bucket, targetShard, 3) + otherShard := fixedShardEntries(t, bucket, (targetShard+1)%16, 3) + all := append([]*bootstrap.Entry{}, inShard...) + all = append(all, otherShard...) + + snap := snapshotForBucketRule(t, bucket, 7) // 30-day-old objects expire + d := &recordingDispatcher{} + err := WalkBuckets(context.Background(), snap, targetShard, []string{bucket}, + bootstrap.EntryCallback(all), d) + require.NoError(t, err) + + gotPaths := make([]string, 0, len(d.calls)) + for _, c := range d.calls { + gotPaths = append(gotPaths, c.path) + assert.Equal(t, bucket, c.bucket) + } + sort.Strings(gotPaths) + + wantPaths := make([]string, 0, len(inShard)) + for _, e := range inShard { + wantPaths = append(wantPaths, e.Path) + } + sort.Strings(wantPaths) + assert.Equal(t, wantPaths, gotPaths, "walker must dispatch exactly the in-shard entries") +} + +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)) +} + +func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) { + // Two buckets, first ListFunc errors. WalkBuckets should still + // process the second bucket but return the first bucket's error. + listErr := errors.New("filer flake") + list := func(_ context.Context, bucket, _ string, _ func(*bootstrap.Entry) error) error { + if bucket == "bad" { + return listErr + } + return nil + } + + snap := snapshotForBucketRule(t, "good", 7) + err := WalkBuckets(context.Background(), snap, 0, []string{"bad", "good"}, list, &recordingDispatcher{}) + require.Error(t, err) + assert.ErrorIs(t, err, listErr) +} + +func TestWalkBuckets_HonorsContextCancellationBetweenBuckets(t *testing.T) { + // Pre-cancel ctx before invocation. Even an empty bucket list + // must surface ctx.Err early so a long walk in progress can + // short-circuit. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), 0, []string{"b"}, + bootstrap.EntryCallback(nil), &recordingDispatcher{}) + assert.ErrorIs(t, err, context.Canceled) +} + +func TestEntryShardID_MPUUsesDestKey(t *testing.T) { + bucket := "bkt" + // Force a (Path, DestKey) pair that maps to different shards + // so we can prove the filter picks the right one. + for i := 0; i < 4096; i++ { + path := ".uploads/" + intToStr(i) + destKey := "user/" + intToStr(i) + if s3lifecycle.ShardID(bucket, path) != s3lifecycle.ShardID(bucket, destKey) { + e := &bootstrap.Entry{Path: path, DestKey: destKey, IsMPUInit: true} + assert.Equal(t, s3lifecycle.ShardID(bucket, destKey), entryShardID(bucket, e), + "MPU init must use DestKey for shard, not the .uploads/ path") + return + } + } + t.Fatal("could not find an MPU path/destkey pair with diverging shards in 4096 attempts") +}