volume: bound each FetchAndWriteNeedle replica write (#12415)

* seaweed-volume: bound each FetchAndWriteNeedle replica write

A replica that accepted the connection and then went silent held the
request until TCP gave up, because the shared outgoing client has no
request timeout. Each replica write now fails after 30 s, from connect
to the response status, matching the Go volume server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: bound each FetchAndWriteNeedle replica write

The default replica path uploads through the guarded peer client, which
bounds only the dial and has no response header timeout, and the filer
calls FetchAndWriteNeedle with no deadline. A replica that accepted the
connection and went silent held the request for minutes. Each replica
write, retries included, now has 30 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: require the silent-peer test to fail at the deadline

A connection error that came back early also passed, since the test
only checked an upper bound. It now also requires the paused clock to
have reached FETCH_AND_WRITE_REPLICA_TIMEOUT.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* seaweed-volume: grow the FetchAndWriteNeedle replica bound with the chunk

The reqwest timeout includes sending the body, and a remote-cache chunk
is a thousandth of the object once it passes the 1000-chunk cap, so a
fixed 30 s would fail a large chunk on a link under ~35 MB/s. The bound
is now 30 s plus 1 s per MiB.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* volume: grow the FetchAndWriteNeedle replica bound with the chunk

The context deadline includes sending the body, so a fixed 30 s would
fail a large remote-cache chunk on a slow link. The bound is now 30 s
plus 1 s per MiB, matching the Rust volume server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

seaweed-volume/src/server/grpc_server.rs

@@ -5618,27 +5618,7 @@ impl VolumeServer for VolumeGrpcService {
                 let needle_id = req.needle_id;
                 let size = req.size;
                 handles.push(tokio::spawn(async move {
-                    let form = reqwest::multipart::Form::new()
-                        .part("file", reqwest::multipart::Part::bytes(data_clone));
-                    let resp = client_clone
-                        .post(&url)
-                        .multipart(form)
-                        .send()
-                        .await
-                        .map_err(|e| {
-                            format!("remote write needle {} size {}: {}", needle_id, size, e)
-                        })?;
-                    if resp.status().is_success() {
-                        Ok(())
-                    } else {
-                        Err(format!(
-                            "remote write needle {} size {}: {} returned status {}",
-                            needle_id,
-                            size,
-                            url,
-                            resp.status()
-                        ))
-                    }
+                    fetch_and_write_replica(&client_clone, &url, data_clone, needle_id, size).await
                 }));
             }
         }
@@ -7906,6 +7886,51 @@ impl EcDecodeJob {
     }
 }
 
+/// The base bound on one `fetch_and_write_needle` replica write, from connect to
+/// the response status. Mirrors the Go volume server's fetchAndWriteReplicaTimeout.
+const FETCH_AND_WRITE_REPLICA_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// The slowest link a replica write's chunk may cross within its bound. The
+/// reqwest timeout includes sending the body, and a remote-cache chunk can be
+/// a thousandth of a very large object.
+const REPLICA_MIN_BYTES_PER_SEC: u64 = 1 << 20;
+
+fn fetch_and_write_replica_timeout(body_len: usize) -> std::time::Duration {
+    FETCH_AND_WRITE_REPLICA_TIMEOUT
+        + std::time::Duration::from_millis(
+            (body_len as u64).saturating_mul(1000) / REPLICA_MIN_BYTES_PER_SEC,
+        )
+}
+
+async fn fetch_and_write_replica(
+    client: &reqwest::Client,
+    url: &str,
+    data: Vec<u8>,
+    needle_id: u64,
+    size: i64,
+) -> Result<(), String> {
+    let timeout = fetch_and_write_replica_timeout(data.len());
+    let form = reqwest::multipart::Form::new().part("file", reqwest::multipart::Part::bytes(data));
+    let resp = client
+        .post(url)
+        .multipart(form)
+        .timeout(timeout)
+        .send()
+        .await
+        .map_err(|e| format!("remote write needle {} size {}: {}", needle_id, size, e))?;
+    if resp.status().is_success() {
+        Ok(())
+    } else {
+        Err(format!(
+            "remote write needle {} size {}: {} returned status {}",
+            needle_id,
+            size,
+            url,
+            resp.status()
+        ))
+    }
+}
+
 /// Bytes compacted between two `VacuumVolumeCompact` progress reports.
 const COMPACT_REPORT_INTERVAL: i64 = 128 * 1024 * 1024;
 
