diff --git a/weed/admin/handlers/file_browser_grpc.go b/weed/admin/handlers/file_browser_grpc.go index 0b2f9eb98d..9239688ed0 100644 --- a/weed/admin/handlers/file_browser_grpc.go +++ b/weed/admin/handlers/file_browser_grpc.go @@ -153,6 +153,9 @@ func (h *FileBrowserHandlers) uploadFileGrpc(ctx context.Context, filePath strin ChunkSize: adminUploadChunkSize, MimeType: mimeType, AssignFunc: assignFunc, + LookupJwtFunc: func(lookupCtx context.Context, fileId string) security.EncodedJwt { + return operation.LookupJwt(lookupCtx, h.adminServer.GetMasterClient().GetMaster(lookupCtx), h.adminServer.GetGrpcDialOption(), fileId) + }, }) if err != nil { // Partial chunks come back even on error so we can clean them up rather diff --git a/weed/command/benchmark.go b/weed/command/benchmark.go index ad0dbadc4c..0ee3f03136 100644 --- a/weed/command/benchmark.go +++ b/weed/command/benchmark.go @@ -238,7 +238,7 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) { } var jwtAuthorization security.EncodedJwt if isSecure { - jwtAuthorization = operation.LookupJwt(b.masterClient.GetMaster(context.Background()), b.grpcDialOption, df.fp.Fid) + jwtAuthorization = operation.LookupJwt(context.Background(), b.masterClient.GetMaster(context.Background()), b.grpcDialOption, df.fp.Fid) } if e := util_http.Delete(fmt.Sprintf("http://%s/%s", df.fp.Server, df.fp.Fid), string(jwtAuthorization)); e == nil { s.completed++ diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index 4fb2581581..3e9c29f813 100644 --- a/weed/operation/assign_file_id.go +++ b/weed/operation/assign_file_id.go @@ -157,11 +157,11 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO return ret, lastError } -func LookupJwt(master pb.ServerAddress, grpcDialOption grpc.DialOption, fileId string) (token security.EncodedJwt) { +func LookupJwt(ctx context.Context, master pb.ServerAddress, grpcDialOption grpc.DialOption, fileId string) (token security.EncodedJwt) { - WithMasterServerClient(context.Background(), false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + WithMasterServerClient(ctx, false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error { - resp, grpcErr := masterClient.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{ + resp, grpcErr := masterClient.LookupVolume(ctx, &master_pb.LookupVolumeRequest{ VolumeOrFileIds: []string{fileId}, }) if grpcErr != nil { diff --git a/weed/operation/upload_chunked.go b/weed/operation/upload_chunked.go index 244a553f96..8124500dfa 100644 --- a/weed/operation/upload_chunked.go +++ b/weed/operation/upload_chunked.go @@ -9,6 +9,7 @@ import ( "fmt" "hash" "io" + "net/http" "sort" "sync" "time" @@ -45,6 +46,7 @@ type ChunkedUploadOption struct { Cipher bool // encrypt data on volume servers AssignFunc func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) UploadFunc func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) // Optional: for testing + LookupJwtFunc func(ctx context.Context, fileId string) security.EncodedJwt // Optional: renews the volume JWT for a cleanup delete } var chunkBufferPool = sync.Pool{ @@ -213,7 +215,7 @@ uploadLoop: // failed write can still leave a copy behind. No chunk will name // this fid whether we retry or give up here, and an unreferenced // needle is not garbage vacuum can find, so drop it either way. - deleteChunkFromHolders(chunkHolders(assignResult), assignResult.Fid, jwt) + deleteChunkFromHolders(chunkHolders(assignResult), assignResult.Fid, jwt, opt) if attempt == chunkAssignAttempts || !ShouldReassignUpload(uploadResultErr) || objectFailed() { break } @@ -413,19 +415,46 @@ func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data if firstErr != nil { // A failed fan-out records no chunk, so roll back the copies that landed // before the cancel rather than leaking them as orphans. - deleteChunkFromHolders(succeeded, fid, jwt) + deleteChunkFromHolders(succeeded, fid, jwt, opt) return nil, firstErr } return first, nil } +// cleanupJwtLookupTimeout bounds a cleanup-token renewal, covering the wait +// for a master plus the lookup RPC, so a master outage cannot stall the +// rollback of a failed fan-out. +const cleanupJwtLookupTimeout = 10 * time.Second + // deleteChunkFromHolders best-effort removes a needle from each holder it landed // on, using type=replicate so the volume drops only its local copy. A failed // delete falls back to vacuum reclaiming the orphan. -func deleteChunkFromHolders(hosts []string, fid string, jwt security.EncodedJwt) { +func deleteChunkFromHolders(hosts []string, fid string, jwt security.EncodedJwt, opt *ChunkedUploadOption) { + // An unresponsive holder can stall the upload past the write JWT's expiry, + // so renew the token whenever a holder rejects the delete as unauthorized. + // Each holder's delete is retried at most once, and a slow holder that + // outlives the renewed token can still trigger another renewal. + deleteJwt := jwt for _, host := range hosts { - if err := util_http.Delete(fmt.Sprintf("http://%s/%s?type=replicate", host, fid), string(jwt)); err != nil { + deleteUrl := fmt.Sprintf("http://%s/%s?type=replicate", host, fid) + body, status, err := util_http.DeleteProxied(deleteUrl, string(deleteJwt)) + if status == http.StatusUnauthorized && opt.LookupJwtFunc != nil { + lookupCtx, cancel := context.WithTimeout(context.Background(), cleanupJwtLookupTimeout) + token := opt.LookupJwtFunc(lookupCtx, fid) + cancel() + if token != "" { + deleteJwt = token + body, status, err = util_http.DeleteProxied(deleteUrl, string(deleteJwt)) + } + } + if err != nil { glog.Warningf("replica fan-out cleanup: delete %s from %s: %v", fid, host, err) + continue + } + switch status { + case http.StatusNotFound, http.StatusNoContent, http.StatusAccepted, http.StatusOK: + default: + glog.Warningf("replica fan-out cleanup: delete %s from %s: status %d: %s", fid, host, status, body) } } } diff --git a/weed/operation/upload_chunked_test.go b/weed/operation/upload_chunked_test.go index b1accb3b75..87ab218ebc 100644 --- a/weed/operation/upload_chunked_test.go +++ b/weed/operation/upload_chunked_test.go @@ -26,13 +26,11 @@ func TestUploadReaderInChunksReturnsPartialResultsOnError(t *testing.T) { testData := bytes.Repeat([]byte("test data for chunk upload failure testing"), 1000) // ~40KB reader := bytes.NewReader(testData) - uploadAttempts := 0 + var uploadAttempts int32 // Create a mock assign function that succeeds for first chunk, then fails assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) { - uploadAttempts++ - - if uploadAttempts == 1 { + if atomic.AddInt32(&uploadAttempts, 1) == 1 { // First chunk succeeds return nil, &AssignResult{ Fid: "test-fid-1,1234", @@ -679,3 +677,90 @@ func TestUploadReaderInChunksAppendsFsyncWhenAssigned(t *testing.T) { t.Errorf("Expected the upload URL to carry ?fsync=true when the assignment says fsync, got %q", gotUploadUrl) } } + +// A holder that stalls past the assignment JWT's expiry answers the rollback +// delete with 401; the cleanup renews the token instead of leaking the copy. +func TestDeleteChunkFromHoldersRenewsExpiredJwt(t *testing.T) { + var staleDeletes, freshDeletes int32 + holder := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + w.WriteHeader(http.StatusCreated) + return + } + switch r.Header.Get("Authorization") { + case "Bearer fresh-jwt": + atomic.AddInt32(&freshDeletes, 1) + w.WriteHeader(http.StatusAccepted) + default: + atomic.AddInt32(&staleDeletes, 1) + w.WriteHeader(http.StatusUnauthorized) + fmt.Fprint(w, `{"error":"wrong jwt"}`) + } + })) + defer holder.Close() + + lookups := 0 + deleteChunkFromHolders( + []string{strings.TrimPrefix(holder.URL, "http://")}, + "1,0a0b0c0d", + "stale-jwt", + &ChunkedUploadOption{ + LookupJwtFunc: func(ctx context.Context, fileId string) security.EncodedJwt { + lookups++ + return "fresh-jwt" + }, + }, + ) + + if lookups != 1 { + t.Fatalf("expected one jwt renewal, got %d", lookups) + } + if atomic.LoadInt32(&freshDeletes) != 1 { + t.Fatalf("expected the delete retried with the fresh jwt, got %d fresh deletes", freshDeletes) + } +} + +// A later holder that rejects the renewed token can trigger another renewal: +// the one-retry limit is per holder, not per cleanup pass. +func TestDeleteChunkFromHoldersRenewsPerHolder(t *testing.T) { + var acceptedDeletes int32 + holder := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + w.WriteHeader(http.StatusCreated) + return + } + if r.Header.Get("Authorization") == "Bearer jwt-2" { + atomic.AddInt32(&acceptedDeletes, 1) + w.WriteHeader(http.StatusAccepted) + return + } + w.WriteHeader(http.StatusUnauthorized) + fmt.Fprint(w, `{"error":"wrong jwt"}`) + })) + defer holder.Close() + + tokens := []security.EncodedJwt{"jwt-1", "jwt-2"} + lookups := 0 + deleteChunkFromHolders( + []string{ + strings.TrimPrefix(holder.URL, "http://"), + strings.TrimPrefix(holder.URL, "http://"), + }, + "1,0a0b0c0d", + "stale-jwt", + &ChunkedUploadOption{ + LookupJwtFunc: func(ctx context.Context, fileId string) security.EncodedJwt { + token := tokens[lookups] + lookups++ + return token + }, + }, + ) + + if lookups != 2 { + t.Fatalf("expected a renewal per rejecting holder, got %d", lookups) + } + if atomic.LoadInt32(&acceptedDeletes) != 1 { + t.Fatalf("expected the second holder's retried delete to land, got %d", acceptedDeletes) + } +} diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 1b18f0c560..b38a205f16 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -696,6 +696,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader MimeType: r.Header.Get("Content-Type"), Cipher: s3a.cipher, // encrypt data on volume servers AssignFunc: assignFunc, + LookupJwtFunc: s3a.lookupVolumeJwt, }) if err != nil { glog.Errorf("putToFiler: chunked upload failed: %v", err) @@ -3022,6 +3023,15 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReads(r *http.Request, bucket, return s3a.validateConditionalHeadersForReads(r, headers, entry, bucket, object) } +// lookupVolumeJwt mints a fresh volume-server JWT for an existing file id, so a +// cleanup delete that runs past the assignment JWT's expiry still authenticates. +func (s3a *S3ApiServer) lookupVolumeJwt(ctx context.Context, fileId string) security.EncodedJwt { + if s3a.masterClient == nil { + return "" + } + return operation.LookupJwt(ctx, s3a.masterClient.GetMaster(ctx), s3a.option.GrpcDialOption, fileId) +} + // deleteOrphanedChunks attempts to delete chunks that were uploaded but whose entry creation failed // This prevents resource leaks and wasted storage. Errors are logged but don't prevent cleanup attempts. func (s3a *S3ApiServer) deleteOrphanedChunks(chunks []*filer_pb.FileChunk) { diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index d06fed4424..86b1025081 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -105,6 +105,7 @@ type S3ApiServer struct { randomClientId int32 filerGuard *security.Guard filerClient *wdclient.FilerClient + masterClient *wdclient.MasterClient client util_http_client.HTTPClientInterface bucketRegistry *BucketRegistry credentialManager *credential.CredentialManager @@ -345,6 +346,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl randomClientId: util.RandomInt32(), filerGuard: security.NewGuard([]string{}, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec), filerClient: filerClient, + masterClient: masterClient, cb: NewCircuitBreaker(option), credentialManager: iam.credentialManager, bucketConfigCache: NewBucketConfigCache(60 * time.Minute), // Increased TTL since cache is now event-driven