s3store: fix silent truncation of deferred-length uploads

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:

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