@@ -15143,4 +15168,82 @@ mod tests {
             assert_eq!(resp.results[0].error, "not found");
         }
     }
+
+    #[tokio::test(start_paused = true)]
+    async fn fetch_and_write_replica_fails_when_the_peer_goes_silent() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let url = format!(
+            "http://{}/3,01637037d6?type=replicate",
+            listener.local_addr().unwrap()
+        );
+        let client = reqwest::Client::new();
+        let started = tokio::time::Instant::now();
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(600),
+            fetch_and_write_replica(&client, &url, b"needle".to_vec(), 1, 6),
+        )
+        .await
+        .expect("a replica that never answers must not hold the request");
+        let err = result.expect_err("a silent replica must fail the write");
+        assert!(err.contains("remote write needle 1 size 6"), "{err}");
+        let elapsed = started.elapsed();
+        assert!(
+            elapsed >= FETCH_AND_WRITE_REPLICA_TIMEOUT
+                && elapsed < FETCH_AND_WRITE_REPLICA_TIMEOUT + std::time::Duration::from_secs(1),
+            "failed after {elapsed:?}: {err}"
+        );
+        drop(listener);
+    }
+
+    #[test]
+    fn fetch_and_write_replica_timeout_grows_with_the_chunk() {
+        let mib = REPLICA_MIN_BYTES_PER_SEC as usize;
+        assert_eq!(
+            fetch_and_write_replica_timeout(0),
+            FETCH_AND_WRITE_REPLICA_TIMEOUT
+        );
+        assert_eq!(
+            fetch_and_write_replica_timeout(1024 * mib),
+            FETCH_AND_WRITE_REPLICA_TIMEOUT + std::time::Duration::from_secs(1024)
+        );
+        assert_eq!(
+            fetch_and_write_replica_timeout(mib / 2),
+            FETCH_AND_WRITE_REPLICA_TIMEOUT + std::time::Duration::from_millis(500)
+        );
+    }
+
+    #[tokio::test]
+    async fn fetch_and_write_replica_reports_the_peer_status() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        for (status_line, ok) in [("201 Created", true), ("500 Internal Server Error", false)] {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let url = format!(
+                "http://{}/3,01637037d6?type=replicate",
+                listener.local_addr().unwrap()
+            );
+            let peer = tokio::spawn(async move {
+                let (mut conn, _) = listener.accept().await.unwrap();
+                let mut request = Vec::new();
+                let mut buf = [0u8; 4096];
+                while !request.ends_with(b"--\r\n") {
+                    let n = conn.read(&mut buf).await.unwrap();
+                    assert!(n > 0, "request ended early");
+                    request.extend_from_slice(&buf[..n]);
+                }
+                let reply = format!("HTTP/1.1 {status_line}\r\nContent-Length: 0\r\n\r\n");
+                conn.write_all(reply.as_bytes()).await.unwrap();
+                String::from_utf8_lossy(&request).into_owned()
+            });
+            let result =
+                fetch_and_write_replica(&reqwest::Client::new(), &url, b"needle".to_vec(), 1, 6)
+                    .await;
+            let request = peer.await.unwrap();
+            assert!(
+                request.starts_with("POST /3,01637037d6?type=replicate "),
+                "{request}"
+            );
+            assert!(request.contains("needle"), "{request}");
+            assert_eq!(result.is_ok(), ok, "{status_line}: {result:?}");
+        }
+    }
 }

weed/server/volume_grpc_remote.go

@@ -760,7 +760,7 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 					uploader = operation.NewUploaderWithHttpClient(newGuardedPeerHTTPClient(targetVolumeServer))
 				}
 
