Ensure that we don't send massive error message over gRPC

The gRPC servers are configured with a max size of 4MB,
this patch adds more resiliency for potentially large error
message that would bust the gRPC message size limit.

Author: QnJ1c2kNCg

crates/arroyo-worker/src/lib.rs

@@ -44,7 +44,7 @@ use tonic::{Request, Response, Status};
 use tracing::{debug, error, info, warn};
 
 use crate::job_controller::controller::WorkerJobController;
-use crate::utils::to_d2;
+use crate::utils::{MAX_TASK_ERROR_FIELD_BYTES, maybe_truncate, to_d2};
 use arroyo_datastream::logical::LogicalProgram;
 use arroyo_planner::physical::new_registry;
 use arroyo_rpc::config::config;
@@ -576,11 +576,14 @@ impl WorkerState {
                                 error: Some(rpc::TaskError {
                                     task_id,
                                     subtask_idx,
-                                    error: error.message,
+                                    error: maybe_truncate(error.message, MAX_TASK_ERROR_FIELD_BYTES),
                                     error_domain: rpc::ErrorDomain::from(error.domain) as i32,
                                     retry_hint: rpc::RetryHint::from(error.retry_hint) as i32,
                                     operator_id: error.operator_id.unwrap_or_default(),
-                                    details: error.details.unwrap_or_default(),
+                                    details: maybe_truncate(
+                                        error.details.unwrap_or_default(),
+                                        MAX_TASK_ERROR_FIELD_BYTES
+                                    ),
                                 }),
                             };
                             send_control_rpc!(
@@ -599,10 +602,10 @@ impl WorkerState {
                                     task_id,
                                     operator_id,
                                     subtask_idx,
-                                    error: message,
+                                    error: maybe_truncate(message, MAX_TASK_ERROR_FIELD_BYTES),
                                     error_domain: rpc::ErrorDomain::External as i32,
                                     retry_hint: rpc::RetryHint::NoRetry as i32,
-                                    details,
+                                    details: maybe_truncate(details, MAX_TASK_ERROR_FIELD_BYTES),
                                 }),
                             };
                             send_control_rpc!(

crates/arroyo-worker/src/utils.rs

@@ -5,6 +5,7 @@ use arroyo_operator::operator::Registry;
 use arroyo_planner::physical::new_registry;
 use std::fmt::Write;
 use std::sync::Arc;
+use tracing::warn;
 
 fn format_arrow_schema_fields(schema: &Schema) -> Vec<(String, String)> {
     schema
@@ -135,3 +136,55 @@ pub async fn to_d2(logical: &LogicalProgram) -> anyhow::Result<String> {
 
     Ok(d2)
 }
+
+pub(crate) const MAX_TASK_ERROR_FIELD_BYTES: usize = 64 * 1024;
+
+pub(crate) fn maybe_truncate(mut value: String, max_size_bytes: usize) -> String {
+    let original_bytes = value.len();
+    if original_bytes <= max_size_bytes {
+        return value;
+    }
+
+    let suffix = format!(" [truncated; original_bytes={original_bytes}]");
+    let mut end = max_size_bytes - suffix.len();
+    while !value.is_char_boundary(end) {
+        end -= 1;
+    }
+
+    value.truncate(end);
+    value.push_str(&suffix);
+
+    warn!(
+        "Truncated oversized String from {} bytes to {} bytes: {}",
+        original_bytes,
+        value.len(),
+        value
+    );
+    value
+}
+
+#[cfg(test)]
+mod tests {
+    use super::maybe_truncate;
+
+    #[test]
+    fn maybe_truncate_preserves_value_at_limit() {
+        let value = "a".repeat(64);
+
+        assert_eq!(maybe_truncate(value.clone(), value.len()), value);
+    }
+
+    #[test]
+    fn maybe_truncate_respects_limit() {
+        let value = "a".repeat(100);
+        let max_size_bytes = 64;
+
+        let truncated = maybe_truncate(value, max_size_bytes);
+
+        assert_eq!(truncated.len(), max_size_bytes);
+        assert_eq!(
+            truncated,
+            format!("{} [truncated; original_bytes=100]", "a".repeat(32))
+        );
+    }
+}