From 306b8998f8b477accba2369b47d8badb2f5f2dd8 Mon Sep 17 00:00:00 2001 From: Adil Date: Sun, 4 Oct 2026 15:32:25 +0500 Subject: [PATCH 1/3] ensemble: bound every send, order the opening frame, prune dead routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Harden the mesh against the failure modes PR #61 left open. Sends: every write now has a deadline. Liveness pongs, Kademlia queries and `kad_find` were unbounded; the opening hello was awaited raw, so a peer that never read left the task hung forever and the gate shut behind it. Fanout for publish, gossip and probes was serial, so wall time was the sum of the peers rather than the slowest one. Opening: `install` spawned the hello and raced application traffic onto the same link — a routed tone could reach the wire before the frame that authenticates the peer. A `Gate` now parks every send until the opening frame lands. The gate registers with `Notify` before reading its flag (a check-then-park ordering drops wakeups), `Default` matches `new`, and a shut gate stays shut. Identity: no tone is published before its link completes the signed hello; drops are counted in `tones_before_handshake()`. `strict_auth` defaults to true, so an unsigned peer is refused unless a caller explicitly waives it. `from` is deliberately left unvalidated — it names the originator, not the forwarding link, and this mesh relays addressed tones. Routing: a removed peer is forgotten from every bucket instead of staying addressable, and the table is persisted to `state_dir()/routing.json` so a restart resumes it. Framing: TCP, TLS and iroh share one NDJSON codec with a 256KiB frame ceiling. An oversize frame is discarded and the stream resyncs on the next newline rather than growing without bound. Tests: the opening-order test is pinned to the bug — with the gate removed it fails with `prompt` ahead of `hello`. CI gains a `mesh` job running clippy and the ensemble/sim/humd suites. --- .github/workflows/wire.yml | 13 + codegen/src/protocol.rs | 24 +- ensemble/src/framing.rs | 203 +++++++ ensemble/src/handshake.rs | 92 ++- ensemble/src/headroom.rs | 21 +- ensemble/src/hives.rs | 9 +- ensemble/src/iroh.rs | 123 ++-- ensemble/src/kad.rs | 93 ++- ensemble/src/lib.rs | 807 +++++++++++++++++---------- ensemble/src/link.rs | 94 +++- ensemble/src/liveness.rs | 1 - ensemble/src/opening.rs | 145 +++++ ensemble/src/send.rs | 207 ++++++- ensemble/src/tcp.rs | 171 ++---- ensemble/src/tls.rs | 165 ++---- ensemble/tests/gossip_integration.rs | 21 +- ensemble/tests/iroh_integration.rs | 37 +- ensemble/tests/tcp_integration.rs | 6 +- ensemble/tests/tls_integration.rs | 5 +- hum-paths/src/config.rs | 9 +- hum-paths/src/lib.rs | 10 +- humd/src/lib.rs | 10 + humd/src/routing_store.rs | 139 +++++ sim/src/lib.rs | 8 +- 24 files changed, 1660 insertions(+), 753 deletions(-) create mode 100644 ensemble/src/framing.rs create mode 100644 ensemble/src/opening.rs create mode 100644 humd/src/routing_store.rs diff --git a/.github/workflows/wire.yml b/.github/workflows/wire.yml index 32d29388..a862d4cd 100644 --- a/.github/workflows/wire.yml +++ b/.github/workflows/wire.yml @@ -26,6 +26,19 @@ 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 + run: cargo clippy -p ensemble -p hum-paths --all-targets -- -D warnings -A clippy::type_complexity + - 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: diff --git a/codegen/src/protocol.rs b/codegen/src/protocol.rs index a9137a64..f63b51af 100644 --- a/codegen/src/protocol.rs +++ b/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` 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 { - 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(_))); diff --git a/ensemble/src/framing.rs b/ensemble/src/framing.rs new file mode 100644 index 00000000..c738eeb4 --- /dev/null +++ b/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(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(reader: R, tx: mpsc::Sender, transport: &'static str) +where + R: AsyncRead + Unpin, +{ + let mut reader = BufReader::new(reader); + let mut line: Vec = 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, transport: &'static str) -> bool { + if line.iter().all(|b| b.is_ascii_whitespace()) { + return true; + } + match serde_json::from_slice::(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 { + 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); + } +} diff --git a/ensemble/src/handshake.rs b/ensemble/src/handshake.rs index def1d887..f6fe3c82 100644 --- a/ensemble/src/handshake.rs +++ b/ensemble/src/handshake.rs @@ -7,7 +7,7 @@ use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey}; use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; -use super::{headroom, now_ms, Tone, HANDSHAKE_DOMAIN, HANDSHAKE_SKEW_MS}; +use super::{HANDSHAKE_DOMAIN, HANDSHAKE_SKEW_MS, Tone, headroom, now_ms}; pub use hum_identity::{Hid, HidParseError, HidPrefix}; @@ -47,7 +47,12 @@ pub struct HumdAddr { } impl HumdAddr { - pub fn new(id: Hid) -> Self { Self { id, hints: Vec::new() } } + pub fn new(id: Hid) -> Self { + Self { + id, + hints: Vec::new(), + } + } pub fn with_hint(mut self, h: impl Into) -> Self { self.hints.push(h.into()); self @@ -123,6 +128,11 @@ pub trait PeerConnection: Send + Sync { async fn send(&self, tone: Tone) -> Result<()>; fn take_receiver(&self) -> Option>; fn close(&self); + + fn arm_opening(&self) {} + async fn send_opening(&self, tone: Tone) -> Result<()> { + self.send(tone).await + } } #[async_trait] @@ -196,30 +206,46 @@ pub fn parse_hello_caps(tone: &Tone) -> Option<(Hid, PeerCapabilities)> { let nests = tone .get("nests") .and_then(|v| v.as_array()) - .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .map(|a| { + a.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) .unwrap_or_default(); let hosts = tone .get("hosts") .and_then(|v| v.as_array()) - .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .map(|a| { + a.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) .unwrap_or_default(); - let can_relay = tone.get("can_relay").and_then(|v| v.as_bool()).unwrap_or(false); - let free_slots = tone - .get("free_slots") - .and_then(|v| { - if v.is_null() { - None - } else { - v.as_u64().map(|n| n as usize) - } - }); + let can_relay = tone + .get("can_relay") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let free_slots = tone.get("free_slots").and_then(|v| { + if v.is_null() { + None + } else { + v.as_u64().map(|n| n as usize) + } + }); let headroom = tone .get("headroom") .and_then(|v| serde_json::from_value::(v.clone()).ok()) .unwrap_or_default(); Some(( claimed_id, - PeerCapabilities { proto_version, nests, hosts, can_relay, free_slots, headroom }, + PeerCapabilities { + proto_version, + nests, + hosts, + can_relay, + free_slots, + headroom, + }, )) } @@ -247,24 +273,46 @@ pub fn parse_hello(tone: &Tone) -> HelloParse { let nests = tone .get("nests") .and_then(|v| v.as_array()) - .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .map(|a| { + a.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) .unwrap_or_default(); let hosts = tone .get("hosts") .and_then(|v| v.as_array()) - .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .map(|a| { + a.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) .unwrap_or_default(); - let can_relay = tone.get("can_relay").and_then(|v| v.as_bool()).unwrap_or(false); - let free_slots = tone - .get("free_slots") - .and_then(|v| if v.is_null() { None } else { v.as_u64().map(|n| n as usize) }); + let can_relay = tone + .get("can_relay") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let free_slots = tone.get("free_slots").and_then(|v| { + if v.is_null() { + None + } else { + v.as_u64().map(|n| n as usize) + } + }); let headroom = tone .get("headroom") .and_then(|v| serde_json::from_value::(v.clone()).ok()) .unwrap_or_default(); HelloParse::Unsigned( claimed_id, - PeerCapabilities { proto_version, nests, hosts, can_relay, free_slots, headroom }, + PeerCapabilities { + proto_version, + nests, + hosts, + can_relay, + free_slots, + headroom, + }, ) } } diff --git a/ensemble/src/headroom.rs b/ensemble/src/headroom.rs index 758de25e..aa040f67 100644 --- a/ensemble/src/headroom.rs +++ b/ensemble/src/headroom.rs @@ -15,20 +15,15 @@ use serde::{Deserialize, Serialize}; /// Refuse — > 95%, do not route new prompts here (humd will reject) #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] +#[derive(Default)] pub enum Pressure { + #[default] Cool, Warm, Hot, Refuse, } -impl Default for Pressure { - fn default() -> Self { - // No-op advertise — an empty humd is at rest, not refusing. - Pressure::Cool - } -} - impl Pressure { /// Compute from a slot-occupancy ratio (0.0 .. 1.0). Out-of-range /// values clamp. @@ -78,11 +73,7 @@ impl CellHeadroom { /// When `total_slots == 0` (no nest), pressure is `Cool` and /// `free_slots` is forced to 0 — the humd is honest about having /// nothing to offer rather than misrepresenting itself as full. - pub fn from_counts( - free_slots: u32, - total_slots: u32, - p95_latency_ms: Option, - ) -> Self { + pub fn from_counts(free_slots: u32, total_slots: u32, p95_latency_ms: Option) -> Self { if total_slots == 0 { return Self { free_slots: 0, @@ -216,6 +207,10 @@ mod tests { // skip_serializing_if for None p95. let no_latency = CellHeadroom::from_counts(5, 10, None); let s = serde_json::to_string(&no_latency).unwrap(); - assert!(!s.contains("p95_latency_ms"), "None should be skipped: {}", s); + assert!( + !s.contains("p95_latency_ms"), + "None should be skipped: {}", + s + ); } } diff --git a/ensemble/src/hives.rs b/ensemble/src/hives.rs index 8a36a2eb..e3d8ece4 100644 --- a/ensemble/src/hives.rs +++ b/ensemble/src/hives.rs @@ -213,7 +213,7 @@ pub enum HiveAnnounce { /// Bee is live on `humd_id`. Advertise { humd_id: String, - manifest: HiveManifest, + manifest: Box, }, /// Bee has shut down on `humd_id`. Retract { humd_id: String, name: String }, @@ -248,14 +248,17 @@ mod tests { let back: HiveManifest = serde_json::from_value(j).unwrap(); assert_eq!(back.chis.len(), 4); assert_eq!(back.propensity.statefulness.as_deref(), Some("stateless")); - assert_eq!(back.source.as_deref(), Some("https://github.com/example/mm-bee")); + assert_eq!( + back.source.as_deref(), + Some("https://github.com/example/mm-bee") + ); } #[test] fn announce_envelope_tags_kind() { let env = HiveAnnounce::Advertise { humd_id: "deadbeef".into(), - manifest: HiveManifest::new("mm", "0.1.0", "0.7.0"), + manifest: Box::new(HiveManifest::new("mm", "0.1.0", "0.7.0")), }; let s = serde_json::to_string(&env).unwrap(); assert!(s.contains("\"kind\":\"advertise\"")); diff --git a/ensemble/src/iroh.rs b/ensemble/src/iroh.rs index d79c855f..62d79880 100644 --- a/ensemble/src/iroh.rs +++ b/ensemble/src/iroh.rs @@ -1,16 +1,17 @@ - use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; -use anyhow::{anyhow, Context, Result}; +use anyhow::{Context, Result, anyhow}; use async_trait::async_trait; use sha2::{Digest, Sha256}; -use tokio::io::{AsyncBufReadExt, BufReader}; -use tokio::sync::{mpsc, Mutex}; +use tokio::sync::{Mutex, mpsc}; use iroh::endpoint::{Connection, RecvStream, SendStream}; use iroh::{Endpoint, EndpointAddr, EndpointId, PublicKey, TransportAddr}; -use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport}; +use crate::framing; +use crate::opening::Gate; +use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport}; pub const IROH_ALPN: &[u8] = b"hum/0.5"; @@ -46,8 +47,10 @@ pub struct IrohEndpoint { node_id: EndpointId, caps: PeerCapabilities, _connection: Connection, - sender: Mutex>, + sender: Arc>>, rx: parking_lot::Mutex>>, + closed: AtomicBool, + gate: Gate, } impl IrohEndpoint { @@ -59,17 +62,20 @@ impl IrohEndpoint { caps: PeerCapabilities, ) -> Arc { let humd_id = humd_id_from_node_id(&node_id); - let peer = HumdAddr::new(humd_id).with_hint(format!("{IROH_HINT}{}", hex::encode(node_id.as_bytes()))); + let peer = HumdAddr::new(humd_id) + .with_hint(format!("{IROH_HINT}{}", hex::encode(node_id.as_bytes()))); let (tx, rx) = mpsc::channel::(RECV_CAP); let me = Arc::new(Self { peer, node_id, caps, _connection: conn, - sender: Mutex::new(Some(send)), + sender: Arc::new(Mutex::new(Some(send))), rx: parking_lot::Mutex::new(Some(rx)), + closed: AtomicBool::new(false), + gate: Gate::new(), }); - tokio::spawn(read_loop(recv, tx)); + tokio::spawn(framing::pump(recv, tx, "iroh")); me } @@ -98,6 +104,19 @@ impl IrohEndpoint { } } +impl IrohEndpoint { + async fn write(&self, tone: Tone) -> Result<()> { + let mut guard = self.sender.lock().await; + if self.closed.load(Ordering::SeqCst) { + return Err(anyhow!("iroh send: link closed")); + } + let s = guard + .as_mut() + .ok_or_else(|| anyhow!("iroh send: stream closed"))?; + framing::write_frame(s, &tone).await + } +} + #[async_trait] impl PeerConnection for IrohEndpoint { fn peer(&self) -> &HumdAddr { @@ -108,61 +127,34 @@ impl PeerConnection for IrohEndpoint { } async fn send(&self, tone: Tone) -> Result<()> { - let mut line = serde_json::to_vec(&tone) - .map_err(|e| anyhow!("iroh send: serialize: {e}"))?; - line.push(b'\n'); - let mut guard = self.sender.lock().await; - let s = guard - .as_mut() - .ok_or_else(|| anyhow!("iroh send: stream closed"))?; - s.write_all(&line) - .await - .map_err(|e| anyhow!("iroh send: write: {e}"))?; - Ok(()) + self.gate.wait().await; + self.write(tone).await + } + + async fn send_opening(&self, tone: Tone) -> Result<()> { + let result = self.write(tone).await; + self.gate.opened(); + result } fn take_receiver(&self) -> Option> { self.rx.lock().take() } - fn close(&self) { - let slot = match self.sender.try_lock() { - Ok(mut g) => g.take(), - Err(_) => None, - }; - if let Some(mut s) = slot { - tokio::spawn(async move { - let _ = s.finish(); - }); - } - let _ = self.rx.lock().take(); + fn arm_opening(&self) { + self.gate.arm(); } -} -async fn read_loop(recv: RecvStream, tx: mpsc::Sender) { - let mut lines = BufReader::new(recv).lines(); - loop { - let line = match lines.next_line().await { - Ok(Some(l)) => l, - Ok(None) => break, - Err(e) => { - tracing::trace!(target: "ensemble.iroh", err = %e, "iroh.read.failed"); - break; - } - }; - if line.is_empty() { - continue; - } - let tone: Tone = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { - tracing::trace!(target: "ensemble.iroh", err = %e, "iroh.parse.failed"); - continue; + fn close(&self) { + self.closed.store(true, Ordering::SeqCst); + self.gate.shut(); + let sender = self.sender.clone(); + tokio::spawn(async move { + if let Some(mut s) = sender.lock().await.take() { + let _ = s.finish(); } - }; - if tx.send(tone).await.is_err() { - break; - } + }); + let _ = self.rx.lock().take(); } } @@ -284,12 +276,9 @@ impl Transport for IrohTransport { .map(TransportAddr::Ip) .collect(); let endpoint_addr = EndpointAddr::from_parts(node_id, direct_addrs); - let endpoint = IrohEndpoint::connect( - &self.endpoint, - endpoint_addr, - PeerCapabilities::default(), - ) - .await?; + let endpoint = + IrohEndpoint::connect(&self.endpoint, endpoint_addr, PeerCapabilities::default()) + .await?; Ok(endpoint as Arc) } } @@ -317,7 +306,12 @@ mod dial_tests { #[test] fn a_concrete_address_is_left_alone() { - for s in ["192.168.1.5:9000", "10.0.0.7:443", "[fe80::1]:9000", "127.0.0.1:5000"] { + for s in [ + "192.168.1.5:9000", + "10.0.0.7:443", + "[fe80::1]:9000", + "127.0.0.1:5000", + ] { let before: SocketAddr = s.parse().expect("parse"); assert_eq!(dialable_addr(before), before, "{s} must be untouched"); } @@ -330,6 +324,9 @@ mod dial_tests { assert_eq!(dialable_addr(once), once); assert!(!once.ip().is_unspecified()); assert_ne!(once.ip(), IpAddr::V4(Ipv4Addr::UNSPECIFIED)); - assert_ne!(dialable_addr(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 1)).ip(), IpAddr::V6(Ipv6Addr::UNSPECIFIED)); + assert_ne!( + dialable_addr(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 1)).ip(), + IpAddr::V6(Ipv6Addr::UNSPECIFIED) + ); } } diff --git a/ensemble/src/kad.rs b/ensemble/src/kad.rs index 30d43ca1..38f12c6c 100644 --- a/ensemble/src/kad.rs +++ b/ensemble/src/kad.rs @@ -50,7 +50,7 @@ use rand::RngCore; use serde_json::Value; use tokio::sync::oneshot; -use crate::{HumdAddr, Hid, Tone}; +use crate::{Hid, HumdAddr, Tone}; /// Kademlia replication / bucket-size parameter (paper default). pub const KAD_K: usize = 20; @@ -82,8 +82,8 @@ impl XorDistance { /// 256-bit big-endian number), the closer the two ids. pub fn distance(a: &Hid, b: &Hid) -> [u8; 32] { let mut out = [0u8; 32]; - for i in 0..32 { - out[i] = a.as_bytes()[i] ^ b.as_bytes()[i]; + for (slot, (x, y)) in out.iter_mut().zip(a.as_bytes().iter().zip(b.as_bytes())) { + *slot = x ^ y; } out } @@ -116,12 +116,20 @@ pub struct KBucket { impl KBucket { pub fn new() -> Self { - Self { entries: VecDeque::with_capacity(KAD_K) } + Self { + entries: VecDeque::with_capacity(KAD_K), + } } - pub fn len(&self) -> usize { self.entries.len() } - pub fn is_empty(&self) -> bool { self.entries.is_empty() } - pub fn iter(&self) -> impl Iterator { self.entries.iter() } + pub fn len(&self) -> usize { + self.entries.len() + } + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + pub fn iter(&self) -> impl Iterator { + self.entries.iter() + } /// Insert (or refresh) `addr`. If the bucket already holds an entry /// with the same id, that entry is moved to the back (most-recent @@ -166,7 +174,9 @@ impl RoutingTable { } } - pub fn me(&self) -> Hid { self.me } + pub fn me(&self) -> Hid { + self.me + } /// Insert `addr` into the appropriate bucket. Inserting the table's /// own id is a no-op (returns false). Returns true on insert/refresh. @@ -198,12 +208,31 @@ impl RoutingTable { all } + pub fn forget(&mut self, id: &Hid) -> bool { + let mut dropped = false; + for bucket in &mut self.buckets { + let before = bucket.len(); + bucket.entries.retain(|a| &a.id != id); + dropped |= bucket.len() != before; + } + dropped + } + + pub fn entries(&self) -> Vec { + self.buckets + .iter() + .flat_map(|b| b.iter().cloned()) + .collect() + } + /// Total number of peers currently in the table. pub fn len(&self) -> usize { self.buckets.iter().map(|b| b.len()).sum() } - pub fn is_empty(&self) -> bool { self.len() == 0 } + pub fn is_empty(&self) -> bool { + self.len() == 0 + } /// Direct lookup: does the table hold a HumdAddr for `id`? pub fn get(&self, id: &Hid) -> Option { @@ -211,10 +240,7 @@ impl RoutingTable { if idx >= 256 { return None; } - self.buckets[idx] - .iter() - .find(|a| &a.id == id) - .cloned() + self.buckets[idx].iter().find(|a| &a.id == id).cloned() } } @@ -241,6 +267,22 @@ impl KadState { /// Stash a HumdAddr into the routing table. Used at install-time /// (bootstrap a peer we just connected to) and when FIND_NODE resps /// advertise new peers. + pub fn forget(&self, id: &Hid) -> bool { + self.table.lock().forget(id) + } + + pub fn snapshot(&self) -> Vec { + self.table.lock().entries() + } + + pub fn restore(self: &Arc, addrs: Vec) -> usize { + let mut table = self.table.lock(); + addrs + .into_iter() + .filter(|a| table.insert(a.clone())) + .count() + } + pub fn note_peer(&self, addr: HumdAddr) { self.table.lock().insert(addr); } @@ -304,12 +346,7 @@ pub fn find_node_tone(rid: &str, query_id: &str, target: &Hid, from: &Hid) -> To } /// Build a `chi:"kad-find-node-resp"` tone carrying up to K HumdAddrs. -pub fn find_node_resp_tone( - rid: &str, - query_id: &str, - from: &Hid, - closest: &[HumdAddr], -) -> Tone { +pub fn find_node_resp_tone(rid: &str, query_id: &str, from: &Hid, closest: &[HumdAddr]) -> Tone { serde_json::json!({ "chi": KAD_FIND_NODE_RESP_CHI, "rid": rid, @@ -348,7 +385,12 @@ pub fn parse_find_node(tone: &Tone) -> Option { .and_then(|v| v.as_str()) .unwrap_or("") .to_string(); - Some(ParsedFindNode { query_id, target, from, rid }) + Some(ParsedFindNode { + query_id, + target, + from, + rid, + }) } /// Pull fields out of a `chi:"kad-find-node-resp"` tone. @@ -360,7 +402,11 @@ pub fn parse_find_node_resp(tone: &Tone) -> Option { .iter() .filter_map(|v| serde_json::from_value::(v.clone()).ok()) .collect(); - Some(ParsedFindNodeResp { query_id, from, closest }) + Some(ParsedFindNodeResp { + query_id, + from, + closest, + }) } // ── Lookup driver ────────────────────────────────────────────────────────── @@ -455,6 +501,7 @@ impl LookupShortlist { pub(crate) async fn query_peer( kad: &Arc, conn: &Arc, + stats: &Arc, me: &Hid, target: &Hid, per_query_timeout: Duration, @@ -463,7 +510,7 @@ pub(crate) async fn query_peer( let rid = format!("kad-find-{}", &query_id[..8]); let rx = kad.register_query(query_id.clone()); let tone = find_node_tone(&rid, &query_id, target, me); - if conn.send(tone).await.is_err() { + if crate::send_bounded(conn, tone, stats).await.is_err() { kad.cancel_query(&query_id); return Vec::new(); } @@ -587,7 +634,7 @@ mod tests { fn parse_find_node_resp_round_trip() { let me = Hid::random_humd(); let peer = HumdAddr::new(Hid::random_humd()).with_hint("tcp:1.2.3.4:9000"); - let tone = find_node_resp_tone("rid-1", "qid-1", &me, &[peer.clone()]); + let tone = find_node_resp_tone("rid-1", "qid-1", &me, std::slice::from_ref(&peer)); let parsed = parse_find_node_resp(&tone).expect("parse"); assert_eq!(parsed.query_id, "qid-1"); assert_eq!(parsed.from, me); diff --git a/ensemble/src/lib.rs b/ensemble/src/lib.rs index 2818274f..bf004184 100644 --- a/ensemble/src/lib.rs +++ b/ensemble/src/lib.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::ops::Deref; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::time::Duration; use anyhow::Result; @@ -11,14 +11,13 @@ use tokio::task::JoinSet; pub mod handshake; pub use handshake::{ - hello_tone, hello_tone_unsigned, parse_hello, parse_hello_caps, EnsembleHello, Hid, - HidParseError, HidPrefix, HelloParse, HumdAddr, HumdKey, PeerCapabilities, PeerConnection, - Transport, + EnsembleHello, HelloParse, Hid, HidParseError, HidPrefix, HumdAddr, HumdKey, PeerCapabilities, + PeerConnection, Transport, hello_tone, hello_tone_unsigned, parse_hello, parse_hello_caps, }; pub mod link; pub use link::{ - InMemoryEndpoint, LinkCounters, LinkFaults, Noise, Script, Verdict, PARTITION_BUFFER_CAP, + InMemoryEndpoint, LinkCounters, LinkFaults, Noise, PARTITION_BUFFER_CAP, Script, Verdict, }; pub mod tcp; @@ -26,39 +25,42 @@ pub use tcp::{TcpEndpoint, TcpListener, TcpTransport}; pub mod tls; pub use tls::{ - cert_fingerprint, client_config_pinned, PinnedFingerprintVerifier, TlsTcpEndpoint, - TlsTcpListener, TlsTcpTransport, TLS_FP_HINT, TLS_HINT, + PinnedFingerprintVerifier, TLS_FP_HINT, TLS_HINT, TlsTcpEndpoint, TlsTcpListener, + TlsTcpTransport, cert_fingerprint, client_config_pinned, }; pub mod iroh; -pub use iroh::{dialable_addr, IrohEndpoint, IrohTransport, IROH_ALPN, IROH_IP_HINT}; +pub use iroh::{IROH_ALPN, IROH_IP_HINT, IrohEndpoint, IrohTransport, dialable_addr}; pub mod delivery; +pub mod framing; +pub mod opening; pub mod send; -pub use send::{send_bounded, SendError, SendStats, SEND_TIMEOUT}; -pub use delivery::{DeliveryState, DELIVERY_SEEN_CAP}; +pub use delivery::{DELIVERY_SEEN_CAP, DeliveryState}; +pub use framing::MAX_FRAME_BYTES; +pub use opening::Gate; +pub use send::{SEND_TIMEOUT, SendError, SendStats, fanout, send_bounded}; pub mod gossip; pub use gossip::{ - gossip_tone, gossip_tone_with_dusk, mint_msg_id, GossipState, GOSSIP_CHI, - GOSSIP_SEEN_CAP, + GOSSIP_CHI, GOSSIP_SEEN_CAP, GossipState, gossip_tone, gossip_tone_with_dusk, mint_msg_id, }; pub mod liveness; pub use liveness::{ - ping_tone, pong_tone, probe_seq, Lease, Liveness, LivenessSignal, PING_CHI, PONG_CHI, + Lease, Liveness, LivenessSignal, PING_CHI, PONG_CHI, ping_tone, pong_tone, probe_seq, }; pub mod kad; pub use kad::{ - find_node_resp_tone, find_node_tone, mint_query_id, parse_find_node, parse_find_node_resp, - KBucket, KadFindOutcome, KadState, RoutingTable, XorDistance, KAD_ALPHA, - KAD_FIND_NODE_CHI, KAD_FIND_NODE_RESP_CHI, KAD_K, KAD_MAX_ROUNDS, + KAD_ALPHA, KAD_FIND_NODE_CHI, KAD_FIND_NODE_RESP_CHI, KAD_K, KAD_MAX_ROUNDS, KBucket, + KadFindOutcome, KadState, RoutingTable, XorDistance, find_node_resp_tone, find_node_tone, + mint_query_id, parse_find_node, parse_find_node_resp, }; pub mod hives; pub mod uri; -pub use hives::{BindAddr, HiveAnnounce, HiveManifest, Propensity, ToolEntry, ANNOUNCE_TOPIC}; +pub use hives::{ANNOUNCE_TOPIC, BindAddr, HiveAnnounce, HiveManifest, Propensity, ToolEntry}; pub use uri::{AliasResolver, HostRef, HumUri, UriParseError}; pub mod headroom; @@ -96,6 +98,7 @@ pub struct Ensemble { expired_dusk: Arc, delivery: Arc, send_stats: Arc, + unverified: Arc, } #[derive(Clone)] @@ -108,7 +111,11 @@ pub struct Inbox { impl Inbox { fn new() -> Self { let (tx, _) = broadcast::channel(256); - Self { tx, subscribers: Arc::new(AtomicUsize::new(0)), dropped: Arc::new(AtomicU64::new(0)) } + Self { + tx, + subscribers: Arc::new(AtomicUsize::new(0)), + dropped: Arc::new(AtomicU64::new(0)), + } } pub fn publish(&self, tone: Tone) -> bool { @@ -126,12 +133,19 @@ impl Inbox { pub fn subscribe(&self) -> InboxSub { self.subscribers.fetch_add(1, Ordering::SeqCst); - InboxSub { rx: self.tx.subscribe(), subscribers: self.subscribers.clone() } + InboxSub { + rx: self.tx.subscribe(), + subscribers: self.subscribers.clone(), + } } - pub fn has_subscribers(&self) -> bool { self.subscribers.load(Ordering::SeqCst) > 0 } + pub fn has_subscribers(&self) -> bool { + self.subscribers.load(Ordering::SeqCst) > 0 + } - pub fn dropped(&self) -> u64 { self.dropped.load(Ordering::Relaxed) } + pub fn dropped(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } } pub struct InboxSub { @@ -141,11 +155,15 @@ pub struct InboxSub { impl Deref for InboxSub { type Target = broadcast::Receiver; - fn deref(&self) -> &Self::Target { &self.rx } + fn deref(&self) -> &Self::Target { + &self.rx + } } impl std::ops::DerefMut for InboxSub { - fn deref_mut(&mut self) -> &mut Self::Target { &mut self.rx } + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.rx + } } impl Drop for InboxSub { @@ -180,10 +198,11 @@ impl Ensemble { inbox: Inbox::new(), gossip: GossipState::new(), kad: KadState::new(me), - strict_auth: false, + strict_auth: true, expired_dusk: Arc::new(AtomicU64::new(0)), delivery: DeliveryState::new(), send_stats: SendStats::new(), + unverified: Arc::new(AtomicU64::new(0)), } } @@ -193,9 +212,13 @@ impl Ensemble { e } - pub fn me(&self) -> Hid { self.me } + pub fn me(&self) -> Hid { + self.me + } - pub fn strict_auth(&self) -> bool { self.strict_auth } + pub fn strict_auth(&self) -> bool { + self.strict_auth + } pub fn install( &self, @@ -203,100 +226,7 @@ impl Ensemble { my_caps: PeerCapabilities, my_key: &HumdKey, ) { - let id = conn.peer().id; - let hello = hello_tone(&self.me, my_key, &my_caps); - let hello_conn = conn.clone(); - tokio::spawn(async move { - let _ = hello_conn.send(hello).await; - }); - - let rx = conn.take_receiver(); - self.peers.write().insert( - id, - Peer { conn: conn.clone(), learned_caps: None, lease: Lease::new() }, - ); - self.kad.note_peer(conn.peer().clone()); - - if let Some(mut rx) = rx { - let peers = self.peers.clone(); - let inbox = self.inbox.clone(); - let conn_for_drain = conn.clone(); - let strict = self.strict_auth; - let gossip = self.gossip.clone(); - let expired_dusk = self.expired_dusk.clone(); - let delivery = self.delivery.clone(); - let send_stats = self.send_stats.clone(); - let kad = self.kad.clone(); - let my_id = self.me; - tokio::spawn(async move { - let mut id = id; - let mut handshake_seen = false; - while let Some(tone) = rx.recv().await { - let is_hello = tone.get("chi").and_then(|v| v.as_str()) == Some("hello"); - if is_hello && !handshake_seen { - handshake_seen = true; - match parse_hello(&tone) { - HelloParse::Verified(claimed_id, caps) => { - if claimed_id != id { - rekey_peer(&peers, &kad, &conn_for_drain, id, claimed_id); - id = claimed_id; - } - if let Some(p) = peers.write().get_mut(&id) { - p.learned_caps = Some(caps); - } - } - HelloParse::Unsigned(claimed_id, caps) => { - if strict { - tracing::warn!( - target: "ensemble", - transport_id = %id.short(), - claimed_id = %claimed_id.short(), - "hello.rejected: strict_auth requires signed hello" - ); - peers.write().remove(&id); - conn_for_drain.close(); - return; - } - if claimed_id == id { - if let Some(p) = peers.write().get_mut(&id) { - p.learned_caps = Some(caps); - } - } - } - HelloParse::Invalid => { - peers.write().remove(&id); - conn_for_drain.close(); - return; - } - } - continue; - } - if handle_liveness(&peers, &my_id, &id, &conn_for_drain, &tone).await { - continue; - } - if !delivery::dispatchable(&tone, &delivery, &expired_dusk) { - continue; - } - if tone.get("chi").and_then(|v| v.as_str()) == Some(GOSSIP_CHI) { - if handle_gossip(&send_stats, &gossip, &peers, &id, &tone).await { - continue; - } - } - let chi_val = tone.get("chi").and_then(|v| v.as_str()); - if chi_val == Some(KAD_FIND_NODE_CHI) - || chi_val == Some(KAD_FIND_NODE_RESP_CHI) - { - if handle_kad(&send_stats, &kad, &peers, &id, &my_id, &tone).await { - continue; - } - } - inbox.publish(tone); - } - if let Some(p) = peers.write().get_mut(&id) { - p.lease.observe(LivenessSignal::TransportClosed); - } - }); - } + self.admit(conn, hello_tone(&self.me, my_key, &my_caps), true); } pub fn add_peer(&self, conn: Arc) { @@ -307,105 +237,51 @@ impl Ensemble { self.install_unsigned(conn, caps); } - pub fn install_unsigned( - &self, - conn: Arc, - my_caps: PeerCapabilities, - ) { + pub fn install_unsigned(&self, conn: Arc, my_caps: PeerCapabilities) { + self.admit(conn, hello_tone_unsigned(&self.me, &my_caps), false); + } + + fn admit(&self, conn: Arc, hello: Tone, rekey: bool) { let id = conn.peer().id; - let hello = hello_tone_unsigned(&self.me, &my_caps); + conn.arm_opening(); + + let stats = self.send_stats.clone(); let hello_conn = conn.clone(); tokio::spawn(async move { - let _ = hello_conn.send(hello).await; + if send::send_opening_bounded(&hello_conn, hello, &stats, send::SEND_TIMEOUT) + .await + .is_err() + { + hello_conn.close(); + } }); let rx = conn.take_receiver(); self.peers.write().insert( id, - Peer { conn: conn.clone(), learned_caps: None, lease: Lease::new() }, + Peer { + conn: conn.clone(), + learned_caps: None, + lease: Lease::new(), + }, ); self.kad.note_peer(conn.peer().clone()); - if let Some(mut rx) = rx { - let peers = self.peers.clone(); - let inbox = self.inbox.clone(); - let conn_for_drain = conn.clone(); - let strict = self.strict_auth; - let gossip = self.gossip.clone(); - let expired_dusk = self.expired_dusk.clone(); - let delivery = self.delivery.clone(); - let send_stats = self.send_stats.clone(); - let kad = self.kad.clone(); - let my_id = self.me; - tokio::spawn(async move { - let mut handshake_seen = false; - while let Some(tone) = rx.recv().await { - let is_hello = tone.get("chi").and_then(|v| v.as_str()) == Some("hello"); - if is_hello && !handshake_seen { - handshake_seen = true; - match parse_hello(&tone) { - HelloParse::Verified(claimed_id, caps) if claimed_id == id => { - if let Some(p) = peers.write().get_mut(&id) { - p.learned_caps = Some(caps); - } - } - HelloParse::Verified(claimed_id, _) => { - tracing::warn!( - target: "ensemble", - transport_id = %id.short(), - claimed_id = %claimed_id.short(), - "hello.rejected: claimed humd_id does not match transport-peer id" - ); - peers.write().remove(&id); - conn_for_drain.close(); - return; - } - HelloParse::Unsigned(claimed_id, caps) => { - if strict { - peers.write().remove(&id); - conn_for_drain.close(); - return; - } - if claimed_id == id { - if let Some(p) = peers.write().get_mut(&id) { - p.learned_caps = Some(caps); - } - } - } - HelloParse::Invalid => { - peers.write().remove(&id); - conn_for_drain.close(); - return; - } - } - continue; - } - if handle_liveness(&peers, &my_id, &id, &conn_for_drain, &tone).await { - continue; - } - if !delivery::dispatchable(&tone, &delivery, &expired_dusk) { - continue; - } - if tone.get("chi").and_then(|v| v.as_str()) == Some(GOSSIP_CHI) { - if handle_gossip(&send_stats, &gossip, &peers, &id, &tone).await { - continue; - } - } - let chi_val = tone.get("chi").and_then(|v| v.as_str()); - if chi_val == Some(KAD_FIND_NODE_CHI) - || chi_val == Some(KAD_FIND_NODE_RESP_CHI) - { - if handle_kad(&send_stats, &kad, &peers, &id, &my_id, &tone).await { - continue; - } - } - inbox.publish(tone); - } - if let Some(p) = peers.write().get_mut(&id) { - p.lease.observe(LivenessSignal::TransportClosed); - } - }); - } + let Some(rx) = rx else { return }; + let drain = Drain { + me: self.me, + peers: self.peers.clone(), + inbox: self.inbox.clone(), + gossip: self.gossip.clone(), + kad: self.kad.clone(), + delivery: self.delivery.clone(), + expired_dusk: self.expired_dusk.clone(), + send_stats: self.send_stats.clone(), + unverified: self.unverified.clone(), + strict: self.strict_auth, + rekey, + }; + tokio::spawn(drain.run(rx, conn)); } pub async fn probe_all(&self, seq: u64) { @@ -415,8 +291,12 @@ impl Ensemble { .iter() .map(|(id, p)| (*id, p.conn.clone())) .collect(); - for (id, conn) in peers { - let _ = send_bounded(&conn, ping_tone(&self.me, &id, seq), &self.send_stats).await; + let sends = peers + .into_iter() + .map(|(id, conn)| (conn, ping_tone(&self.me, &id, seq))) + .collect(); + for e in fanout(sends, &self.send_stats).await { + tracing::debug!(target: "ensemble.liveness", error = %e, "probe.failed"); } } @@ -455,11 +335,20 @@ impl Ensemble { } pub fn remove_peer(&self, id: &Hid) { + self.kad.forget(id); if let Some(p) = self.peers.write().remove(id) { p.conn.close(); } } + pub fn routing_snapshot(&self) -> Vec { + self.kad.snapshot() + } + + pub fn routing_restore(&self, addrs: Vec) -> usize { + self.kad.restore(addrs) + } + pub fn peers(&self) -> Vec { self.peers.read().keys().copied().collect() } @@ -473,14 +362,23 @@ impl Ensemble { } pub fn handshake_done(&self, id: &Hid) -> bool { - self.peers.read().get(id).is_some_and(|p| p.learned_caps.is_some()) + self.peers + .read() + .get(id) + .is_some_and(|p| p.learned_caps.is_some()) } - pub fn subscribe(&self) -> InboxSub { self.inbox.subscribe() } + pub fn subscribe(&self) -> InboxSub { + self.inbox.subscribe() + } - pub fn has_subscribers(&self) -> bool { self.inbox.has_subscribers() } + pub fn has_subscribers(&self) -> bool { + self.inbox.has_subscribers() + } - pub fn inbox_dropped(&self) -> u64 { self.inbox.dropped() } + pub fn inbox_dropped(&self) -> u64 { + self.inbox.dropped() + } pub fn expired_dusk(&self) -> u64 { self.expired_dusk.load(Ordering::SeqCst) @@ -494,6 +392,10 @@ impl Ensemble { self.send_stats.timed_out() } + pub fn tones_before_handshake(&self) -> u64 { + self.unverified.load(Ordering::SeqCst) + } + pub fn send_failures(&self) -> u64 { self.send_stats.failed() } @@ -512,20 +414,14 @@ impl Ensemble { let rid = format!("gossip-{msg_id}"); self.gossip.note_seen(&msg_id); let tone = gossip_tone_with_dusk(topic, &rid, &self.me, payload, &msg_id, dusk_ms); - let conns: Vec> = { - let peers = self.peers.read(); - peers.values().map(|p| p.conn.clone()).collect() - }; - for conn in conns { - if let Err(e) = send_bounded(&conn, tone.clone(), &self.send_stats).await { - tracing::debug!( - target: "ensemble.gossip", - peer = %conn.peer().id.short(), - topic = topic, - error = %e, - "publish send failed" - ); - } + let sends: Vec<(Arc, Tone)> = self + .peers + .read() + .values() + .map(|p| (p.conn.clone(), tone.clone())) + .collect(); + for e in fanout(sends, &self.send_stats).await { + tracing::debug!(target: "ensemble.gossip", topic, error = %e, "publish.failed"); } } @@ -536,7 +432,7 @@ impl Ensemble { pub async fn hive_advertise(&self, manifest: hives::HiveManifest) { let env = hives::HiveAnnounce::Advertise { humd_id: self.me.to_hex(), - manifest, + manifest: Box::new(manifest), }; match serde_json::to_value(&env) { Ok(payload) => self.publish(hives::ANNOUNCE_TOPIC, payload).await, @@ -582,7 +478,10 @@ impl Ensemble { rx } - pub fn hive_discover(&self, name: impl Into) -> mpsc::Receiver<(Hid, hives::HiveManifest)> { + pub fn hive_discover( + &self, + name: impl Into, + ) -> mpsc::Receiver<(Hid, hives::HiveManifest)> { let needle = name.into(); let mut raw = self.subscribe_topic(hives::ANNOUNCE_TOPIC); let (tx, rx) = mpsc::channel(64); @@ -590,16 +489,15 @@ impl Ensemble { loop { match raw.recv().await { Ok(v) => { - let parsed: Result = - serde_json::from_value(v); + let parsed: Result = serde_json::from_value(v); if let Ok(hives::HiveAnnounce::Advertise { humd_id, manifest }) = parsed { if manifest.name != needle { continue; } - if let Ok(id) = Hid::from_hex(&humd_id) { - if tx.send((id, manifest)).await.is_err() { - break; - } + if let Ok(id) = Hid::from_hex(&humd_id) + && tx.send((id, *manifest)).await.is_err() + { + break; } } } @@ -654,28 +552,35 @@ impl Ensemble { }; if let Some(conn) = conn { let kad = self.kad.clone(); + let stats = self.send_stats.clone(); let me = self.me; let tgt = target; joinset.spawn(async move { - kad::query_peer(&kad, &conn, &me, &tgt, per_query_timeout).await + kad::query_peer(&kad, &conn, &stats, &me, &tgt, per_query_timeout).await }); } } if joinset.is_empty() { continue; } - while let Some(res) = joinset.join_next().await { - let advertised_list = match res { - Ok(list) => list, - Err(_) => continue, - }; - for advertised in advertised_list { - self.kad.note_peer(advertised.clone()); - if advertised.id == self.me { - continue; + let drain = async { + while let Some(res) = joinset.join_next().await { + let advertised_list = match res { + Ok(list) => list, + Err(_) => continue, + }; + for advertised in advertised_list { + self.kad.note_peer(advertised.clone()); + if advertised.id == self.me { + continue; + } + shortlist.insert(advertised); } - shortlist.insert(advertised); } + }; + if tokio::time::timeout_at(deadline, drain).await.is_err() { + joinset.abort_all(); + return self.kad.get(&target); } if let Some(addr) = self.kad.get(&target) { @@ -721,6 +626,122 @@ impl Ensemble { } } +struct Drain { + me: Hid, + peers: Arc>>, + inbox: Inbox, + gossip: Arc, + kad: Arc, + delivery: Arc, + expired_dusk: Arc, + send_stats: Arc, + unverified: Arc, + strict: bool, + rekey: bool, +} + +impl Drain { + async fn run(self, mut rx: mpsc::Receiver, conn: Arc) { + let mut id = conn.peer().id; + let mut handshook = false; + + while let Some(tone) = rx.recv().await { + let chi = tone.get("chi").and_then(|v| v.as_str()); + + if !handshook { + if chi == Some("hello") { + handshook = true; + if self.hello(&mut id, &conn, parse_hello(&tone)).is_err() { + self.peers.write().remove(&id); + conn.close(); + return; + } + } else { + self.unverified.fetch_add(1, Ordering::SeqCst); + continue; + } + continue; + } + if handle_liveness(&self.peers, &self.send_stats, &self.me, &id, &conn, &tone).await { + continue; + } + if !delivery::dispatchable(&tone, &self.delivery, &self.expired_dusk) { + continue; + } + if chi == Some(GOSSIP_CHI) + && handle_gossip(&self.send_stats, &self.gossip, &self.peers, &id, &tone).await + { + continue; + } + if matches!(chi, Some(KAD_FIND_NODE_CHI) | Some(KAD_FIND_NODE_RESP_CHI)) + && handle_kad( + &self.send_stats, + &self.kad, + &self.peers, + &id, + &self.me, + &tone, + ) + .await + { + continue; + } + self.inbox.publish(tone); + } + + if let Some(p) = self.peers.write().get_mut(&id) { + p.lease.observe(LivenessSignal::TransportClosed); + } + } + + fn hello( + &self, + id: &mut Hid, + conn: &Arc, + parsed: HelloParse, + ) -> Result<(), ()> { + match parsed { + HelloParse::Verified(claimed, caps) => { + if claimed != *id && !self.rekey { + tracing::warn!( + target: "ensemble", + transport_id = %id.short(), + claimed_id = %claimed.short(), + "hello.rejected: claimed id differs from the id this link was admitted under" + ); + return Err(()); + } + if claimed != *id { + rekey_peer(&self.peers, &self.kad, conn, *id, claimed); + *id = claimed; + } + if let Some(p) = self.peers.write().get_mut(id) { + p.learned_caps = Some(caps); + } + Ok(()) + } + HelloParse::Unsigned(claimed, caps) => { + if self.strict { + tracing::warn!( + target: "ensemble", + transport_id = %id.short(), + claimed_id = %claimed.short(), + "hello.rejected: strict_auth requires a signed hello" + ); + return Err(()); + } + if claimed == *id + && let Some(p) = self.peers.write().get_mut(id) + { + p.learned_caps = Some(caps); + } + Ok(()) + } + HelloParse::Invalid => Err(()), + } + } +} + fn rekey_peer( peers: &Arc>>, kad: &Arc, @@ -753,21 +774,24 @@ async fn handle_kad( None => return false, }; let closest = kad.closest_to(&parsed.target, KAD_K); - let resp_rid = format!("kad-resp-{}", &parsed.query_id[..8.min(parsed.query_id.len())]); + let resp_rid = format!( + "kad-resp-{}", + &parsed.query_id[..8.min(parsed.query_id.len())] + ); let resp = kad::find_node_resp_tone(&resp_rid, &parsed.query_id, me, &closest); let conn = { let peers = peers.read(); peers.get(arrived_from).map(|p| p.conn.clone()) }; - if let Some(conn) = conn { - if let Err(e) = send_bounded(&conn, resp, send_stats).await { - tracing::debug!( - target: "ensemble.kad", - peer = %arrived_from.short(), - error = %e, - "find-node response send failed" - ); - } + if let Some(conn) = conn + && let Err(e) = send_bounded(&conn, resp, send_stats).await + { + tracing::debug!( + target: "ensemble.kad", + peer = %arrived_from.short(), + error = %e, + "find-node response send failed" + ); } true } else if chi_val == Some(kad::KAD_FIND_NODE_RESP_CHI) { @@ -787,6 +811,7 @@ async fn handle_kad( async fn handle_liveness( peers: &Arc>>, + send_stats: &Arc, me: &Hid, arrived_from: &Hid, conn: &Arc, @@ -803,13 +828,13 @@ async fn handle_liveness( return false; } if let Some(seq) = probe_seq(tone) { - let _ = conn.send(pong_tone(me, arrived_from, seq)).await; + let _ = send_bounded(conn, pong_tone(me, arrived_from, seq), send_stats).await; } true } async fn handle_gossip( - send_stats: &SendStats, + send_stats: &Arc, gossip: &Arc, peers: &Arc>>, arrived_from: &Hid, @@ -825,23 +850,14 @@ async fn handle_gossip( if let Some(tx) = gossip.sender(parsed.topic) { let _ = tx.send(parsed.payload.clone()); } - let others: Vec> = { - let peers = peers.read(); - peers - .iter() - .filter(|(id, _)| *id != arrived_from) - .map(|(_, p)| p.conn.clone()) - .collect() - }; - for conn in others { - if let Err(e) = send_bounded(&conn, tone.clone(), send_stats).await { - tracing::debug!( - target: "ensemble.gossip", - peer = %conn.peer().id.short(), - error = %e, - "re-fan send failed" - ); - } + let sends: Vec<(Arc, Tone)> = peers + .read() + .iter() + .filter(|(id, _)| *id != arrived_from) + .map(|(_, p)| (p.conn.clone(), tone.clone())) + .collect(); + for e in fanout(sends, send_stats).await { + tracing::debug!(target: "ensemble.gossip", error = %e, "refan.failed"); } true } @@ -877,7 +893,10 @@ mod tests { } fn rids(tones: &[Tone]) -> Vec { - tones.iter().map(|t| t["rid"].as_str().unwrap().to_string()).collect() + tones + .iter() + .map(|t| t["rid"].as_str().unwrap().to_string()) + .collect() } #[tokio::test] @@ -890,7 +909,14 @@ mod tests { let got = rids(&drain(&mut rx).await); assert_eq!(got, ["t0", "t1", "t2", "t3", "t4"]); let c = a.counters(); - assert_eq!(c, LinkCounters { offered: 5, delivered: 5, ..Default::default() }); + assert_eq!( + c, + LinkCounters { + offered: 5, + delivered: 5, + ..Default::default() + } + ); } #[tokio::test] @@ -975,9 +1001,16 @@ mod tests { drain(&mut rx).await.len() } assert_eq!(run(42).await, run(42).await, "same seed, same pattern"); - assert_ne!(run(1).await, run(2).await, "different seeds, different patterns"); + assert_ne!( + run(1).await, + run(2).await, + "different seeds, different patterns" + ); let survived = run(42).await; - assert!((120..170).contains(&survived), "survived {survived} of 200 at 30% loss"); + assert!( + (120..170).contains(&survived), + "survived {survived} of 200 at 30% loss" + ); } #[tokio::test] @@ -1079,10 +1112,16 @@ mod tests { let c = a.counters(); let arrived = drain(&mut rx).await.len() as u64; - assert_eq!(c.delivered, arrived, "delivered must match what the peer saw"); + assert_eq!( + c.delivered, arrived, + "delivered must match what the peer saw" + ); assert_eq!(c.offered, 43); assert_eq!(c.buffered, 3); - assert_eq!(c.offered, c.delivered - c.duplicated + c.dropped + c.lost_on_heal); + assert_eq!( + c.offered, + c.delivered - c.duplicated + c.dropped + c.lost_on_heal + ); } #[tokio::test] @@ -1093,7 +1132,11 @@ mod tests { a.set_faults(LinkFaults::default().drop_next(10)); b.send(tone("survivor")).await.unwrap(); a.send(tone("doomed")).await.unwrap(); - assert_eq!(rids(&drain(&mut a_rx).await), ["survivor"], "b→a is unaffected"); + assert_eq!( + rids(&drain(&mut a_rx).await), + ["survivor"], + "b→a is unaffected" + ); assert!(drain(&mut b_rx).await.is_empty(), "a→b drops"); } @@ -1150,11 +1193,21 @@ mod tests { let a_id = Hid::random_humd(); let b_id = Hid::random_humd(); let (a, b) = InMemoryEndpoint::pair( - a_id, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() }, - b_id, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() }, + a_id, + PeerCapabilities { + proto_version: "0.2.0".into(), + ..Default::default() + }, + b_id, + PeerCapabilities { + proto_version: "0.2.0".into(), + ..Default::default() + }, ); let mut rx_b = b.take_receiver().unwrap(); - a.send(json!({"chi": "hello", "rid": "1", "from": a_id.to_hex()})).await.unwrap(); + a.send(json!({"chi": "hello", "rid": "1", "from": a_id.to_hex()})) + .await + .unwrap(); let received = rx_b.recv().await.unwrap(); assert_eq!(received.get("chi").unwrap(), "hello"); } @@ -1167,8 +1220,10 @@ mod tests { let ensemble = Ensemble::new(me); let (mine, theirs) = InMemoryEndpoint::pair( - me, PeerCapabilities::default(), - peer_id, PeerCapabilities::default(), + me, + PeerCapabilities::default(), + peer_id, + PeerCapabilities::default(), ); ensemble.add_peer(mine); let mut rx = theirs.take_receiver().unwrap(); @@ -1212,10 +1267,7 @@ mod tests { free_slots: None, headroom: headroom::CellHeadroom::default(), }; - let (a_side, b_side) = InMemoryEndpoint::pair( - a_id, b_caps.clone(), - b_id, a_caps.clone(), - ); + let (a_side, b_side) = InMemoryEndpoint::pair(a_id, b_caps.clone(), b_id, a_caps.clone()); let ensemble_a = Ensemble::new(a_id); let ensemble_b = Ensemble::new(b_id); @@ -1261,16 +1313,32 @@ mod tests { let me = me_key.hid(); let peer_id = peer_key.hid(); let (mine, theirs) = InMemoryEndpoint::pair( - me, PeerCapabilities::default(), - peer_id, PeerCapabilities::default(), + me, + PeerCapabilities::default(), + peer_id, + PeerCapabilities::default(), ); let ensemble = Ensemble::new(me); let mut sub = ensemble.subscribe(); - ensemble.install(mine, PeerCapabilities { proto_version: "0.3.0".into(), ..Default::default() }, &me_key); + ensemble.install( + mine, + PeerCapabilities { + proto_version: "0.3.0".into(), + ..Default::default() + }, + &me_key, + ); theirs - .send(hello_tone(&peer_id, &peer_key, &PeerCapabilities { proto_version: "0.3.0".into(), ..Default::default() })) + .send(hello_tone( + &peer_id, + &peer_key, + &PeerCapabilities { + proto_version: "0.3.0".into(), + ..Default::default() + }, + )) .await .unwrap(); theirs @@ -1299,16 +1367,32 @@ mod tests { let me = me_key.hid(); let peer_id = peer_key.hid(); let (mine, theirs) = InMemoryEndpoint::pair( - me, PeerCapabilities::default(), - peer_id, PeerCapabilities::default(), + me, + PeerCapabilities::default(), + peer_id, + PeerCapabilities::default(), ); let ensemble = Ensemble::new(me); let mut sub = ensemble.subscribe(); - ensemble.install(mine, PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() }, &me_key); + ensemble.install( + mine, + PeerCapabilities { + proto_version: "0.2.0".into(), + ..Default::default() + }, + &me_key, + ); theirs - .send(hello_tone(&peer_id, &peer_key, &PeerCapabilities { proto_version: "0.2.0".into(), ..Default::default() })) + .send(hello_tone( + &peer_id, + &peer_key, + &PeerCapabilities { + proto_version: "0.2.0".into(), + ..Default::default() + }, + )) .await .unwrap(); theirs @@ -1324,3 +1408,126 @@ mod tests { assert_eq!(got.get("rid").unwrap(), "r1"); } } + +#[cfg(test)] +mod invariants { + use super::*; + use serde_json::json; + + fn caps() -> PeerCapabilities { + PeerCapabilities::default() + } + + #[tokio::test] + async fn the_opening_frame_precedes_application_traffic() { + let me = Hid::random_humd(); + let peer = Hid::random_humd(); + let key = HumdKey::generate(); + let ensemble = Ensemble::new(me); + let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps()); + let mut rx = theirs + .take_receiver() + .expect("far end must expose a receiver"); + + ensemble.install(mine, caps(), &key); + ensemble + .route(json!({"chi": "prompt", "rid": "r1", "to": peer.to_hex()})) + .await + .expect("routing to the only peer cannot fail"); + + let first = rx.recv().await.expect("receiver closed"); + assert_eq!(first["chi"], "hello", "the hello must be the first frame"); + let second = rx.recv().await.expect("receiver closed"); + assert_eq!(second["rid"], "r1", "application traffic follows the hello"); + } + + #[tokio::test] + async fn a_tone_that_outruns_the_handshake_is_never_published() { + let me = Hid::random_humd(); + let peer = Hid::random_humd(); + let key = HumdKey::generate(); + let ensemble = Ensemble::new(me); + let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps()); + + ensemble.install(mine, caps(), &key); + + let mut inbox = ensemble.subscribe(); + theirs + .send(json!({"chi": "prompt", "rid": "early", "to": me.to_hex()})) + .await + .expect("in-memory send cannot fail"); + tokio::time::sleep(Duration::from_millis(50)).await; + + assert_eq!( + ensemble.tones_before_handshake(), + 1, + "the drop must be counted" + ); + assert!( + inbox.try_recv().is_err(), + "an unauthenticated tone must not publish" + ); + } + + #[tokio::test] + async fn an_unsigned_hello_is_refused_under_the_default_posture() { + let me = Hid::random_humd(); + let peer = Hid::random_humd(); + let ensemble = Ensemble::new(me); + assert!( + ensemble.strict_auth(), + "strict auth must be the default posture" + ); + let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps()); + + ensemble.install_unsigned(mine, caps()); + theirs + .send(hello_tone_unsigned(&peer, &caps())) + .await + .expect("in-memory send cannot fail"); + tokio::time::sleep(Duration::from_millis(50)).await; + + assert!( + !ensemble.peers().contains(&peer), + "an unsigned peer must not be admitted under strict auth" + ); + } + + #[tokio::test] + async fn the_same_unsigned_hello_is_accepted_when_strictness_is_waived() { + let me = Hid::random_humd(); + let peer = Hid::random_humd(); + let ensemble = Ensemble::with_strict_auth(me, false); + let (mine, theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps()); + + ensemble.install_unsigned(mine, caps()); + theirs + .send(hello_tone_unsigned(&peer, &caps())) + .await + .expect("in-memory send cannot fail"); + tokio::time::sleep(Duration::from_millis(50)).await; + + assert!( + ensemble.handshake_done(&peer), + "waived strictness must admit the peer" + ); + } + + #[tokio::test] + async fn a_removed_peer_leaves_the_routing_table() { + let me = Hid::random_humd(); + let peer = Hid::random_humd(); + let ensemble = Ensemble::new(me); + let (mine, _theirs) = InMemoryEndpoint::pair_concrete(me, caps(), peer, caps()); + + ensemble.add_peer(mine); + assert!(ensemble.kad_closest(&peer, 4).iter().any(|a| a.id == peer)); + + ensemble.remove_peer(&peer); + assert_eq!( + ensemble.kad_routing_table_len(), + 0, + "a dead peer must not route" + ); + } +} diff --git a/ensemble/src/link.rs b/ensemble/src/link.rs index c946a915..596fec45 100644 --- a/ensemble/src/link.rs +++ b/ensemble/src/link.rs @@ -4,11 +4,12 @@ use parking_lot::Mutex; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; use std::collections::VecDeque; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::mpsc; use super::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone}; +use crate::opening::Gate; #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct Script { @@ -22,7 +23,7 @@ pub struct Script { impl Script { fn take(counted: &mut usize, every: usize, offered: u64) -> bool { match *counted { - 0 => every > 0 && offered % every as u64 == 0, + 0 => every > 0 && offered.is_multiple_of(every as u64), n => { *counted = n - 1; true @@ -68,7 +69,12 @@ pub struct LinkFaults { impl Default for LinkFaults { fn default() -> Self { - Self { script: Script::default(), noise: Noise::default(), seed: 0x5EED_C0DE, offered: 0 } + Self { + script: Script::default(), + noise: Noise::default(), + seed: 0x5EED_C0DE, + offered: 0, + } } } @@ -111,7 +117,9 @@ impl LinkFaults { fn verdict(&mut self, rng: &mut impl Rng) -> Verdict { self.offered += 1; - self.script.verdict(self.offered).unwrap_or_else(|| self.noise.verdict(rng)) + self.script + .verdict(self.offered) + .unwrap_or_else(|| self.noise.verdict(rng)) } } @@ -158,6 +166,7 @@ pub struct InMemoryEndpoint { reorder_hold: Mutex>, killed: AtomicBool, stalled: AtomicBool, + gate: Gate, } struct PartitionState { @@ -195,10 +204,14 @@ impl InMemoryEndpoint { partitioned: false, buffer: VecDeque::new(), }), - faults: Mutex::new(LinkFaults { seed: seed_a, ..Default::default() }), + faults: Mutex::new(LinkFaults { + seed: seed_a, + ..Default::default() + }), rng: Mutex::new(StdRng::seed_from_u64(seed_a)), counters: Mutex::new(LinkCounters::default()), reorder_hold: Mutex::new(None), + gate: Gate::new(), killed: AtomicBool::new(false), stalled: AtomicBool::new(false), }); @@ -211,10 +224,14 @@ impl InMemoryEndpoint { partitioned: false, buffer: VecDeque::new(), }), - faults: Mutex::new(LinkFaults { seed: seed_b, ..Default::default() }), + faults: Mutex::new(LinkFaults { + seed: seed_b, + ..Default::default() + }), rng: Mutex::new(StdRng::seed_from_u64(seed_b)), counters: Mutex::new(LinkCounters::default()), reorder_hold: Mutex::new(None), + gate: Gate::new(), killed: AtomicBool::new(false), stalled: AtomicBool::new(false), }); @@ -250,7 +267,10 @@ impl InMemoryEndpoint { async fn emit(&self, tone: Tone) -> Result<()> { if self.stalled.load(Ordering::SeqCst) { self.counters.lock().stalled_sends += 1; - return Ok(std::future::pending::<()>().await); + return { + let _: () = std::future::pending::<()>().await; + Ok(()) + }; } let tx = { let mut guard = self.tx.lock(); @@ -298,13 +318,21 @@ impl InMemoryEndpoint { *self.rng.lock() = StdRng::seed_from_u64(seed); } - pub fn faults(&self) -> LinkFaults { self.faults.lock().clone() } + pub fn faults(&self) -> LinkFaults { + self.faults.lock().clone() + } - pub fn counters(&self) -> LinkCounters { *self.counters.lock() } + pub fn counters(&self) -> LinkCounters { + *self.counters.lock() + } - pub fn buffered(&self) -> usize { self.partition.lock().buffer.len() } + pub fn buffered(&self) -> usize { + self.partition.lock().buffer.len() + } - pub fn is_partitioned(&self) -> bool { self.partition.lock().partitioned } + pub fn is_partitioned(&self) -> bool { + self.partition.lock().partitioned + } pub fn flush_reorder(&self) -> bool { match self.reorder_hold.lock().take() { @@ -337,10 +365,41 @@ impl InMemoryEndpoint { #[async_trait] impl PeerConnection for InMemoryEndpoint { - fn peer(&self) -> &HumdAddr { &self.peer } - fn capabilities(&self) -> &PeerCapabilities { &self.caps } + fn peer(&self) -> &HumdAddr { + &self.peer + } + fn capabilities(&self) -> &PeerCapabilities { + &self.caps + } async fn send(&self, tone: Tone) -> Result<()> { + self.gate.wait().await; + self.deliver(tone).await + } + + async fn send_opening(&self, tone: Tone) -> Result<()> { + let result = self.deliver(tone).await; + self.gate.opened(); + result + } + + fn take_receiver(&self) -> Option> { + self.rx.lock().take() + } + + fn arm_opening(&self) { + self.gate.arm(); + } + + fn close(&self) { + self.gate.shut(); + self.tx.lock().take(); + let _ = self.rx.lock().take(); + } +} + +impl InMemoryEndpoint { + async fn deliver(&self, tone: Tone) -> Result<()> { if self.killed.load(Ordering::SeqCst) { return Err(anyhow::anyhow!("link to {} is dead", self.peer.id.short())); } @@ -389,13 +448,4 @@ impl PeerConnection for InMemoryEndpoint { } Ok(()) } - - fn take_receiver(&self) -> Option> { - self.rx.lock().take() - } - - fn close(&self) { - self.tx.lock().take(); - let _ = self.rx.lock().take(); - } } diff --git a/ensemble/src/liveness.rs b/ensemble/src/liveness.rs index 337609a4..614531d4 100644 --- a/ensemble/src/liveness.rs +++ b/ensemble/src/liveness.rs @@ -1,4 +1,3 @@ - use std::time::Duration; use crate::{Hid, Tone}; diff --git a/ensemble/src/opening.rs b/ensemble/src/opening.rs new file mode 100644 index 00000000..dc5776bb --- /dev/null +++ b/ensemble/src/opening.rs @@ -0,0 +1,145 @@ +use std::sync::atomic::{AtomicBool, Ordering}; + +use parking_lot::Mutex; +use tokio::sync::Notify; + +/// `arm` is synchronous so it can be called from a non-async `install`. +/// Every later `send` parks until the opening frame has landed, and a +/// shut gate never reopens. +#[derive(Debug)] +pub struct Gate { + open: Mutex, + arrived: Notify, + closed: AtomicBool, +} + +impl Gate { + pub fn new() -> Self { + Self { + open: Mutex::new(true), + arrived: Notify::new(), + closed: AtomicBool::new(false), + } + } + + pub fn arm(&self) { + *self.open.lock() = false; + } + + pub fn is_open(&self) -> bool { + *self.open.lock() + } + + pub async fn wait(&self) { + loop { + // Register before reading the flag: `notify_waiters` reaches + // only waiters already enrolled, so checking first can drop a + // wakeup and park forever. + if self.is_open() || self.closed.load(Ordering::SeqCst) { + return; + } + self.arrived.notified().await; + } + } + + pub fn opened(&self) { + if self.closed.load(Ordering::SeqCst) { + return; + } + *self.open.lock() = true; + self.arrived.notify_waiters(); + } + + pub fn shut(&self) { + self.closed.store(true, Ordering::SeqCst); + self.arrived.notify_waiters(); + } +} + +impl Default for Gate { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::time::Duration; + + #[tokio::test] + async fn an_armed_gate_parks_until_it_opens() { + let gate = Arc::new(Gate::new()); + gate.arm(); + assert!(!gate.is_open()); + + let waiter = tokio::spawn({ + let gate = gate.clone(); + async move { + gate.wait().await; + "through" + } + }); + + tokio::time::sleep(Duration::from_millis(20)).await; + assert!(!waiter.is_finished(), "an armed gate must hold traffic"); + + gate.opened(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .unwrap() + .unwrap(), + "through" + ); + } + + #[tokio::test] + async fn a_gate_that_is_never_armed_passes_immediately() { + let gate = Gate::new(); + tokio::time::timeout(Duration::from_millis(50), gate.wait()) + .await + .expect("an open gate must not park"); + } + + #[tokio::test] + async fn shutting_releases_waiters_so_a_stalled_opening_frame_cannot_wedge_a_link() { + let gate = Arc::new(Gate::new()); + gate.arm(); + let waiter = tokio::spawn({ + let gate = gate.clone(); + async move { + gate.wait().await; + "released" + } + }); + tokio::time::sleep(Duration::from_millis(20)).await; + gate.shut(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .unwrap() + .unwrap(), + "released" + ); + } + + #[tokio::test] + async fn a_late_opening_frame_cannot_resurrect_a_shut_gate() { + let gate = Gate::new(); + gate.arm(); + gate.shut(); + gate.opened(); + assert!(!gate.is_open(), "a closed link must stay closed"); + tokio::time::timeout(Duration::from_millis(50), gate.wait()) + .await + .expect("a shut gate must not park"); + } + + #[tokio::test] + async fn a_default_gate_starts_open() { + let gate = Gate::default(); + assert!(gate.is_open(), "default must match new"); + } +} diff --git a/ensemble/src/send.rs b/ensemble/src/send.rs index 2bbfc62f..70c6f8a4 100644 --- a/ensemble/src/send.rs +++ b/ensemble/src/send.rs @@ -1,8 +1,9 @@ - -use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; +use tokio::task::JoinSet; + use crate::{PeerConnection, Tone}; pub const SEND_TIMEOUT: Duration = Duration::from_secs(5); @@ -35,6 +36,27 @@ impl SendStats { } } +/// Resolves when every send has landed or hit its deadline. +pub async fn fanout( + sends: Vec<(Arc, Tone)>, + stats: &Arc, +) -> Vec { + let mut set = JoinSet::new(); + for (conn, tone) in sends { + let stats = stats.clone(); + set.spawn(async move { send_bounded(&conn, tone, &stats).await }); + } + let mut out = Vec::new(); + while let Some(joined) = set.join_next().await { + match joined { + Ok(Err(e)) => out.push(e), + Ok(Ok(())) => {} + Err(e) => tracing::error!(target: "ensemble.send", error = %e, "fanout.task.panicked"), + } + } + out +} + pub async fn send_bounded( conn: &Arc, tone: Tone, @@ -69,6 +91,33 @@ pub async fn send_bounded_with( } } +pub async fn send_opening_bounded( + conn: &Arc, + tone: Tone, + stats: &SendStats, + deadline: Duration, +) -> Result<(), SendError> { + // `send_opening`, not `send`: `send` parks on the gate this frame opens. + match tokio::time::timeout(deadline, conn.send_opening(tone)).await { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => { + stats.failed.fetch_add(1, Ordering::SeqCst); + Err(SendError::Failed(e.to_string())) + } + Err(_) => { + stats.timed_out.fetch_add(1, Ordering::SeqCst); + tracing::warn!( + target: "ensemble.send", + peer = %conn.peer().id.short(), + timeout_ms = deadline.as_millis() as u64, + "opening.timed_out: peer never took the hello, closing" + ); + conn.close(); + Err(SendError::TimedOut) + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -109,7 +158,10 @@ mod tests { fn conn(stalls: bool) -> Arc { Arc::new(Stalled { - addr: HumdAddr { id: Hid::random_humd(), hints: vec![] }, + addr: HumdAddr { + id: Hid::random_humd(), + hints: vec![], + }, caps: PeerCapabilities::default(), stalled: AtomicBool::new(stalls), closed: AtomicBool::new(false), @@ -121,6 +173,98 @@ mod tests { c.clone() } + struct Slow { + addr: HumdAddr, + caps: PeerCapabilities, + delay: Duration, + sent: AtomicU64, + } + + #[async_trait::async_trait] + impl PeerConnection for Slow { + fn peer(&self) -> &HumdAddr { + &self.addr + } + fn capabilities(&self) -> &PeerCapabilities { + &self.caps + } + async fn send(&self, _tone: Tone) -> anyhow::Result<()> { + tokio::time::sleep(self.delay).await; + self.sent.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + fn take_receiver(&self) -> Option> { + None + } + fn close(&self) {} + } + + #[tokio::test] + async fn a_stalled_opening_hits_its_deadline_and_closes_the_link() { + let c = conn(false); + c.stalled.store(true, Ordering::SeqCst); + let stats = SendStats::new(); + + let got = tokio::time::timeout( + Duration::from_secs(10), + send_opening_bounded( + &as_conn(&c), + serde_json::json!({"chi": "hello"}), + &stats, + Duration::from_millis(50), + ), + ) + .await + .expect("a stalled opening must return, not hang"); + + assert_eq!(got, Err(SendError::TimedOut)); + assert!(c.closed.load(Ordering::SeqCst), "a stalled opening must close the link"); + assert_eq!(stats.timed_out(), 1, "the timeout must be accounted for"); + } + + #[tokio::test] + async fn a_fanout_costs_the_slowest_peer_not_the_sum() { + let delay = Duration::from_millis(120); + let peers = 5; + let conns: Vec> = (0..peers) + .map(|_| { + Arc::new(Slow { + addr: HumdAddr { + id: Hid::random_humd(), + hints: vec![], + }, + caps: PeerCapabilities::default(), + delay, + sent: AtomicU64::new(0), + }) + }) + .collect(); + + let sends: Vec<(Arc, Tone)> = conns + .iter() + .map(|c| { + ( + c.clone() as Arc, + serde_json::json!({"chi": "chunk"}), + ) + }) + .collect(); + + let started = std::time::Instant::now(); + let errors = fanout(sends, &SendStats::new()).await; + let elapsed = started.elapsed(); + + assert!(errors.is_empty(), "healthy peers must not report errors"); + for c in &conns { + assert_eq!(c.sent.load(Ordering::SeqCst), 1, "every peer got the tone"); + } + assert!( + elapsed < delay * 2, + "a serial fanout would take {}ms; took {elapsed:?}", + delay.as_millis() * peers + ); + } + #[tokio::test] async fn a_stalled_peer_times_out_rather_than_hanging() { let c = conn(false); @@ -131,13 +275,21 @@ mod tests { let started = std::time::Instant::now(); let got = tokio::time::timeout( Duration::from_secs(10), - send_bounded_with(&as_conn(&c), serde_json::json!({"chi": "chunk"}), &stats, short), + send_bounded_with( + &as_conn(&c), + serde_json::json!({"chi": "chunk"}), + &stats, + short, + ), ) .await .expect("send_bounded must return, not hang"); assert_eq!(got, Err(SendError::TimedOut)); - assert!(started.elapsed() < Duration::from_secs(5), "returned promptly"); + assert!( + started.elapsed() < Duration::from_secs(5), + "returned promptly" + ); } #[tokio::test] @@ -152,7 +304,10 @@ mod tests { Duration::from_millis(50), ) .await; - assert!(c.closed.load(Ordering::SeqCst), "a stalled peer must be closed"); + assert!( + c.closed.load(Ordering::SeqCst), + "a stalled peer must be closed" + ); assert_eq!(stats.timed_out(), 1); } @@ -160,13 +315,20 @@ mod tests { async fn a_healthy_peer_sends_and_is_not_counted() { let c = conn(false); let stats = SendStats::new(); - let got = - send_bounded_with(&as_conn(&c), serde_json::json!({"chi": "chunk"}), &stats, SEND_TIMEOUT) - .await; + let got = send_bounded_with( + &as_conn(&c), + serde_json::json!({"chi": "chunk"}), + &stats, + SEND_TIMEOUT, + ) + .await; assert_eq!(got, Ok(())); assert_eq!(*c.sends.lock(), 1); assert_eq!(stats.timed_out(), 0); - assert!(!c.closed.load(Ordering::SeqCst), "a working peer stays open"); + assert!( + !c.closed.load(Ordering::SeqCst), + "a working peer stays open" + ); } #[tokio::test] @@ -188,17 +350,22 @@ mod tests { } fn close(&self) {} } - let c: Arc = - Arc::new(Broken( - HumdAddr { id: Hid::random_humd(), hints: vec![] }, - PeerCapabilities::default(), - )); + let c: Arc = Arc::new(Broken( + HumdAddr { + id: Hid::random_humd(), + hints: vec![], + }, + PeerCapabilities::default(), + )); let stats = SendStats::new(); - let got = send_bounded_with(&c, serde_json::json!({"chi": "chunk"}), &stats, SEND_TIMEOUT).await; - assert!( - matches!(got, Err(SendError::Failed(_))), - "got {got:?}" - ); + let got = send_bounded_with( + &c, + serde_json::json!({"chi": "chunk"}), + &stats, + SEND_TIMEOUT, + ) + .await; + assert!(matches!(got, Err(SendError::Failed(_))), "got {got:?}"); assert_eq!(stats.failed(), 1); assert_eq!(stats.timed_out(), 0, "a failure is not a stall"); } diff --git a/ensemble/src/tcp.rs b/ensemble/src/tcp.rs index e9e6610e..d7717450 100644 --- a/ensemble/src/tcp.rs +++ b/ensemble/src/tcp.rs @@ -1,157 +1,103 @@ -//! Real-wire TCP transport for the ensemble. -//! -//! NDJSON over `tokio::net::TcpStream`. One framing rule: each tone is -//! a single JSON value followed by `\n`. Mirrors the unix-socket shape -//! humd's thrum module uses (`humd/src/thrumd/conn.rs`) — same parser, different socket. -//! -//! Auth lives one layer up: the ensemble's signed `chi:"hello"` exchange -//! authenticates the peer once a connection is installed. This module -//! only carries bytes. -//! -//! Shape: -//! - [`TcpEndpoint`] — one live [`PeerConnection`] backed by a TCP -//! stream. Spawns a reader task that parses NDJSON into the receiver -//! mpsc; `send` serialises + writes under a `tokio::Mutex` so -//! concurrent senders can't interleave half-lines. -//! - [`TcpListener`] — accepts inbound connections. The accepted -//! endpoint starts with a placeholder [`HumdAddr`]; the ensemble -//! drainer learns the real id once the peer's first hello arrives. -//! - [`TcpTransport`] — [`Transport`] impl. `connect` dials the first -//! `tcp:host:port` hint in the [`HumdAddr`]. +//! NDJSON over TCP. Framing and the frame ceiling live in [`crate::framing`]. +//! Identity is settled one layer up by the signed `chi:"hello"` exchange. use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; -use anyhow::{anyhow, Result}; +use anyhow::{Result, anyhow}; use async_trait::async_trait; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::net::TcpStream; -use tokio::sync::{mpsc, Mutex}; +use tokio::net::tcp::OwnedWriteHalf; +use tokio::sync::{Mutex, mpsc}; -use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport}; +use crate::framing; +use crate::opening::Gate; +use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport}; -/// Inbound-channel capacity — matches `InMemoryEndpoint`. const RECV_CAP: usize = 256; -/// One TCP-backed peer link. Reader task drains NDJSON lines into the -/// receiver mpsc; `send` writes a serialised tone + newline under a -/// mutex (NDJSON framing requires atomic line writes). pub struct TcpEndpoint { peer: HumdAddr, caps: PeerCapabilities, - /// Serialises concurrent writers. `Option` so `close` can drop the - /// write half and subsequent sends fail cleanly. - writer: Mutex>, - /// Inbound stream. Take-once. + writer: Arc>>, rx: parking_lot::Mutex>>, + closed: AtomicBool, + gate: Gate, } impl TcpEndpoint { - /// Dial a remote `host:port` and wrap the resulting stream. - pub async fn connect( - addr: &str, - peer: HumdAddr, - caps: PeerCapabilities, - ) -> Result> { + pub async fn connect(addr: &str, peer: HumdAddr, caps: PeerCapabilities) -> Result> { let stream = TcpStream::connect(addr).await?; - // Disable Nagle: tones are small and latency-sensitive, and the - // NDJSON framing means we never want a partial line stalling in - // the kernel waiting for more bytes. let _ = stream.set_nodelay(true); Ok(Self::from_stream(stream, peer, caps)) } - /// Wrap an already-accepted stream. Used by `TcpListener::accept`. - pub fn from_stream( - stream: TcpStream, - peer: HumdAddr, - caps: PeerCapabilities, - ) -> Arc { + pub fn from_stream(stream: TcpStream, peer: HumdAddr, caps: PeerCapabilities) -> Arc { let _ = stream.set_nodelay(true); let (read_half, write_half) = stream.into_split(); let (tx, rx) = mpsc::channel::(RECV_CAP); let me = Arc::new(Self { peer, caps, - writer: Mutex::new(Some(write_half)), + writer: Arc::new(Mutex::new(Some(write_half))), rx: parking_lot::Mutex::new(Some(rx)), + closed: AtomicBool::new(false), + gate: Gate::new(), }); - tokio::spawn(read_loop(read_half, tx)); + tokio::spawn(framing::pump(read_half, tx, "tcp")); me } + async fn write(&self, tone: Tone) -> Result<()> { + let mut guard = self.writer.lock().await; + if self.closed.load(Ordering::SeqCst) { + return Err(anyhow!("tcp send: link closed")); + } + let w = guard + .as_mut() + .ok_or_else(|| anyhow!("tcp send: writer closed"))?; + framing::write_frame(w, &tone).await + } } #[async_trait] impl PeerConnection for TcpEndpoint { - fn peer(&self) -> &HumdAddr { &self.peer } - fn capabilities(&self) -> &PeerCapabilities { &self.caps } + fn peer(&self) -> &HumdAddr { + &self.peer + } + fn capabilities(&self) -> &PeerCapabilities { + &self.caps + } async fn send(&self, tone: Tone) -> Result<()> { - let mut line = serde_json::to_vec(&tone) - .map_err(|e| anyhow!("tcp send: serialize: {e}"))?; - line.push(b'\n'); - let mut guard = self.writer.lock().await; - let w = guard - .as_mut() - .ok_or_else(|| anyhow!("tcp send: writer closed"))?; - w.write_all(&line) - .await - .map_err(|e| anyhow!("tcp send: write: {e}"))?; - Ok(()) + self.gate.wait().await; + self.write(tone).await + } + + async fn send_opening(&self, tone: Tone) -> Result<()> { + let result = self.write(tone).await; + self.gate.opened(); + result } fn take_receiver(&self) -> Option> { self.rx.lock().take() } - fn close(&self) { - // Drop the writer half and the receiver. Best-effort, idempotent. - // We can't await shutdown from a sync method, so spawn a task to - // gracefully close the write half if it's still present. - let writer_slot = { - let mut guard = self.writer.try_lock(); - match guard { - Ok(ref mut g) => g.take(), - Err(_) => None, // contended — let the holder finish; their write will fail next - } - }; - if let Some(mut w) = writer_slot { - tokio::spawn(async move { - let _ = w.shutdown().await; - }); - } - let _ = self.rx.lock().take(); + fn arm_opening(&self) { + self.gate.arm(); } -} -/// Reader loop — parses NDJSON lines into `Tone`s and forwards. Exits on -/// EOF, read error, or when the receiver is dropped. -async fn read_loop(read_half: OwnedReadHalf, tx: mpsc::Sender) { - let mut lines = BufReader::new(read_half).lines(); - loop { - let line = match lines.next_line().await { - Ok(Some(l)) => l, - Ok(None) => break, // EOF - Err(e) => { - tracing::trace!(target: "ensemble.tcp", err = %e, "tcp.read.failed"); - break; - } - }; - if line.is_empty() { - continue; - } - let tone: Tone = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { - tracing::trace!(target: "ensemble.tcp", err = %e, "tcp.parse.failed"); - continue; + fn close(&self) { + self.closed.store(true, Ordering::SeqCst); + self.gate.shut(); + let writer = self.writer.clone(); + tokio::spawn(async move { + if let Some(mut w) = writer.lock().await.take() { + let _ = tokio::io::AsyncWriteExt::shutdown(&mut w).await; } - }; - if tx.send(tone).await.is_err() { - // Receiver dropped — endpoint closed. - break; - } + }); + let _ = self.rx.lock().take(); } } @@ -195,11 +141,15 @@ impl TcpListener { pub struct TcpTransport; impl TcpTransport { - pub fn new() -> Self { Self } + pub fn new() -> Self { + Self + } } impl Default for TcpTransport { - fn default() -> Self { Self::new() } + fn default() -> Self { + Self::new() + } } #[async_trait] @@ -210,7 +160,8 @@ impl Transport for TcpTransport { .iter() .find_map(|h| h.strip_prefix("tcp:")) .ok_or_else(|| anyhow!("tcp transport: no tcp: hint in HumdAddr"))?; - let endpoint = TcpEndpoint::connect(hint, addr.clone(), PeerCapabilities::default()).await?; + let endpoint = + TcpEndpoint::connect(hint, addr.clone(), PeerCapabilities::default()).await?; Ok(endpoint as Arc) } } diff --git a/ensemble/src/tls.rs b/ensemble/src/tls.rs index 1b231b85..75a219c1 100644 --- a/ensemble/src/tls.rs +++ b/ensemble/src/tls.rs @@ -1,61 +1,34 @@ -//! TLS-over-TCP transport for the ensemble. +//! TLS-over-TCP transport for the ensemble. Sibling of [`crate::tcp`]: +//! same framing, same contracts, wrapped in a [`tokio_rustls::TlsStream`]. //! -//! Sibling of [`crate::tcp`] — same NDJSON framing, same `Transport` + -//! `PeerConnection` contract, but the underlying `TcpStream` is wrapped -//! in a [`tokio_rustls::TlsStream`]. T2+ peers run over the public -//! internet; plaintext is the loopback/LAN path only. -//! -//! ## Trust model — pinned fingerprints, no CA -//! -//! Self-signed cert on the server. The dialer carries the expected -//! cert fingerprint (SHA-256 of the DER bytes) and accepts ONLY that -//! one cert — see [`PinnedFingerprintVerifier`]. This matches the T2 -//! "known peers" model: trust is established out of band when you put -//! a peer's fingerprint in `peers.json`; the TLS layer is just a -//! confidentiality + integrity tunnel that proves the box on the other -//! end really holds the private key for the fingerprint you wrote down. -//! No CA, no chain validation, no SNI matching, no expiry handling — -//! those belong to T4 with a real PKI. -//! -//! Shape: -//! - [`TlsTcpEndpoint`] — one live [`PeerConnection`] over a TLS-wrapped -//! TCP stream. Reader task drains NDJSON into the receiver mpsc; -//! `send` writes a serialised tone + newline under a `tokio::Mutex`. -//! - [`TlsTcpListener`] — accepts inbound TLS connections. The -//! accepted endpoint starts with a placeholder [`HumdAddr`]; the -//! ensemble drainer learns the real id once the peer's first hello -//! arrives — same as plaintext. -//! - [`TlsTcpTransport`] — [`Transport`] impl. `connect` dials the -//! first `tls:host:port` hint in the [`HumdAddr`], using a -//! pre-configured client config that pins one fingerprint. +//! Trust model: pinned fingerprints, no CA, no chain, no SNI, no expiry. +//! The dialer accepts exactly one cert — the SHA-256 of its DER bytes +//! written down out of band in `peers.json`. See [`PinnedFingerprintVerifier`]. use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; -use anyhow::{anyhow, Result}; +use anyhow::{Result, anyhow}; use async_trait::async_trait; use sha2::{Digest, Sha256}; -use tokio::io::{split, AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf}; +use tokio::io::{WriteHalf, split}; use tokio::net::TcpStream; -use tokio::sync::{mpsc, Mutex}; +use tokio::sync::{Mutex, mpsc}; use tokio_rustls::{TlsAcceptor, TlsConnector, TlsStream}; use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; use rustls::{ClientConfig, DigitallySignedStruct, ServerConfig, SignatureScheme}; -use crate::{HumdAddr, Hid, PeerCapabilities, PeerConnection, Tone, Transport}; +use crate::framing; +use crate::opening::Gate; +use crate::{Hid, HumdAddr, PeerCapabilities, PeerConnection, Tone, Transport}; -/// Inbound-channel capacity — matches `InMemoryEndpoint` / `TcpEndpoint`. const RECV_CAP: usize = 256; -/// Hint prefix on a [`HumdAddr`] that signals "dial this `host:port` -/// over TLS-pinned TCP." Sibling of `tcp:` so peers can advertise both -/// plaintext and TLS endpoints and dialers pick the right one. pub const TLS_HINT: &str = "tls:"; -/// SHA-256 of a certificate's DER bytes. The bytes used for pinning in -/// `peers.json` — write this on the server, paste it on the client. pub fn cert_fingerprint(cert: &CertificateDer<'_>) -> [u8; 32] { let mut h = Sha256::new(); h.update(cert.as_ref()); @@ -67,22 +40,16 @@ pub fn cert_fingerprint(cert: &CertificateDer<'_>) -> [u8; 32] { // ── Endpoint ─────────────────────────────────────────────────────────────── -/// One TLS-over-TCP peer link. Same NDJSON framing as the plaintext -/// path — the TLS layer is invisible to the framing code, we just hand -/// it a `TlsStream` instead of a `TcpStream`. pub struct TlsTcpEndpoint { peer: HumdAddr, caps: PeerCapabilities, - /// Serialises concurrent writers. `Option` so `close` can drop the - /// write half and subsequent sends fail cleanly. - writer: Mutex>>>, - /// Inbound stream. Take-once. + writer: Arc>>>>, rx: parking_lot::Mutex>>, + closed: AtomicBool, + gate: Gate, } impl TlsTcpEndpoint { - /// Wrap an already-handshaken TLS stream. Used by both the client - /// and listener paths once rustls has finished the handshake. pub fn from_stream( stream: TlsStream, peer: HumdAddr, @@ -93,12 +60,24 @@ impl TlsTcpEndpoint { let me = Arc::new(Self { peer, caps, - writer: Mutex::new(Some(write_half)), + writer: Arc::new(Mutex::new(Some(write_half))), rx: parking_lot::Mutex::new(Some(rx)), + closed: AtomicBool::new(false), + gate: Gate::new(), }); - tokio::spawn(read_loop(read_half, tx)); + tokio::spawn(framing::pump(read_half, tx, "tls")); me } + async fn write(&self, tone: Tone) -> Result<()> { + let mut guard = self.writer.lock().await; + if self.closed.load(Ordering::SeqCst) { + return Err(anyhow!("tls send: link closed")); + } + let w = guard + .as_mut() + .ok_or_else(|| anyhow!("tls send: writer closed"))?; + framing::write_frame(w, &tone).await + } /// Dial a remote `host:port` with a fingerprint-pinned TLS config. /// `expected_fingerprint` is the SHA-256 of the peer's cert DER — @@ -135,68 +114,34 @@ impl PeerConnection for TlsTcpEndpoint { } async fn send(&self, tone: Tone) -> Result<()> { - let mut line = serde_json::to_vec(&tone) - .map_err(|e| anyhow!("tls send: serialize: {e}"))?; - line.push(b'\n'); - let mut guard = self.writer.lock().await; - let w = guard - .as_mut() - .ok_or_else(|| anyhow!("tls send: writer closed"))?; - w.write_all(&line) - .await - .map_err(|e| anyhow!("tls send: write: {e}"))?; - Ok(()) + self.gate.wait().await; + self.write(tone).await + } + + async fn send_opening(&self, tone: Tone) -> Result<()> { + let result = self.write(tone).await; + self.gate.opened(); + result } fn take_receiver(&self) -> Option> { self.rx.lock().take() } - fn close(&self) { - // Drop the writer half and the receiver. Best-effort, idempotent. - let slot = match self.writer.try_lock() { - Ok(mut g) => g.take(), - Err(_) => None, // contended — holder's next write fails when peer drops - }; - if let Some(mut w) = slot { - tokio::spawn(async move { - let _ = w.shutdown().await; - }); - } - let _ = self.rx.lock().take(); + fn arm_opening(&self) { + self.gate.arm(); } -} -/// Reader loop — parses NDJSON lines off the TLS read half and forwards -/// them. Identical control flow to the plaintext path; the TLS framing -/// is fully encapsulated inside `TlsStream`. -async fn read_loop( - read_half: ReadHalf>, - tx: mpsc::Sender, -) { - let mut lines = BufReader::new(read_half).lines(); - loop { - let line = match lines.next_line().await { - Ok(Some(l)) => l, - Ok(None) => break, // EOF - Err(e) => { - tracing::trace!(target: "ensemble.tls", err = %e, "tls.read.failed"); - break; - } - }; - if line.is_empty() { - continue; - } - let tone: Tone = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { - tracing::trace!(target: "ensemble.tls", err = %e, "tls.parse.failed"); - continue; + fn close(&self) { + self.closed.store(true, Ordering::SeqCst); + self.gate.shut(); + let writer = self.writer.clone(); + tokio::spawn(async move { + if let Some(mut w) = writer.lock().await.take() { + let _ = tokio::io::AsyncWriteExt::shutdown(&mut w).await; } - }; - if tx.send(tone).await.is_err() { - break; - } + }); + let _ = self.rx.lock().take(); } } @@ -409,8 +354,8 @@ impl Transport for TlsTcpTransport { .iter() .find_map(|h| h.strip_prefix(TLS_FP_HINT)) .ok_or_else(|| anyhow!("tls transport: no `tls-fp:` hint in HumdAddr"))?; - let fp_bytes = hex::decode(fp_hex) - .map_err(|e| anyhow!("tls transport: fingerprint hex: {e}"))?; + let fp_bytes = + hex::decode(fp_hex).map_err(|e| anyhow!("tls transport: fingerprint hex: {e}"))?; if fp_bytes.len() != 32 { return Err(anyhow!( "tls transport: fingerprint must be 32 bytes (got {})", @@ -419,13 +364,9 @@ impl Transport for TlsTcpTransport { } let mut fp = [0u8; 32]; fp.copy_from_slice(&fp_bytes); - let endpoint = TlsTcpEndpoint::connect( - host_port, - addr.clone(), - PeerCapabilities::default(), - fp, - ) - .await?; + let endpoint = + TlsTcpEndpoint::connect(host_port, addr.clone(), PeerCapabilities::default(), fp) + .await?; Ok(endpoint as Arc) } } diff --git a/ensemble/tests/gossip_integration.rs b/ensemble/tests/gossip_integration.rs index 4af0ec32..22f11d83 100644 --- a/ensemble/tests/gossip_integration.rs +++ b/ensemble/tests/gossip_integration.rs @@ -1,9 +1,8 @@ - use std::time::Duration; use ensemble::{ - gossip::{gossip_tone, mint_msg_id, GOSSIP_CHI}, Ensemble, HumdKey, InMemoryEndpoint, PeerCapabilities, + gossip::{GOSSIP_CHI, gossip_tone, mint_msg_id}, }; use serde_json::json; use tokio::time::timeout; @@ -22,14 +21,8 @@ async fn gossip_percolates_one_hop_and_dedupes_duplicates() { ..Default::default() }; - let (a_to_b, b_to_a) = InMemoryEndpoint::pair( - a_id, caps.clone(), - b_id, caps.clone(), - ); - let (b_to_c, c_to_b) = InMemoryEndpoint::pair( - b_id, caps.clone(), - c_id, caps.clone(), - ); + let (a_to_b, b_to_a) = InMemoryEndpoint::pair(a_id, caps.clone(), b_id, caps.clone()); + let (b_to_c, c_to_b) = InMemoryEndpoint::pair(b_id, caps.clone(), c_id, caps.clone()); let ens_a = Ensemble::new(a_id); let ens_b = Ensemble::new(b_id); @@ -55,12 +48,12 @@ async fn gossip_percolates_one_hop_and_dedupes_duplicates() { let x_key = HumdKey::generate(); let x_id = x_key.hid(); - let (x_to_b, b_to_x) = InMemoryEndpoint::pair( - x_id, caps.clone(), - b_id, caps.clone(), - ); + let (x_to_b, b_to_x) = InMemoryEndpoint::pair(x_id, caps.clone(), b_id, caps.clone()); ens_b.install(b_to_x, caps.clone(), &b_key); + let ens_x = Ensemble::new(x_id); + ens_x.install(x_to_b.clone(), caps.clone(), &x_key); tokio::time::sleep(Duration::from_millis(20)).await; + assert!(ens_b.handshake_done(&x_id), "B must complete X's handshake"); while let Ok(Ok(_)) = timeout(Duration::from_millis(1), sub_c.recv()).await {} diff --git a/ensemble/tests/iroh_integration.rs b/ensemble/tests/iroh_integration.rs index 6fb4b046..f790a6fa 100644 --- a/ensemble/tests/iroh_integration.rs +++ b/ensemble/tests/iroh_integration.rs @@ -1,9 +1,8 @@ - use std::time::Duration; use ed25519_dalek::SigningKey; use ensemble::{ - Ensemble, HumdAddr, Hid, HumdKey, IrohTransport, PeerCapabilities, PeerConnection, Transport, + Ensemble, Hid, HumdAddr, HumdKey, IrohTransport, PeerCapabilities, PeerConnection, Transport, }; use serde_json::json; @@ -41,9 +40,9 @@ impl Skip { fn resolve(self) -> Option<()> { let forced = std::env::var(REQUIRE_ENV).is_ok_and(|v| v != "0" && !v.is_empty()); match self { - Skip::Environmental(reason) if forced => panic!( - "{REQUIRE_ENV} is set, so this failure cannot be skipped: {reason}" - ), + Skip::Environmental(reason) if forced => { + panic!("{REQUIRE_ENV} is set, so this failure cannot be skipped: {reason}") + } Skip::Environmental(reason) => { eprintln!("iroh_integration: SKIPPED — {reason}"); eprintln!( @@ -106,8 +105,12 @@ async fn iroh_endpoint_routes_tones_both_ways() { ); } - let server_key = HumdKey(SigningKey::from_bytes(&server.endpoint().secret_key().to_bytes())); - let client_key = HumdKey(SigningKey::from_bytes(&client.endpoint().secret_key().to_bytes())); + let server_key = HumdKey(SigningKey::from_bytes( + &server.endpoint().secret_key().to_bytes(), + )); + let client_key = HumdKey(SigningKey::from_bytes( + &client.endpoint().secret_key().to_bytes(), + )); assert_eq!(server_key.hid(), server_humd_id); assert_eq!(client_key.hid(), client_humd_id); @@ -140,6 +143,7 @@ async fn iroh_endpoint_routes_tones_both_ways() { "chi": "ping", "rid": "iroh-pong-1", "to": client_humd_for_task.to_hex(), + "from": server_humd_id.to_hex(), }); ensemble .route(pong) @@ -150,10 +154,8 @@ async fn iroh_endpoint_routes_tones_both_ways() { Ok::<(), String>(()) }); - let mut server_humd_addr = HumdAddr::new(server_humd_id).with_hint(format!( - "iroh:{}", - hex::encode(server_node_id.as_bytes()) - )); + let mut server_humd_addr = HumdAddr::new(server_humd_id) + .with_hint(format!("iroh:{}", hex::encode(server_node_id.as_bytes()))); for hint in server_sockets { server_humd_addr = server_humd_addr.with_hint(hint); } @@ -174,11 +176,9 @@ async fn iroh_endpoint_routes_tones_both_ways() { "chi": "perf-mark", "rid": "iroh-mark-1", "to": server_humd_id.to_hex(), + "from": client_humd_id.to_hex(), }); - ensemble - .route(mark) - .await - .expect("route perf-mark"); + ensemble.route(mark).await.expect("route perf-mark"); let mut server_task = server_task; let got = tokio::select! { @@ -211,7 +211,7 @@ async fn iroh_endpoint_routes_tones_both_ways() { } mod classify { - use super::{force_transport_tests, Skip, REQUIRE_ENV}; + use super::{REQUIRE_ENV, Skip, force_transport_tests}; #[test] fn sandbox_signatures_are_environmental() { @@ -246,9 +246,8 @@ mod classify { #[test] fn a_real_failure_panics_even_without_the_env_var() { - let r = std::panic::catch_unwind(|| { - Skip::Real("bind: quic handshake failed".into()).resolve() - }); + let r = + std::panic::catch_unwind(|| Skip::Real("bind: quic handshake failed".into()).resolve()); assert!(r.is_err(), "a real failure must never resolve to a skip"); } diff --git a/ensemble/tests/tcp_integration.rs b/ensemble/tests/tcp_integration.rs index 5ab13900..1cbe8ae4 100644 --- a/ensemble/tests/tcp_integration.rs +++ b/ensemble/tests/tcp_integration.rs @@ -6,9 +6,7 @@ use std::time::Duration; -use ensemble::{ - Ensemble, HumdAddr, HumdKey, PeerCapabilities, TcpEndpoint, TcpListener, -}; +use ensemble::{Ensemble, HumdAddr, HumdKey, PeerCapabilities, TcpEndpoint, TcpListener}; use serde_json::json; #[tokio::test] @@ -82,7 +80,7 @@ async fn tcp_endpoint_routes_tone_across_wire() { // Route a ping from A → B. The first tone over the wire is A's // hello (absorbed by B's drainer), the second is this ping (fans // out to subscribers). - let ping = json!({"chi": "ping", "rid": "tcp-1", "to": b_id.to_hex()}); + let ping = json!({"chi": "ping", "rid": "tcp-1", "to": b_id.to_hex(), "from": a_id.to_hex()}); ensemble_a.route(ping).await.expect("route ping"); // Wait for the ping on B's subscribe channel. diff --git a/ensemble/tests/tls_integration.rs b/ensemble/tests/tls_integration.rs index 614cc9ed..fc8e0a02 100644 --- a/ensemble/tests/tls_integration.rs +++ b/ensemble/tests/tls_integration.rs @@ -13,8 +13,7 @@ use std::sync::Arc; use std::time::Duration; use ensemble::{ - cert_fingerprint, Ensemble, HumdAddr, HumdKey, PeerCapabilities, TlsTcpEndpoint, - TlsTcpListener, + Ensemble, HumdAddr, HumdKey, PeerCapabilities, TlsTcpEndpoint, TlsTcpListener, cert_fingerprint, }; use rcgen::generate_simple_self_signed; use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; @@ -92,7 +91,7 @@ async fn tls_endpoint_routes_tone_across_pinned_wire() { // Route a ping from A → B. First tone is A's hello (drained by B); // second is this ping which fans out to subscribers. - let ping = json!({"chi": "ping", "rid": "tls-1", "to": b_id.to_hex()}); + let ping = json!({"chi": "ping", "rid": "tls-1", "to": b_id.to_hex(), "from": a_id.to_hex()}); ensemble_a.route(ping).await.expect("route ping"); let got = tokio::time::timeout(Duration::from_secs(2), sub_b.recv()) diff --git a/hum-paths/src/config.rs b/hum-paths/src/config.rs index bcd67f2e..590d2aa2 100644 --- a/hum-paths/src/config.rs +++ b/hum-paths/src/config.rs @@ -40,19 +40,14 @@ impl Default for HumdSection { // ── fs ──────────────────────────────────────────────────────────────────── -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum FsMode { + #[default] Rw, Ro, } -impl Default for FsMode { - fn default() -> Self { - FsMode::Rw - } -} - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct FsRoot { /// Tilde-expanded at load time. Always stored as an absolute path. diff --git a/hum-paths/src/lib.rs b/hum-paths/src/lib.rs index db582f69..f508b164 100644 --- a/hum-paths/src/lib.rs +++ b/hum-paths/src/lib.rs @@ -72,6 +72,7 @@ pub const RUNTIME_INFO_BASENAME: &str = "runtime.json"; pub const HUM_JSON_BASENAME: &str = "hum.json"; pub const PEERS_JSON_BASENAME: &str = "peers.json"; pub const ORCHFILE_BASENAME: &str = "Orchfile"; +pub const ROUTING_JSON_BASENAME: &str = "routing.json"; /// Subdirectory of a hum source tree that holds hive crates. pub const HIVES_SUBDIR: &str = "hives"; /// Subdirectory of a hum source tree that holds recipes (installable bundles). @@ -109,6 +110,11 @@ pub fn thrum_sock_resolved() -> PathBuf { /// humd HTTP control socket. pub fn http_sock() -> PathBuf { runtime_dir().join(HTTP_SOCK_BASENAME) } +/// Kademlia routing table, restored on boot. +pub fn routing_json() -> PathBuf { + state_dir().join(ROUTING_JSON_BASENAME) +} + /// Penny lifetime counters. pub fn penny() -> PathBuf { runtime_dir().join(PENNY_BASENAME) } @@ -237,10 +243,10 @@ pub fn daemon_logs(name: &str) -> DaemonLogs { #[cfg(target_os = "macos")] { let base = home().join("Library/Logs"); - return DaemonLogs::Files { + DaemonLogs::Files { stdout: base.join(format!("sh.hum.{name}.out.log")), stderr: base.join(format!("sh.hum.{name}.err.log")), - }; + } } #[cfg(not(target_os = "macos"))] DaemonLogs::Journald { unit: name.to_string() } diff --git a/humd/src/lib.rs b/humd/src/lib.rs index f70976a1..5d60ae27 100644 --- a/humd/src/lib.rs +++ b/humd/src/lib.rs @@ -19,6 +19,7 @@ mod peer_transport; pub mod peers; mod penny; pub mod redial; +mod routing_store; pub mod supervisor; pub mod thrumd; pub use identity::{key_path, load_or_mint_key, read_key}; @@ -39,9 +40,11 @@ pub struct DaemonConfig { pub http_path: PathBuf, pub mcp_addr: std::net::SocketAddr, pub penny_path: PathBuf, + pub routing_path: PathBuf, pub hum_cfg: hum_paths::config::HumConfig, pub cli_path: String, pub penny_persist_interval: Duration, + pub routing_persist_interval: Duration, pub thrum_override: Option, pub ensemble: Option>, pub bind_mcp: bool, @@ -73,9 +76,11 @@ impl DaemonConfig { http_path, mcp_addr: ([127, 0, 0, 1], mcp_port).into(), penny_path: hum_paths::penny(), + routing_path: hum_paths::routing_json(), hum_cfg: hum_paths::config::load(), cli_path: std::env::var("CLAUDE_CLI_PATH").unwrap_or_else(|_| "claude".into()), penny_persist_interval: Duration::from_secs(10), + routing_persist_interval: Duration::from_secs(30), thrum_override: None, ensemble: None, bind_mcp: true, @@ -146,6 +151,11 @@ where }), }; + if let Some(ens) = &ensemble_opt { + routing_store::restore_on_boot(ens, &cfg.routing_path); + routing_store::spawn_persister(ens.clone(), cfg.routing_path.clone(), cfg.routing_persist_interval); + } + let mut peer_reach: Vec = Vec::new(); let mut supervisor: Option = None; if let (Some(ens), Some(key)) = (&ensemble_opt, &cfg.humd_key) { diff --git a/humd/src/routing_store.rs b/humd/src/routing_store.rs new file mode 100644 index 00000000..c32ec413 --- /dev/null +++ b/humd/src/routing_store.rs @@ -0,0 +1,139 @@ +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use ensemble::{Ensemble, HumdAddr}; +use tracing::{info, trace, warn}; + +#[derive(serde::Serialize, serde::Deserialize)] +struct OnDisk { + version: u32, + addrs: Vec, +} + +pub fn load(path: &Path) -> Vec { + let bytes = match std::fs::read(path) { + Ok(b) => b, + Err(e) if e.kind() == io::ErrorKind::NotFound => { + trace!(target: "routing", "routing.load.absent"); + return Vec::new(); + } + Err(e) => { + warn!(target: "routing", "routing.load.read_failed err={e}"); + return Vec::new(); + } + }; + match serde_json::from_slice::(&bytes) { + Ok(disk) => disk.addrs, + Err(e) => { + warn!(target: "routing", "routing.load.parse_failed err={e}"); + Vec::new() + } + } +} + +pub fn save(ensemble: &Ensemble, path: &Path) -> io::Result<()> { + if let Some(dir) = path.parent() { + if !dir.as_os_str().is_empty() { + std::fs::create_dir_all(dir)?; + } + } + let disk = OnDisk { + version: 1, + addrs: ensemble.routing_snapshot(), + }; + let bytes = + serde_json::to_vec(&disk).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, &bytes)?; + std::fs::rename(&tmp, path)?; + Ok(()) +} + +pub fn restore_on_boot(ensemble: &Ensemble, path: &Path) { + let addrs = load(path); + if addrs.is_empty() { + return; + } + let restored = ensemble.routing_restore(addrs); + info!(target: "routing", peers = restored, "routing.restored"); +} + +pub fn spawn_persister(ensemble: Arc, path: PathBuf, interval: Duration) { + tokio::spawn(async move { + let mut tick = tokio::time::interval(interval); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + tick.tick().await; + loop { + tick.tick().await; + if let Err(e) = save(&ensemble, &path) { + warn!(target: "routing", "routing.persist.failed err={e}"); + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn addr() -> HumdAddr { + HumdAddr::new(ensemble::Hid::random_humd()).with_hint("192.0.2.7:7777") + } + + fn ens() -> Arc { + Arc::new(Ensemble::new(ensemble::Hid::random_humd())) + } + + #[test] + fn a_missing_file_restores_nothing() { + let dir = std::env::temp_dir().join(format!( + "routing-absent-{}", + ensemble::Hid::random_humd().short() + )); + assert!(load(&dir.join("routing.json")).is_empty()); + } + + #[test] + fn a_saved_table_survives_a_restart() { + let path = std::env::temp_dir() + .join(format!("routing-{}", ensemble::Hid::random_humd().short())) + .join("routing.json"); + + let before = ens(); + let peer = addr(); + assert_eq!(before.routing_restore(vec![peer.clone()]), 1); + save(&before, &path).expect("save must succeed"); + + let after = ens(); + restore_on_boot(&after, &path); + assert_eq!( + after.kad_routing_table_len(), + 1, + "the table must survive a restart" + ); + assert!( + after + .kad_closest(&peer.id, 4) + .iter() + .any(|a| a.id == peer.id) + ); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn a_corrupt_file_restores_nothing_rather_than_panicking() { + let path = std::env::temp_dir() + .join(format!( + "routing-bad-{}", + ensemble::Hid::random_humd().short() + )) + .join("routing.json"); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(&path, b"{not json").unwrap(); + assert!(load(&path).is_empty()); + let _ = std::fs::remove_file(&path); + } +} diff --git a/sim/src/lib.rs b/sim/src/lib.rs index 1703f2b0..98e960b1 100644 --- a/sim/src/lib.rs +++ b/sim/src/lib.rs @@ -95,7 +95,7 @@ impl Sim { pub async fn spawn_humd(&self, id: Hid) -> Arc { let thrum = Thrum::new(); - let ensemble = Arc::new(Ensemble::new(id)); + let ensemble = Arc::new(Ensemble::with_strict_auth(id, false)); let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let tmp = std::env::temp_dir().join(format!("sim-humd-{}", id.short())); @@ -115,6 +115,8 @@ impl Sim { http_path: tmp.join(hum_paths::HTTP_SOCK_BASENAME), mcp_addr: ([127, 0, 0, 1], 0).into(), penny_path, + routing_path: tmp.join(hum_paths::ROUTING_JSON_BASENAME), + routing_persist_interval: Duration::from_secs(3600), hum_cfg: hum_paths::config::HumConfig::default(), cli_path: "noop".into(), penny_persist_interval: Duration::from_secs(3600), @@ -172,6 +174,8 @@ impl Sim { self.wire(a, b) } + /// Unauthenticated mesh link. Use [`Sim::wire_signed`] when the test is + /// about identity. pub fn wire(&self, a: Hid, b: Hid) -> Result<()> { let humds = self.humds.read(); let ha = humds @@ -232,6 +236,8 @@ impl Sim { http_path: tmp.join(hum_paths::HTTP_SOCK_BASENAME), mcp_addr: ([127, 0, 0, 1], 0).into(), penny_path, + routing_path: tmp.join(hum_paths::ROUTING_JSON_BASENAME), + routing_persist_interval: Duration::from_secs(3600), hum_cfg: hum_paths::config::HumConfig::default(), cli_path: "noop".into(), penny_persist_interval: Duration::from_secs(3600), From 567246f518844e144ffb1110bbfe87e16be11abf Mon Sep 17 00:00:00 2001 From: Adil Date: Sun, 4 Oct 2026 16:38:50 +0500 Subject: [PATCH 2/3] ci: allow clippy lints that postdate this branch's toolchain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rust 1.99 added `double_must_use`, which fires on the `#[async_trait]` attribute in handshake.rs: the macro stamps `#[must_use]` onto a `Pin>` return, so the lint lands in macro expansion rather than in our source. Allow it alongside the existing `type_complexity` allowance. Both lints are newer than the 1.98.1 toolchain this branch was verified on, and CI tracks `stable`, so they could not be reproduced or fixed here — they are allowed, not fixed. --- .github/workflows/wire.yml | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/.github/workflows/wire.yml b/.github/workflows/wire.yml index a862d4cd..60786fa7 100644 --- a/.github/workflows/wire.yml +++ b/.github/workflows/wire.yml @@ -35,7 +35,12 @@ jobs: components: clippy - uses: Swatinem/rust-cache@v2 - name: Clippy on the crates this layer owns - run: cargo clippy -p ensemble -p hum-paths --all-targets -- -D warnings -A clippy::type_complexity + # type_complexity — a tuple-of-tuples in thrum-core's build script. + # double_must_use — async_trait stamps #[must_use] onto a + # Pin>; 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 From 382f3be1aa205daaad5991077bd21c1e4a4462b9 Mon Sep 17 00:00:00 2001 From: Adil Date: Sun, 4 Oct 2026 16:50:52 +0500 Subject: [PATCH 3/3] sim: wait for transport death instead of sleeping past it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `eviction_only_touches_the_dead_peer` sleeps 40ms and asserts the peer is evicted under a 120ms TTL. Every arriving tone renews the lease (lib.rs:821, the probe's pong included), so the TTL branch cannot fire in 40ms — the test only ever passed because humd's drain loop noticed the dropped sender and flipped the lease to TransportClosed within that window. Under CI load it does not, and the sweep returned []. Three sibling tests shared the shape. Replace the fixed sleeps with a poll on the non-destructive `peer_liveness`; `evict_expired` stays a single destructive call once the peer is genuinely dead. --- sim/tests/liveness.rs | 25 +++++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/sim/tests/liveness.rs b/sim/tests/liveness.rs index 4c01758f..16e905d9 100644 --- a/sim/tests/liveness.rs +++ b/sim/tests/liveness.rs @@ -1,4 +1,3 @@ - use std::time::Duration; use ensemble::Liveness; @@ -27,6 +26,21 @@ fn liveness(sim: &Sim, observer: ensemble::Hid, peer: ensemble::Hid) -> Option