ensemble: bound every send, order the opening frame, prune dead routes

.github/workflows/wire.yml

@@ -26,6 +26,24 @@ jobs:
       - name: Golden fixture round-trips through views
         run: cargo test -p thrum-core
 
+  mesh:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - uses: Swatinem/rust-cache@v2
+      - name: Clippy on the crates this layer owns
+        # type_complexity — a tuple-of-tuples in thrum-core's build script.
+        # double_must_use — async_trait stamps #[must_use] onto a
+        #   Pin<Box<dyn Future>>; the lint fires in macro expansion, not our source.
+        # Both lints landed after this branch's local toolchain (1.98.1), so they
+        # are allowed rather than fixed blind.
+        run: cargo clippy -p ensemble -p hum-paths --all-targets -- -D warnings -A clippy::type_complexity -A clippy::double_must_use
+      - name: Ensemble, daemon and mesh simulation
+        run: cargo test --locked -p ensemble -p sim -p humd -p hum -p humctl
+
   go:
     runs-on: ubuntu-latest
     steps:

codegen/src/protocol.rs

@@ -120,15 +120,12 @@ fn simplify(ty: &Type) -> (bool, Repr) {
     if let Type::Path(p) = ty {
         // Check the leading segment name directly: syn's `is_ident` rejects
         // paths with generic arguments, so `Option<String>` would slip by.
-        if let Some(seg) = p.path.segments.first() {
-            if seg.ident == "Option" {
-                if let syn::PathArguments::AngleBracketed(a) = &seg.arguments {
-                    if let Some(syn::GenericArgument::Type(inner)) = a.args.first() {
+        if let Some(seg) = p.path.segments.first()
+            && seg.ident == "Option"
+                && let syn::PathArguments::AngleBracketed(a) = &seg.arguments
+                    && let Some(syn::GenericArgument::Type(inner)) = a.args.first() {
                         return (true, simplify(inner).1);
                     }
-                }
-            }
-        }
     }
     (false, repr_of(ty))
 }
@@ -145,12 +142,12 @@ fn repr_of(ty: &Type) -> Repr {
         "bool" => Repr::Bool,
         "Value" => Repr::Json,
         "Vec" => {
-            let inner = angle_ty(&seg.arguments, 0).unwrap_or_else(|| Repr::Json);
+            let inner = angle_ty(&seg.arguments, 0).unwrap_or(Repr::Json);
             Repr::Arr(Box::new(inner))
         }
         "BTreeMap" => {
             // The value type is the SECOND generic arg (first is the key).
-            let inner = angle_ty(&seg.arguments, 1).unwrap_or_else(|| Repr::Json);
+            let inner = angle_ty(&seg.arguments, 1).unwrap_or(Repr::Json);
             Repr::Map(Box::new(inner))
         }
         other => {
@@ -169,11 +166,10 @@ fn repr_of(ty: &Type) -> Repr {
 }
 
 fn angle_ty(args: &syn::PathArguments, nth: usize) -> Option<Repr> {
-    if let syn::PathArguments::AngleBracketed(a) = args {
-        if let Some(syn::GenericArgument::Type(t)) = a.args.iter().nth(nth) {
+    if let syn::PathArguments::AngleBracketed(a) = args
+        && let Some(syn::GenericArgument::Type(t)) = a.args.iter().nth(nth) {
             return Some(simplify(t).1);
         }
-    }
     None
 }
 
@@ -576,8 +572,8 @@ mod tests {
         let f = &ts[0].fields;
         assert_eq!(f.len(), 4);
         assert_eq!(f[0].json, "chunkType");
-        assert_eq!(f[0].repr.matches(&Repr::Str), true);
-        assert_eq!(f[1].optional, true);
+        assert!(f[0].repr.matches(&Repr::Str));
+        assert!(f[1].optional);
         assert!(matches!(f[1].repr, Repr::Json));
         assert!(matches!(f[2].repr, Repr::Arr(_)));
         assert!(matches!(f[3].repr, Repr::Map(_)));

ensemble/src/framing.rs

@@ -0,0 +1,203 @@
+use anyhow::{Result, anyhow};
+use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
+use tokio::sync::mpsc;
+
+use crate::Tone;
+
+pub const MAX_FRAME_BYTES: usize = 256 * 1024;
+
+pub async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, tone: &Tone) -> Result<()> {
+    let mut line = serde_json::to_vec(tone).map_err(|e| anyhow!("encode: {e}"))?;
+    line.push(b'\n');
+    writer
+        .write_all(&line)
+        .await
+        .map_err(|e| anyhow!("write: {e}"))
+}
+
+pub async fn pump<R>(reader: R, tx: mpsc::Sender<Tone>, transport: &'static str)
+where
+    R: AsyncRead + Unpin,
+{
+    let mut reader = BufReader::new(reader);
+    let mut line: Vec<u8> = Vec::new();
+    let mut oversize = false;
+    let mut discarded = 0u64;
+
+    loop {
+        let chunk = match reader.fill_buf().await {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::trace!(target: "ensemble.frame", transport, error = %e, "read.failed");
+                break;
+            }
+        };
+        if chunk.is_empty() {
+            if !line.is_empty() && !oversize {
+                forward(&line, &tx, transport).await;
+            }
+            break;
+        }
+        let newline = chunk.iter().position(|b| *b == b'\n');
+        let take = newline.unwrap_or(chunk.len());
+        let consumed = take + usize::from(newline.is_some());
+
+        if line.len() + take > MAX_FRAME_BYTES {
+            if !oversize {
+                oversize = true;
+                tracing::warn!(
+                    target: "ensemble.frame",
+                    transport,
+                    cap = MAX_FRAME_BYTES,
+                    "frame.oversize",
+                );
+            }
+        } else {
+            line.extend_from_slice(&chunk[..take]);
+        }
+        reader.consume(consumed);
+
+        if newline.is_none() {
+            continue;
+        }
+        if oversize {
+            oversize = false;
+            discarded += 1;
+            line.clear();
+            continue;
+        }
+        if !forward(&line, &tx, transport).await {
+            break;
+        }
+        line.clear();
+    }
+
+    if discarded > 0 {
+        tracing::warn!(target: "ensemble.frame", transport, discarded, "frame.oversize.total");
+    }
+}
+
+async fn forward(line: &[u8], tx: &mpsc::Sender<Tone>, transport: &'static str) -> bool {
+    if line.iter().all(|b| b.is_ascii_whitespace()) {
+        return true;
+    }
+    match serde_json::from_slice::<Tone>(line) {
+        Ok(tone) => tx.send(tone).await.is_ok(),
+        Err(e) => {
+            tracing::trace!(target: "ensemble.frame", transport, error = %e, "frame.rejected");
+            true
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    async fn frames(input: &[u8]) -> Vec<Tone> {
+        let (tx, mut rx) = mpsc::channel(64);
+        let (mut w, r) = tokio::io::duplex(input.len() + 64);
+        w.write_all(input).await.unwrap();
+        w.shutdown().await.unwrap();
+        pump(r, tx, "test").await;
+        let mut out = Vec::new();
+        while let Ok(t) = rx.try_recv() {
+            out.push(t);
+        }
+        out
+    }
+
+    #[tokio::test]
+    async fn a_frame_survives_arriving_in_pieces() {
+        let got = frames(b"{\"chi\":\"a\"}\n{\"chi\":\"b\"}\n").await;
+        assert_eq!(got.len(), 2);
+        assert_eq!(got[0]["chi"], "a");
+        assert_eq!(got[1]["chi"], "b");
+    }
+
+    #[tokio::test]
+    async fn one_byte_at_a_time_reassembles() {
+        let raw = b"{\"chi\":\"drip\"}\n";
+        let (tx, mut rx) = mpsc::channel(4);
+        let (mut w, r) = tokio::io::duplex(64);
+        let feeder = tokio::spawn(async move {
+            for b in raw {
+                w.write_all(&[*b]).await.unwrap();
+                tokio::task::yield_now().await;
+            }
+            w.shutdown().await.unwrap();
+        });
+        pump(r, tx, "test").await;
+        feeder.await.unwrap();
+        let tone = rx.try_recv().unwrap();
+        assert_eq!(tone["chi"], "drip");
+    }
+
+    #[tokio::test]
+    async fn a_trailing_frame_without_a_newline_still_arrives() {
+        let got = frames(b"{\"chi\":\"a\"}\n{\"chi\":\"tail\"}").await;
+        assert_eq!(got.len(), 2);
+        assert_eq!(got[1]["chi"], "tail");
+    }
+
+    #[tokio::test]
+    async fn garbage_is_skipped_and_the_next_frame_still_arrives() {
+        let got = frames(b"not json\n{\"chi\":\"ok\"}\n").await;
+        assert_eq!(got.len(), 1);
+        assert_eq!(got[0]["chi"], "ok");
+    }
+
+    #[tokio::test]
+    async fn a_blank_line_is_a_keepalive_not_a_tone() {
+        let got = frames(b"\n\n{\"chi\":\"ok\"}\n").await;
+        assert_eq!(got.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn an_oversize_frame_is_discarded_and_the_link_resyncs() {
+        let mut input = vec![b'x'; MAX_FRAME_BYTES + 4096];
+        input.push(b'\n');
+        input.extend_from_slice(b"{\"chi\":\"after\"}\n");
+        let got = frames(&input).await;
+        assert_eq!(got.len(), 1, "the oversize frame must not be parsed");
+        assert_eq!(got[0]["chi"], "after");
+    }
+
+    #[tokio::test]
+    async fn a_frame_at_exactly_the_cap_is_accepted() {
+        let mut input = br#"{"chi":""#.to_vec();
+        input.extend(std::iter::repeat_n(b'p', MAX_FRAME_BYTES - input.len() - 2));
+        input.extend_from_slice(br#""}"#);
+        assert_eq!(input.len(), MAX_FRAME_BYTES);
+        input.push(b'\n');
+        let got = frames(&input).await;
+        assert_eq!(got.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn pump_stops_when_the_receiver_is_gone() {
+        let (tx, rx) = mpsc::channel(1);
+        drop(rx);
+        let (mut w, r) = tokio::io::duplex(64);
+        w.write_all(b"{\"chi\":\"a\"}\n{\"chi\":\"b\"}\n")
+            .await
+            .unwrap();
+        pump(r, tx, "test").await;
+    }
+
+    #[tokio::test]
+    async fn write_frame_emits_exactly_one_newline_terminated_line() {
+        let (mut w, r) = tokio::io::duplex(256);
+        let mut r = BufReader::new(r);
+        write_frame(&mut w, &json!({"chi": "x", "n": 1}))
+            .await
+            .unwrap();
+        drop(w);
+        let mut first = String::new();
+        r.read_line(&mut first).await.unwrap();
+        assert_eq!(first, "{\"chi\":\"x\",\"n\":1}\n");
+        let mut rest = String::new();
+        assert_eq!(r.read_line(&mut rest).await.unwrap(), 0);
+    }
+}