feat(s3/lifecycle): WalkBuckets composes ListFunc + Dispatcher per shard

Adds dailyrun.WalkBuckets — the composable driver the handler-side WalkerFunc will call. Iterates a bucket list, wraps the supplied bootstrap.ListFunc with a per-shard filter (Path for non-MPU, DestKey for MPU init), and runs bootstrap.Walk per bucket using the supplied Dispatcher. First bucket error wins; remaining buckets log and run to completion so one filer flake doesn't kill the shard. Composable rather than monolithic so callers and tests can swap parts: production uses a filer-backed ListFunc + WalkerDispatcher; tests use bootstrap.EntryCallback + a stub. The filer-backed ListFunc is the next commit. Tests pin: shard filter routes only matching entries, MPU shard uses DestKey not the .uploads/<id> path, single-bucket error propagates while other buckets still run, ctx cancellation short-circuits between buckets, nil guards on view/list/dispatch.
2026-09-20 13:30:46 +02:00 · 2026-05-11 22:11:48 -07:00
parent d2b9f53edb
commit bcef462b6c
2 changed files with 251 additions and 0 deletions
@@ -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)
+}
@@ -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/<id> path")
+			return
+		}
+	}
+	t.Fatal("could not find an MPU path/destkey pair with diverging shards in 4096 attempts")
+}