diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 541b9b797b..5ee4890416 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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, + 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:?}"); + } + } } diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index 3943886899..b30015866b 100644 --- a/weed/server/volume_grpc_remote.go +++ b/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 diff --git a/weed/server/volume_grpc_remote_test.go b/weed/server/volume_grpc_remote_test.go index 50957a0991..72ba24fbcf 100644 --- a/weed/server/volume_grpc_remote_test.go +++ b/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