fix(relay): apply ingest's content checks to workflow SendMessage

crates/buzz-relay/src/handlers/ingest.rs

@@ -2523,27 +2523,62 @@ pub async fn ingest_event(
 ///   client ceiling (60 s, CANVAS_MAX_FUTURE_SKEW_SECS) < canvas relay bound (300 s) < relay general bound (900 s)
 const CANVAS_MAX_INGEST_FUTURE_SECS: i64 = 300;
 
+/// Largest event `content`, in bytes, the relay will store.
+pub(crate) const MAX_EVENT_CONTENT_BYTES: usize = 256 * 1024; // 256 KB
+
+/// Why event content cannot be stored. `Display` omits the `invalid:` prefix
+/// so each write path can map it into its own error type.
+#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+pub(crate) enum StoredContentError {
+    /// Content is longer than [`MAX_EVENT_CONTENT_BYTES`].
+    #[error("content exceeds maximum size of {MAX_EVENT_CONTENT_BYTES} bytes (got {0})")]
+    TooLarge(usize),
+    /// Content contains `\0`, which Postgres `text` cannot store.
+    #[error("content contains a NUL byte")]
+    NulInContent,
+    /// A tag value contains `\0`, which Postgres `text` cannot store.
+    #[error("tag contains a NUL byte")]
+    NulInTag,
+}
+
+/// Refuse content longer than [`MAX_EVENT_CONTENT_BYTES`]. Pure and cheap, so
+/// callers run it before any database read or signature check.
+pub(crate) fn check_content_size(content: &str) -> Result<(), StoredContentError> {
+    if content.len() > MAX_EVENT_CONTENT_BYTES {
+        return Err(StoredContentError::TooLarge(content.len()));
+    }
+    Ok(())
+}
+
+/// Refuse content containing a NUL byte. Postgres `text` cannot store `\0`,
+/// so such content would otherwise fail at insert with an encoding error.
+pub(crate) fn check_content_nul(content: &str) -> Result<(), StoredContentError> {
+    if content.contains('\0') {
+        return Err(StoredContentError::NulInContent);
+    }
+    Ok(())
+}
+
 /// Reject events whose content or any tag value contains a NUL byte.
 /// Postgres `text` cannot store `\0`, so such an event would otherwise fail
 /// at insert with an encoding error and surface as an internal error.
 fn reject_nul_bytes(event: &Event) -> Result<(), IngestError> {
-    if event.content.contains('\0') {
-        return Err(IngestError::Rejected(
-            "invalid: content contains a NUL byte".into(),
-        ));
-    }
+    check_content_nul(&event.content).map_err(stored_content_rejection)?;
     if event
         .tags
         .iter()
         .any(|tag| tag.as_slice().iter().any(|value| value.contains('\0')))
     {
-        return Err(IngestError::Rejected(
-            "invalid: tag contains a NUL byte".into(),
-        ));
+        return Err(stored_content_rejection(StoredContentError::NulInTag));
     }
     Ok(())
 }
 
+/// Map a stored-content refusal to the client-facing `invalid:` rejection.
+fn stored_content_rejection(error: StoredContentError) -> IngestError {
+    IngestError::Rejected(format!("invalid: {error}"))
+}
+
 /// Returns `Ok(())` if the canvas event timestamp is within the allowed future
 /// window, or `Err` with a rejection message otherwise.
 ///
@@ -2605,14 +2640,7 @@ async fn ingest_event_inner(
     // The size check is a pure, cheap reject: it needs neither the DB nor
     // signature verification, so oversized events never cost a writer-pool
     // read or a blocking-pool Schnorr check.
-    const MAX_EVENT_CONTENT_BYTES: usize = 256 * 1024; // 256 KB
-    if event.content.len() > MAX_EVENT_CONTENT_BYTES {
-        return Err(IngestError::Rejected(format!(
-            "invalid: content exceeds maximum size of {} bytes (got {})",
-            MAX_EVENT_CONTENT_BYTES,
-            event.content.len()
-        )));
-    }
+    check_content_size(&event.content).map_err(stored_content_rejection)?;
 
     // Durable community write fence: persistent ingest is a DB write the
     // deletion engine cannot exclude via serving-write leases (those cover

crates/buzz-relay/src/workflow_sink.rs

@@ -222,6 +222,16 @@ impl ActionSink for RelayActionSink {
                 .upgrade()
                 .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?;
 
+            // The rendered text is the only caller-supplied value stored
+            // unparsed (it can carry webhook output), so it gets ingest's
+            // stored-content checks before any database work; every tag below
+            // is built from parsed or stored values. Without them an oversized
+            // body is stored past ingest's limit and a NUL byte fails the
+            // insert as a database error.
+            crate::handlers::ingest::check_content_size(&text)
+                .and_then(|()| crate::handlers::ingest::check_content_nul(&text))
+                .map_err(|e| ActionSinkError::InvalidInput(e.to_string()))?;
+
             // The run carries its owning community (`community_id`); the
             // relay-signed kind:9 message belongs to *that* community, never the
             // deployment default. Re-deriving the tenant from `config.relay_url`
@@ -773,6 +783,59 @@ mod tests {
         assert_eq!(tags.len(), 1);
         assert_eq!(tags[0].as_slice(), ["p", owner.as_str(), "", "owner"]);
     }
+
+    /// Rendered workflow text can carry webhook output (up to 1 MiB, NUL
+    /// bytes kept by lossy UTF-8 decoding). The sink must refuse what ingest
+    /// refuses — oversized or NUL-bearing content — as invalid input before
+    /// it reads or writes the database. The database here is unreachable, so
+    /// any text that gets past the checks fails as `Database` instead.
+    /// Mutation: drop either check in `send_message` and its case fails.
+    #[tokio::test]
+    async fn send_message_refuses_unstorable_text_before_touching_the_database() {
+        let state = crate::state::tests::test_state_with_unreachable_database_url(
+            "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1
+        )
+        .await;
+        let sink = RelayActionSink::new(&state);
+        let community = CommunityId::from_uuid(Uuid::new_v4());
+        let channel = Uuid::new_v4().to_string();
+        let owner = pk('1');
+        let max = crate::handlers::ingest::MAX_EVENT_CONTENT_BYTES;
+
+        let send = |text: String| {
+            let sink = &sink;
+            let channel = channel.clone();
+            let owner = owner.clone();
+            async move {
+                sink.send_message(community, &channel, &text, "{{text}}", &owner, None)
+                    .await
+            }
+        };
+
+        for (text, expected) in [
+            (
+                "a\0b".to_string(),
+                "content contains a NUL byte".to_string(),
+            ),
+            (
+                "x".repeat(max + 1),
+                format!("content exceeds maximum size of {max} bytes"),
+            ),
+        ] {
+            match send(text).await {
+                Err(ActionSinkError::InvalidInput(msg)) => {
+                    assert!(msg.contains(&expected), "{expected}: {msg}");
+                }
+                other => panic!("{expected}: must be refused as invalid input, got {other:?}"),
+            }
+        }
+
+        // Text at the limit passes the checks and reaches the database.
+        match send("x".repeat(max)).await {
+            Err(ActionSinkError::Database(_)) => {}
+            other => panic!("text at the limit must reach the database, got {other:?}"),
+        }
+    }
 }
 
 #[cfg(test)]