runtime-next: bound status trailers so connector errors aren't masked by ResourceExhausted

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<E: Into<anyhow::Error>>(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<_>>(),
+            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<Result<TestSpec, _>> = 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");

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 }

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 {

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::<tonic::Status>() {
+        Ok(status) => bound_status(status),
+        Err(err) => bounded_unknown_status(format!("{err:?}")),
+    }
+}

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::<tonic::Status>() {
-        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()));
     }

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::<tonic::Status>() {
-        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()));
     }