From 7bb0a1c127a0e82667c405036334ecefdf1c28a4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 28 Aug 2026 14:30:21 -0700 Subject: [PATCH] s3: replay a delete whose reply the transport dropped (#11022) * s3: stop retrying a delete the filer refused for a non-empty folder The filer looked and the children are there, so the answer will not change. retryFilerOp spent six attempts and up to 3.1s of backoff on it before the caller could act on the condition it was already holding. Claude-Session: https://claude.ai/code/session_01XqaJrwgXQ5GSUpyzRbe5nD * s3: thread the request context through the unversioned delete path doDeleteEntry issued every DeleteEntry on context.Background(), so an S3 client that hung up left the gateway working on its behalf, out of reach of both cancellation and the per-request retry allowance that DeleteMultipleObjectsHandler installs. Claude-Session: https://claude.ai/code/session_01XqaJrwgXQ5GSUpyzRbe5nD * s3: treat a cancelled filer RPC as terminal, not transient isRetryableFilerErr matched context.Canceled and DeadlineExceeded by sentinel, which only holds while the error is still local. Once it has crossed gRPC it is a status, so an abandoned request was retried six times on behalf of a caller that had already gone. Claude-Session: https://claude.ai/code/session_01XqaJrwgXQ5GSUpyzRbe5nD * s3: replay a delete whose reply the transport dropped A delete is idempotent at the filer, which answers an entry that is already gone with an empty resp.Error, so a reply lost in transit can be reissued rather than surfaced. Surfaced, it becomes a 500 on the bucket delete, which boto3 resends and is then answered NoSuchBucket, or a per-key InternalError inside the 200 of a multi-object delete, which no SDK retries at all. The replay runs through retryFilerOp, so it draws on the allowance the request already installs rather than paying a backoff per key, and stops for a caller that has gone. rm and rmObject re-enter WithFilerClient per attempt, so each one walks the failover list again on a connection the failed attempt had invalidated; the multi-object loop holds one client for the batch, so there the replay reuses it. Classification stays structural. The filer reports its own refusals in resp.Error, which carries no status and has the deleted path - and, for a recursive delete, the children it stopped on - formatted into it, so no key name can steer the decision either way. rm and rmObject now take the caller's context. Cleanup and rollback paths pass context.Background() deliberately: they have to run whether or not the caller is still waiting. Claude-Session: https://claude.ai/code/session_01XqaJrwgXQ5GSUpyzRbe5nD * s3: share one retry allowance across multipart completion cleanup The unused-entry loop deletes once per entry, and each delete now retries, so a filer that stays unavailable held the response for 3.1s per entry after the object was already committed. Claude-Session: https://claude.ai/code/session_01XqaJrwgXQ5GSUpyzRbe5nD --- weed/s3api/filer_multipart.go | 13 +- weed/s3api/filer_util.go | 38 +++-- weed/s3api/filer_util_delete_test.go | 158 ++++++++++++++++-- weed/s3api/s3api_bucket_handlers.go | 7 +- weed/s3api/s3api_bucket_owner_index.go | 3 +- weed/s3api/s3api_directory_marker.go | 4 +- weed/s3api/s3api_internal_lifecycle.go | 4 +- weed/s3api/s3api_object_handlers_copy.go | 4 +- .../s3api_object_handlers_copy_part_sse.go | 2 +- weed/s3api/s3api_object_handlers_delete.go | 16 +- .../s3api_object_handlers_delete_test.go | 11 +- weed/s3api/s3api_object_handlers_put.go | 4 +- weed/s3api/s3api_object_handlers_rename.go | 2 +- weed/s3api/s3api_object_versioning.go | 47 +++--- 14 files changed, 237 insertions(+), 76 deletions(-) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 4752c904e..b1d847fe6 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -2,6 +2,7 @@ package s3api import ( "cmp" + "context" "crypto/md5" "crypto/rand" "encoding/base64" @@ -936,12 +937,16 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl } if completionState != nil { + // The object is already committed and the client is still waiting, so the + // cleanup below runs on its own context but spends one allowance between + // all of it rather than a retry backoff per unused entry. + cleanupCtx := withFilerRetryBudget(context.Background(), filerRetryRequestBudget) for _, deleteEntry := range completionState.deleteEntries { - if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil { + if err := s3a.rm(cleanupCtx, uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil { glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err) } } - if err := s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil { + if err := s3a.rm(cleanupCtx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil { glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err) } if len(completionState.supersededPartManifests) > 0 { @@ -955,7 +960,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl // Metadata-only: the version file's chunks are the still-registered parts' // chunks, which a retried completion needs. func (s3a *S3ApiServer) rollbackMultipartVersion(versionDir, versionFileName string) error { - return s3a.rmObject(versionDir, versionFileName, false, false) + return s3a.rmObject(context.Background(), versionDir, versionFileName, false, false) } func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) { @@ -995,7 +1000,7 @@ func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput return nil, s3err.ErrInternalError } if exists { - err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true) + err = s3a.rm(context.Background(), s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true) } if err != nil { glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err) diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go index 62a1000c3..240e3b18f 100644 --- a/weed/s3api/filer_util.go +++ b/weed/s3api/filer_util.go @@ -94,26 +94,36 @@ func listWithRetry(parentDirectoryPath string, doList func() (entries []*filer_p } -func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { +// A delete is idempotent at the filer, which answers an entry that is already +// gone with an empty resp.Error, so a reply the transport dropped can be +// reissued instead of surfaced: as a 500 on the bucket delete, or as a per-key +// InternalError inside the 200 of a multi-object delete, which no SDK retries. +// Each attempt re-enters WithFilerClient, so it walks the failover list again +// on a connection the failed one had invalidated. +func (s3a *S3ApiServer) rm(ctx context.Context, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { - return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return retryFilerOp(ctx, "rm "+parentDirectoryPath+"/"+entryName, func() error { + return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive) + return doDeleteEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive) + }) }) } -func (s3a *S3ApiServer) rmObject(parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { +func (s3a *S3ApiServer) rmObject(ctx context.Context, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { - return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return retryFilerOp(ctx, "rmObject "+parentDirectoryPath+"/"+entryName, func() error { + return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return deleteObjectEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive) + return deleteObjectEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive) + }) }) } -func deleteObjectEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { - err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive) +func deleteObjectEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error { + err := doDeleteEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive) if err == nil { return nil } @@ -121,10 +131,10 @@ func deleteObjectEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath, return err } - return demoteDirectoryMarkerToImplicitDirectory(client, parentDirectoryPath, entryName) + return demoteDirectoryMarkerToImplicitDirectory(ctx, client, parentDirectoryPath, entryName) } -func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error { +func doDeleteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error { request := &filer_pb.DeleteEntryRequest{ Directory: parentDirectoryPath, Name: entryName, @@ -134,7 +144,7 @@ func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath strin } glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request) - if resp, err := client.DeleteEntry(context.Background(), request); err != nil { + if resp, err := client.DeleteEntry(ctx, request); err != nil { glog.V(1).Infof("delete entry %v: %v", request, err) return fmt.Errorf("delete entry %s/%s: %w", parentDirectoryPath, entryName, err) } else { @@ -147,8 +157,8 @@ func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath strin return nil } -func demoteDirectoryMarkerToImplicitDirectory(client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string) error { - resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ +func demoteDirectoryMarkerToImplicitDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string) error { + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ Directory: parentDirectoryPath, Name: entryName, }) @@ -167,7 +177,7 @@ func demoteDirectoryMarkerToImplicitDirectory(client filer_pb.SeaweedFilerClient clearDirectoryMarkerMetadata(resp.Entry) - if err := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ + if err := filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ Directory: parentDirectoryPath, Entry: resp.Entry, }); err != nil { diff --git a/weed/s3api/filer_util_delete_test.go b/weed/s3api/filer_util_delete_test.go index 47c6b9428..8ee94a4d3 100644 --- a/weed/s3api/filer_util_delete_test.go +++ b/weed/s3api/filer_util_delete_test.go @@ -3,7 +3,9 @@ package s3api import ( "context" "errors" + "fmt" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -24,13 +26,34 @@ type deleteObjectEntryTestClient struct { lookupErr error updateErr error - deleteReq *filer_pb.DeleteEntryRequest - lookupReq *filer_pb.LookupDirectoryEntryRequest - updateReq *filer_pb.UpdateEntryRequest + // deleteCallErrs scripts the RPC itself per attempt, deleteRespErrs the + // filer's own message: the two ways a delete fails, told apart because + // only the first is a transport that dropped a reply. + deleteCallErrs []error + deleteRespErrs []string + + deleteAttempts int + deleteReqs []*filer_pb.DeleteEntryRequest + deleteReq *filer_pb.DeleteEntryRequest + lookupReq *filer_pb.LookupDirectoryEntryRequest + updateReq *filer_pb.UpdateEntryRequest } -func (c *deleteObjectEntryTestClient) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { +func (c *deleteObjectEntryTestClient) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + attempt := c.deleteAttempts + c.deleteAttempts++ c.deleteReq = req + c.deleteReqs = append(c.deleteReqs, req) + + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + if attempt < len(c.deleteCallErrs) && c.deleteCallErrs[attempt] != nil { + return nil, c.deleteCallErrs[attempt] + } + if attempt < len(c.deleteRespErrs) && c.deleteRespErrs[attempt] != "" { + return &filer_pb.DeleteEntryResponse{Error: c.deleteRespErrs[attempt]}, nil + } if c.deleteResp == nil { return &filer_pb.DeleteEntryResponse{}, c.deleteErr } @@ -77,7 +100,7 @@ func TestDeleteObjectEntryDemotesNonEmptyDirectoryMarker(t *testing.T) { }, } - err := deleteObjectEntry(client, "/buckets/test", "photos", true, false) + err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false) require.NoError(t, err) require.NotNil(t, client.lookupReq) require.NotNil(t, client.updateReq) @@ -116,7 +139,7 @@ func TestDeleteObjectEntryDemotesPrefixObject(t *testing.T) { }, } - require.NoError(t, deleteObjectEntry(client, "/buckets/test", "photos", true, false)) + require.NoError(t, deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false)) require.NotNil(t, client.updateReq) updated := client.updateReq.Entry @@ -140,7 +163,7 @@ func TestDeleteObjectEntryTreatsImplicitDirectoryAsSuccessfulNoop(t *testing.T) }, } - err := deleteObjectEntry(client, "/buckets/test", "photos", true, false) + err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false) require.NoError(t, err) require.NotNil(t, client.lookupReq) assert.Nil(t, client.updateReq) @@ -163,7 +186,7 @@ func TestDeleteObjectEntryIgnoresConcurrentUpdateNotFound(t *testing.T) { updateErr: status.Error(codes.NotFound, "already removed"), } - err := deleteObjectEntry(client, "/buckets/test", "photos", true, false) + err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false) require.NoError(t, err) require.NotNil(t, client.lookupReq) require.NotNil(t, client.updateReq) @@ -180,7 +203,7 @@ func TestDeleteObjectEntryIgnoresMarkerSpoofedByKey(t *testing.T) { }, } - err := deleteObjectEntry(client, "/buckets/test", name, true, false) + err := deleteObjectEntry(context.Background(), client, "/buckets/test", name, true, false) require.Error(t, err) assert.Contains(t, err.Error(), "disk full") assert.Nil(t, client.lookupReq) @@ -192,9 +215,124 @@ func TestDeleteObjectEntryPropagatesNonDirectoryDeleteErrors(t *testing.T) { deleteErr: errors.New("boom"), } - err := deleteObjectEntry(client, "/buckets/test", "photos", true, false) + err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false) require.Error(t, err) assert.Contains(t, err.Error(), "boom") assert.Nil(t, client.lookupReq) assert.Nil(t, client.updateReq) } + +// A delete is idempotent at the filer, so a reply the transport dropped is +// reissued rather than surfaced. Issue #7204 saw the surfaced version as a 500 +// on the bucket delete, which boto3 resent and then answered NoSuchBucket; +// issue #7224 saw it as a per-key InternalError inside a 200, which no SDK +// retries, so the object silently stayed. +func TestDeleteUnversionedObjectReplaysADroppedReply(t *testing.T) { + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + client := &deleteObjectEntryTestClient{ + deleteCallErrs: []error{status.Error(codes.Unavailable, "transport is closing"), nil}, + } + + err := s3a.deleteUnversionedObjectWithClient(testRetryCtx(10*time.Millisecond), client, "mybucket", "a/b/c.txt", false) + + require.NoError(t, err) + assert.Equal(t, 2, client.deleteAttempts) + // the replay must ask for the same delete, not a differently shaped one + require.Len(t, client.deleteReqs, 2) + assert.Equal(t, client.deleteReqs[0].String(), client.deleteReqs[1].String()) +} + +// The filer formats the deleted path, and for a recursive delete the child it +// stopped on, into the message it sends back, so both are the client's text. +// Nothing here reads that text, and this pins it: the same failure under a key +// named after a transport condition is replayed exactly as often as under a +// plain one. +func TestDeleteUnversionedObjectReplayIgnoresTheObjectName(t *testing.T) { + attemptsFor := func(key string) int { + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + client := &deleteObjectEntryTestClient{ + deleteRespErrs: []string{"delete file /buckets/b/" + key + ": permission denied", ""}, + } + require.NoError(t, s3a.deleteUnversionedObjectWithClient(testRetryCtx(10*time.Millisecond), client, "b", key, false)) + return client.deleteAttempts + } + + plain := attemptsFor("plain.log") + for _, key := range []string{"transport.log", "unavailable.txt", "slowdown.csv", filer_pb.ErrNotFound.Error()} { + assert.Equal(t, plain, attemptsFor(key), "key %q must not steer the replay", key) + } +} + +// A folder the filer refused because it still has children is an answer about +// the tree, not a hiccup: it is handed straight to the demote path instead of +// being replayed until the allowance runs out. +func TestDeleteUnversionedObjectDoesNotReplayANonEmptyFolder(t *testing.T) { + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + client := &deleteObjectEntryTestClient{ + deleteResp: &filer_pb.DeleteEntryResponse{ + Error: filer.MsgFailDelNonEmptyFolder + ": /buckets/b/photos", + }, + } + + require.NoError(t, s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "photos", false)) + assert.Equal(t, 1, client.deleteAttempts) + assert.NotNil(t, client.lookupReq, "the demote must still run") +} + +// A caller that has gone away is not worth replaying for. +func TestDeleteUnversionedObjectDoesNotReplayForACancelledCaller(t *testing.T) { + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + client := &deleteObjectEntryTestClient{} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := s3a.deleteUnversionedObjectWithClient(ctx, client, "b", "k", false) + + require.Error(t, err) + assert.Equal(t, 1, client.deleteAttempts) +} + +// The client picks how many keys a multi-object delete carries, so the backoff +// is drawn from one allowance held by the request rather than one per key. +func TestDeleteUnversionedObjectDrawsFromTheRequestAllowance(t *testing.T) { + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + ctx := testRetryCtx(20 * time.Millisecond) + + start := time.Now() + var attempts int + for i := 0; i < 50; i++ { + client := &deleteObjectEntryTestClient{deleteErr: status.Error(codes.Unavailable, "transport is closing")} + require.Error(t, s3a.deleteUnversionedObjectWithClient(ctx, client, "b", fmt.Sprintf("k%d", i), false)) + attempts += client.deleteAttempts + } + + assert.Less(t, time.Since(start), time.Second, "one allowance, not one per key") + assert.Greater(t, attempts, 50, "the allowance must still buy a replay") +} + +// testRetryCtx caps the backoff so a test exercises the replay without waiting +// out the real per-request allowance. +func testRetryCtx(allowance time.Duration) context.Context { + return withFilerRetryBudget(context.Background(), allowance) +} + +func TestIsRetryableFilerErr(t *testing.T) { + assert.False(t, isRetryableFilerErr(nil)) + assert.True(t, isRetryableFilerErr(status.Error(codes.Unavailable, "transport is closing"))) + assert.True(t, isRetryableFilerErr(fmt.Errorf("delete entry /b/k: %w", status.Error(codes.Unavailable, "x")))) + // the filer's own message carries no status, and a store error may well clear + assert.True(t, isRetryableFilerErr(filer.DeleteEntryError("delete file /b/k: filer store delete: disk full"))) + + assert.False(t, isRetryableFilerErr(filer.DeleteEntryError(filer.MsgFailDelNonEmptyFolder+": /b/photos"))) + assert.False(t, isRetryableFilerErr(filer_pb.ErrNotFound)) + assert.False(t, isRetryableFilerErr(status.Error(codes.NotFound, "gone"))) + assert.False(t, isRetryableFilerErr(context.Canceled)) + assert.False(t, isRetryableFilerErr(status.Error(codes.Canceled, "context canceled"))) + assert.False(t, isRetryableFilerErr(status.Error(codes.DeadlineExceeded, "context deadline exceeded"))) + + // the message is never read, so the same code classifies the same way + // whatever the client named the key + assert.Equal(t, + isRetryableFilerErr(status.Error(codes.PermissionDenied, "denied for /b/plain.log")), + isRetryableFilerErr(status.Error(codes.PermissionDenied, "denied for /b/transport.log"))) +} diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go index 5cc7afd49..5b335089e 100644 --- a/weed/s3api/s3api_bucket_handlers.go +++ b/weed/s3api/s3api_bucket_handlers.go @@ -370,7 +370,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) // This ensures we don't leave a bucket without the requested Object Lock configuration if objectLockSetupError != nil { glog.Errorf("PutBucketHandler: rolling back bucket %s creation due to Object Lock setup failure: %v", bucket, objectLockSetupError) - if deleteErr := s3a.rm(s3a.option.BucketsPath, bucket, true, true); deleteErr != nil { + if deleteErr := s3a.rm(context.Background(), s3a.option.BucketsPath, bucket, true, true); deleteErr != nil { glog.Errorf("PutBucketHandler: failed to rollback bucket %s after Object Lock setup failure: %v", bucket, deleteErr) } s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) @@ -414,6 +414,9 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque bucket, _ := s3_constants.GetBucketAndObject(r) glog.V(3).Infof("DeleteBucketHandler %s", bucket) + // The teardown below retries, and a failover walk repeats it once per + // filer, so the backoff comes out of one allowance held here. + r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget)) if s3a.isTableBucket(bucket) { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) @@ -463,7 +466,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque // the "collection exists but bucket directory missing" inconsistency that blocks // bucket recreation. An orphaned collection is harmless and will be cleaned up // or reused when the bucket is recreated. - err := s3a.rm(s3a.option.BucketsPath, bucket, false, true) + err := s3a.rm(r.Context(), s3a.option.BucketsPath, bucket, false, true) if err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) return diff --git a/weed/s3api/s3api_bucket_owner_index.go b/weed/s3api/s3api_bucket_owner_index.go index 977aa39f4..b69cbc66a 100644 --- a/weed/s3api/s3api_bucket_owner_index.go +++ b/weed/s3api/s3api_bucket_owner_index.go @@ -1,6 +1,7 @@ package s3api import ( + "context" "net/http" "net/url" "slices" @@ -67,7 +68,7 @@ func (s3a *S3ApiServer) addBucketToOwnerIndex(owner, bucket string, crtime int64 } func (s3a *S3ApiServer) removeBucketFromOwnerIndex(owner, bucket string) error { - return s3a.rm(s3a.bucketOwnerDir(owner), bucket, false, false) + return s3a.rm(context.Background(), s3a.bucketOwnerDir(owner), bucket, false, false) } // maintainBucketOwnerIndex applies a /buckets metadata event to the owner diff --git a/weed/s3api/s3api_directory_marker.go b/weed/s3api/s3api_directory_marker.go index 41577bae7..0c97382a5 100644 --- a/weed/s3api/s3api_directory_marker.go +++ b/weed/s3api/s3api_directory_marker.go @@ -93,7 +93,7 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object st break } } - if rmErr := s3a.rm(markerDir, s3_constants.VersionsFolder, true, true); rmErr != nil { + if rmErr := s3a.rm(r.Context(), markerDir, s3_constants.VersionsFolder, true, true); rmErr != nil { glog.Errorf("deleteDirectoryMarker: failed to remove stale history of %s/%s: %v", bucket, object, rmErr) return s3err.ErrInternalError } @@ -103,7 +103,7 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object st } if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false) + return s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object, false) }); err != nil { glog.Errorf("deleteDirectoryMarker: failed to delete %s/%s: %v", bucket, object, err) return s3err.ErrInternalError diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 164a999ba..f32c478c7 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -88,7 +88,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle return done(), nil default: err := s3a.WithFilerClient(false, func(c filer_pb.SeaweedFilerClient) error { - return s3a.deleteUnversionedObjectWithClient(c, req.Bucket, req.ObjectPath, metadataOnly) + return s3a.deleteUnversionedObjectWithClient(ctx, c, req.Bucket, req.ObjectPath, metadataOnly) }) if err != nil { if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrObjectNotFound) { @@ -179,7 +179,7 @@ func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle if !exists { return noopResolved("NOT_FOUND"), nil } - if err := s3a.rm(uploadsFolder, uploadID, true, true); err != nil { + if err := s3a.rm(ctx, uploadsFolder, uploadID, true, true); err != nil { if errors.Is(err, filer_pb.ErrNotFound) { return noopResolved("NOT_FOUND_AT_DELETE"), nil } diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 775852c35..f1e69b6e9 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -643,7 +643,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion func (s3a *S3ApiServer) rollbackCopyVersion(bucketDir, versionObjectPath string) error { versionPath := util.FullPath(fmt.Sprintf("%s/%s", bucketDir, versionObjectPath)) versionDir, versionName := versionPath.DirAndName() - return s3a.rmObject(versionDir, versionName, true, false) + return s3a.rmObject(context.Background(), versionDir, versionName, true, false) } func (s3a *S3ApiServer) resolveCopySourceEntry(bucket, object, versionId, versioningState string) (*filer_pb.Entry, error) { @@ -1090,7 +1090,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req // Save the part entry to the multipart uploads folder // Check if part exists and remove it first (allow re-copying same part) if exists, _ := s3a.exists(uploadDir, partName, false); exists { - if err := s3a.rm(uploadDir, partName, false, false); err != nil { + if err := s3a.rm(r.Context(), uploadDir, partName, false, false); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) return } diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go index 1bcc05a25..7a4968fa9 100644 --- a/weed/s3api/s3api_object_handlers_copy_part_sse.go +++ b/weed/s3api/s3api_object_handlers_copy_part_sse.go @@ -415,7 +415,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption( func (s3a *S3ApiServer) writeEmptyCopyPart(dstBucket, uploadID string, partID int) (string, s3err.ErrorCode) { uploadDir, partName := s3a.copyPartLocation(dstBucket, uploadID, partID) if exists, _ := s3a.exists(uploadDir, partName, false); exists { - if err := s3a.rm(uploadDir, partName, false, false); err != nil { + if err := s3a.rm(context.Background(), uploadDir, partName, false, false); err != nil { return "", s3err.ErrInternalError } } diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 5f42bc9f5..a21a73fe2 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -1,6 +1,7 @@ package s3api import ( + "context" "encoding/xml" "errors" "io" @@ -189,19 +190,26 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v // relies on the volume's natural TTL to reclaim chunks; pass true only // when the entry's Attributes.TtlSec > 0 so the volume is guaranteed to // drop the chunks on its own. -func (s3a *S3ApiServer) deleteUnversionedObjectWithClient(client filer_pb.SeaweedFilerClient, bucket, object string, metadataOnly bool) error { +func (s3a *S3ApiServer) deleteUnversionedObjectWithClient(ctx context.Context, client filer_pb.SeaweedFilerClient, bucket, object string, metadataOnly bool) error { if !s3_constants.IsValidBucketName(bucket) || !s3_constants.IsValidObjectKey(object) { return errors.New("invalid bucket or object path") } target := util.NewFullPath(s3a.bucketDir(bucket), object) dir, name := target.DirAndName() - return deleteObjectEntry(client, dir, name, !metadataOnly, false) + // The caller holds one client for a whole batch, so a dropped reply is + // replayed on that client rather than by re-entering WithFilerClient. + return retryFilerOp(ctx, "delete "+string(target), func() error { + return deleteObjectEntry(ctx, client, dir, name, !metadataOnly, false) + }) } func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) { bucket, object := s3_constants.GetBucketAndObject(r) glog.Infof("DeleteObjectHandler %s %s", bucket, object) + // The filer ops below each retry, and a failover walk runs the whole set + // once per filer, so the backoff comes out of one allowance held here. + r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget)) if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return @@ -326,7 +334,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque } if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false) + return s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object, false) }); err != nil { glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err) return s3err.ErrInternalError @@ -493,7 +501,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h return s3a.deleteDirectoryMarker(r, bucket, object.Key) } - if err := s3a.deleteUnversionedObjectWithClient(client, bucket, object.Key, false); err != nil { + if err := s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object.Key, false); err != nil { glog.Errorf("DeleteMultipleObjectsHandler: failed to delete %s/%s: %v", bucket, object.Key, err) return s3err.ErrInternalError } diff --git a/weed/s3api/s3api_object_handlers_delete_test.go b/weed/s3api/s3api_object_handlers_delete_test.go index b2104e257..dd680d6c0 100644 --- a/weed/s3api/s3api_object_handlers_delete_test.go +++ b/weed/s3api/s3api_object_handlers_delete_test.go @@ -1,6 +1,7 @@ package s3api import ( + "context" "errors" "testing" @@ -47,7 +48,7 @@ func TestDeleteUnversionedObjectWithClient_MetadataOnlySkipsChunkDelete(t *testi s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} client := &deleteObjectEntryTestClient{} - err := s3a.deleteUnversionedObjectWithClient(client, "b", "k", true) + err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", true) require.NoError(t, err) require.NotNil(t, client.deleteReq) assert.Equal(t, "/buckets/b", client.deleteReq.Directory) @@ -61,7 +62,7 @@ func TestDeleteUnversionedObjectWithClient_FullDeletePreservesIsDeleteData(t *te s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} client := &deleteObjectEntryTestClient{} - err := s3a.deleteUnversionedObjectWithClient(client, "b", "k", false) + err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", false) require.NoError(t, err) require.NotNil(t, client.deleteReq) assert.True(t, client.deleteReq.IsDeleteData, "default delete must keep IsDeleteData true") @@ -75,7 +76,7 @@ func TestDeleteUnversionedObjectWithClient_FullPathFromBucketsRoot(t *testing.T) s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} client := &deleteObjectEntryTestClient{} - err := s3a.deleteUnversionedObjectWithClient(client, "mybucket", "a/b/c.txt", false) + err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "mybucket", "a/b/c.txt", false) require.NoError(t, err) require.NotNil(t, client.deleteReq) assert.Equal(t, "/buckets/mybucket/a/b", client.deleteReq.Directory) @@ -86,7 +87,7 @@ func TestDeleteUnversionedObjectWithClientRejectsTraversal(t *testing.T) { s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} client := &deleteObjectEntryTestClient{} - err := s3a.deleteUnversionedObjectWithClient(client, "source-bucket", "../victim-bucket/secret", false) + err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "source-bucket", "../victim-bucket/secret", false) require.Error(t, err) assert.Nil(t, client.deleteReq, "invalid path must be rejected before a filer delete RPC") @@ -102,7 +103,7 @@ func TestDeleteUnversionedObjectWithClient_PropagatesEntryAttributesIrrelevant(t deleteResp: &filer_pb.DeleteEntryResponse{}, } - require.NoError(t, s3a.deleteUnversionedObjectWithClient(client, "b", "k", true)) + require.NoError(t, s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", true)) require.NotNil(t, client.deleteReq) assert.False(t, client.deleteReq.IsDeleteData) } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index a60231551..53a857272 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -933,7 +933,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader entryCreated = true if finalize != nil && finalize.afterCreate != nil { if afterCreateCode := finalize.afterCreate(entry); afterCreateCode != s3err.ErrNone { - rollbackErr = s3a.rmObject(path.Dir(filePath), path.Base(filePath), true, false) + rollbackErr = s3a.rmObject(context.Background(), path.Dir(filePath), path.Base(filePath), true, false) if rollbackErr != nil { glog.Errorf("putToFiler: failed to rollback created entry for %s after post-create error: %v", filePath, rollbackErr) } else { @@ -1470,7 +1470,7 @@ func (s3a *S3ApiServer) removeNullVersionFile(bucket, object string) { if string(entry.Extended[s3_constants.ExtVersionIdKey]) != "null" { continue } - if rmErr := s3a.rm(versionsDir, entry.Name, true, false); rmErr != nil { + if rmErr := s3a.rm(context.Background(), versionsDir, entry.Name, true, false); rmErr != nil { glog.Warningf("removeNullVersionFile: %s/%s: %v", bucket, object, rmErr) } return diff --git a/weed/s3api/s3api_object_handlers_rename.go b/weed/s3api/s3api_object_handlers_rename.go index a2275c89a..d4d45112c 100644 --- a/weed/s3api/s3api_object_handlers_rename.go +++ b/weed/s3api/s3api_object_handlers_rename.go @@ -289,7 +289,7 @@ func (s3a *S3ApiServer) renameKeyHoldingNestedKeys(bucket, srcObject, dstObject // The destination holds copies now, so the source's own chunks go with it. srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName() - if err := s3a.rmObject(srcDir, srcName, true, false); err != nil { + if err := s3a.rmObject(context.Background(), srcDir, srcName, true, false); err != nil { glog.Errorf("RenameObject %s: strip %s: %v", bucket, srcObject, err) return s3err.ErrInternalError } diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 44ce312c5..29986cbbd 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -1144,7 +1144,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket, // it falls back to for an entry other keys are nested under writes the entry // back under that parent - so a key with a slash in it has to be split first. dir, name := util.NewFullPath(bucketDir, normalizedObject).DirAndName() - deleteErr := s3a.rmObject(dir, name, !metadataOnly, false) + deleteErr := s3a.rmObject(ctx, dir, name, !metadataOnly, false) if deleteErr != nil { // Check if file was already deleted by another process if _, checkErr := s3a.getEntry(bucketDir, normalizedObject); checkErr != nil { @@ -1193,7 +1193,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket, // Attempt to delete the version file // Note: We don't check if the file exists first to avoid race conditions // The deletion operation should be idempotent - deleteErr := s3a.rm(versionsDir, versionFile, !metadataOnly, false) + deleteErr := s3a.rm(ctx, versionsDir, versionFile, !metadataOnly, false) if deleteErr != nil { // Check if file was already deleted by another process (race condition handling) if _, checkErr := s3a.getEntry(versionsDir, versionFile); checkErr != nil { @@ -1212,7 +1212,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket, // down. Non-recursive: any orphan from older code paths leaves // the directory in place for the empty-folder cleaner or our // reconciler to handle. - if rmErr := s3a.rm(s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder, true, false); rmErr != nil { + if rmErr := s3a.rm(ctx, s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder, true, false); rmErr != nil { glog.V(2).Infof("deleteSpecificObjectVersion: deferring .versions/ teardown for %s/%s: %v", bucket, normalizedObject, rmErr) } case isLatestVersion && !prePointerRolled: @@ -1442,6 +1442,8 @@ func (b *filerRetryBudget) take(d time.Duration) (time.Duration, bool) { // - NotFound: the entry genuinely doesn't exist. Retrying won't make // it appear, and callers (e.g. repointLatestBeforeDeletion) want // to act on this directly. +// - non-empty folder: the filer looked and the children are there, so +// the answer will not change; callers act on it directly too. // - context.Canceled / DeadlineExceeded: the request was aborted by // the client or hit a deadline. Continuing to retry just delays // the failure return. @@ -1452,12 +1454,18 @@ func isRetryableFilerErr(err error) bool { if err == nil { return false } - if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound { + if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, filer.ErrNonEmptyFolder) { return false } if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return false } + // the same conditions once they have crossed gRPC, where a cancelled + // context arrives as a status and no longer matches the sentinel above + switch status.Code(err) { + case codes.NotFound, codes.Canceled, codes.DeadlineExceeded: + return false + } return true } @@ -1650,41 +1658,28 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(ctx context.Context, bu // object is correctly absent. glog.V(2).Infof("updateLatestVersionAfterDeletion: no versions left for %s/%s, deleting .versions directory", bucket, object) - rmErr := s3a.rm(bucketDir, versionsObjectPath, true, false) + rmErr := s3a.rm(ctx, bucketDir, versionsObjectPath, true, false) if rmErr == nil { return nil } - // Two ways rm can fail here: "non-empty folder" (orphan entries - // blocking the teardown — fall through to pointer clear) and a - // transient filer error (worth retrying). Distinguish by the - // sentinel; if we can't tell, treat as transient. + // "non-empty folder" means orphan entries are blocking the + // teardown — fall through to the pointer clear. Anything else + // has already exhausted rm's own retries. Either way we still + // clear the stale pointer so readers get a clean miss; the + // directory can be tidied by the reconciler later. if errors.Is(rmErr, filer.ErrNonEmptyFolder) { glog.V(2).Infof("updateLatestVersionAfterDeletion: .versions/ for %s/%s still has orphan entries: %v", bucket, object, rmErr) s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion") return nil } - // Transient — retry the rm a few times before giving up. Even - // if it ultimately fails, we still clear the stale pointer so - // readers get a clean miss; the directory can be tidied by the - // reconciler later. - retryErr := retryFilerOp(ctx, "updateLatestVersionAfterDeletion.rm", func() error { - return s3a.rm(bucketDir, versionsObjectPath, true, false) - }) - if retryErr == nil { - return nil - } - if errors.Is(retryErr, filer.ErrNonEmptyFolder) { - s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion") - return nil - } - versioningHealWarningf("teardown_failed", "bucket=%s key=%s err=%v (fell through to clearStale)", bucket, object, retryErr) + versioningHealWarningf("teardown_failed", "bucket=%s key=%s err=%v (fell through to clearStale)", bucket, object, rmErr) if s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion") { // Pointer is consistent again; reader will get NoSuchKey via // the clean-miss path. Don't emit `produced` or enqueue the // reconciler — there's no stranded state left to heal. return nil } - return fmt.Errorf("delete .versions directory: %w", retryErr) + return fmt.Errorf("delete .versions directory: %w", rmErr) } return nil @@ -2067,7 +2062,7 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s // non-recursive rm makes this safe against a concurrent PUT — a new // child fails the rm, and a directory removed just before that PUT's // version file lands is recreated by the create's parent handling. - if rmErr := s3a.rm(bucketDir, versionsObjectPath, true, false); rmErr == nil { + if rmErr := s3a.rm(context.Background(), bucketDir, versionsObjectPath, true, false); rmErr == nil { versioningHealInfof("healed", "bucket=%s key=%s mode=empty_dir_removed", bucket, normalizedObject) } else { // Orphan entries (files in .versions/ that lack the version-id