diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index e16558cb7..6f98846fa 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -40,6 +40,7 @@ type UploadOption struct { Md5 string BytesBuffer *bytes.Buffer SourceUrl string // optional: for logging when reading from a remote source + MaxAttempts int // <=0 uses the default } type UploadResult struct { @@ -247,15 +248,26 @@ func (uploader *Uploader) doUpload(ctx context.Context, reader io.Reader, option } func (uploader *Uploader) retriedUploadData(ctx context.Context, data []byte, option *UploadOption) (uploadResult *UploadResult, err error) { - for i := 0; i < 3; i++ { + maxAttempts := option.MaxAttempts + if maxAttempts <= 0 { + maxAttempts = 3 + } + for i := 0; i < maxAttempts; i++ { if i > 0 { - time.Sleep(time.Millisecond * time.Duration(237*(i+1))) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(time.Millisecond * time.Duration(237*(i+1))): + } } uploadResult, err = uploader.doUploadData(ctx, data, option) if err == nil { uploadResult.RetryCount = i return } + if ctx.Err() != nil { + return nil, ctx.Err() + } glog.WarningfCtx(ctx, "uploading %d to %s: %v", i, option.UploadUrl, err) } return diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index a045eea03..7694eecdc 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -10,6 +10,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/security" ) @@ -198,3 +199,101 @@ func TestUploadRewindsBodyOnConnectionReset(t *testing.T) { }) } } + +// deadReplicaClient simulates an unreachable replica: every request blocks for +// dialDelay (standing in for the TCP dial timeout) and then fails, unless the +// request context is cancelled first. +type deadReplicaClient struct { + mu sync.Mutex + calls int + dialDelay time.Duration +} + +func (c *deadReplicaClient) attempts() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +func (c *deadReplicaClient) Do(req *http.Request) (*http.Response, error) { + c.mu.Lock() + c.calls++ + c.mu.Unlock() + select { + case <-time.After(c.dialDelay): + return nil, fmt.Errorf("dial tcp %s: i/o timeout", req.URL.Host) + case <-req.Context().Done(): + return nil, req.Context().Err() + } +} + +// TestUploadToDeadReplicaRetriesThreeTimes reproduces the dead-replica upload +// stall: a synchronous replica write to an unreachable host pays the dial +// timeout three times over, so one dead replica stalls the caller for ~3x the +// dial timeout. In production dialDelay is the 10s dialer timeout, so the +// caller blocks ~30s before the failure surfaces. +func TestUploadToDeadReplicaRetriesThreeTimes(t *testing.T) { + client := &deadReplicaClient{dialDelay: 100 * time.Millisecond} + uploader := newUploader(client) + + start := time.Now() + _, err := uploader.UploadData(context.Background(), []byte("hello"), &UploadOption{ + UploadUrl: "http://dead-replica:8080/3,01", + Filename: "test.bin", + }) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected an error uploading to a dead replica") + } + if got := client.attempts(); got != 3 { + t.Fatalf("dial attempts = %d, want 3 (each attempt pays the full dial timeout)", got) + } + t.Logf("dead replica stalled the caller for %v across %d attempts", elapsed, client.attempts()) +} + +// TestUploadToDeadReplicaSingleAttempt verifies the fix: a synchronous replica +// write makes a single attempt (MaxAttempts=1), so a dead replica fails after +// one dial timeout instead of three. The outer client write still retries. +func TestUploadToDeadReplicaSingleAttempt(t *testing.T) { + client := &deadReplicaClient{dialDelay: 100 * time.Millisecond} + uploader := newUploader(client) + + _, err := uploader.UploadData(context.Background(), []byte("hello"), &UploadOption{ + UploadUrl: "http://dead-replica:8080/3,01", + Filename: "test.bin", + MaxAttempts: 1, + }) + if err == nil { + t.Fatal("expected an error uploading to a dead replica") + } + if got := client.attempts(); got != 1 { + t.Fatalf("dial attempts = %d, want 1", got) + } +} + +// TestUploadRetryStopsOnContextCancel verifies the retry loop honors context +// cancellation so DistributedOperation can abandon a slow replica once the +// write outcome is already decided by another replica. +func TestUploadRetryStopsOnContextCancel(t *testing.T) { + client := &deadReplicaClient{dialDelay: time.Hour} // would block ~forever per attempt + uploader := newUploader(client) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already decided elsewhere + + start := time.Now() + _, err := uploader.UploadData(ctx, []byte("hello"), &UploadOption{ + UploadUrl: "http://dead-replica:8080/3,01", + Filename: "test.bin", + }) + if err == nil { + t.Fatal("expected a context error") + } + if elapsed := time.Since(start); elapsed > 10*time.Second { + t.Fatalf("upload did not abort on cancellation: took %v", elapsed) + } + if got := client.attempts(); got > 1 { + t.Fatalf("dial attempts = %d, want at most 1 after cancellation", got) + } +} diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index 6547c0b5c..147a1383d 100644 --- a/weed/topology/store_replicate.go +++ b/weed/topology/store_replicate.go @@ -70,7 +70,7 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi inFlightGauge.Inc() defer inFlightGauge.Dec() - err = DistributedOperation(remoteLocations, func(location operation.Location) error { + err = DistributedOperation(ctx, remoteLocations, func(ctx context.Context, location operation.Location) error { u := url.URL{ Scheme: "http", Host: location.Url, @@ -115,6 +115,7 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi Jwt: jwt, Md5: contentMd5, BytesBuffer: bytesBuffer, + MaxAttempts: 1, // fail fast on a dead replica; the client write retries } uploader, err := operation.NewUploader() @@ -160,7 +161,8 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp } if len(remoteLocations) > 0 { //send to other replica locations - if err = DistributedOperation(remoteLocations, func(location operation.Location) error { + // background, not r.Context(): a client disconnect must not orphan replica deletes + if err = DistributedOperation(context.Background(), remoteLocations, func(ctx context.Context, location operation.Location) error { return util_http.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", string(jwt)) }); err != nil { size = 0 @@ -189,18 +191,22 @@ type RemoteResult struct { Error error } -func DistributedOperation(locations []operation.Location, op func(location operation.Location) error) error { +func DistributedOperation(ctx context.Context, locations []operation.Location, op func(ctx context.Context, location operation.Location) error) error { length := len(locations) if length == 0 { return nil } + // cancel outstanding replica ops once the outcome is decided + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // buffered so a straggler (e.g. a replica stalled on a TCP dial timeout) can // still deliver its result and exit after we have already returned. resultCh := make(chan RemoteResult, length) for _, location := range locations { go func(location operation.Location) { - resultCh <- RemoteResult{location.Url, op(location)} + resultCh <- RemoteResult{location.Url, op(ctx, location)} }(location) } diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go new file mode 100644 index 000000000..f5f96dc48 --- /dev/null +++ b/weed/topology/store_replicate_test.go @@ -0,0 +1,55 @@ +package topology + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/operation" +) + +// TestDistributedOperationCancelsSiblingsOnFirstError verifies that once one +// replica fails, an outstanding replica still stalled in a dial timeout is +// cancelled rather than gating the caller until it times out. +func TestDistributedOperationCancelsSiblingsOnFirstError(t *testing.T) { + locations := []operation.Location{{Url: "fast"}, {Url: "slow"}} + cancelled := make(chan struct{}, 1) + + start := time.Now() + err := DistributedOperation(context.Background(), locations, func(ctx context.Context, location operation.Location) error { + if location.Url == "fast" { + return errors.New("connection refused") + } + // slow: a replica stalled in a dial timeout + select { + case <-ctx.Done(): + cancelled <- struct{}{} + return ctx.Err() + case <-time.After(10 * time.Second): + return nil + } + }) + + if err == nil { + t.Fatal("expected an error from the fast-failing replica") + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("did not fail fast: took %v", elapsed) + } + select { + case <-cancelled: + case <-time.After(2 * time.Second): + t.Fatal("slow replica was not cancelled after the first error") + } +} + +func TestDistributedOperationEmpty(t *testing.T) { + err := DistributedOperation(context.Background(), nil, func(ctx context.Context, location operation.Location) error { + t.Fatal("op should not be called when there are no locations") + return nil + }) + if err != nil { + t.Fatalf("expected nil for no locations, got %v", err) + } +}