mount: re-assign to a live volume when a write can't land (#10239)

With replication 001 and one node down, ~2/3 of volumes have their
replica on the dead node. The primary write lands locally but the
replica forward fails, so the volume returns 500 "failed to write to
replicas". #9744 made that upload fail fast so the client re-assigns,
but the client retry never fired: the reassign gate matched a fixed
list of error substrings that didn't include this one. The mount
surfaced I/O error and dropped the chunk, leaving missing lines and
null-byte gaps in an append workload while a node rebooted.

Decide reassignment by HTTP status instead of matching message text.
On the write path a volume server only 5xxs on a ReplicatedWrite
failure (local disk, replica peer down, under-replication) — all of
which a different volume dodges — so any 5xx reassigns; a no-response
transport failure (the assigned target itself is down) reassigns too;
a 4xx is a genuine client error and is surfaced. doUploadData tags its
errors with the response status via uploadStatusError, and the gate
moves from util.MultiRetry(errList) to util.RetryOnError(predicate).

This drops the fragile substring list (and the message-prefix
constants and guard test it needed); store_replicate.go is untouched.

weed/operation/upload_content.go

@@ -6,6 +6,7 @@ import (
 	"crypto/md5"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"mime"
@@ -118,11 +119,35 @@ var (
 	once        sync.Once
 )
 
-var uploadRetryableAssignErrList = []string{
-	"transport",
-	"is read only",
-	"failed to write to local disk",
-	"Volume Size ",
+// uploadStatusError carries the volume-server HTTP status of a failed upload so
+// the retry gate can decide on the status code instead of the message text.
+// StatusCode 0 means the request never got a response — a transport failure.
+type uploadStatusError struct {
+	StatusCode int
+	err        error
+}
+
+func (e *uploadStatusError) Error() string { return e.err.Error() }
+func (e *uploadStatusError) Unwrap() error { return e.err }
+
+// shouldReassignUpload reports whether an upload error means the client should
+// ask for a fresh volume assignment and retry on another volume.
+//
+// On the write path a volume server only 5xxs on a ReplicatedWrite failure
+// (local disk, replica peer down, or under-replication) — all of which a
+// different volume dodges — so any 5xx is reassignable. A status of 0 means the
+// assigned target never answered (down/unreachable), also reassignable. A 4xx
+// is a genuine client error and is surfaced. Errors without a status come from
+// the AssignVolume RPC or request setup; retry only transient transport ones.
+func shouldReassignUpload(err error) bool {
+	if err == nil {
+		return false
+	}
+	var se *uploadStatusError
+	if errors.As(err, &se) {
+		return se.StatusCode == 0 || se.StatusCode >= 500
+	}
+	return strings.Contains(err.Error(), "transport")
 }
 
 // assignVolumeTimeout bounds a single AssignVolume RPC so an overwhelmed filer
@@ -195,7 +220,7 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho
 			return true
 		})
 	} else {
-		err = util.MultiRetry("uploadWithRetry", uploadRetryableAssignErrList, doUploadFunc)
+		err = util.RetryOnError("uploadWithRetry", shouldReassignUpload, doUploadFunc)
 	}
 
 	return
@@ -515,7 +540,7 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction
 	}
 	if post_err != nil {
 		stats.UploadErrorCounter.WithLabelValues("0").Inc()
-		return nil, fmt.Errorf("upload %s %d bytes to %v: %v", option.Filename, originalDataSize, option.UploadUrl, post_err)
+		return nil, &uploadStatusError{StatusCode: 0, err: fmt.Errorf("upload %s %d bytes to %v: %v", option.Filename, originalDataSize, option.UploadUrl, post_err)}
 	}
 	// print("-")
 
@@ -530,18 +555,18 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction
 	resp_body, ra_err := io.ReadAll(resp.Body)
 	if ra_err != nil {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
-		return nil, fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)}
 	}
 
 	unmarshal_err := json.Unmarshal(resp_body, &ret)
 	if unmarshal_err != nil {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
 		glog.ErrorfCtx(ctx, "unmarshal %s: %v", option.UploadUrl, string(resp_body))
-		return nil, fmt.Errorf("unmarshal %v: %w", option.UploadUrl, unmarshal_err)
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("unmarshal %v: %w", option.UploadUrl, unmarshal_err)}
 	}
 	if ret.Error != "" {
 		stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
-		return nil, fmt.Errorf("unmarshalled error %v: %v", option.UploadUrl, ret.Error)
+		return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("unmarshalled error %v: %v", option.UploadUrl, ret.Error)}
 	}
 	ret.ETag = etag
 	ret.ContentMd5 = resp.Header.Get("Content-MD5")

weed/operation/upload_content_test.go

@@ -30,18 +30,6 @@ type scriptedHTTPClient struct {
 	calls     []string
 }
 
-func testIsUploadRetryableAssignError(err error) bool {
-	if err == nil {
-		return false
-	}
-	for _, retryable := range uploadRetryableAssignErrList {
-		if strings.Contains(err.Error(), retryable) {
-			return true
-		}
-	}
-	return false
-}
-
 func (c *scriptedHTTPClient) Do(req *http.Request) (*http.Response, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -63,23 +51,26 @@ func (c *scriptedHTTPClient) Do(req *http.Request) (*http.Response, error) {
 	}, nil
 }
 
