topology: fail replica writes fast when a replica is unreachable (#9744)

* operation: bound upload retries and honor context cancellation

retriedUploadData hardcoded 3 attempts and an uninterruptible backoff
sleep. A synchronous replica write to a dead host therefore paid the
full dial timeout three times over before failing.

Add UploadOption.MaxAttempts (<=0 keeps the default of 3) so callers can
cap attempts, and make the loop return as soon as the context is
cancelled so an abandoned upload unwinds instead of retrying.

* topology: fail replica writes fast when a replica is unreachable

DistributedOperation already returns on the first error, but a single
dead replica is itself the slow result: its goroutine retries the upload
three times through the dial timeout (~30s) before any error surfaces,
stalling the originating client write the whole time.

Make the replica write a single attempt (MaxAttempts=1) so a dead
replica fails after one dial timeout instead of three, and thread a
context into DistributedOperation that is cancelled once the outcome is
decided, so a healthy replica is no longer held hostage by one stalled
in a dial. The originating client write is what retries.

* topology: keep replica deletes off the client request context

ReplicatedDelete runs after the local needle is already deleted. Driving
the replica deletes off r.Context() means a client disconnect cancels
them and orphans needles on the replicas, so use a background context.

* operation, topology: trim comments on the replica fail-fast path

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

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)
+	}
+}

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)
 	}
 

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)
+	}
+}