From 23598afb3132bf1864b06d90e70ea501f8e038c9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 5 Nov 2026 04:05:33 -0800 Subject: [PATCH] volume: bound each replica forward so a silent peer cannot hold a write (#12417) * volume: bound replica forwards so a silent peer cannot hold a write The Rust volume server forwards each replicated write to its peers with the shared outgoing reqwest client, which had no connect or request timeout. When a peer is silently partitioned (packets dropped, no reset) the forward waits until TCP gives up, minutes later, and the write waits with it. A chaos test (replication 001, one volume server partitioned) saw every write on a volume with a copy there block. - Replica forwards go through replica_request, bounded at REPLICA_REQUEST_TIMEOUT = 10 s from connect to the response. - Both outgoing clients start from outgoing_client_builder, which adds a 5 s connect timeout, so new connections to a peer that never answers fail fast. The shared client gets no overall request timeout, since it also serves lookups and other transfers. Tests: a replica that receives the forward and never answers fails the whole replicated write at the bound, naming that replica; a silent peer fails the request builder at the bound on a paused clock; a connect that stalls in the resolver fails at the 5 s connect bound. Co-Authored-By: Claude Opus 5.5 * volume: give a replica forward more time for a larger body The 10 s replica forward bound runs from connect to the end of the response, so it also covers sending the body. A write near the 256 MB file size limit needs about 26 MB/s to reach its replica in 10 s, so on a link that is slower or busy the forward failed and the write returned 500 even though the peer was answering. A forward now gets REPLICA_REQUEST_TIMEOUT plus one second per MiB of body. A bodiless forward keeps the 10 s bound (a 1 MiB write gets 11 s). Co-Authored-By: Claude Opus 5.5 * volume: bound each Go replica forward like the Rust one The Go volume server forwards a replicated write with a 10 s dial bound and the transport's 30 s response header timeout. A replica that takes the forward and never answers held the write for 30 s, and a body larger than the socket buffers to a silently partitioned peer was held until TCP gave up, since nothing bounds sending the body. Each forward now runs under replicaRequestBound: 10 s plus one second per MiB of body, the same bound the Rust volume server puts on its forwards. Test: a replica that receives the forward and never answers fails the write at the bound (300 ms in the test); without it the write fails at 30 s on the response header timeout. Co-Authored-By: Claude Opus 5.5 * volume: keep proxy variables out of the stalled-connect test reqwest reads HTTP_PROXY and ALL_PROXY by default, so with a proxy set the test's request went to the proxy instead of the stalled resolver. Co-Authored-By: Claude Opus 5.5 * volume: count a replica forward that hit its bound as a timeout DistributedOperationResult joined the replica errors into a plain string, and the uploader reports a cancelled request without wrapping the context error, so a forward cut off at replicaRequestBound was counted under server_error in the replication failure metric, as was a response header timeout before it. The joined error now unwraps to each replica's error, and a forward that ran out of its bound reports context.DeadlineExceeded. Test: the silent replica write raises the timeout failure count by one; with the flattened join it stays put. Co-Authored-By: Claude Opus 5.5 * volume: test that a large Go replica forward gets its body's bound A replica answers after the bodiless bound but within the bound for a 1 MiB body, so the write fails if replicateTo drops the body scaling. Co-Authored-By: Claude Opus 5.5 * volume: run the Rust whole-path replica test with a 1 MiB body The forward then has to wait the bound for its body, so the test fails if do_replicated_request stops passing the body length to the bound. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: pingqiu Co-authored-by: Claude Opus 5.5 --- seaweed-volume/src/main.rs | 44 +++++++- seaweed-volume/src/server/handlers.rs | 154 +++++++++++++++++++++++++- weed/topology/store_replicate.go | 44 +++++++- weed/topology/store_replicate_test.go | 117 +++++++++++++++++++ 4 files changed, 350 insertions(+), 9 deletions(-) diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 57fedb75d9..3216e1ebeb 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -118,6 +118,17 @@ fn reject_https_downgrade(attempt: reqwest::redirect::Attempt) -> reqwest::redir } } +/// A peer that never answers a SYN (silent partition) must not stall new +/// outgoing connections; requests set their own overall bounds as needed. +const OUTGOING_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +/// Base of every outgoing HTTP client: redirect policy and connect bound. +fn outgoing_client_builder(redirect_policy: reqwest::redirect::Policy) -> reqwest::ClientBuilder { + reqwest::Client::builder() + .redirect(redirect_policy) + .connect_timeout(OUTGOING_CONNECT_TIMEOUT) +} + fn build_outgoing_http_client( config: &VolumeServerConfig, ) -> Result<(reqwest::Client, String), Box> { @@ -129,14 +140,12 @@ fn build_outgoing_http_client( let redirect_policy = reqwest::redirect::Policy::custom(reject_https_downgrade); if !config.https_client_enabled { return Ok(( - reqwest::Client::builder() - .redirect(redirect_policy) - .build()?, + outgoing_client_builder(redirect_policy).build()?, scheme.to_string(), )); } - let mut builder = reqwest::Client::builder().redirect(redirect_policy); + let mut builder = outgoing_client_builder(redirect_policy); if !config.https_client_ca_file.is_empty() { let ca_pem = std::fs::read(&config.https_client_ca_file).map_err(|e| { format!( @@ -1115,6 +1124,33 @@ mod tests { }; use seaweed_volume::security::tls::TlsPolicy; + /// A connect that never finishes fails at the 5 s bound. It stalls in the + /// resolver, which the connect bound also covers, because that needs no + /// routing: a real unanswered SYN depends on the host's network. + #[tokio::test(start_paused = true)] + async fn outgoing_client_bounds_a_stalled_connect() { + struct Stalled; + impl reqwest::dns::Resolve for Stalled { + fn resolve(&self, _: reqwest::dns::Name) -> reqwest::dns::Resolving { + Box::pin(std::future::pending()) + } + } + let client = super::outgoing_client_builder(reqwest::redirect::Policy::none()) + .no_proxy() + .dns_resolver(std::sync::Arc::new(Stalled)) + .build() + .unwrap(); + let started = tokio::time::Instant::now(); + let sent = tokio::time::timeout( + std::time::Duration::from_secs(600), + client.get("http://silent-peer.invalid:8080/status").send(), + ) + .await + .expect("a stalled connect must not outlive its bound"); + assert!(sent.is_err(), "a stalled connect cannot answer"); + assert_eq!(started.elapsed(), super::OUTGOING_CONNECT_TIMEOUT); + } + fn write_pem(dir: &tempfile::TempDir, name: &str, body: &str) -> String { let path = dir.path().join(name); std::fs::write(&path, body).unwrap(); diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 846e42e95f..54eacac554 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -713,6 +713,36 @@ fn override_query_fields(query: &str, overrides: &[(&str, String)]) -> Result std::time::Duration { + REPLICA_REQUEST_TIMEOUT + + std::time::Duration::from_millis( + (body_len as u64).saturating_mul(1000) / REPLICA_MIN_BYTES_PER_SEC, + ) +} + +fn replica_request( + client: &reqwest::Client, + method: axum::http::Method, + url: &str, + body_len: usize, +) -> reqwest::RequestBuilder { + client + .request(method, url) + .timeout(replica_request_timeout(body_len)) +} + /// Helper to synchronously replicate a request to peer volume servers. async fn do_replicated_request( state: &VolumeServerState, @@ -787,7 +817,8 @@ async fn do_replicated_request( )?; let client = state.http_client.clone(); - let mut req_builder = client.request(method.clone(), &url); + let body_len = body.as_ref().map_or(0, |b| b.len()); + let mut req_builder = replica_request(&client, method.clone(), &url, body_len); // Forward relevant headers if let Some(ct) = headers.get(axum::http::header::CONTENT_TYPE) { @@ -4955,6 +4986,127 @@ fn parse_content_disposition_filename(value: &str) -> Option { mod tests { use super::*; + /// A replica that reads the forward and never answers. Each request's + /// first line is sent on the returned channel. + async fn spawn_silent_replica() -> ( + std::net::SocketAddr, + tokio::sync::mpsc::UnboundedReceiver, + ) { + use tokio::io::AsyncReadExt; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (received, receipts) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + loop { + let (mut socket, _) = listener.accept().await.unwrap(); + let received = received.clone(); + tokio::spawn(async move { + let mut request = vec![0u8; 4096]; + let n = socket.read(&mut request).await.unwrap_or(0); + let text = String::from_utf8_lossy(&request[..n]).into_owned(); + let _ = received.send(text.lines().next().unwrap_or("").to_string()); + std::future::pending::<()>().await; + }); + } + }); + (addr, receipts) + } + + /// The whole replicated write path, not only the request builder: the + /// replica receives the forward, never answers, and the write fails at + /// the bound for its body with an error naming that replica. + #[tokio::test] + async fn test_replicated_write_fails_at_the_bound_when_a_peer_never_answers() { + let tmp = tempfile::TempDir::new().unwrap(); + let (peer, mut receipts) = spawn_silent_replica().await; + let state = remote_manifest_state(&tmp, spawn_lookup(peer).await); + let body = bytes::Bytes::from(vec![0u8; REPLICA_MIN_BYTES_PER_SEC as usize]); + let bound = replica_request_timeout(body.len()); + let started = std::time::Instant::now(); + let sent = do_replicated_request( + &state, + 1, + axum::http::Method::POST, + "/1,01ab", + "", + &axum::http::HeaderMap::new(), + Some(body), + ) + .await; + let took = started.elapsed(); + let forwarded = receipts + .try_recv() + .expect("the replica never received the forward"); + assert!( + forwarded.starts_with("POST /1,01ab?") && forwarded.contains("type=replicate"), + "{forwarded}" + ); + let error = sent.expect_err("a silent replica cannot ack the write"); + assert!(error.contains(&peer.to_string()), "{error}"); + assert!( + took >= bound && took < bound + std::time::Duration::from_secs(5), + "{took:?}" + ); + } + + /// A peer that accepts the connection but never answers (a silent + /// partition after connect) must fail the forward at the bound, not hold + /// the write until TCP gives up. + #[tokio::test(start_paused = true)] + async fn test_replica_request_times_out_on_a_silent_peer() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let _held = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + std::future::pending::<()>().await; + drop(socket); + }); + let client = reqwest::Client::new(); + let url = format!("http://{}/3,01ab?type=replicate", addr); + let body = vec![0u8; 1 << 20]; + let bound = replica_request_timeout(body.len()); + let started = tokio::time::Instant::now(); + let sent = tokio::time::timeout( + std::time::Duration::from_secs(600), + replica_request(&client, axum::http::Method::POST, &url, body.len()) + .body(body) + .send(), + ) + .await + .expect("replica forward must not outlive its bound"); + let err = sent.expect_err("a silent peer cannot ack the replica"); + assert!(err.is_timeout(), "{err}"); + assert!(started.elapsed() >= bound && started.elapsed() < bound * 2); + } + + /// A forward's bound grows with its body at REPLICA_MIN_BYTES_PER_SEC, so + /// a write near the file size limit is not cut off at the bodiless bound. + #[test] + fn test_replica_request_bound_grows_with_the_body() { + assert_eq!(replica_request_timeout(0), REPLICA_REQUEST_TIMEOUT); + let mib = REPLICA_MIN_BYTES_PER_SEC as usize; + assert_eq!( + replica_request_timeout(256 * mib), + REPLICA_REQUEST_TIMEOUT + std::time::Duration::from_secs(256) + ); + assert_eq!( + replica_request_timeout(mib / 2), + REPLICA_REQUEST_TIMEOUT + std::time::Duration::from_millis(500) + ); + let request = replica_request( + &reqwest::Client::new(), + axum::http::Method::POST, + "http://127.0.0.1:1/3,01ab?type=replicate", + 64 * mib, + ) + .build() + .unwrap(); + assert_eq!( + request.timeout(), + Some(&(REPLICA_REQUEST_TIMEOUT + std::time::Duration::from_secs(64))) + ); + } + /// The upload handler reads fsync off the decoded query fields rather than /// matching the raw string, because Go's r.FormValue decodes and a raw /// match would silently drop a percent-encoded value. diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index d130d3200a..e8851a76d6 100644 --- a/weed/topology/store_replicate.go +++ b/weed/topology/store_replicate.go @@ -177,7 +177,7 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi glog.Errorf("replication-UploadData, err:%v, url:%s", uploaderErr, u.String()) return uploaderErr } - _, uploadErr := uploader.UploadData(ctx, replicaData, uploadOption) + uploadErr := replicateTo(ctx, uploader, replicaData, uploadOption) if uploadErr != nil { glog.Errorf("replication-UploadData, err:%v, url:%s", uploadErr, u.String()) } @@ -211,6 +211,30 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi return localIsUnchanged, nil } +// replicaRequestTimeout bounds one replica forward with no body, from connect +// to the end of the response, so a silently partitioned peer cannot hold the +// write. Mirrors the Rust volume server's REPLICA_REQUEST_TIMEOUT. +var replicaRequestTimeout = 10 * time.Second + +// replicaMinBytesPerSec is the slowest link a forward's body may cross within +// its bound, so a large write is not cut off on a link slower than a LAN. +const replicaMinBytesPerSec = 1 << 20 + +func replicaRequestBound(size int) time.Duration { + return replicaRequestTimeout + time.Duration(int64(size)*1000/replicaMinBytesPerSec)*time.Millisecond +} + +func replicateTo(ctx context.Context, uploader *operation.Uploader, data []byte, option *operation.UploadOption) error { + bound := replicaRequestBound(len(data)) + ctx, cancel := context.WithTimeout(ctx, bound) + defer cancel() + _, err := uploader.UploadData(ctx, data, option) + if err != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) && !errors.Is(err, context.DeadlineExceeded) { + return fmt.Errorf("%w after %v: %v", context.DeadlineExceeded, bound, err) + } + return err +} + // ReplicatedDelete deletes a needle from the local volume and sends delete // requests to all remote replica locations. Replica deletes use // context.Background() so that a client disconnect does not orphan replica @@ -275,18 +299,30 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp type DistributedOperationResult map[string]error func (dr DistributedOperationResult) Error() error { - var errs []string + var msgs []string + var errs []error for k, v := range dr { if v != nil { - errs = append(errs, fmt.Sprintf("[%s]: %v", k, v)) + msgs = append(msgs, fmt.Sprintf("[%s]: %v", k, v)) + errs = append(errs, v) } } if len(errs) == 0 { return nil } - return errors.New(strings.Join(errs, "\n")) + return &distributedOperationError{msg: strings.Join(msgs, "\n"), errs: errs} } +// distributedOperationError keeps each replica's error reachable through +// errors.Is, so a forward that hit its bound is classified as a timeout. +type distributedOperationError struct { + msg string + errs []error +} + +func (e *distributedOperationError) Error() string { return e.msg } +func (e *distributedOperationError) Unwrap() []error { return e.errs } + type RemoteResult struct { Host string Error error diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go index d25c574028..f11cffe42e 100644 --- a/weed/topology/store_replicate_test.go +++ b/weed/topology/store_replicate_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "net" "net/http" "net/http/httptest" @@ -13,6 +14,7 @@ import ( "testing" "time" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -450,3 +452,118 @@ func TestReplicatedWriteRejectsWriteWithNoTarget(t *testing.T) { }) } } + +// TestReplicatedWriteFailsAtTheBoundWhenAReplicaNeverAnswers: a replica that +// takes the forward and never answers fails the write at the forward bound, +// not at the transport's 30 s response header timeout. +func TestReplicatedWriteFailsAtTheBoundWhenAReplicaNeverAnswers(t *testing.T) { + received := make(chan struct{}, 1) + release := make(chan struct{}) + replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- struct{}{} + <-release + })) + defer replica.Close() + defer close(release) + + defer func(d time.Duration) { replicaRequestTimeout = d }(replicaRequestTimeout) + replicaRequestTimeout = 300 * time.Millisecond + + masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{ + locations: []*master_pb.Location{{Url: strings.TrimPrefix(replica.URL, "http://")}}, + }) + volumeId := needle.VolumeId(1) + operation.InvalidateVolumeIdLocationCache(volumeId.String()) + r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/1,01637037d6", nil) + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + n := &needle.Needle{Id: 1, Data: []byte("test data"), Ttl: needle.EMPTY_TTL} + + timeouts := stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpWrite, stats.FailureTimeout) + timeoutsBefore := testutil.ToFloat64(timeouts) + start := time.Now() + _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, &storage.Store{}, volumeId, n, r, "") + elapsed := time.Since(start) + select { + case <-received: + default: + t.Fatal("the replica never received the forward") + } + if err == nil { + t.Fatal("a replica that never answers cannot ack the write") + } + if elapsed < replicaRequestTimeout || elapsed > 5*time.Second { + t.Fatalf("write failed after %v, want about the %v forward bound: %v", elapsed, replicaRequestTimeout, err) + } + if got := testutil.ToFloat64(timeouts) - timeoutsBefore; got != 1 { + t.Fatalf("replication timeout failures went up by %v, want 1: %v", got, err) + } +} + +// TestReplicatedWriteGivesALargerBodyMoreTime: a replica that answers after +// the bodiless bound, but within the bound for its body, still acks the write. +func TestReplicatedWriteGivesALargerBodyMoreTime(t *testing.T) { + defer func(d time.Duration) { replicaRequestTimeout = d }(replicaRequestTimeout) + replicaRequestTimeout = 300 * time.Millisecond + data := make([]byte, replicaMinBytesPerSec) + answerAfter := (replicaRequestTimeout + replicaRequestBound(len(data))) / 2 + + received := make(chan int, 1) + replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mr, err := r.MultipartReader() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + part, err := mr.NextPart() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + body, _ := io.ReadAll(part) + received <- len(body) + time.Sleep(answerAfter) + w.WriteHeader(http.StatusCreated) + fmt.Fprint(w, "{}") + })) + defer replica.Close() + + masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{ + locations: []*master_pb.Location{{Url: strings.TrimPrefix(replica.URL, "http://")}}, + }) + volumeId := needle.VolumeId(1) + operation.InvalidateVolumeIdLocationCache(volumeId.String()) + r := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:8080/1,01637037d6", nil) + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + n := &needle.Needle{Id: 1, Data: data, Ttl: needle.EMPTY_TTL} + + if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, &storage.Store{}, volumeId, n, r, ""); err != nil { + t.Fatalf("a replica answering within the %v bound for a %d-byte body failed the write: %v", replicaRequestBound(len(data)), len(data), err) + } + select { + case got := <-received: + if got != len(data) { + t.Fatalf("the replica received %d bytes, want %d", got, len(data)) + } + default: + t.Fatal("the replica never received the forward") + } +} + +func TestReplicaRequestBoundGrowsWithTheBody(t *testing.T) { + for _, tc := range []struct { + size int + want time.Duration + }{ + {0, replicaRequestTimeout}, + {replicaMinBytesPerSec / 2, replicaRequestTimeout + 500*time.Millisecond}, + {256 * replicaMinBytesPerSec, replicaRequestTimeout + 256*time.Second}, + } { + if got := replicaRequestBound(tc.size); got != tc.want { + t.Errorf("replicaRequestBound(%d) = %v, want %v", tc.size, got, tc.want) + } + } +}