From bb0d4859ec2aa5e4925135ef97234e8495870fef Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 22:56:49 -0700 Subject: [PATCH] test(s3/lifecycle): integration coverage for AbortIncompleteMultipartUpload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exercises the lifecycleAbortMPU handler path that the prefix-based expiration tests can't reach — routing keys off of .uploads// directory events, not regular object events, and the dispatcher uses a different RPC path (rm on the .uploads// folder). Setup: AbortIncompleteMultipartUpload rule with DaysAfterInitiation=1, CreateMultipartUpload, UploadPart (so the directory carries the right shape), backdate the .uploads// directory entry 30 days, run the worker. The upload must drop out of ListMultipartUploads. Helpers added: putAbortMPULifecycle, backdateUploadDir. --- test/s3/lifecycle/s3_lifecycle_mpu_test.go | 136 +++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 test/s3/lifecycle/s3_lifecycle_mpu_test.go diff --git a/test/s3/lifecycle/s3_lifecycle_mpu_test.go b/test/s3/lifecycle/s3_lifecycle_mpu_test.go new file mode 100644 index 000000000..22b1439fa --- /dev/null +++ b/test/s3/lifecycle/s3_lifecycle_mpu_test.go @@ -0,0 +1,136 @@ +// AbortIncompleteMultipartUpload integration scenario. +package lifecycle + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/require" +) + +// putAbortMPULifecycle wires AbortIncompleteMultipartUpload on a prefix. +// AWS rejects DaysAfterInitiation < 1, so the test backdates the upload +// directory to age it past the threshold. +func putAbortMPULifecycle(t *testing.T, c *s3.Client, bucket, prefix string, days int32) { + t.Helper() + _, err := c.PutBucketLifecycleConfiguration(context.Background(), &s3.PutBucketLifecycleConfigurationInput{ + Bucket: aws.String(bucket), + LifecycleConfiguration: &types.BucketLifecycleConfiguration{ + Rules: []types.LifecycleRule{ + { + ID: aws.String("abort-stale-mpu"), + Status: types.ExpirationStatusEnabled, + Filter: &types.LifecycleRuleFilter{Prefix: aws.String(prefix)}, + AbortIncompleteMultipartUpload: &types.AbortIncompleteMultipartUpload{ + DaysAfterInitiation: aws.Int32(days), + }, + }, + }, + }, + }) + require.NoError(t, err) +} + +// backdateUploadDir ages the .uploads// directory entry, which is +// what the lifecycle worker keys ABORT_MPU off of (the upload's init-time +// directory carries the destination key in Extended). +func backdateUploadDir(t *testing.T, fc filer_pb.SeaweedFilerClient, bucket, uploadID string, daysOld int) { + t.Helper() + dir := bucketsPath + "/" + bucket + "/.uploads" + resp, err := fc.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{ + Directory: dir, Name: uploadID, + }) + require.NoError(t, err, "lookup .uploads/%s", uploadID) + require.NotNil(t, resp.Entry) + require.NotNil(t, resp.Entry.Attributes) + require.True(t, resp.Entry.IsDirectory, ".uploads/%s should be a directory", uploadID) + + resp.Entry.Attributes.Mtime = time.Now().Add(-time.Duration(daysOld) * 24 * time.Hour).Unix() + resp.Entry.Attributes.MtimeNs = 0 + _, err = fc.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ + Directory: dir, Entry: resp.Entry, + }) + require.NoError(t, err) +} + +// TestLifecycleAbortIncompleteMultipartUpload: an MPU left incomplete past +// the AbortIncompleteMultipartUpload.DaysAfterInitiation threshold must be +// aborted by the lifecycle worker. Exercises the lifecycleAbortMPU handler +// path which is otherwise unreachable from the prefix-based expiration +// tests (the routing keys off of `.uploads//` directory events, not +// regular object events). +func TestLifecycleAbortIncompleteMultipartUpload(t *testing.T) { + c := s3Client(t) + fc, fcClose := filerClient(t) + defer fcClose() + + bucket := uniqueBucket("abort-mpu") + mustCreateBucket(t, c, bucket) + putAbortMPULifecycle(t, c, bucket, "uploads/", 1) + + // Initiate an MPU, upload a single part, then leave it. CompleteMultipart + // would consume the upload; AbortMultipart would explicitly abort. We do + // neither so the lifecycle worker is the path that ends it. + const key = "uploads/big.bin" + createOut, err := c.CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), Key: aws.String(key), + }) + require.NoError(t, err) + uploadID := aws.ToString(createOut.UploadId) + require.NotEmpty(t, uploadID) + + // Upload one part so the directory carries the right shape. + _, err = c.UploadPart(context.Background(), &s3.UploadPartInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: aws.String(uploadID), + PartNumber: aws.Int32(1), + Body: strings.NewReader("part 1 contents"), + }) + require.NoError(t, err) + + // Sanity: ListMultipartUploads sees our pending upload before the + // worker runs. + listOut, err := c.ListMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{ + Bucket: aws.String(bucket), + }) + require.NoError(t, err) + var foundBefore bool + for _, u := range listOut.Uploads { + if aws.ToString(u.UploadId) == uploadID { + foundBefore = true + break + } + } + require.True(t, foundBefore, "upload %s should be visible before the worker runs", uploadID) + + // Backdate the upload directory entry so the rule's DaysAfterInitiation + // threshold is satisfied. The router walks .uploads/ and emits an + // MPU-init event for each entry. + backdateUploadDir(t, fc, bucket, uploadID, 30) + + out := runLifecycleShard(t) + t.Logf("shell output:\n%s", out) + + // The upload must no longer appear in ListMultipartUploads. + require.Eventuallyf(t, func() bool { + listOut, err := c.ListMultipartUploads(context.Background(), &s3.ListMultipartUploadsInput{ + Bucket: aws.String(bucket), + }) + if err != nil { + return false + } + for _, u := range listOut.Uploads { + if aws.ToString(u.UploadId) == uploadID { + return false + } + } + return true + }, 30*time.Second, 500*time.Millisecond, "upload %s must be aborted by the lifecycle worker", uploadID) +}