From e521a2a7b9adca388ada054afecf99d2934561f7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 6 Jul 2026 11:41:26 -0700 Subject: [PATCH] mount: re-assign to a live volume when a write can't land (#10239) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 45 +++++++++++++---- weed/operation/upload_content_test.go | 69 ++++++++++++++++++++------- weed/util/retry.go | 25 ++++++++++ 3 files changed, 111 insertions(+), 28 deletions(-) diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 946149ade..d65fbbb86 100644 --- a/weed/operation/upload_content.go +++ b/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") diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index 2b91c0056..6a65668f7 100644 --- a/weed/operation/upload_content_test.go +++ b/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. diff --git a/weed/util/retry.go b/weed/util/retry.go index 66e3e5e07..4443cc27f 100644 --- a/weed/util/retry.go +++ b/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