refactor: extract iroh wire helpers

This commit is contained in:
Eric Wendland 2026-07-05 17:35:40 +02:00
commit fd3651b25f
3 changed files with 53 additions and 44 deletions

View file

@ -1,6 +1,7 @@
mod runtime; mod runtime;
pub mod service; pub mod service;
mod sync; mod sync;
mod wire;
use base64::Engine; use base64::Engine;
use futures::StreamExt; use futures::StreamExt;
@ -72,6 +73,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream}; use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
use tokio::sync::{mpsc, oneshot}; use tokio::sync::{mpsc, oneshot};
use tun_rs::{DeviceBuilder, Layer}; use tun_rs::{DeviceBuilder, Layer};
use wire::{finish_iroh_send, read_iroh_line};
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum NodeError { 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<String, NodeError> {
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) { fn spawn_iroh_control_accept_loop(node: LocalNode, endpoint: GethIrohEndpoint) {
let raw_endpoint = endpoint.endpoint(); let raw_endpoint = endpoint.endpoint();
tokio::spawn(async move { tokio::spawn(async move {

View file

@ -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<String, NodeError> {
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(())
}

View file

@ -53,6 +53,8 @@ behavior.
- `[ ]` Extract protected peer-control routing. - `[ ]` Extract protected peer-control routing.
Acceptance criteria: 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 - `[ ]` Iroh control ALPN handling, nonce checks, peer-card validation, and
endpoint-binding validation are centralized. endpoint-binding validation are centralized.
- `[ ]` Feature handlers receive authenticated caller context rather than - `[ ]` Feature handlers receive authenticated caller context rather than