diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 30bfdee..ffbd2cd 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -1,6 +1,7 @@ mod runtime; pub mod service; mod sync; +mod wire; use base64::Engine; use futures::StreamExt; @@ -72,6 +73,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream}; use tokio::sync::{mpsc, oneshot}; use tun_rs::{DeviceBuilder, Layer}; +use wire::{finish_iroh_send, read_iroh_line}; #[derive(Debug, thiserror::Error)] pub enum NodeError { @@ -4332,50 +4334,6 @@ async fn request_overlay_wire( } } -async fn read_iroh_line( - recv: &mut iroh::endpoint::RecvStream, - max_len: usize, -) -> Result { - let mut bytes = Vec::new(); - let mut byte = [0_u8; 1]; - loop { - if bytes.len() >= max_len { - return Err(NodeError::IrohPeer(format!( - "iroh response line exceeded {max_len} bytes" - ))); - } - let Some(n) = recv - .read(&mut byte) - .await - .map_err(|error| NodeError::IrohPeer(error.to_string()))? - else { - if bytes.is_empty() { - return Err(NodeError::IrohPeer( - "iroh stream closed before response line".to_owned(), - )); - } - break; - }; - if n == 0 { - continue; - } - bytes.push(byte[0]); - if byte[0] == b'\n' { - break; - } - } - String::from_utf8(bytes).map_err(|error| NodeError::IrohPeer(error.to_string())) -} - -async fn finish_iroh_send(send: &mut iroh::endpoint::SendStream) -> Result<(), NodeError> { - send.finish() - .map_err(|error| NodeError::IrohPeer(error.to_string()))?; - send.stopped() - .await - .map_err(|error| NodeError::IrohPeer(error.to_string()))?; - Ok(()) -} - fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) { let raw_endpoint = endpoint.endpoint(); tokio::spawn(async move { diff --git a/crates/geth-node/src/wire.rs b/crates/geth-node/src/wire.rs new file mode 100644 index 0000000..ce36b74 --- /dev/null +++ b/crates/geth-node/src/wire.rs @@ -0,0 +1,49 @@ +//! Shared Iroh stream helpers. + +use crate::NodeError; + +pub(crate) async fn read_iroh_line( + recv: &mut iroh::endpoint::RecvStream, + max_len: usize, +) -> Result { + let mut bytes = Vec::new(); + let mut byte = [0_u8; 1]; + loop { + if bytes.len() >= max_len { + return Err(NodeError::IrohPeer(format!( + "iroh response line exceeded {max_len} bytes" + ))); + } + let Some(n) = recv + .read(&mut byte) + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))? + else { + if bytes.is_empty() { + return Err(NodeError::IrohPeer( + "iroh stream closed before response line".to_owned(), + )); + } + break; + }; + if n == 0 { + continue; + } + bytes.push(byte[0]); + if byte[0] == b'\n' { + break; + } + } + String::from_utf8(bytes).map_err(|error| NodeError::IrohPeer(error.to_string())) +} + +pub(crate) async fn finish_iroh_send( + send: &mut iroh::endpoint::SendStream, +) -> Result<(), NodeError> { + send.finish() + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + send.stopped() + .await + .map_err(|error| NodeError::IrohPeer(error.to_string()))?; + Ok(()) +} diff --git a/docs/production-readiness-roadmap.md b/docs/production-readiness-roadmap.md index 8ea65d3..b40c035 100644 --- a/docs/production-readiness-roadmap.md +++ b/docs/production-readiness-roadmap.md @@ -53,6 +53,8 @@ behavior. - `[ ]` Extract protected peer-control routing. Acceptance criteria: + - `[x]` Shared bounded Iroh line-read and send-finish helpers live outside + the main feature handler module. - `[ ]` Iroh control ALPN handling, nonce checks, peer-card validation, and endpoint-binding validation are centralized. - `[ ]` Feature handlers receive authenticated caller context rather than