operation: renew the write JWT when a fan-out rollback delete is rejected (#11970)

* operation: renew the cleanup JWT when a fan-out rollback delete is rejected

A stalled holder can out-wait the assignment JWT's 10s expiry, so the
type=replicate delete that rolls back landed copies comes back 401 and the
needle leaks. Retry the delete once with a fresh token minted through the new
LookupJwtFunc hook when a holder answers unauthorized.

* s3api: wire a cleanup-JWT renewer into chunked uploads

Keep the master client on the server and use it to mint fresh volume JWTs for
rollback deletes that run past the assignment token's expiry.

* admin: wire a cleanup-JWT renewer into file-browser uploads

* operation: bound the cleanup-JWT renewal with a per-attempt deadline

A rollback delete that hits 401 renewed the token through a background
context, so a master outage could park the wait for a connection and the
lookup RPC inside the failed upload's cleanup path. Thread a context
through LookupJwtFunc and LookupJwt, and bound each renewal with a 10s
deadline covering both the master wait and the lookup call.

* operation: let every holder's 401 retry with a renewed cleanup JWT

A single renewed flag meant only the first rejecting holder refreshed the
token; a holder that stalled past the renewed JWT's lifetime left later
copies stranded with a 401 they could not resolve. The retry limit is now
per holder — each delete is attempted once with the current token and once
more after a bounded renewal — so a slow middle holder cannot strand the
rest.

* operation: log the volume server's error body on a failed cleanup delete

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* operation: count test upload attempts atomically

The chunk upload loop fans AssignFunc out to parallel goroutines, so the
plain int counter in the partial-results test raced under -race.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

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++

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 {

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)
 		}
 	}
 }

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)
+	}
+}

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) {

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