Skip to content

Commit cf55e64

Browse files
committed
Test with Claude
1 parent 740c8be commit cf55e64

1 file changed

Lines changed: 131 additions & 33 deletions

File tree

  • registry/storage/driver/s3-aws

‎registry/storage/driver/s3-aws/s3.go‎

Lines changed: 131 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -682,6 +682,93 @@ func (d *driver) Writer(ctx context.Context, path string, appendMode bool) (stor
682682
return d.newWriter(ctx, key, *resp.UploadId, nil), nil
683683
}
684684

685+
// Look for the multipart upload that should already exist for this key
686+
// (created moments earlier by the non-append branch above, in a
687+
// separate request). Some S3-compatible backends (e.g. Bunny Storage)
688+
// do not make a just-created multipart upload immediately visible via
689+
// ListMultipartUploads, so retry with backoff before concluding that no
690+
// upload exists. Without this, a fast POST-then-PUT resume can lose the
691+
// race and be mistaken for a request to append to a nonexistent path.
692+
w, err := d.findMultipartUploadWithRetry(ctx, path, key)
693+
if err != nil {
694+
return nil, err
695+
}
696+
if w != nil {
697+
return w, nil
698+
}
699+
700+
// No matching multipart upload found even after retries. Stat is
701+
// expected to find a zero-byte marker object at this path if the
702+
// upload genuinely has no data yet; a NotFound here is retried too,
703+
// for the same read-after-write consistency reasons as above.
704+
fi, err := d.statWithRetry(ctx, path)
705+
if err != nil {
706+
return nil, parseError(path, err)
707+
}
708+
709+
if fi.Size() == 0 {
710+
resp, err := d.S3.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
711+
Bucket: aws.String(d.Bucket),
712+
Key: aws.String(key),
713+
ContentType: d.getContentType(),
714+
ACL: d.getACL(),
715+
ServerSideEncryption: d.getEncryptionMode(),
716+
SSEKMSKeyId: d.getSSEKMSKeyID(),
717+
StorageClass: d.getStorageClass(),
718+
})
719+
if err != nil {
720+
return nil, err
721+
}
722+
return d.newWriter(ctx, key, *resp.UploadId, nil), nil
723+
}
724+
return nil, storagedriver.Error{
725+
DriverName: driverName,
726+
Detail: fmt.Errorf("append to zero-size path %s unsupported", path),
727+
}
728+
}
729+
730+
// listMultipartRetryAttempts and listMultipartRetryBaseDelay control
731+
// findMultipartUploadWithRetry's backoff. Total worst-case delay across all
732+
// attempts is roughly 1.5s (150+300+600+... capped), chosen to comfortably
733+
// cover the ListMultipartUploads visibility lag observed against Bunny
734+
// Storage without materially slowing down the common case where the upload
735+
// is already visible on the first call.
736+
const listMultipartRetryAttempts = 4
737+
const listMultipartRetryBaseDelay = 150 * time.Millisecond
738+
739+
// findMultipartUploadWithRetry looks for an in-progress multipart upload
740+
// matching key, retrying with backoff if none is found on the first pass.
741+
// Returns a nil writer and nil error if no matching upload is found after
742+
// all retries are exhausted — this is not itself an error, since the upload
743+
// may genuinely not exist yet (see the zero-size fallback in Writer).
744+
func (d *driver) findMultipartUploadWithRetry(ctx context.Context, path, key string) (storagedriver.FileWriter, error) {
745+
delay := listMultipartRetryBaseDelay
746+
for attempt := 0; attempt < listMultipartRetryAttempts; attempt++ {
747+
w, err := d.findMultipartUploadOnce(ctx, path, key)
748+
if err != nil {
749+
return nil, err
750+
}
751+
if w != nil {
752+
return w, nil
753+
}
754+
if attempt == listMultipartRetryAttempts-1 {
755+
break
756+
}
757+
select {
758+
case <-ctx.Done():
759+
return nil, ctx.Err()
760+
case <-time.After(delay):
761+
}
762+
delay *= 2
763+
}
764+
return nil, nil
765+
}
766+
767+
// findMultipartUploadOnce performs a single, fully paginated pass over
768+
// ListMultipartUploads looking for an upload matching key. Returns a nil
769+
// writer and nil error (not an error) if no matching upload is found in
770+
// this pass.
771+
func (d *driver) findMultipartUploadOnce(ctx context.Context, path, key string) (storagedriver.FileWriter, error) {
685772
listMultipartUploadsInput := &s3.ListMultipartUploadsInput{
686773
Bucket: aws.String(d.Bucket),
687774
Prefix: aws.String(key),
@@ -692,37 +779,6 @@ func (d *driver) Writer(ctx context.Context, path string, appendMode bool) (stor
692779
return nil, parseError(path, err)
693780
}
694781

695-
// resp.Uploads can only be empty on the first call
696-
// if there were no more results to return after the first call, resp.IsTruncated would have been false
697-
// and the loop would be exited without recalling ListMultipartUploads
698-
if len(resp.Uploads) == 0 {
699-
fi, err := d.Stat(ctx, path)
700-
if err != nil {
701-
return nil, parseError(path, err)
702-
}
703-
704-
if fi.Size() == 0 {
705-
resp, err := d.S3.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
706-
Bucket: aws.String(d.Bucket),
707-
Key: aws.String(key),
708-
ContentType: d.getContentType(),
709-
ACL: d.getACL(),
710-
ServerSideEncryption: d.getEncryptionMode(),
711-
SSEKMSKeyId: d.getSSEKMSKeyID(),
712-
StorageClass: d.getStorageClass(),
713-
})
714-
if err != nil {
715-
return nil, err
716-
}
717-
return d.newWriter(ctx, key, *resp.UploadId, nil), nil
718-
}
719-
return nil, storagedriver.Error{
720-
DriverName: driverName,
721-
Detail: fmt.Errorf("append to zero-size path %s unsupported", path),
722-
}
723-
}
724-
725-
var allParts []*s3.Part
726782
for _, multi := range resp.Uploads {
727783
if key != *multi.Key {
728784
continue
@@ -736,7 +792,7 @@ func (d *driver) Writer(ctx context.Context, path string, appendMode bool) (stor
736792
if err != nil {
737793
return nil, parseError(path, err)
738794
}
739-
allParts = append(allParts, partsList.Parts...)
795+
allParts := partsList.Parts
740796
for *partsList.IsTruncated {
741797
partsList, err = d.S3.ListPartsWithContext(ctx, &s3.ListPartsInput{
742798
Bucket: aws.String(d.Bucket),
@@ -761,7 +817,7 @@ func (d *driver) Writer(ctx context.Context, path string, appendMode bool) (stor
761817
break
762818
}
763819
}
764-
return nil, storagedriver.PathNotFoundError{Path: path}
820+
return nil, nil
765821
}
766822

767823
func (d *driver) statHead(ctx context.Context, path string) (*storagedriver.FileInfoFields, error) {
@@ -836,6 +892,48 @@ func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo,
836892
return storagedriver.FileInfoInternal{FileInfoFields: *fi}, nil
837893
}
838894

895+
// statRetryAttempts and statRetryBaseDelay control statWithRetry's backoff.
896+
// Total worst-case delay across all attempts is roughly 700ms
897+
// (100+200+400), which comfortably covers the replication lag observed
898+
// against Bunny Storage without materially slowing down the common case
899+
// where the object is already visible.
900+
const statRetryAttempts = 4
901+
const statRetryBaseDelay = 100 * time.Millisecond
902+
903+
// statWithRetry calls Stat and, if it fails with a PathNotFoundError,
904+
// retries with exponential backoff before giving up. This exists because
905+
// some S3-compatible backends (e.g. Bunny Storage) do not guarantee
906+
// read-after-write consistency: an object written moments ago by a
907+
// different request may not yet be visible to a HeadObject/ListObjects
908+
// call. Errors other than PathNotFoundError are returned immediately
909+
// without retrying, since retrying won't help a permissions or network
910+
// error and would only slow down the failure.
911+
func (d *driver) statWithRetry(ctx context.Context, path string) (storagedriver.FileInfo, error) {
912+
var fi storagedriver.FileInfo
913+
var err error
914+
915+
delay := statRetryBaseDelay
916+
for attempt := 0; attempt < statRetryAttempts; attempt++ {
917+
fi, err = d.Stat(ctx, path)
918+
if err == nil {
919+
return fi, nil
920+
}
921+
if _, ok := err.(storagedriver.PathNotFoundError); !ok {
922+
return nil, err
923+
}
924+
if attempt == statRetryAttempts-1 {
925+
break
926+
}
927+
select {
928+
case <-ctx.Done():
929+
return nil, ctx.Err()
930+
case <-time.After(delay):
931+
}
932+
delay *= 2
933+
}
934+
return nil, err
935+
}
936+
839937
// List returns a list of the objects that are direct descendants of the given path.
840938
func (d *driver) List(ctx context.Context, opath string) ([]string, error) {
841939
path := opath

0 commit comments

Comments
 (0)