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/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"); 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/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 68a1ab2e826..74a99a73434 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::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. @@ -225,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" + ); + } } 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]