diff --git a/docs/_storage-backends/aws-s3.md b/docs/_storage-backends/aws-s3.md index e45d96d11..04d215f2e 100644 --- a/docs/_storage-backends/aws-s3.md +++ b/docs/_storage-backends/aws-s3.md @@ -95,6 +95,27 @@ If the metadata contains a `filetype` key, its value is used to set the `Content When receiving a `PATCH` request, parts of its body will be temporarily stored on disk before they can be transferred to S3. This is necessary to meet the minimum part size for an S3 multipart upload enforced by S3 and to allow the AWS SDK to calculate a checksum. Once the part has been uploaded to S3, the temporary file will be removed immediately. Therefore, please ensure that the server running this storage backend has enough disk space available to hold these temporary files. +### Consistency requirements + +Tusd requires **strong read-after-write consistency** from its object store. +When a chunk smaller than the minimum part size arrives, tusd stores it in a +temporary object and reads it back on a later request to assemble the final +object; completing an upload likewise reads back that temporary object to +promote it into the multipart upload. If the store serves a stale or missing +read of an object tusd just wrote, uploads can fail or be assembled incorrectly. + +AWS S3 provides this for all operations. Most S3-compatible stores do too, but +if you run one, confirm its consistency guarantees: + +- [AWS S3](https://aws.amazon.com/s3/consistency/) +- [Cloudflare R2](https://developers.cloudflare.com/r2/reference/consistency/) +- [Ceph RadosGW](https://docs.ceph.com/en/reef/dev/radosgw/bucket_index/) + +If you run multiple tusd instances against the same bucket, also configure a +distributed [lock provider]({{ site.baseurl }}/advanced-topics/locks/) so that requests for a single +upload are serialized. Tusd's storage layer relies on that serialization and +does not perform conditional (compare-and-swap) writes of its own. + ## Usage with MinIO [MinIO](https://min.io/) is an object storage solution that provides an S3-compatible API, making it suitable as a replacement for AWS S3 during development/testing or even in production. To get started, please install the MinIO server according to their documentation. There are different installation methods available (Docker, package managers or direct download), but we will not go further into them. We assume that the `minio` (server) and `mc` (client) commands are installed. First, start MinIO with example credentials: diff --git a/pkg/s3store/s3store.go b/pkg/s3store/s3store.go index 33ae52861..ebbb6658b 100644 --- a/pkg/s3store/s3store.go +++ b/pkg/s3store/s3store.go @@ -17,6 +17,10 @@ // the HTTP endpoint used for sending requests to, adjust the `BaseEndpoint` // option in the AWS SDK For Go V2 (https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/s3#Options). // +// S3Store requires strong read-after-write consistency from the backing store; +// for multi-instance deployments a distributed Locker must serialize requests +// per upload (the storage layer performs no conditional writes of its own). +// // # Implementation // // Once a new tus upload is initiated, multiple objects in S3 are created: @@ -430,6 +434,11 @@ func (upload *s3Upload) WriteChunk(ctx context.Context, offset int64, src io.Rea if err := store.deleteIncompletePartForUpload(ctx, upload.objectId); err != nil { return 0, err } + // The ".part" object is gone now. Keep the cached size consistent so a + // FinishUpload later in this same request (on this same upload object) + // does not act on a stale non-zero incompletePartSize. uploadParts below + // re-sets it if it parks a new incomplete part. + upload.incompletePartSize = 0 // Prepend an incomplete part, if necessary and adapt the offset src = io.MultiReader(incompletePartFile, src) @@ -815,7 +824,6 @@ func (upload s3Upload) Terminate(ctx context.Context) error { Quiet: aws.Bool(true), }, }) - if err != nil { errCh <- err return @@ -842,12 +850,79 @@ func (upload s3Upload) Terminate(ctx context.Context) error { func (upload s3Upload) FinishUpload(ctx context.Context) error { store := upload.store - // Get uploaded parts - _, parts, _, err := upload.getInternalInfo(ctx) + // Get uploaded parts and whether an incomplete part is lingering. + info, parts, incompletePartSize, err := upload.getInternalInfo(ctx) if err != nil { return err } + // A sub-MinPartSize tail can be parked in a ".part" object instead of a real + // multipart part (e.g. a deferred-length upload whose final chunk was written + // while the length was still deferred). Completing without it would silently + // drop its bytes, so when such an incomplete part lingers, use the now-known + // declared size to decide what to do with it. When there is no incomplete part + // (the common case, and the concatenation path, which builds parts without + // real sizes), complete with the multipart parts as-is. + if incompletePartSize > 0 { + var partsSize int64 + for _, part := range parts { + partsSize += part.size + } + + switch { + case partsSize == info.Size: + // The declared size is already covered by the multipart parts, so the + // leftover incomplete part is stale (e.g. a completion that was retried + // after the part was promoted but before the ".part" object was deleted). + // Remove it, but do not promote it again. + if err := store.deleteIncompletePartForUpload(ctx, upload.objectId); err != nil { + return err + } + case partsSize+incompletePartSize == info.Size: + // Promote the parked tail into a real final part before completing. + incompletePartFile, err := store.downloadIncompletePartForUpload(ctx, upload.objectId) + if err != nil { + return err + } + if incompletePartFile == nil { + return fmt.Errorf("s3store: expected an incomplete part file for upload %s but did not get any", upload.objectId) + } + defer cleanUpTempFile(incompletePartFile) + + // Number the promoted part after the last real part. Using the last + // part's number (rather than len(parts)) keeps this robust to any gaps + // in what ListParts returned. + partNumber := int32(1) + if len(parts) > 0 { + partNumber = parts[len(parts)-1].number + 1 + } + + t := time.Now() + etag, err := upload.putPartForUpload(ctx, &s3.UploadPartInput{ + Bucket: aws.String(store.Bucket), + Key: store.keyWithPrefix(upload.objectId), + UploadId: aws.String(upload.multipartId), + PartNumber: aws.Int32(partNumber), + }, incompletePartFile, incompletePartSize) + store.observeRequestDuration(t, metricUploadPart) + if err != nil { + return err + } + + parts = append(parts, &s3Part{ + etag: etag, + number: partNumber, + size: incompletePartSize, + }) + + if err := store.deleteIncompletePartForUpload(ctx, upload.objectId); err != nil { + return err + } + default: + return fmt.Errorf("s3store: cannot finish upload %s: multipart parts (%d bytes) plus incomplete part (%d bytes) do not match the declared size (%d bytes)", upload.objectId, partsSize, incompletePartSize, info.Size) + } + } + if len(parts) == 0 { // AWS expects at least one part to be present when completing the multipart // upload. So if the tus upload has a size of 0, we create an empty part diff --git a/pkg/s3store/s3store_deferred_length_repro_test.go b/pkg/s3store/s3store_deferred_length_repro_test.go new file mode 100644 index 000000000..97ae842a8 --- /dev/null +++ b/pkg/s3store/s3store_deferred_length_repro_test.go @@ -0,0 +1,330 @@ +package s3store + +import ( + "bytes" + "context" + "io" + "testing" + + "github.com/golang/mock/gomock" + "github.com/stretchr/testify/assert" + + "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/tus/tusd/v2/pkg/handler" +) + +// These tests cover the fix for silent truncation of deferred-length uploads +// (Upload-Defer-Length / the IETF resumable-upload draft's Upload-Complete tail +// marker). When a sub-MinPartSize tail is written while the length is still +// deferred, it is stashed in a side ".part" object instead of a real +// multipart part. FinishUpload must promote that object into a real final part +// before completing, otherwise its bytes are dropped. See tus/tusd#396 and #798. + +// smallPartStore returns a store whose part-size knobs are tiny so tests can use +// a handful of bytes instead of multi-MiB parts. +func smallPartStore(s3obj *MockS3API) S3Store { + store := New("bucket", s3obj) + store.MaxPartSize = 8 + store.MinPartSize = 4 + store.PreferredPartSize = 4 + store.MaxMultipartParts = 10000 + store.MaxObjectSize = 5 * 1024 * 1024 * 1024 * 1024 + return store +} + +// TestFinishUploadPromotesIncompletePart: a deferred upload with one real part +// (>= MinPartSize) and a smaller tail stashed as an incomplete part. FinishUpload +// must promote the tail into part 2 and complete with BOTH parts. +func TestFinishUploadPromotesIncompletePart(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + assert := assert.New(t) + + s3obj := NewMockS3API(mockCtrl) + store := smallPartStore(s3obj) + + deferredInfoJSON := `{"ID":"uploadId+multipartId","Size":0,"SizeIsDeferred":true,"Offset":0,"MetaData":null,"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"uploadId","Type":"s3store"}}` + finishedInfoJSON := `{"ID":"uploadId+multipartId","Size":6,"SizeIsDeferred":false,"Offset":6,"MetaData":null,"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"uploadId","Type":"s3store"}}` + + // 1. NewUpload(SizeIsDeferred: true) + s3obj.EXPECT().CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{ + Bucket: aws.String("bucket"), + Key: aws.String("uploadId"), + Metadata: map[string]string{}, + }).Return(&s3.CreateMultipartUploadOutput{UploadId: aws.String("multipartId")}, nil) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), + Key: aws.String("uploadId.info"), + Body: bytes.NewReader([]byte(deferredInfoJSON)), + ContentLength: aws.Int64(int64(len(deferredInfoJSON))), + })).Return(nil, nil) + + // 2. WriteChunk 4 bytes -> real part 1 (>= MinPartSize). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(deferredInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{}}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(nil, &types.NotFound{}) + s3obj.EXPECT().UploadPart(context.Background(), NewUploadPartInputMatcher(&s3.UploadPartInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + PartNumber: aws.Int32(1), Body: bytes.NewReader([]byte("1234")), + })).Return(&s3.UploadPartOutput{ETag: aws.String("etag-1")}, nil) + + // 3. WriteChunk 2 bytes -> stashed as incomplete part (< MinPartSize, deferred). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(deferredInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{ + {Size: aws.Int64(4), ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + }}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(nil, &types.NotFound{}) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), Body: bytes.NewReader([]byte("56")), + })).Return(nil, nil) + + // 4. DeclareLength(6). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(deferredInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{ + {Size: aws.Int64(4), ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + }}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.HeadObjectOutput{ContentLength: aws.Int64(2)}, nil) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), + Key: aws.String("uploadId.info"), + Body: bytes.NewReader([]byte(finishedInfoJSON)), + ContentLength: aws.Int64(int64(len(finishedInfoJSON))), + })).Return(nil, nil) + + // 5. FinishUpload -> must promote the incomplete part into part 2. + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(finishedInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{ + {Size: aws.Int64(4), ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + }}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.HeadObjectOutput{ContentLength: aws.Int64(2)}, nil) + // downloadIncompletePartForUpload GETs the .part object (ContentLength required). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.GetObjectOutput{ + Body: io.NopCloser(bytes.NewReader([]byte("56"))), + ContentLength: aws.Int64(2), + }, nil) + // The promoted part is uploaded as part 2. + s3obj.EXPECT().UploadPart(context.Background(), NewUploadPartInputMatcher(&s3.UploadPartInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + PartNumber: aws.Int32(2), Body: bytes.NewReader([]byte("56")), + })).Return(&s3.UploadPartOutput{ETag: aws.String("etag-2")}, nil) + // The now-consumed .part object is deleted. + s3obj.EXPECT().DeleteObject(context.Background(), &s3.DeleteObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.DeleteObjectOutput{}, nil) + // Completion includes BOTH parts. + s3obj.EXPECT().CompleteMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{ + {ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + {ETag: aws.String("etag-2"), PartNumber: aws.Int32(2)}, + }}, + }).Return(nil, nil) + + ctx := context.Background() + + _, err := store.NewUpload(ctx, handler.FileInfo{ID: "uploadId", SizeIsDeferred: true}) + assert.Nil(err) + + upload1, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + n, err := upload1.WriteChunk(ctx, 0, bytes.NewReader([]byte("1234"))) + assert.Nil(err) + assert.Equal(int64(4), n) + + upload2, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + n, err = upload2.WriteChunk(ctx, 4, bytes.NewReader([]byte("56"))) + assert.Nil(err) + assert.Equal(int64(2), n) + + upload3, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + err = store.AsLengthDeclarableUpload(upload3).DeclareLength(ctx, 6) + assert.Nil(err) + + upload4, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + err = upload4.FinishUpload(ctx) + assert.Nil(err) +} + +// TestFinishUploadPromotesIncompletePartWhenNoRealParts: the whole deferred +// upload is smaller than MinPartSize, so there are no real parts at all — only +// the incomplete part. FinishUpload must complete with the real bytes as part 1, +// NOT an empty part. +func TestFinishUploadPromotesIncompletePartWhenNoRealParts(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + assert := assert.New(t) + + s3obj := NewMockS3API(mockCtrl) + store := smallPartStore(s3obj) + + deferredInfoJSON := `{"ID":"uploadId+multipartId","Size":0,"SizeIsDeferred":true,"Offset":0,"MetaData":null,"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"uploadId","Type":"s3store"}}` + finishedInfoJSON := `{"ID":"uploadId+multipartId","Size":2,"SizeIsDeferred":false,"Offset":2,"MetaData":null,"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"uploadId","Type":"s3store"}}` + + // 1. NewUpload(SizeIsDeferred: true) + s3obj.EXPECT().CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), Metadata: map[string]string{}, + }).Return(&s3.CreateMultipartUploadOutput{UploadId: aws.String("multipartId")}, nil) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + Body: bytes.NewReader([]byte(deferredInfoJSON)), ContentLength: aws.Int64(int64(len(deferredInfoJSON))), + })).Return(nil, nil) + + // 2. WriteChunk 2 bytes -> stashed as incomplete part (no real part created). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(deferredInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{}}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(nil, &types.NotFound{}) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), Body: bytes.NewReader([]byte("56")), + })).Return(nil, nil) + + // 3. DeclareLength(2). + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(deferredInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{}}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.HeadObjectOutput{ContentLength: aws.Int64(2)}, nil) + s3obj.EXPECT().PutObject(context.Background(), NewPutObjectInputMatcher(&s3.PutObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + Body: bytes.NewReader([]byte(finishedInfoJSON)), ContentLength: aws.Int64(int64(len(finishedInfoJSON))), + })).Return(nil, nil) + + // 4. FinishUpload -> promote the incomplete part into part 1. + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(finishedInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{}}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.HeadObjectOutput{ContentLength: aws.Int64(2)}, nil) + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.GetObjectOutput{ + Body: io.NopCloser(bytes.NewReader([]byte("56"))), ContentLength: aws.Int64(2), + }, nil) + s3obj.EXPECT().UploadPart(context.Background(), NewUploadPartInputMatcher(&s3.UploadPartInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + PartNumber: aws.Int32(1), Body: bytes.NewReader([]byte("56")), + })).Return(&s3.UploadPartOutput{ETag: aws.String("etag-1")}, nil) + s3obj.EXPECT().DeleteObject(context.Background(), &s3.DeleteObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.DeleteObjectOutput{}, nil) + s3obj.EXPECT().CompleteMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{ + {ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + }}, + }).Return(nil, nil) + + ctx := context.Background() + + _, err := store.NewUpload(ctx, handler.FileInfo{ID: "uploadId", SizeIsDeferred: true}) + assert.Nil(err) + + upload1, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + n, err := upload1.WriteChunk(ctx, 0, bytes.NewReader([]byte("56"))) + assert.Nil(err) + assert.Equal(int64(2), n) + + upload2, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + err = store.AsLengthDeclarableUpload(upload2).DeclareLength(ctx, 2) + assert.Nil(err) + + upload3, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + err = upload3.FinishUpload(ctx) + assert.Nil(err) +} + +// TestFinishUploadDoesNotDuplicateAlreadyPromotedPart: a retried FinishUpload +// where the tail was already promoted to a real part on a prior attempt but the +// ".part" object was not yet deleted (crash between UploadPart and DeleteObject). +// The declared size already equals the sum of the real parts, so FinishUpload +// must delete the stale ".part" and NOT upload it again. +func TestFinishUploadDoesNotDuplicateAlreadyPromotedPart(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + assert := assert.New(t) + + s3obj := NewMockS3API(mockCtrl) + store := smallPartStore(s3obj) + + finishedInfoJSON := `{"ID":"uploadId+multipartId","Size":6,"SizeIsDeferred":false,"Offset":6,"MetaData":null,"IsPartial":false,"IsFinal":false,"PartialUploads":null,"Storage":{"Bucket":"bucket","Key":"uploadId","Type":"s3store"}}` + + // getInternalInfo: both parts are already real; a stale .part still lingers. + s3obj.EXPECT().GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.info"), + }).Return(&s3.GetObjectOutput{Body: io.NopCloser(bytes.NewReader([]byte(finishedInfoJSON)))}, nil) + s3obj.EXPECT().ListParts(context.Background(), &s3.ListPartsInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), PartNumberMarker: nil, + }).Return(&s3.ListPartsOutput{Parts: []types.Part{ + {Size: aws.Int64(4), ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + {Size: aws.Int64(2), ETag: aws.String("etag-2"), PartNumber: aws.Int32(2)}, + }}, nil) + s3obj.EXPECT().HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.HeadObjectOutput{ContentLength: aws.Int64(2)}, nil) + // Stale .part is deleted, NOT re-uploaded. + s3obj.EXPECT().DeleteObject(context.Background(), &s3.DeleteObjectInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId.part"), + }).Return(&s3.DeleteObjectOutput{}, nil) + s3obj.EXPECT().CompleteMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{ + Bucket: aws.String("bucket"), Key: aws.String("uploadId"), UploadId: aws.String("multipartId"), + MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{ + {ETag: aws.String("etag-1"), PartNumber: aws.Int32(1)}, + {ETag: aws.String("etag-2"), PartNumber: aws.Int32(2)}, + }}, + }).Return(nil, nil) + + ctx := context.Background() + + upload, err := store.GetUpload(ctx, "uploadId+multipartId") + assert.Nil(err) + err = upload.FinishUpload(ctx) + assert.Nil(err) +}