mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-08 15:41:15 +02:00
s3: an abort answered mid-part no longer leaves the upload completable (#11025)
* s3: reject a part whose upload was aborted while its body was in flight The upload-exists check runs before the part body is read. An abort answered during the read deletes the upload directory, and the part write that follows re-creates it, so the aborted upload is listed nowhere yet completes. Re-check after the write: only createMultipartUpload stamps the destination key on .uploads/<id>, so a directory without it is one the part write resurrected. Drop it along with the part and answer NoSuchUpload. Claude-Session: https://claude.ai/code/session_01ByZ49KQdUtHhmjG6SpNczT * s3: reject a copied part whose upload was aborted mid-copy UploadPartCopy has the same window as UploadPart: the upload-exists check runs before the bytes are copied, and the part write that follows re-creates the directory an abort removed. Both the re-encryption and the raw-copy path re-check before answering. Claude-Session: https://claude.ai/code/session_01ByZ49KQdUtHhmjG6SpNczT * s3: do not complete an upload whose directory holds no upload record A .uploads/<id> directory that a part write created rather than createMultipartUpload carries no destination key, no owner and no encryption settings. Completing one turned stray parts into an object; answer NoSuchUpload instead. Claude-Session: https://claude.ai/code/session_01ByZ49KQdUtHhmjG6SpNczT * s3: log the part left behind when the resurrected directory survives abortMultipartUpload can fail to remove what the part write re-created. The client still hears NoSuchUpload, since the upload is gone either way and a retry would only write another part, but the leftover is worth a line. Claude-Session: https://claude.ai/code/session_01ByZ49KQdUtHhmjG6SpNczT
This commit is contained in:
@@ -411,6 +411,11 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
|||||||
}
|
}
|
||||||
return nil, nil, s3err.ErrInternalError
|
return nil, nil, s3err.ErrInternalError
|
||||||
}
|
}
|
||||||
|
// only createMultipartUpload stamps the key; a directory a part write left behind is not an upload
|
||||||
|
if !isMultipartUploadEntry(pentry) {
|
||||||
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||||
|
return nil, nil, s3err.ErrNoSuchUpload
|
||||||
|
}
|
||||||
|
|
||||||
deleteEntries := make([]*filer_pb.Entry, 0)
|
deleteEntries := make([]*filer_pb.Entry, 0)
|
||||||
partEntries := make(map[int][]*filer_pb.Entry, len(entries))
|
partEntries := make(map[int][]*filer_pb.Entry, len(entries))
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
package s3api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||||
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestIsMultipartUploadEntry(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
entry *filer_pb.Entry
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"aborted upload", nil, false},
|
||||||
|
{"directory re-created by a part write", &filer_pb.Entry{IsDirectory: true}, false},
|
||||||
|
{"key emptied", &filer_pb.Entry{IsDirectory: true, Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: {}}}, false},
|
||||||
|
{"open upload", &filer_pb.Entry{IsDirectory: true, Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("a.bin")}}, true},
|
||||||
|
}
|
||||||
|
for _, tc := range tests {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
if got := isMultipartUploadEntry(tc.entry); got != tc.want {
|
||||||
|
t.Errorf("isMultipartUploadEntry() = %v, want %v", got, tc.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1028,6 +1028,10 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
|||||||
s3err.WriteErrorResponse(w, r, errCode)
|
s3err.WriteErrorResponse(w, r, errCode)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// the copy above re-creates a directory an abort removed mid-copy
|
||||||
|
if !s3a.checkUploadStillOpen(w, r, dstBucket, dstObject, uploadID) {
|
||||||
|
return
|
||||||
|
}
|
||||||
setEtag(w, "\""+strings.Trim(etag, "\"")+"\"")
|
setEtag(w, "\""+strings.Trim(etag, "\"")+"\"")
|
||||||
// Mirror PutObjectPartHandler: write x-amz-server-side-encryption /
|
// Mirror PutObjectPartHandler: write x-amz-server-side-encryption /
|
||||||
// x-amz-server-side-encryption-aws-kms-key-id headers on the response
|
// x-amz-server-side-encryption-aws-kms-key-id headers on the response
|
||||||
@@ -1104,6 +1108,11 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// the copy above re-creates a directory an abort removed mid-copy
|
||||||
|
if !s3a.checkUploadStillOpen(w, r, dstBucket, dstObject, uploadID) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Calculate ETag for the part
|
// Calculate ETag for the part
|
||||||
etag := copyEntryETag(dstEntry)
|
etag := copyEntryETag(dstEntry)
|
||||||
setEtag(w, etag)
|
setEtag(w, etag)
|
||||||
|
|||||||
@@ -469,6 +469,11 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// the write above re-creates a directory an abort removed mid-body
|
||||||
|
if !s3a.checkUploadStillOpen(w, r, bucket, object, uploadID) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
glog.V(2).Infof("PutObjectPart: SUCCESS - bucket=%s, object=%s, partNumber=%d, etag=%s, sseType=%s",
|
glog.V(2).Infof("PutObjectPart: SUCCESS - bucket=%s, object=%s, partNumber=%d, etag=%s, sseType=%s",
|
||||||
bucket, object, partID, etag, sseMetadata.SSEType)
|
bucket, object, partID, etag, sseMetadata.SSEType)
|
||||||
|
|
||||||
@@ -485,6 +490,38 @@ func (s3a *S3ApiServer) genUploadsFolder(bucket string) string {
|
|||||||
return fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), s3_constants.MultipartUploadsFolder)
|
return fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), s3_constants.MultipartUploadsFolder)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isMultipartUploadEntry tells the .uploads/<id> directory createMultipartUpload
|
||||||
|
// made from the one a part write re-created on its way to the filer: only the
|
||||||
|
// former carries the destination object key.
|
||||||
|
func isMultipartUploadEntry(entry *filer_pb.Entry) bool {
|
||||||
|
return entry != nil && len(entry.Extended[s3_constants.ExtMultipartObjectKey]) > 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkUploadStillOpen re-checks the upload after a part landed, and removes the
|
||||||
|
// part along with the directory it resurrected when an abort was answered while
|
||||||
|
// the part was in flight. It reports whether the caller may answer success.
|
||||||
|
func (s3a *S3ApiServer) checkUploadStillOpen(w http.ResponseWriter, r *http.Request, bucket, object, uploadID string) bool {
|
||||||
|
entry, err := s3a.getEntry(s3a.genUploadsFolder(bucket), uploadID)
|
||||||
|
if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
||||||
|
glog.Errorf("checkUploadStillOpen %s/%s: %v", bucket, uploadID, err)
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if isMultipartUploadEntry(entry) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
// the upload is gone either way, so the client still hears NoSuchUpload
|
||||||
|
if _, code := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{
|
||||||
|
Bucket: aws.String(bucket),
|
||||||
|
Key: objectKey(aws.String(object)),
|
||||||
|
UploadId: aws.String(uploadID),
|
||||||
|
}); code != s3err.ErrNone {
|
||||||
|
glog.Warningf("checkUploadStillOpen %s/%s: part left behind, cleanup failed", bucket, uploadID)
|
||||||
|
}
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchUpload)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// getMultipartSSEAlgorithm returns the canonical SSE algorithm ("AES256" or
|
// getMultipartSSEAlgorithm returns the canonical SSE algorithm ("AES256" or
|
||||||
// "aws:kms") that was stored when the multipart upload was initiated, or ""
|
// "aws:kms") that was stored when the multipart upload was initiated, or ""
|
||||||
// if the upload entry is not found or had no SSE. It is used by the bucket
|
// if the upload entry is not found or had no SSE. It is used by the bucket
|
||||||
|
|||||||
Reference in New Issue
Block a user