-func TestIsUploadRetryableAssignError(t *testing.T) {
+func TestShouldReassignUpload(t *testing.T) {
 	testCases := []struct {
 		name string
 		err  error
 		want bool
 	}{
 		{name: "nil", err: nil, want: false},
-		{name: "transport", err: fmt.Errorf("transport is closing"), want: true},
-		{name: "read only", err: fmt.Errorf("volume 1 is read only"), want: true},
-		{name: "volume full", err: fmt.Errorf("failed to write to local disk: append to volume 1 size 0 actualSize 0: Volume Size 33555976 Exceeded 33554432"), want: true},
+		{name: "no response (transport)", err: &uploadStatusError{StatusCode: 0, err: fmt.Errorf("dial tcp: no such host")}, want: true},
+		{name: "500 replica write failed", err: &uploadStatusError{StatusCode: 500, err: fmt.Errorf("failed to write to replicas")}, want: true},
+		{name: "503 service unavailable", err: &uploadStatusError{StatusCode: 503, err: fmt.Errorf("busy")}, want: true},
+		{name: "400 bad request", err: &uploadStatusError{StatusCode: 400, err: fmt.Errorf("bad needle")}, want: false},
+		{name: "401 unauthorized", err: &uploadStatusError{StatusCode: 401, err: fmt.Errorf("wrong jwt")}, want: false},
+		{name: "assign rpc transport", err: fmt.Errorf("assign volume: transport is closing"), want: true},
 		{name: "other permanent", err: fmt.Errorf("mismatching cookie"), want: false},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := testIsUploadRetryableAssignError(tc.err); got != tc.want {
-				t.Fatalf("testIsUploadRetryableAssignError(%v) = %v, want %v", tc.err, got, tc.want)
+			if got := shouldReassignUpload(tc.err); got != tc.want {
+				t.Fatalf("shouldReassignUpload(%v) = %v, want %v", tc.err, got, tc.want)
 			}
 		})
 	}
@@ -129,6 +120,48 @@ func TestUploadWithRetryDataReassignsOnVolumeSizeExceeded(t *testing.T) {
 	}
 }
 
+// A volume whose replica peer is down reports "failed to write to replicas":
+// the primary write lands but replication fails. The client must re-assign to a
+// volume on live nodes and retry rather than losing the write.
+func TestUploadWithRetryDataReassignsOnReplicaWriteFailure(t *testing.T) {
+	replicaErr := `{"error":"failed to write to replicas for volume 1: [volume-down:8080]: dial tcp: lookup volume-down: no such host"}`
+	httpClient := &scriptedHTTPClient{
+		responses: map[string][]scriptedHTTPResponse{
+			"http://volume-a/1,first": {
+				{status: http.StatusInternalServerError, body: replicaErr},
+				{status: http.StatusInternalServerError, body: replicaErr},
+				{status: http.StatusInternalServerError, body: replicaErr},
+			},
+			"http://volume-b/2,second": {
+				{status: http.StatusCreated, body: `{"name":"test.bin","size":3}`},
+			},
+		},
+	}
+	uploader := newUploader(httpClient)
+
+	assignCalls := 0
+	fileID, uploadResult, err := uploader.uploadWithRetryData(func() (string, string, security.EncodedJwt, error) {
+		assignCalls++
+		if assignCalls == 1 {
+			return "1,first", "volume-a", "", nil
+		}
+		return "2,second", "volume-b", "", nil
+	}, &UploadOption{Filename: "test.bin"}, []byte("abc"))
+
+	if err != nil {
+		t.Fatalf("expected success after reassignment, got %v", err)
+	}
+	if fileID != "2,second" {
+		t.Fatalf("expected reassigned file id, got %s", fileID)
+	}
+	if assignCalls != 2 {
+		t.Fatalf("expected 2 assign attempts, got %d", assignCalls)
+	}
+	if uploadResult == nil || uploadResult.Name != "test.bin" {
+		t.Fatalf("expected successful upload result, got %#v", uploadResult)
+	}
+}
+
 // bodyCapturingHTTPClient drains req.Body on every Do, optionally failing the
 // first attempt with a transport error so we can verify upload_content rewinds
 // the body before retrying.

weed/util/retry.go

@@ -58,6 +58,31 @@ func MultiRetry(name string, errList []string, job func() error) (err error) {
 	return err
 }
 
+// RetryOnError retries job with the same bounded backoff as MultiRetry, but
+// decides retriability with a predicate instead of an error-substring list.
+func RetryOnError(name string, shouldRetry func(error) bool, job func() error) (err error) {
+	waitTime := time.Second
+	hasErr := false
+	for waitTime < RetryWaitTime {
+		err = job()
+		if err == nil {
+			if hasErr {
+				glog.V(0).Infof("retry %s successfully", name)
+			}
+			break
+		}
+		if shouldRetry(err) {
+			hasErr = true
+			glog.V(0).Infof("retry %s: err: %v", name, err)
+		} else {
+			break
+		}
+		time.Sleep(waitTime)
+		waitTime += waitTime / 2
+	}
+	return err
+}
+
 // RetryUntil retries until the job returns no error or onErrFn returns false
 func RetryUntil(name string, job func() error, onErrFn func(err error) (shouldContinue bool)) error {
 	waitTime := time.Second