operation: re-assign chunk upload when replica volume is full (#10588)

* operation: re-assign chunk upload when replica volume is full

When a volume reaches MaxPossibleVolumeSize, needle writes return
'Volume Size Exceeded' and the fan-out in uploadChunkToHolders fails.
Previously the error propagated immediately, killing the entire chunked
upload even though the master has other writable volumes.

Fix: detect 'Volume Size' errors on the fan-out path, call AssignFunc
again to get a fresh volume, and retry (up to 3 attempts). This avoids
backup failures while a single replica volume is full and waiting for GC.

Also add 'volume size exceeded' to the transient error messages so any
retry path that checks IsTransientError recognises it.

* util: fix transient error pattern for volume size errors

The actual error message is 'Volume Size 34361499680 Exceeded 34359738368'
where the numeric size separates 'Volume Size' from 'Exceeded'. The
previous pattern 'volume size exceeded' would never match. Change to
'volume size' which correctly matches any capacity-full error.

* operation: fix reassignment loop nits

- Case-insensitive volume size match (strings.ToLower)
- Propagate AssignFunc error so caller sees reassignment failure
- Reset JWT fallback before each reassignment to avoid retaining stale auth

* util: stop classifying a full volume as a transient error

A volume at capacity does not become writable on the next attempt, so the
entry only bought a retry loop's worth of sleeping before the same failure.
It also reached four consumers that all retry the same target — the deletion
queue, the replication sink, volume lookups, and Retry/MultiRetry — none of
which reassign, and the widened "volume size" substring swallowed the
replica-receive rejection from WriteNeedleBlob too.

The chunked upload path recovers by asking for a different volume instead.

* operation: reassign a chunk with the shared upload gate

shouldReassignUpload already answers this question for the non-chunked path,
keyed off the status the volume server returned rather than its message text.
Reusing it covers a lost replica peer and an unreachable target as well as a
full volume, and it stops a 4xx from being retried on a second volume that
would reject it identically.

Pulling the single attempt out into uploadChunk keeps the retry loop readable
now that it wraps both the fan-out and the relay path.

* test: cover chunk reassignment across volumes

Pins the three outcomes the gate decides: a full volume moves the chunk to a
fresh assignment, a 4xx stays put, and the loop gives up after
chunkAssignAttempts volumes.

* operation: roll back the fid a reassigned chunk abandons

ReplicatedWrite commits the needle locally before it replicates, so a 5xx can
leave a copy behind on a volume the chunk is about to walk away from. Nothing
will ever reference that fid, and an unreferenced needle is not garbage vacuum
can find — it is dead space until the volume is destroyed.

uploadChunkToHolders rolls back only the holders that reported success, which
misses the one whose write landed but whose response did not, and the relay
path had no rollback at all. Delete from every holder of the abandoned
assignment instead; deleting a needle that never landed is a no-op.

* operation: stop the reassignment loop from multiplying work

retriedUploadData already retries a chunk three times against the same URL, so
wrapping it in three assignments made nine POSTs for one chunk. On the relay
path that inner retry is redundant — the loop retries everything it would, and
on a different volume — so cap it at one attempt per assignment and leave the
budget where it was. The fan-out path keeps its inner retries: absorbing a blip
on one holder beats cancelling the rest and re-uploading the whole chunk.

Nothing bounded any of it by time. weed/s3api passes context.Background() so a
chunk survives client disconnect, which also means no deadline cuts the loop
short, and a chunk goroutine holds one of four buffer slots while it spins.
Break out once another chunk has already failed the object.

* operation: keep the reassignment gate's inputs deterministic

uploadChunkToHolders reported whichever holder error won the channel race.
That was cosmetic while the value was only logged; now it decides whether the
chunk moves to another volume, so a 400 and a 500 arriving in either order
made the retry behavior depend on scheduling. Prefer an error the caller can
act on, and the same failure always retries the same way.

A failed reassignment also overwrote the upload error that prompted it, which
buried a full volume behind whatever the filer happened to say. Keep both in
the chain.

The tests grew a JWT per assignment, since the loop re-derives one and nothing
covered it, and the bound is now spelled out rather than compared against the
constant that defines it.

* operation: roll back the last abandoned fid too

The rollback ran only on the path that goes on to reassign, so the attempt
that exhausts the budget — or stops because another chunk already failed the
object, or because the error is not one a different volume fixes — left its
fid behind. That is the case that matters most: no chunk names it, the caller
gets no fid to clean up, and a 5xx can still mean the needle was committed.

Roll back on every failed attempt instead, before deciding whether to retry.

---------

Co-authored-by: timolow <timolow@users.noreply.github.com>
Co-authored-by: timolow <tim@timolow.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/operation/upload_chunked.go

@@ -71,6 +71,15 @@ func UploadReaderInChunks(ctx context.Context, reader io.Reader, opt *ChunkedUpl
 	const bytesBufferCounter = 4
 	bytesBufferLimitChan := make(chan struct{}, bytesBufferCounter)
 
+	// objectFailed reports whether another chunk has already doomed the upload,
+	// so an in-flight chunk stops spending its retry budget on bytes nobody
+	// will reference.
+	objectFailed := func() bool {
+		uploadErrLock.Lock()
+		defer uploadErrLock.Unlock()
+		return uploadErr != nil
+	}
+
 uploadLoop:
 	for {
 		// Throttle buffer usage
@@ -192,35 +201,31 @@ uploadLoop:
 			var uploadResult *UploadResult
 			var uploadResultErr error
 
-			holders := chunkHolders(assignResult)
-			// Fan out to every holder, except for cipher: per-call encryption
-			// would give each replica different bytes, so keep its relay path.
-			if opt.UploadFunc == nil && !opt.Cipher && len(holders) > 1 {
-				uploadResult, uploadResultErr = uploadChunkToHolders(ctx, holders, assignResult.Fid, buf.Bytes(), jwt, chunkMd5B64, opt)
-			} else {
-				uploadOption := &UploadOption{
-					UploadUrl:         fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid),
-					Cipher:            opt.Cipher,
-					IsInputCompressed: false,
-					MimeType:          opt.MimeType,
-					PairMap:           nil,
-					Jwt:               jwt,
-					Md5:               chunkMd5B64,
+			// A target that fills up, loses its replica peer, or goes away fails
+			// every attempt against the same fid, so ask for a fresh assignment and
+			// retry there rather than losing the whole object to one bad volume.
+			for attempt := 1; ; attempt++ {
+				uploadResult, uploadResultErr = uploadChunk(ctx, assignResult, buf.Bytes(), jwt, chunkMd5B64, opt)
+				if uploadResultErr == nil {
+					break
 				}
-				// Use mock upload function if provided (for testing), otherwise use real uploader
-				if opt.UploadFunc != nil {
-					uploadResult, uploadResultErr = opt.UploadFunc(ctx, buf.Bytes(), uploadOption)
-				} else {
-					uploader, uploaderErr := NewUploader()
-					if uploaderErr != nil {
-						uploadErrLock.Lock()
-						if uploadErr == nil {
-							uploadErr = fmt.Errorf("create uploader: %w", uploaderErr)
-						}
-						uploadErrLock.Unlock()
-						return
-					}
-					uploadResult, uploadResultErr = uploader.UploadData(ctx, buf.Bytes(), uploadOption)
+				// The volume server commits the needle before replicating, so a
+				// 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)
+				if attempt == chunkAssignAttempts || !shouldReassignUpload(uploadResultErr) || objectFailed() {
+					break
+				}
+				glog.V(2).Infof("re-assigning chunk at offset %d after attempt %d/%d: %v", offset, attempt, chunkAssignAttempts, uploadResultErr)
+				_, assignResult, assignErr = opt.AssignFunc(ctx, 1, uint64(size))
+				if assignErr != nil {
+					uploadResultErr = fmt.Errorf("reassign volume after %w: %w", uploadResultErr, assignErr)
+					break
+				}
+				jwt = opt.Jwt
+				if assignResult.Auth != "" {
+					jwt = assignResult.Auth
 				}
 			}
 
@@ -291,6 +296,46 @@ uploadLoop:
 	}, nil
 }
 
+// chunkAssignAttempts bounds how many volumes one chunk may be offered to
+// before the upload gives up.
+const chunkAssignAttempts = 3
+
+// uploadChunk writes one chunk to its assigned volume. It fans out to every
+// holder, except for cipher: per-call encryption would give each replica
+// different bytes, so keep its relay path.
+func uploadChunk(ctx context.Context, assignResult *AssignResult, data []byte, jwt security.EncodedJwt, md5b64 string, opt *ChunkedUploadOption) (*UploadResult, error) {
+	holders := chunkHolders(assignResult)
+	if opt.UploadFunc == nil && !opt.Cipher && len(holders) > 1 {
+		return uploadChunkToHolders(ctx, holders, assignResult.Fid, data, jwt, md5b64, opt)
+	}
+	uploadOption := &UploadOption{
+		UploadUrl:         fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid),
+		Cipher:            opt.Cipher,
+		IsInputCompressed: false,
+		MimeType:          opt.MimeType,
+		PairMap:           nil,
+		Jwt:               jwt,
+		Md5:               md5b64,
+		// One upload call per assignment: the caller retries everything a
+		// same-URL retry would, and a different volume is the better second try.
+		// This bounds retriedUploadData only — doUploadData still reissues once
+		// on a connection reset, with a rewound body, which is a transport
+		// stutter rather than a fresh attempt at the volume. The fan-out path
+		// keeps its retries, where absorbing a blip locally beats cancelling
+		// every holder and re-uploading the chunk.
+		MaxAttempts: 1,
+	}
+	// Use mock upload function if provided (for testing), otherwise use real uploader
+	if opt.UploadFunc != nil {
+		return opt.UploadFunc(ctx, data, uploadOption)
+	}
+	uploader, err := NewUploader()
+	if err != nil {
+		return nil, fmt.Errorf("create uploader: %w", err)
+	}
+	return uploader.UploadData(ctx, data, uploadOption)
+}
+
 // chunkHolders returns the assigned volume plus its replica holders.
 func chunkHolders(assignResult *AssignResult) []string {
 	hosts := []string{assignResult.Url}
@@ -341,9 +386,15 @@ func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data
 	for range hosts {
 		o := <-outcomes
 		if o.err != nil {
+			// Once one holder fails the rest are cancelled, so errors arrive in
+			// no fixed order. Prefer one the caller can act on, or the choice of
+			// which host to report — and whether to retry elsewhere — turns on
+			// goroutine scheduling.
 			if firstErr == nil {
 				firstErr = o.err
 				cancel()
+			} else if !shouldReassignUpload(firstErr) && shouldReassignUpload(o.err) {
+				firstErr = o.err
 			}
 		} else {
 			succeeded = append(succeeded, o.host)

weed/operation/upload_chunked_test.go

@@ -13,6 +13,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/security"
 )
 
 // TestUploadReaderInChunksReturnsPartialResultsOnError verifies that when
@@ -415,3 +417,225 @@ func TestUploadReaderInChunksTagsTruncatedBody(t *testing.T) {
 		t.Errorf("expected io.ErrUnexpectedEOF to remain in the chain, got %v", err)
 	}
 }
+
+// uploadSingleChunk runs a single-chunk upload against uploadFunc, handing out a
+// fresh volume on every assignment, and reports how many assignments it took.
+func uploadSingleChunk(t *testing.T, uploadFunc func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error)) (*ChunkedUploadResult, int, error) {
+	t.Helper()
+
+	assigns := 0
+	assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) {
+		assigns++
+		return nil, &AssignResult{
+			Fid:   fmt.Sprintf("%d,0a0b0c0d", assigns),
+			Url:   fmt.Sprintf("volume-%d:8080", assigns),
+			Auth:  security.EncodedJwt(fmt.Sprintf("jwt-%d", assigns)),
+			Count: 1,
+		}, nil
+	}
+
+	result, err := UploadReaderInChunks(context.Background(), bytes.NewReader(bytes.Repeat([]byte("x"), 4096)), &ChunkedUploadOption{
+		ChunkSize:  8 * 1024,
+		Collection: "test",
+		AssignFunc: assignFunc,
+		UploadFunc: uploadFunc,
+	})
+	return result, assigns, err
+}
+
+// A volume at capacity answers every attempt against the same fid with a 500,
+// so the chunk has to move to a freshly assigned volume rather than take the
+// whole object down with it.
+func TestUploadReaderInChunksReassignsOnFullVolume(t *testing.T) {
+	var jwts []security.EncodedJwt
+	result, assigns, err := uploadSingleChunk(t, func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) {
+		jwts = append(jwts, option.Jwt)
+		if strings.Contains(option.UploadUrl, "volume-1:") {
+			return nil, &uploadStatusError{
+				StatusCode: http.StatusInternalServerError,
+				err:        errors.New("failed to write to local disk: Volume Size 34361499680 Exceeded 34359738368"),
+			}
+		}
+		return &UploadResult{Size: uint32(len(data))}, nil
+	})
+
+	if err != nil {
+		t.Fatalf("expected the chunk to land on the reassigned volume, got %v", err)
+	}
+	if assigns != 2 {
+		t.Fatalf("expected 2 assignments, got %d", assigns)
+	}
+	// A secured cluster mints a JWT per assignment; presenting the full volume's
+	// token to the new one would 401.
+	if len(jwts) != 2 || jwts[0] != "jwt-1" || jwts[1] != "jwt-2" {
+		t.Errorf("expected each attempt to carry its own assignment JWT, got %v", jwts)
+	}
+	if len(result.FileChunks) != 1 {
+		t.Fatalf("expected 1 chunk, got %d", len(result.FileChunks))
+	}
+	if got := result.FileChunks[0].FileId; got != "2,0a0b0c0d" {
+		t.Errorf("expected the chunk to record the reassigned fid, got %s", got)
+	}
+}
+
+func TestUploadReaderInChunksDoesNotReassignOnClientError(t *testing.T) {
+	_, assigns, err := uploadSingleChunk(t, func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) {
+		return nil, &uploadStatusError{StatusCode: http.StatusBadRequest, err: errors.New("mismatching cookie")}
+	})
+
+	if err == nil {
+		t.Fatal("expected a 4xx to fail the upload")
+	}
+	// Another volume would reject the same request the same way.
+	if assigns != 1 {
+		t.Fatalf("expected 1 assignment, got %d", assigns)
+	}
+}
+
+func TestUploadReaderInChunksBoundsReassignment(t *testing.T) {
+	_, assigns, err := uploadSingleChunk(t, func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) {
+		return nil, &uploadStatusError{StatusCode: http.StatusInternalServerError, err: errors.New("volume full")}
+	})
+
+	if err == nil {
+		t.Fatal("expected the upload to fail once every volume it is offered is full")
+	}
+	// Spelled out rather than compared to chunkAssignAttempts: the budget is
+	// what bounds this against retriedUploadData's own attempts, so raising it
+	// should fail here, not pass silently.
+	if assigns != 3 {
+		t.Fatalf("expected 3 assignments, got %d", assigns)
+	}
+}
+
+// The fan-out path is what a real multi-replica cluster takes, and it behaves
+// differently from the relay path under failure: it cancels its siblings and
+// rolls back the copies that landed. Drive the reassignment loop through it.
+func TestUploadReaderInChunksReassignsAcrossHolders(t *testing.T) {
+	var primaryDeletes, fullDeletes int32
+
+	healthy := func(deletes *int32) *httptest.Server {
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method == http.MethodDelete {
+				atomic.AddInt32(deletes, 1)
+				w.WriteHeader(http.StatusAccepted)
+				return
+			}
+			w.WriteHeader(http.StatusCreated)
+			fmt.Fprint(w, `{"name":"chunk","size":4096}`)
+		}))
+	}
+
+	full := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodDelete {
+			atomic.AddInt32(&fullDeletes, 1)
+			w.WriteHeader(http.StatusAccepted)
+			return
+		}
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprint(w, `{"error":"failed to write to local disk: Volume Size 34361499680 Exceeded 34359738368"}`)
+	}))
+	defer full.Close()
+
+	var secondaryDeletes int32
+	primary, secondary := healthy(&primaryDeletes), healthy(&secondaryDeletes)
+	defer primary.Close()
+	defer secondary.Close()
+
+	host := func(s *httptest.Server) string { return strings.TrimPrefix(s.URL, "http://") }
+
+	assigns := 0
+	assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) {
+		assigns++
+		if assigns == 1 {
+			// Two holders, so uploadChunk takes the fan-out path; one is full.
+			return nil, &AssignResult{
+				Fid:      "1,0a0b0c0d",
+				Url:      host(primary),
+				Replicas: []Location{{Url: host(full)}},
+				Count:    1,
+			}, nil
+		}
+		return nil, &AssignResult{
+			Fid:      "2,0a0b0c0d",
+			Url:      host(primary),
+			Replicas: []Location{{Url: host(secondary)}},
+			Count:    1,
+		}, nil
+	}
+
+	result, err := UploadReaderInChunks(context.Background(), bytes.NewReader(bytes.Repeat([]byte("x"), 4096)), &ChunkedUploadOption{
+		ChunkSize:  8 * 1024,
+		Collection: "test",
+		AssignFunc: assignFunc,
+	})
+
+	if err != nil {
+		t.Fatalf("expected the chunk to land on the reassigned volume, got %v", err)
+	}
+	if assigns != 2 {
+		t.Fatalf("expected 2 assignments, got %d", assigns)
+	}
+	if len(result.FileChunks) != 1 || result.FileChunks[0].FileId != "2,0a0b0c0d" {
+		t.Fatalf("expected the chunk to record the reassigned fid, got %+v", result.FileChunks)
+	}
+	// The copy that landed on the healthy holder before its peer reported full
+	// is referenced by nothing, so it has to be rolled back.
+	if atomic.LoadInt32(&primaryDeletes) == 0 {
+		t.Error("expected the copy that landed to be rolled back")
+	}
+	// The full holder is not in uploadChunkToHolders' succeeded set, so only the
+	// loop's own rollback reaches it — and it has to, because a 5xx there can
+	// still mean the needle was committed before replication failed.
+	if atomic.LoadInt32(&fullDeletes) == 0 {
+		t.Error("expected the abandoned fid to be deleted from the failed holder too")
+	}
+	if atomic.LoadInt32(&secondaryDeletes) != 0 {
+		t.Error("the reassigned volume kept the chunk; it must not be rolled back")
+	}
+}
+
+// retriedUploadData retries the same URL three times by default. On the relay
+// path the reassignment loop retries everything that would, so leaving both in
+// place would multiply into nine upload calls for one chunk.
+//
+// This counts calls that reached a handler and answered. doUploadData reissues
+// once more on a connection reset before any of them return, which no HTTP
+// status can provoke; upload_content_test covers that separately.
+func TestUploadReaderInChunksDoesNotMultiplyRelayAttempts(t *testing.T) {
+	var posts, deletes int32
+	dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodDelete {
+			atomic.AddInt32(&deletes, 1)
+			w.WriteHeader(http.StatusAccepted)
+			return
+		}
+		atomic.AddInt32(&posts, 1)
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprint(w, `{"error":"failed to write to local disk: Volume Size 34361499680 Exceeded 34359738368"}`)
+	}))
+	defer dead.Close()
+
+	assigns := 0
+	_, err := UploadReaderInChunks(context.Background(), bytes.NewReader(bytes.Repeat([]byte("x"), 4096)), &ChunkedUploadOption{
+		ChunkSize:  8 * 1024,
+		Collection: "test",
+		AssignFunc: func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) {
+			assigns++
+			// One holder, so uploadChunk takes the relay path.
+			return nil, &AssignResult{Fid: fmt.Sprintf("%d,0a0b0c0d", assigns), Url: strings.TrimPrefix(dead.URL, "http://"), Count: 1}, nil
+		},
+	})
+
+	if err == nil {
+		t.Fatal("expected the upload to fail once every volume it is offered is full")
+	}
+	if got := atomic.LoadInt32(&posts); got != 3 {
+		t.Errorf("expected 3 upload calls, one per assignment, got %d", got)
+	}
+	// Including the last one: giving up does not make the needle it may have
+	// committed anyone else's to find, and no chunk will ever name that fid.
+	if got := atomic.LoadInt32(&deletes); got != 3 {
+		t.Errorf("expected every abandoned fid to be rolled back, got %d deletes", got)
+	}
+}