From 2a8c6e606da3d9ae6050c6b55a8ffa4d9ff33a9d Mon Sep 17 00:00:00 2001 From: Jacob Marble Date: Thu, 13 Aug 2026 15:40:16 +0000 Subject: [PATCH 1/4] proto-grpc: home the status bounding shared by runtime crates `MAX_STATUS_MESSAGE_LEN` and its helpers live in `runtime-next`, but three other crates map errors onto the same wire and need the same ceiling. Move them to `proto-grpc`, which each of those crates already depends on, and re-export from `runtime-next` so callers are unchanged. No behavior change: `runtime-next` bounds exactly what it bounded before. --- Cargo.lock | 2 + crates/proto-grpc/Cargo.toml | 2 + crates/proto-grpc/src/lib.rs | 3 ++ crates/proto-grpc/src/status.rs | 69 ++++++++++++++++++++++++++++++++ crates/runtime-next/src/lib.rs | 71 +++------------------------------ 5 files changed, 81 insertions(+), 66 deletions(-) create mode 100644 crates/proto-grpc/src/status.rs diff --git a/Cargo.lock b/Cargo.lock index 3b17495a5aa..49e8d603881 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6494,6 +6494,8 @@ dependencies = [ name = "proto-grpc" version = "0.0.0" dependencies = [ + "anyhow", + "bytes", "labels", "proto-build", "proto-flow", diff --git a/crates/proto-grpc/Cargo.toml b/crates/proto-grpc/Cargo.toml index 140a1f9c9c3..5a1c18eaa24 100644 --- a/crates/proto-grpc/Cargo.toml +++ b/crates/proto-grpc/Cargo.toml @@ -14,6 +14,8 @@ proto-flow = { path = "../proto-flow", optional = true } proto-gazette = { path = "../proto-gazette" } tokens = { path = "../tokens" } +anyhow = { workspace = true } +bytes = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tonic-prost = { workspace = true } diff --git a/crates/proto-grpc/src/lib.rs b/crates/proto-grpc/src/lib.rs index f3f60660a32..5ce386cf19d 100644 --- a/crates/proto-grpc/src/lib.rs +++ b/crates/proto-grpc/src/lib.rs @@ -10,6 +10,9 @@ mod protocol; pub mod runtime; pub mod shuffle; +mod status; +pub use status::{MAX_STATUS_MESSAGE_LEN, anyhow_to_status, bound_status, bounded_unknown_status}; + // The `protocol` package is publicly exported as `broker`. #[cfg(any(feature = "broker_client", feature = "broker_server"))] pub mod broker { diff --git a/crates/proto-grpc/src/status.rs b/crates/proto-grpc/src/status.rs new file mode 100644 index 00000000000..7ae98dab3df --- /dev/null +++ b/crates/proto-grpc/src/status.rs @@ -0,0 +1,69 @@ +//! Bounding of `tonic::Status` messages, so that a status survives its trip +//! over the wire. + +/// Maximum byte length of a `tonic::Status` message that we build from a +/// formatted error. gRPC status text rides in an HTTP/2 trailer; an oversized +/// trailer forces the header block across many CONTINUATION frames, tripping +/// `h2`'s `too_many_continuations` guard, which aborts the connection and +/// *replaces* the real status text with an opaque transport error +/// +/// The ceiling exists so that *any* error, anticipated or not, stays within a +/// single HTTP/2 frame (default `max_frame_size` is 16 KiB) and never needs a +/// CONTINUATION frame at all. The `grpc-message` trailer is percent-encoded, so +/// a byte can expand up to 3x (`%XX`); 4 KiB raw is ≤ 12 KiB encoded, fitting +/// one frame even in the worst case. +pub const MAX_STATUS_MESSAGE_LEN: usize = 4096; + +/// Build a `tonic::Status::unknown` whose message is bounded to +/// [`MAX_STATUS_MESSAGE_LEN`] bytes. The anyhow debug format leads with the +/// error's top-level context and appends lower-level detail, so truncating the +/// tail preserves the human-meaningful prefix. +pub fn bounded_unknown_status(message: String) -> tonic::Status { + tonic::Status::unknown(bound_message(message)) +} + +/// Bound the message of a `Status` we did not format ourselves — one round +/// tripped from a peer or produced by a connector — preserving its code, +/// details, and metadata. Guards the same h2 trailer limit as +/// [`bounded_unknown_status`] so that e.g. a misbehaving connector emitting a +/// huge message can't produce a status that's dropped in transit. Returns the +/// status untouched when its message already fits. +pub fn bound_status(status: tonic::Status) -> tonic::Status { + if status.message().len() <= MAX_STATUS_MESSAGE_LEN { + return status; + } + tonic::Status::with_details_and_metadata( + status.code(), + bound_message(status.message().to_string()), + bytes::Bytes::copy_from_slice(status.details()), + status.metadata().clone(), + ) +} + +/// Truncate `message` to at most [`MAX_STATUS_MESSAGE_LEN`] bytes, cutting the +/// tail on a UTF-8 boundary and marking the elision. Returns it unchanged when +/// it already fits. +fn bound_message(mut message: String) -> String { + if message.len() > MAX_STATUS_MESSAGE_LEN { + const SUFFIX: &str = "… [truncated]"; + // Reserve room for SUFFIX and back off to a UTF-8 char boundary. + let mut end = MAX_STATUS_MESSAGE_LEN - SUFFIX.len(); + while !message.is_char_boundary(end) { + end -= 1; + } + message.truncate(end); + message.push_str(SUFFIX); + } + message +} + +/// Map an `anyhow::Error` into a `tonic::Status`. +/// If the error is already a Status, it's downcast (and bounded). +/// Otherwise an Unknown status wraps the formatted error chain, bounded to +/// [`MAX_STATUS_MESSAGE_LEN`]. +pub fn anyhow_to_status(err: anyhow::Error) -> tonic::Status { + match err.downcast::() { + Ok(status) => bound_status(status), + Err(err) => bounded_unknown_status(format!("{err:?}")), + } +} diff --git a/crates/runtime-next/src/lib.rs b/crates/runtime-next/src/lib.rs index 68a1ab2e826..314ea212c55 100644 --- a/crates/runtime-next/src/lib.rs +++ b/crates/runtime-next/src/lib.rs @@ -148,72 +148,11 @@ impl RuntimeProtocol { } } -/// Maximum byte length of a `tonic::Status` message that we build from a -/// formatted error. gRPC status text rides in an HTTP/2 trailer; an oversized -/// trailer forces the header block across many CONTINUATION frames, tripping -/// `h2`'s `too_many_continuations` guard, which aborts the connection and -/// *replaces* the real status text with an opaque transport error -/// -/// The ceiling exists so that *any* error, anticipated or not, stays within a -/// single HTTP/2 frame (default `max_frame_size` is 16 KiB) and never needs a -/// CONTINUATION frame at all. The `grpc-message` trailer is percent-encoded, so -/// a byte can expand up to 3x (`%XX`); 4 KiB raw is ≤ 12 KiB encoded, fitting -/// one frame even in the worst case. -pub const MAX_STATUS_MESSAGE_LEN: usize = 4096; - -// Map an anyhow::Error into a tonic::Status. -// If the error is already a Status, it's downcast (and bounded). -// Otherwise, an internal error is used to wrap a formatted anyhow::Error chain, -// bounded to MAX_STATUS_MESSAGE_LEN. -pub fn anyhow_to_status(err: anyhow::Error) -> tonic::Status { - match err.downcast::() { - Ok(status) => bound_status(status), - Err(err) => bounded_unknown_status(format!("{err:?}")), - } -} - -/// Build a `tonic::Status::unknown` whose message is bounded to -/// [`MAX_STATUS_MESSAGE_LEN`] bytes. The anyhow debug format leads with the -/// error's top-level context and appends lower-level detail, so truncating the -/// tail preserves the human-meaningful prefix. -pub(crate) fn bounded_unknown_status(message: String) -> tonic::Status { - tonic::Status::unknown(bound_message(message)) -} - -/// Bound the message of a `Status` we did not format ourselves — one round -/// tripped from a peer or produced by a connector — preserving its code, -/// details, and metadata. Guards the same h2 trailer limit as -/// [`bounded_unknown_status`] so that e.g. a misbehaving connector emitting a -/// huge message can't produce a status that's dropped in transit. Returns the -/// status untouched when its message already fits. -pub(crate) fn bound_status(status: tonic::Status) -> tonic::Status { - if status.message().len() <= MAX_STATUS_MESSAGE_LEN { - return status; - } - tonic::Status::with_details_and_metadata( - status.code(), - bound_message(status.message().to_string()), - bytes::Bytes::copy_from_slice(status.details()), - status.metadata().clone(), - ) -} - -/// Truncate `message` to at most [`MAX_STATUS_MESSAGE_LEN`] bytes, cutting the -/// tail on a UTF-8 boundary and marking the elision. Returns it unchanged when -/// it already fits. -fn bound_message(mut message: String) -> String { - if message.len() > MAX_STATUS_MESSAGE_LEN { - const SUFFIX: &str = "… [truncated]"; - // Reserve room for SUFFIX and back off to a UTF-8 char boundary. - let mut end = MAX_STATUS_MESSAGE_LEN - SUFFIX.len(); - while !message.is_char_boundary(end) { - end -= 1; - } - message.truncate(end); - message.push_str(SUFFIX); - } - message -} +// Status bounding lives in `proto-grpc`, which every crate speaking this +// protocol already depends on. See `proto_grpc::MAX_STATUS_MESSAGE_LEN` for +// why an unbounded status can't survive its trip over the wire. +pub(crate) use proto_grpc::{bound_status, bounded_unknown_status}; +pub use proto_grpc::{MAX_STATUS_MESSAGE_LEN, anyhow_to_status}; // Map a tonic::Status into an anyhow::Error. // If the status is an internal error, its message is extracted into a dynamic anyhow::Error. From 20b572b5e08e78632f9f34e19f7b25cde4da3bbf Mon Sep 17 00:00:00 2001 From: Jacob Marble Date: Thu, 13 Aug 2026 15:40:37 +0000 Subject: [PATCH 2/4] runtime, shuffle: bound statuses mapped from anyhow errors Both crates carried their own copy of `anyhow_to_status` which formatted an error chain into a status message of unbounded length. A gRPC status rides in an HTTP/2 trailer, and an oversized trailer is not merely truncated: it spills across more CONTINUATION frames than `h2` allows, which tears down the connection and replaces the status text with an opaque transport error. Use the shared bounded implementation, as `runtime-next` already did. --- crates/runtime/src/lib.rs | 11 ++++------- crates/shuffle/src/lib.rs | 12 ++++-------- 2 files changed, 8 insertions(+), 15 deletions(-) diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 677bbf1aefc..475a44bff38 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -86,14 +86,11 @@ impl RuntimeProtocol { } // Map an anyhow::Error into a tonic::Status. -// If the error is already a Status, it's downcast. +// If the error is already a Status, it's downcast (and bounded). // Otherwise, an internal error is used to wrap a formatted anyhow::Error chain. -pub fn anyhow_to_status(err: anyhow::Error) -> tonic::Status { - match err.downcast::() { - Ok(status) => status, - Err(err) => tonic::Status::unknown(format!("{err:?}")), - } -} +// Bounding keeps an oversized status from being replaced in transit by an +// opaque transport error; see `proto_grpc::MAX_STATUS_MESSAGE_LEN`. +pub use proto_grpc::anyhow_to_status; // Map a tonic::Status into an anyhow::Error. // If the status is an internal error, its message is extracted into a dynamic anyhow::Error. diff --git a/crates/shuffle/src/lib.rs b/crates/shuffle/src/lib.rs index aa6456d238f..8e297cc3c93 100644 --- a/crates/shuffle/src/lib.rs +++ b/crates/shuffle/src/lib.rs @@ -167,14 +167,10 @@ fn opening_aborted(source: &str, peer: &str, msg: Option>) - } } -// Map an anyhow::Error into a tonic::Status. -#[inline] -fn anyhow_to_status(err: anyhow::Error) -> tonic::Status { - match err.downcast::() { - Ok(status) => status, - Err(err) => tonic::Status::unknown(format!("{err:?}")), - } -} +// Map an anyhow::Error into a tonic::Status, bounding its message so that an +// oversized status isn't replaced in transit by an opaque transport error. +// See `proto_grpc::MAX_STATUS_MESSAGE_LEN`. +use proto_grpc::anyhow_to_status; // Map a tonic::Status into an anyhow::Error. #[inline] From ae99226f3ac0629ae05793b85a880bd40afef663 Mon Sep 17 00:00:00 2001 From: Jacob Marble Date: Thu, 13 Aug 2026 15:41:01 +0000 Subject: [PATCH 3/4] connector-init: bound the connector's terminal status trailer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a connector exits non-zero we report its final stderr log as the RPC's terminal Status: the log's message as `grpc-message`, and the whole encoded `ops::Log` again as `last-log-bin` metadata. Neither was bounded. That status is the sole carrier of the connector's error, and it travels entirely in an HTTP/2 trailer. hyper pins `max_header_list_size` to 16 KiB, which gives `h2` a budget of five CONTINUATION frames; past that it trips `too_many_continuations` and answers with a connection-level GOAWAY(ENHANCE_YOUR_CALM). Every stream on the connection then fails with an opaque `ResourceExhausted: h2 protocol error` — so a connector which prints a large diagnostic as it dies destroys the message explaining why it died. Bound the log before it reaches the wire. The budget is small enough that even worst-case encoding (3x percent-encoding of the message, 4/3 base64 of the metadata) fits a single frame, needing no CONTINUATION frame at all. Part of #3353. --- crates/connector-init/src/rpc.rs | 174 ++++++++++++++++++++++++++++++- 1 file changed, 173 insertions(+), 1 deletion(-) diff --git a/crates/connector-init/src/rpc.rs b/crates/connector-init/src/rpc.rs index 329c7a97b6c..95e6e9662ca 100644 --- a/crates/connector-init/src/rpc.rs +++ b/crates/connector-init/src/rpc.rs @@ -123,6 +123,12 @@ where if !status.success() { tracing::error!(%status, "connector failed"); + // Bound before the Log reaches the wire: this Status is the sole + // carrier of the connector's terminal error, and an oversized + // trailer destroys it outright. See MAX_TERMINAL_LOG_LEN. + let mut last_log = last_log; + bound_log(&mut last_log); + let mut status = Status::unknown(&last_log.message); status.metadata_mut().insert_bin( "last-log-bin", @@ -221,10 +227,69 @@ fn map_status>(message: &'static str, err: E) -> Status { Status::internal(format!("{:#}", anyhow::anyhow!(err).context(message))) } +/// Maximum byte length of the connector's final log message, which is sent +/// both as the terminal Status message and (encoded) as its `last-log-bin` +/// metadata. +/// +/// A gRPC status rides entirely in an HTTP/2 trailer. hyper pins +/// `SETTINGS_MAX_HEADER_LIST_SIZE` to 16 KiB, which gives `h2` a budget of five +/// CONTINUATION frames; a trailer that needs more trips `too_many_continuations`, +/// and h2 answers with a *connection*-level GOAWAY(ENHANCE_YOUR_CALM). Every +/// stream on the connection then fails with an opaque +/// `ResourceExhausted: h2 protocol error` — and the connector's real error, +/// which is the whole point of this Status, is lost in transit. +/// +/// So the trailer must fit a single 16 KiB frame. The message rides in +/// percent-encoded `grpc-message` (up to 3x) and again in base64 +/// `last-log-bin` (4/3x, plus the Log's other fields), so the worst case is +/// ~4.4x this bound: 2 KiB raw stays under 9 KiB encoded, with room to spare +/// for the Log's shard, timestamp, and remaining fields. +/// +/// `runtime-next` bounds the statuses *it* formats for the same reason; see +/// `runtime_next::MAX_STATUS_MESSAGE_LEN`. +const MAX_TERMINAL_LOG_LEN: usize = 2048; + +/// Truncate `log` so that it fits a single HTTP/2 frame once encoded into a +/// gRPC status trailer. See [`MAX_TERMINAL_LOG_LEN`]. +fn bound_log(log: &mut ops::Log) { + const SUFFIX: &str = "… [truncated]"; + + if log.message.len() > MAX_TERMINAL_LOG_LEN { + // Reserve room for SUFFIX and back off to a UTF-8 char boundary. + let mut end = MAX_TERMINAL_LOG_LEN - SUFFIX.len(); + while !log.message.is_char_boundary(end) { + end -= 1; + } + log.message.truncate(end); + log.message.push_str(SUFFIX); + } + + // Structured fields are arbitrary connector-supplied JSON and are not + // individually meaningful enough to be worth a truncation scheme: drop + // them wholesale if they'd push the trailer over budget. + let fields_len: usize = log + .fields_json_map + .iter() + .map(|(name, value)| name.len() + value.len()) + .sum(); + + if fields_len > MAX_TERMINAL_LOG_LEN { + log.fields_json_map.clear(); + log.fields_json_map.insert( + "fieldsElided".to_string(), + format!("\"{fields_len} bytes of structured fields were elided\"").into(), + ); + } + + // Spans nest arbitrarily deep and carry no diagnostic value here. + log.spans.clear(); +} + #[cfg(test)] mod test { - use super::{Codec, bidi, new_command, process_logs, unary}; + use super::{Codec, MAX_TERMINAL_LOG_LEN, bidi, bound_log, new_command, process_logs, unary}; use futures::{StreamExt, TryStreamExt}; + use prost::Message; use proto_flow::flow::TestSpec; #[tokio::test] @@ -528,6 +593,113 @@ mod test { } } + #[test] + fn test_bound_log() { + // Multi-byte characters straddle the truncation point, and structured + // fields blow the budget on their own. + let mut log = ops::Log { + level: ops::LogLevel::Error as i32, + message: "π".repeat(MAX_TERMINAL_LOG_LEN), + fields_json_map: [( + "detail".to_string(), + format!("\"{}\"", "x".repeat(MAX_TERMINAL_LOG_LEN)).into(), + )] + .into_iter() + .collect(), + spans: vec![ops::Log::default()], + ..Default::default() + }; + bound_log(&mut log); + + assert!(log.message.len() <= MAX_TERMINAL_LOG_LEN); + assert!(log.message.ends_with("… [truncated]")); + assert_eq!( + log.fields_json_map.keys().collect::>(), + vec!["fieldsElided"] + ); + assert!(log.spans.is_empty()); + + // A Log which already fits is passed through untouched. + let mut log = ops::Log { + message: "all is well".to_string(), + fields_json_map: [("k".to_string(), "1".into())].into_iter().collect(), + ..Default::default() + }; + let expect = log.clone(); + bound_log(&mut log); + assert_eq!(log, expect); + } + + #[test] + fn test_bounded_trailer_fits_one_frame() { + // A gRPC status trailer must fit one 16 KiB HTTP/2 frame, or `h2` + // tears down the whole connection. Model the worst case: every byte of + // the message percent-encodes to three bytes in `grpc-message`, and + // the encoded Log base64s to 4/3 in `last-log-bin`. + let mut log = ops::Log { + level: ops::LogLevel::Error as i32, + // '\n' is percent-encoded by tonic, and is a 1-byte character, so + // this maximizes both the raw length and the expansion factor. + message: "\n".repeat(64 * 1024), + fields_json_map: [("detail".to_string(), "\"boom\"".into())] + .into_iter() + .collect(), + ..Default::default() + }; + bound_log(&mut log); + + let message_len = 3 * log.message.len(); + let metadata_len = 4 * log.encode_to_vec().len().div_ceil(3); + + assert!( + message_len + metadata_len < 16 * 1024, + "encoded trailer is {message_len} + {metadata_len} bytes" + ); + } + + #[tokio::test] + async fn test_bidi_bounds_a_large_terminal_log() { + let requests = futures::stream::repeat_with(|| { + Ok(TestSpec { + name: "hello world".to_string(), + ..Default::default() + }) + }); // Unbounded stream. + + // Model a connector which writes a large diagnostic to stderr as it fails. + let responses: Vec> = bidi( + new_command(&[ + "sh".to_string(), + "-c".to_string(), + "i=0; while [ $i -lt 500 ]; do printf 'a failed walrus appears' >&2; \ + i=$((i+1)); done; exit 1" + .to_string(), + ]), + Codec::Proto, + requests, + ops::stderr_log_handler, + ) + .unwrap() + .collect() + .await; + + let [Err(status)] = responses.as_slice() else { + panic!("expected a single terminal error, got {responses:?}"); + }; + assert!(status.message().starts_with("a failed walrus appears")); + assert!(status.message().ends_with("… [truncated]")); + assert!(status.message().len() <= MAX_TERMINAL_LOG_LEN); + + let metadata = status + .metadata() + .get_bin("last-log-bin") + .expect("last-log-bin is set") + .to_bytes() + .unwrap(); + let log = ops::Log::decode(metadata).unwrap(); + assert_eq!(log.message, status.message()); + } + fn strip_log(mut status: tonic::Status) -> tonic::Status { if let Some(m) = status.metadata_mut().get_bin_mut("last-log-bin") { *m = tonic::metadata::MetadataValue::from_bytes(b"val"); From ea39819921d40601327a0324878992954da16446 Mon Sep 17 00:00:00 2001 From: Jacob Marble Date: Thu, 13 Aug 2026 15:41:13 +0000 Subject: [PATCH 4/4] runtime-next: report the connector's error, not a peer's transport failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A shard which fails takes its leader's session down with it, and the leader broadcasts that failure to every shard of the task. Two things then went wrong with what an operator sees. The leader forwarded the failed peer's `Status` verbatim. A `Status` describes the stream it arrived on, not the streams it's relayed to, so one shard's transport error became every shard's reported cause — and its metadata rode along, so an oversized trailer that had already broken one connection was relayed onto every other. Format the failure into a message instead, which names the peer that failed and carries no metadata. The broadcast also returns to the shard whose connector caused it. The `biased` select prefers the connector arm, but only if the loop polls again; if the leader's echo lands first, a shard reports the echo of its own failure rather than the connector error behind it. Prefer a connector error which is already ready. Part of #3353. --- .../runtime-next/src/leader/derive/handler.rs | 7 +- .../src/leader/materialize/handler.rs | 14 ++-- crates/runtime-next/src/lib.rs | 11 ++- crates/runtime-next/src/shard/derive/actor.rs | 20 +++++- .../src/shard/materialize/actor.rs | 69 ++++++++++++++++++- 5 files changed, 109 insertions(+), 12 deletions(-) diff --git a/crates/runtime-next/src/leader/derive/handler.rs b/crates/runtime-next/src/leader/derive/handler.rs index ea0246b3a76..eb62f997253 100644 --- a/crates/runtime-next/src/leader/derive/handler.rs +++ b/crates/runtime-next/src/leader/derive/handler.rs @@ -245,10 +245,9 @@ where handler.finish_err(&format!("{err:#}")); // Best-effort broadcast of terminal error to all shards. - let status = match err.downcast_ref::() { - Some(status) => crate::bound_status(status.clone()), - None => crate::bounded_unknown_status(format!("{err:?}")), - }; + // See the materialize handler for why the peer's Status isn't forwarded. + let status = crate::bounded_unknown_status(format!("leader session failed: {err:?}")); + for tx in error_tx { let _ = tx.send(Err(status.clone())); } diff --git a/crates/runtime-next/src/leader/materialize/handler.rs b/crates/runtime-next/src/leader/materialize/handler.rs index 85c70476e65..e9b8cd80be3 100644 --- a/crates/runtime-next/src/leader/materialize/handler.rs +++ b/crates/runtime-next/src/leader/materialize/handler.rs @@ -266,10 +266,16 @@ where handler.finish_err(&format!("{err:#}")); // Best-effort broadcast of terminal error to all shards. - let status = match err.downcast_ref::() { - Some(status) => crate::bound_status(status.clone()), - None => crate::bounded_unknown_status(format!("{err:?}")), - }; + // + // A peer's Status is never forwarded verbatim: it describes that peer's + // stream, not the recipient's, so a transport failure of one shard would + // otherwise be reported identically by every shard of the task — masking + // whichever shard actually failed first. Its metadata is dropped for the + // same reason, and because relaying an oversized trailer (such as a + // connector's `last-log-bin`) onward breaks further connections. + // The formatted chain names the failed peer, via `crate::verify`. + let status = crate::bounded_unknown_status(format!("leader session failed: {err:?}")); + for tx in error_tx { let _ = tx.send(Err(status.clone())); } diff --git a/crates/runtime-next/src/lib.rs b/crates/runtime-next/src/lib.rs index 314ea212c55..74a99a73434 100644 --- a/crates/runtime-next/src/lib.rs +++ b/crates/runtime-next/src/lib.rs @@ -151,7 +151,7 @@ impl RuntimeProtocol { // Status bounding lives in `proto-grpc`, which every crate speaking this // protocol already depends on. See `proto_grpc::MAX_STATUS_MESSAGE_LEN` for // why an unbounded status can't survive its trip over the wire. -pub(crate) use proto_grpc::{bound_status, bounded_unknown_status}; +pub(crate) use proto_grpc::bounded_unknown_status; pub use proto_grpc::{MAX_STATUS_MESSAGE_LEN, anyhow_to_status}; // Map a tonic::Status into an anyhow::Error. @@ -164,6 +164,15 @@ pub fn status_to_anyhow(status: tonic::Status) -> anyhow::Error { tonic::Code::Unknown => anyhow::anyhow!(status.message().to_owned()), // For all other Status types, pass through the Status in order to preserve a // capability to lossless-ly downcast back to the Status later. + // + // A locally-produced transport Status renders opaquely — tonic formats + // hyper's Display, which is only `h2 protocol error: http2 error` — but it + // keeps the h2 error as its `source`, which names the reason and h2's debug + // data (`connection error received: ENHANCE_YOUR_CALM (b"too_many_continuations")`). + // anyhow walks that chain, so the detail reaches the log unaided. A Status + // which round-tripped over the wire has *no* source — only its message + // survived — which is why a peer's failure must be formatted into a message + // rather than relayed as a Status. _ => anyhow::Error::new(status), } } diff --git a/crates/runtime-next/src/shard/derive/actor.rs b/crates/runtime-next/src/shard/derive/actor.rs index 3352b2f34c0..4016c6c8670 100644 --- a/crates/runtime-next/src/shard/derive/actor.rs +++ b/crates/runtime-next/src/shard/derive/actor.rs @@ -199,8 +199,9 @@ impl Actor { } // Next, a leader message. msg = leader_rx.next() => { - let (next, stopped) = - self.on_leader_message(phase, &mut accumulator, &mut accumulator_idle, msg)?; + let (next, stopped) = self + .on_leader_message(phase, &mut accumulator, &mut accumulator_idle, msg) + .map_err(|err| prefer_connector_error(connector_rx, err))?; phase = next; if stopped { @@ -574,6 +575,21 @@ impl Actor { } } +/// Replace a leader-stream failure with the connector's own terminal error, +/// if the connector has already failed and its error is immediately ready. +/// See the materialize twin for why this is needed. +fn prefer_connector_error(connector_rx: &mut Conn, err: anyhow::Error) -> anyhow::Error +where + Conn: futures::Stream> + Unpin, +{ + match connector_rx.next().now_or_never() { + Some(Some(Err(status))) => { + crate::verify("Derive", "connector response", "connector").fail_status(status) + } + _ => err, + } +} + async fn maybe_fut(opt: &mut Option>) -> T { match opt.as_mut() { Some(fut) => { diff --git a/crates/runtime-next/src/shard/materialize/actor.rs b/crates/runtime-next/src/shard/materialize/actor.rs index 37b213a8c53..070d0cb98cc 100644 --- a/crates/runtime-next/src/shard/materialize/actor.rs +++ b/crates/runtime-next/src/shard/materialize/actor.rs @@ -236,7 +236,9 @@ impl Actor { } // Next, a leader message. msg = leader_rx.next() => { - let (next, stopped) = self.on_leader_message(phase, msg)?; + let (next, stopped) = self + .on_leader_message(phase, msg) + .map_err(|err| prefer_connector_error(connector_rx, err))?; phase = next; if stopped { @@ -664,6 +666,30 @@ impl Actor { } } +/// Replace a leader-stream failure with the connector's own terminal error, +/// if the connector has already failed and its error is immediately ready. +/// +/// The leader fails a session when any of its shards does, and broadcasts that +/// failure to every shard — including the shard whose connector caused it. Both +/// errors are then ready at once, and while the `biased` select prefers the +/// connector arm, that only helps if the loop polls again: an error surfaced by +/// the leader arm in the meantime would report the leader's echo of this +/// shard's own failure, rather than the connector error which caused it. +/// +/// A ready *response* is discarded rather than handled: the session is failing +/// either way, and the only question is which error describes why. +fn prefer_connector_error(connector_rx: &mut Conn, err: anyhow::Error) -> anyhow::Error +where + Conn: futures::Stream> + Unpin, +{ + match connector_rx.next().now_or_never() { + Some(Some(Err(status))) => { + crate::verify("Materialize", "connector response", "connector").fail_status(status) + } + _ => err, + } +} + async fn maybe_fut(opt: &mut Option>) -> T { match opt.as_mut() { Some(fut) => { @@ -1278,4 +1304,45 @@ mod tests { "a corrupt document UUID fails a truncating binding's transaction" ); } + + #[tokio::test] + async fn connector_error_wins_over_the_leaders_echo() { + let leader_err = || anyhow::anyhow!("leader session failed: some peer shard failed"); + + // A connector error which is already ready replaces the leader's error: + // the leader is echoing this shard's own failure back at it. + let (tx, rx) = mpsc::channel(1); + tx.send(Err(tonic::Status::unknown( + "commit failed: refusing to commit store table", + ))) + .await + .unwrap(); + let mut connector_rx = ReceiverStream::new(rx); + + let err = prefer_connector_error(&mut connector_rx, leader_err()); + assert_eq!( + format!("{err:#}"), + "Materialize error (expected connector response) from connector: \ + commit failed: refusing to commit store table" + ); + + // A healthy connector leaves the leader's error in place, as does a + // connector which has merely reached EOF. + let (_tx, rx) = mpsc::channel::>(1); + let mut connector_rx = ReceiverStream::new(rx); + let err = prefer_connector_error(&mut connector_rx, leader_err()); + assert_eq!( + format!("{err:#}"), + "leader session failed: some peer shard failed" + ); + + let (tx, rx) = mpsc::channel::>(1); + drop(tx); + let mut connector_rx = ReceiverStream::new(rx); + let err = prefer_connector_error(&mut connector_rx, leader_err()); + assert_eq!( + format!("{err:#}"), + "leader session failed: some peer shard failed" + ); + } }