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