From 196c41d21aa90ff97a29e3542619abb3b77aadda Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 8 May 2026 22:14:45 -0700 Subject: [PATCH] test(s3/lifecycle): cover scheduler/bootstrap walker + MPU detection (#9380) Locks down isMPUInitDir, walkBucketDir (regular/nested/MPU/error/cancel paths), and BucketBootstrapper.KickOffNew (per-bucket fanout and in-process dedup) against a fake SeaweedFilerClient. --- .../s3lifecycle/scheduler/bootstrap_test.go | 437 ++++++++++++++++++ 1 file changed, 437 insertions(+) create mode 100644 weed/s3api/s3lifecycle/scheduler/bootstrap_test.go diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go new file mode 100644 index 000000000..30c674108 --- /dev/null +++ b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go @@ -0,0 +1,437 @@ +package scheduler + +import ( + "context" + "errors" + "io" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +// fakeListStream implements grpc.ServerStreamingClient[filer_pb.ListEntriesResponse]. +type fakeListStream struct { + responses []*filer_pb.ListEntriesResponse + index int + ctx context.Context +} + +func (s *fakeListStream) Recv() (*filer_pb.ListEntriesResponse, error) { + if s.ctx != nil { + if err := s.ctx.Err(); err != nil { + return nil, err + } + } + if s.index >= len(s.responses) { + return nil, io.EOF + } + resp := s.responses[s.index] + s.index++ + return resp, nil +} + +func (s *fakeListStream) Header() (metadata.MD, error) { return metadata.MD{}, nil } +func (s *fakeListStream) Trailer() metadata.MD { return metadata.MD{} } +func (s *fakeListStream) CloseSend() error { return nil } +func (s *fakeListStream) Context() context.Context { + if s.ctx != nil { + return s.ctx + } + return context.Background() +} +func (s *fakeListStream) SendMsg(any) error { return nil } +func (s *fakeListStream) RecvMsg(any) error { return nil } + +// fakeFilerClient embeds SeaweedFilerClient (nil interface) and overrides +// only ListEntries. Calling any other method panics, which is fine for +// these tests. +type fakeFilerClient struct { + filer_pb.SeaweedFilerClient + + mu sync.Mutex + tree map[string][]*filer_pb.Entry // dir path -> immediate children + listed []string // dirs the walker asked about, in order + listedN int32 // atomic counter for cross-goroutine reads +} + +func (c *fakeFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + c.mu.Lock() + c.listed = append(c.listed, in.Directory) + c.mu.Unlock() + atomic.AddInt32(&c.listedN, 1) + + children := c.tree[in.Directory] + responses := make([]*filer_pb.ListEntriesResponse, 0, len(children)) + for _, e := range children { + responses = append(responses, &filer_pb.ListEntriesResponse{Entry: e}) + } + return &fakeListStream{responses: responses, ctx: ctx}, nil +} + +func (c *fakeFilerClient) listedCopy() []string { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]string, len(c.listed)) + copy(out, c.listed) + return out +} + +// recordingInjector captures injected events; optionally returns a +// pre-set error to test propagation. +type recordingInjector struct { + mu sync.Mutex + events []*reader.Event + err error +} + +func (r *recordingInjector) InjectEvent(_ context.Context, ev *reader.Event) error { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, ev) + return r.err +} + +func (r *recordingInjector) snapshot() []*reader.Event { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*reader.Event, len(r.events)) + copy(out, r.events) + return out +} + +func dirEntry(name string, extended map[string][]byte) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: name, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: extended, + } +} + +func fileEntry(name string) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: name, + IsDirectory: false, + Attributes: &filer_pb.FuseAttributes{}, + } +} + +// ---------- isMPUInitDir ---------- + +func TestIsMPUInitDir_DetectsValidInitDir(t *testing.T) { + e := dirEntry("upload-id-1", map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte("foo/bar.txt"), + }) + assert.True(t, isMPUInitDir(s3_constants.MultipartUploadsFolder+"/upload-id-1", e)) +} + +func TestIsMPUInitDir_RejectsUploadsRoot(t *testing.T) { + e := dirEntry(s3_constants.MultipartUploadsFolder, nil) + // key = ".uploads/" -> rest = "" -> false + assert.False(t, isMPUInitDir(s3_constants.MultipartUploadsFolder+"/", e)) +} + +func TestIsMPUInitDir_RejectsPartFile(t *testing.T) { + e := fileEntry("part-0001") + assert.False(t, isMPUInitDir(s3_constants.MultipartUploadsFolder+"/upload-id-1/part-0001", e)) +} + +func TestIsMPUInitDir_RejectsRegularKey(t *testing.T) { + e := fileEntry("key.txt") + assert.False(t, isMPUInitDir("regular/object/key", e)) +} + +func TestIsMPUInitDir_RejectsEmptyExtended(t *testing.T) { + e := dirEntry("upload-id-1", map[string][]byte{}) + assert.False(t, isMPUInitDir(s3_constants.MultipartUploadsFolder+"/upload-id-1", e)) +} + +func TestIsMPUInitDir_RejectsEmptyMultipartObjectKeyValue(t *testing.T) { + e := dirEntry("upload-id-1", map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte(""), + }) + assert.False(t, isMPUInitDir(s3_constants.MultipartUploadsFolder+"/upload-id-1", e)) +} + +// ---------- walkBucketDir ---------- + +const testBucketRoot = "/buckets/b1" + +func TestWalkBucketDir_SingleRegularFile(t *testing.T) { + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {fileEntry("a.txt")}, + }, + } + var keys []string + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + keys = append(keys, key) + return nil + }) + require.NoError(t, err) + assert.Equal(t, []string{"a.txt"}, keys) +} + +func TestWalkBucketDir_NestedDirectoriesUseBucketRelativeKeys(t *testing.T) { + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {dirEntry("sub", nil)}, + testBucketRoot + "/sub": {dirEntry("deeper", nil), fileEntry("mid.txt")}, + testBucketRoot + "/sub/deeper": {fileEntry("leaf.txt")}, + }, + } + var keys []string + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + keys = append(keys, key) + return nil + }) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"sub/deeper/leaf.txt", "sub/mid.txt"}, keys) +} + +func TestWalkBucketDir_MPUInitDirEmittedOnceAndNotRecursed(t *testing.T) { + uploadsDir := testBucketRoot + "/" + s3_constants.MultipartUploadsFolder + initDir := uploadsDir + "/upload-id-1" + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {dirEntry(s3_constants.MultipartUploadsFolder, nil)}, + uploadsDir: {dirEntry("upload-id-1", map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte("foo/bar.txt"), + })}, + // If recursed (it shouldn't), this part-0001 file would be emitted. + initDir: {fileEntry("part-0001")}, + }, + } + var seen []string + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + seen = append(seen, key) + return nil + }) + require.NoError(t, err) + assert.Equal(t, []string{s3_constants.MultipartUploadsFolder + "/upload-id-1"}, seen) + + // Walker must not have descended into the init dir. + for _, d := range client.listedCopy() { + assert.NotEqual(t, initDir, d, "walker must not list init dir contents") + } +} + +func TestWalkBucketDir_NonMPUDirUnderUploadsRecurses(t *testing.T) { + // Directory under .uploads/ that lacks ExtMultipartObjectKey: walker + // recurses normally and emits inner files. + uploadsDir := testBucketRoot + "/" + s3_constants.MultipartUploadsFolder + notInitDir := uploadsDir + "/missing-key" + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {dirEntry(s3_constants.MultipartUploadsFolder, nil)}, + uploadsDir: {dirEntry("missing-key", nil)}, + notInitDir: {fileEntry("part-0001")}, + }, + } + var seen []string + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + seen = append(seen, key) + return nil + }) + require.NoError(t, err) + assert.Equal(t, []string{s3_constants.MultipartUploadsFolder + "/missing-key/part-0001"}, seen) +} + +func TestWalkBucketDir_MixedTreeBucketRelativeKeys(t *testing.T) { + uploadsDir := testBucketRoot + "/" + s3_constants.MultipartUploadsFolder + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: { + fileEntry("top.txt"), + dirEntry("nested", nil), + dirEntry(s3_constants.MultipartUploadsFolder, nil), + }, + testBucketRoot + "/nested": {fileEntry("inner.txt")}, + uploadsDir: {dirEntry("upload-id-1", map[string][]byte{ + s3_constants.ExtMultipartObjectKey: []byte("destkey"), + })}, + }, + } + var seen []string + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + seen = append(seen, key) + return nil + }) + require.NoError(t, err) + assert.ElementsMatch(t, []string{ + "top.txt", + "nested/inner.txt", + s3_constants.MultipartUploadsFolder + "/upload-id-1", + }, seen) +} + +func TestWalkBucketDir_CallbackErrorPropagates(t *testing.T) { + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {fileEntry("a.txt"), fileEntry("b.txt")}, + }, + } + cbErr := errors.New("boom") + calls := 0 + err := walkBucketDir(context.Background(), client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, _ string) error { + calls++ + return cbErr + }) + require.Error(t, err) + assert.ErrorIs(t, err, cbErr) + // Walk must stop at the first failing callback. + assert.Equal(t, 1, calls) +} + +func TestWalkBucketDir_ContextCancelledStopsWalk(t *testing.T) { + // Big tree under root; cancel context up front so list call returns + // the context error immediately. + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + testBucketRoot: {fileEntry("a.txt"), dirEntry("d", nil)}, + testBucketRoot + "/d": {fileEntry("inner.txt")}, + }, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + var seen []string + err := walkBucketDir(ctx, client, testBucketRoot, testBucketRoot, func(_ *filer_pb.Entry, key string) error { + seen = append(seen, key) + return nil + }) + require.Error(t, err) + assert.Empty(t, seen, "no entries should be emitted after context cancel") +} + +// ---------- BucketBootstrapper.KickOffNew ---------- + +// emptyClient: a filer client that always returns no children. Walks +// finish almost immediately. +func newEmptyFilerClient() *fakeFilerClient { + return &fakeFilerClient{tree: map[string][]*filer_pb.Entry{}} +} + +// waitFor polls until cond returns true or fails the test. +func waitFor(t *testing.T, cond func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatalf("timeout waiting: %s", msg) +} + +func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) { + client := newEmptyFilerClient() + inj := &recordingInjector{} + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + } + + b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"}) + + // Both walks must hit ListEntries once each (empty trees -> no recursion). + waitFor(t, func() bool { + return atomic.LoadInt32(&client.listedN) >= 2 + }, "both bucket walks to start") + + listed := client.listedCopy() + assert.ElementsMatch(t, []string{"/buckets/bucketA", "/buckets/bucketB"}, listed) + + b.mu.Lock() + defer b.mu.Unlock() + assert.True(t, b.known["bucketA"]) + assert.True(t, b.known["bucketB"]) + assert.Len(t, b.known, 2) +} + +func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) { + client := newEmptyFilerClient() + inj := &recordingInjector{} + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + } + + b.KickOffNew(context.Background(), []string{"bucketA", "bucketB"}) + waitFor(t, func() bool { + return atomic.LoadInt32(&client.listedN) >= 2 + }, "first wave to complete") + + firstWave := atomic.LoadInt32(&client.listedN) + + // Second call: bucketA is already known, bucketC is new. Only one + // new walk should fire. + b.KickOffNew(context.Background(), []string{"bucketA", "bucketC"}) + waitFor(t, func() bool { + return atomic.LoadInt32(&client.listedN) >= firstWave+1 + }, "bucketC walk to start") + + // Give a moment for any spurious bucketA walk to also tick. + time.Sleep(20 * time.Millisecond) + + listed := client.listedCopy() + // Count distinct buckets walked. + bucketCount := map[string]int{} + for _, d := range listed { + bucketCount[d]++ + } + assert.Equal(t, 1, bucketCount["/buckets/bucketA"], "bucketA must be walked exactly once across both calls") + assert.Equal(t, 1, bucketCount["/buckets/bucketB"]) + assert.Equal(t, 1, bucketCount["/buckets/bucketC"]) + + b.mu.Lock() + defer b.mu.Unlock() + assert.Len(t, b.known, 3) + assert.True(t, b.known["bucketC"]) +} + +func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) { + client := newEmptyFilerClient() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: nil, + } + require.NotPanics(t, func() { + b.KickOffNew(context.Background(), []string{"bucketA"}) + }) + // No walks must have been kicked off, and known must remain empty. + time.Sleep(20 * time.Millisecond) + assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN)) + b.mu.Lock() + defer b.mu.Unlock() + assert.Empty(t, b.known) +} + +func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) { + client := newEmptyFilerClient() + inj := &recordingInjector{} + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + } + b.KickOffNew(context.Background(), nil) + b.KickOffNew(context.Background(), []string{}) + + time.Sleep(20 * time.Millisecond) + assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN)) + assert.Empty(t, inj.snapshot()) +}