-				if _, replicaWriteErr := uploader.UploadData(ctx, data, uploadOption); replicaWriteErr != nil {
+				if replicaWriteErr := fetchAndWriteReplica(ctx, uploader, data, uploadOption); replicaWriteErr != nil {
 					replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, replicaWriteErr)
 				}
 			}(i, replica.Url)
@@ -780,6 +780,27 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 	return resp, err
 }
 
+// fetchAndWriteReplicaTimeout is the base bound on one FetchAndWriteNeedle
+// replica write, retries included, so a peer that stops answering cannot hold
+// the request.
+var fetchAndWriteReplicaTimeout = 30 * time.Second
+
+// replicaMinBytesPerSec is the slowest link a replica write's chunk may cross
+// within its bound. The bound includes sending the body, and a remote-cache
+// chunk can be a thousandth of a very large object.
+const replicaMinBytesPerSec = 1 << 20
+
+func fetchAndWriteReplicaBound(size int) time.Duration {
+	return fetchAndWriteReplicaTimeout + time.Duration(int64(size)*1000/replicaMinBytesPerSec)*time.Millisecond
+}
+
+func fetchAndWriteReplica(ctx context.Context, uploader *operation.Uploader, data []byte, option *operation.UploadOption) error {
+	ctx, cancel := context.WithTimeout(ctx, fetchAndWriteReplicaBound(len(data)))
+	defer cancel()
+	_, err := uploader.UploadData(ctx, data, option)
+	return err
+}
+
 // volumeRemoteReadError keeps a confirmed missing object distinguishable across
 // gRPC, where a wrapped sentinel would arrive as codes.Unknown text. The
 // sentinel's text in the message is what the filer matches, so a NotFound for

weed/server/volume_grpc_remote_test.go

@@ -902,6 +902,52 @@ func TestGuardedPeerHTTPClientUploadsOverTLS(t *testing.T) {
 	}
 }
 
+// TestFetchAndWriteReplicaBoundsASilentPeer: a replica that accepts the
+// connection and never answers fails the write at the bound.
+func TestFetchAndWriteReplicaBoundsASilentPeer(t *testing.T) {
+	listener, err := net.Listen("tcp", net.JoinHostPort(peerIPv4(t).String(), "0"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+	defer func(original time.Duration) { fetchAndWriteReplicaTimeout = original }(fetchAndWriteReplicaTimeout)
+	fetchAndWriteReplicaTimeout = time.Second
+
+	addr := listener.Addr().String()
+	uploader := operation.NewUploaderWithHttpClient(newGuardedPeerHTTPClient(addr))
+	done := make(chan error, 1)
+	go func() {
+		done <- fetchAndWriteReplica(t.Context(), uploader, []byte("needle-data"), &operation.UploadOption{
+			UploadUrl:     "http://" + addr + "/1,01637037d6?type=replicate",
+			IsReplication: true,
+		})
+	}()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("silent replica: got %v, want %v", err, context.DeadlineExceeded)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the write to a silent replica is still waiting")
+	}
+}
+
+func TestFetchAndWriteReplicaBoundGrowsWithTheChunk(t *testing.T) {
+	const mib = replicaMinBytesPerSec
+	for _, tc := range []struct {
+		size int
+		want time.Duration
+	}{
+		{0, fetchAndWriteReplicaTimeout},
+		{mib / 2, fetchAndWriteReplicaTimeout + 500*time.Millisecond},
+		{1024 * mib, fetchAndWriteReplicaTimeout + 1024*time.Second},
+	} {
+		if got := fetchAndWriteReplicaBound(tc.size); got != tc.want {
+			t.Errorf("fetchAndWriteReplicaBound(%d) = %v, want %v", tc.size, got, tc.want)
+		}
+	}
+}
+
 // TestGuardedPeerHTTPClientRejectsHTTPSDowngrade pins the downgrade guard the
 // global client carries: an HTTPS peer answering with a redirect to a
 // reachable http server must fail the upload rather than resend the body and