From 0a52be794c861038063df9f343cf25f659cc30f9 Mon Sep 17 00:00:00 2001 From: tube0013 Date: Wed, 15 Jul 2026 11:15:59 -0400 Subject: [PATCH 1/2] Support connecting to an RCP over a raw TCP socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ziggurat-spinel's SpinelClient was hard-coded to tokio_serial::SerialStream. Generalize it to a boxed RcpTransport (any AsyncRead + AsyncWrite), and teach ziggurat-server to open a TcpStream when --device is given as tcp://host:port, falling back to the existing serial-port path otherwise. Rebased onto dev (through #46, "Granular network state events for backup/restore"). The wire protocol was rewritten from JSON-over-WebSocket to a COBS-framed binary protocol (#38, ziggurat-protocol) since this was last rebased; only ziggurat-spinel/src/client.rs and the transport-opening half of ziggurat-server/src/main.rs (SerialConfig, open_transport, fn phy and its callers) needed re-applying — the request/response dispatch layer itself was untouched by this change and is left as upstream has it. Co-Authored-By: Claude Sonnet 4.6 --- Cargo.lock | 1 - crates/ziggurat-server/src/main.rs | 106 +++++++++++++++++++++------ crates/ziggurat-spinel/Cargo.toml | 1 - crates/ziggurat-spinel/src/client.rs | 16 ++-- 4 files changed, 91 insertions(+), 33 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 83465d8..3f036fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2074,7 +2074,6 @@ dependencies = [ "rand 0.10.1", "thiserror", "tokio", - "tokio-serial", "tracing", "ziggurat-ieee-802154", ] diff --git a/crates/ziggurat-server/src/main.rs b/crates/ziggurat-server/src/main.rs index daa2528..37c10e0 100644 --- a/crates/ziggurat-server/src/main.rs +++ b/crates/ziggurat-server/src/main.rs @@ -5,8 +5,8 @@ use std::time::Duration; use clap::{Parser, ValueEnum}; use futures_util::{SinkExt, StreamExt}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::net::{TcpListener, UnixListener}; -use tokio::sync::{broadcast, mpsc}; +use tokio::net::{TcpListener, TcpStream, UnixListener}; +use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc}; use tokio::task::JoinHandle; use tokio_serial::{FlowControl, SerialPortBuilderExt}; use tokio_tungstenite::tungstenite::Message; @@ -21,7 +21,7 @@ use ziggurat_driver::ziggurat_ieee_802154::types::{Eui64, Nwk, PanId}; use ziggurat_phy::{RadioConfig, RadioPhy, Receiver}; use ziggurat_phy_spinel::SpinelPhy; use ziggurat_protocol::{self as proto}; -use ziggurat_spinel::client::SpinelClient; +use ziggurat_spinel::client::{RcpTransport, SpinelClient}; /// Outbound frames a connection can queue before it is considered too slow and /// disconnected. Received frames dominate the traffic; a client that cannot keep up @@ -61,7 +61,7 @@ pub struct ZigguratServer { /// The radio transport owns the serial port for the lifetime of the process: it is /// opened lazily by the first command that needs it and never reopened, so stack /// replacement cannot race a straggling port handle (`EBUSY`) - phy: Mutex>>, + phy: AsyncMutex>>, stack: Mutex>>>, started: AtomicBool, notification_tx: broadcast::Sender, @@ -76,7 +76,7 @@ impl ZigguratServer { Self { serial, - phy: Mutex::new(None), + phy: AsyncMutex::new(None), stack: Mutex::new(None), started: AtomicBool::new(false), notification_tx, @@ -140,20 +140,15 @@ impl ZigguratServer { } /// The process-lifetime radio transport, opening the serial port on first use. - fn phy(&self) -> Result, tokio_serial::Error> { - let mut phy = self.phy.lock().unwrap(); + async fn phy(&self) -> std::io::Result> { + let mut phy = self.phy.lock().await; if let Some(phy) = &*phy { return Ok(phy.clone()); } - // Without flow control the RCP's UART drops bytes under load, corrupting - // host->RCP frames ("Framing error" + command timeout) - let port = tokio_serial::new(&self.serial.device, self.serial.baudrate) - .flow_control(self.serial.flow_control.into()) - .open_native_async()?; - - let new_phy = Arc::new(SpinelPhy::new(Arc::new(SpinelClient::new(port)))); + let transport = open_transport(&self.serial).await?; + let new_phy = Arc::new(SpinelPhy::new(Arc::new(SpinelClient::new(transport)))); *phy = Some(new_phy.clone()); drop(phy); @@ -551,7 +546,7 @@ impl ZigguratServer { payload: proto::ResetPayload, ) -> Result { if payload.hard { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reset() .await .map_err(|e| proto::Error::new(proto::Status::RadioError, &e.to_string()))?; @@ -561,7 +556,7 @@ impl ZigguratServer { } async fn handle_get_hw_address(&self) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let ieee = phy .hw_address() .await @@ -573,7 +568,7 @@ impl ZigguratServer { async fn handle_shutdown(&self) -> Result { self.teardown_stack().await; - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.set_frame_pending_table(&[], &[]) .await .map_err(|e| proto::Error::new(proto::Status::RadioError, &e.to_string()))?; @@ -593,7 +588,7 @@ impl ZigguratServer { self.teardown_stack().await; tracing::info!("Initializing Zigbee stack with new settings..."); - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let aps_frame_counter = payload.state.aps_frame_counter; let stack = ZigbeeStack::new( @@ -700,7 +695,7 @@ impl ZigguratServer { ) -> Result { // An energy detect is a radio operation, not a network one: it drives the // radio directly and needs no configured stack. - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; let duration = Duration::from_millis(u64::from(payload.duration_per_channel_ms)); for channel in payload.channels { @@ -773,7 +768,7 @@ impl ZigguratServer { payload: proto::ChannelPayload, outbound: &mpsc::Sender>, ) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reconfigure(&capture_config(payload.channel)) .await @@ -807,7 +802,7 @@ impl ZigguratServer { &self, payload: proto::ChannelPayload, ) -> Result { - let phy = self.phy().map_err(radio_error)?; + let phy = self.phy().await.map_err(radio_error)?; phy.reconfigure(&capture_config(payload.channel)) .await @@ -841,6 +836,27 @@ pub struct SerialConfig { flow_control: FlowControlMode, } +/// Connects to the RCP per `serial.device`: a `tcp://host:port` address for a raw TCP +/// socket (e.g. a network-attached RCP or a `ser2net`-style serial-to-TCP bridge), or a +/// path for a local serial device. +async fn open_transport(serial: &SerialConfig) -> std::io::Result> { + if let Some(addr) = serial.device.strip_prefix("tcp://") { + let stream = TcpStream::connect(addr).await?; + // The Spinel control plane is latency-sensitive request/response traffic in + // small frames; Nagle's algorithm would needlessly delay them. + stream.set_nodelay(true)?; + return Ok(Box::new(stream)); + } + + // Without flow control the RCP's UART drops bytes under load, corrupting + // host->RCP frames ("Framing error" + command timeout) + let port = tokio_serial::new(&serial.device, serial.baudrate) + .flow_control(serial.flow_control.into()) + .open_native_async() + .map_err(std::io::Error::other)?; + Ok(Box::new(port)) +} + /// How the Zigbee API is exposed to clients. #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] enum ApiMode { @@ -860,15 +876,16 @@ struct Args { #[arg(long, value_enum, default_value_t = ApiMode::Ws)] api: ApiMode, - /// Serial device of the 802.15.4 RCP + /// RCP transport: a serial device path, or `tcp://host:port` for a raw TCP socket #[arg(long)] device: String, - /// Serial baudrate + /// Serial baudrate; ignored for a `tcp://` device #[arg(long, default_value_t = 460_800)] baudrate: u32, - /// Serial flow control; the RCP UART drops bytes under load without it + /// Serial flow control; the RCP UART drops bytes under load without it. Ignored + /// for a `tcp://` device #[arg(long, value_enum, default_value_t = FlowControlMode::Hardware)] flow_control: FlowControlMode, @@ -919,3 +936,44 @@ fn main() -> Result<(), Box> { Ok(()) }) } + +#[cfg(test)] +mod tests { + use super::{FlowControlMode, SerialConfig, open_transport}; + use tokio::net::TcpListener; + + #[tokio::test] + async fn open_transport_connects_to_a_tcp_device() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let _ = listener.accept().await; + }); + + let serial = SerialConfig { + device: format!("tcp://{addr}"), + baudrate: 115_200, + flow_control: FlowControlMode::None, + }; + + open_transport(&serial) + .await + .expect("tcp:// device should connect over TCP"); + } + + #[tokio::test] + async fn open_transport_rejects_an_unreachable_tcp_device() { + // Port 0 is never a valid connect target, so this fails fast without relying on + // a real port being closed. + let serial = SerialConfig { + device: "tcp://127.0.0.1:0".to_string(), + baudrate: 115_200, + flow_control: FlowControlMode::None, + }; + + assert!( + open_transport(&serial).await.is_err(), + "connecting to port 0 should fail" + ); + } +} diff --git a/crates/ziggurat-spinel/Cargo.toml b/crates/ziggurat-spinel/Cargo.toml index 3eee68e..564ab84 100644 --- a/crates/ziggurat-spinel/Cargo.toml +++ b/crates/ziggurat-spinel/Cargo.toml @@ -16,7 +16,6 @@ tracing = "0.1" num_enum = "0.7.3" thiserror = "2.0.12" tokio = { version = "1.43.0", features = ["rt", "time", "sync", "io-util"] } -tokio-serial = "5.4" [dev-dependencies] hex-literal = "1.1.0" diff --git a/crates/ziggurat-spinel/src/client.rs b/crates/ziggurat-spinel/src/client.rs index aad7689..5a5e82d 100644 --- a/crates/ziggurat-spinel/src/client.rs +++ b/crates/ziggurat-spinel/src/client.rs @@ -5,18 +5,22 @@ use crate::{ }; use std::string::String; use thiserror::Error; -use tokio_serial::SerialStream; use ziggurat_ieee_802154::FrameBytes; use ziggurat_ieee_802154::types::Eui64; use std::sync::Arc; use std::sync::Mutex; use std::sync::atomic::{AtomicU32, Ordering}; -use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; use tokio::sync::Mutex as AsyncMutex; use tokio::sync::mpsc; use tokio::time::{Duration, timeout}; +/// A link to the RCP: a serial port or a TCP socket. Boxed so [`SpinelClient`] stays +/// transport-agnostic regardless of which one a caller opens. +pub trait RcpTransport: AsyncRead + AsyncWrite + Send + Unpin + 'static {} +impl RcpTransport for T {} + /// A local serial link answers in milliseconds; anything beyond this is a failure. const TIMEOUT: Duration = Duration::from_secs(2); @@ -227,17 +231,15 @@ pub enum SpinelError { /// The writer half of the port plus its serialization scratch: frames go out one at a /// time, so two persistent buffers cover every TX without per-frame allocation. -#[derive(Debug)] struct SpinelWriter { - port: WriteHalf, + port: WriteHalf>, frame_scratch: Vec, hdlc_scratch: Vec, } -#[derive(Debug)] pub struct SpinelClient { /// The reader half of the port, owned by the task spawned in `spawn_reader`. - reader: Mutex>>, + reader: Mutex>>>, /// The writer half of the port. The mutex also serializes outbound HDLC writes so /// concurrent commands cannot interleave partial frames inside the byte stream. writer: AsyncMutex, @@ -249,7 +251,7 @@ pub struct SpinelClient { } impl SpinelClient { - pub fn new(port: SerialStream) -> Self { + pub fn new(port: Box) -> Self { let (reader, writer) = tokio::io::split(port); Self { From 8caf1ea235f87e40e51abafe8ac3e36482c3cbaa Mon Sep 17 00:00:00 2001 From: tube0013 Date: Wed, 15 Jul 2026 13:24:24 -0400 Subject: [PATCH 2/2] Update crates/ziggurat-server/src/main.rs Co-authored-by: puddly <32534428+puddly@users.noreply.github.com> --- crates/ziggurat-server/src/main.rs | 41 ------------------------------ 1 file changed, 41 deletions(-) diff --git a/crates/ziggurat-server/src/main.rs b/crates/ziggurat-server/src/main.rs index 37c10e0..cf536f8 100644 --- a/crates/ziggurat-server/src/main.rs +++ b/crates/ziggurat-server/src/main.rs @@ -936,44 +936,3 @@ fn main() -> Result<(), Box> { Ok(()) }) } - -#[cfg(test)] -mod tests { - use super::{FlowControlMode, SerialConfig, open_transport}; - use tokio::net::TcpListener; - - #[tokio::test] - async fn open_transport_connects_to_a_tcp_device() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - let _ = listener.accept().await; - }); - - let serial = SerialConfig { - device: format!("tcp://{addr}"), - baudrate: 115_200, - flow_control: FlowControlMode::None, - }; - - open_transport(&serial) - .await - .expect("tcp:// device should connect over TCP"); - } - - #[tokio::test] - async fn open_transport_rejects_an_unreachable_tcp_device() { - // Port 0 is never a valid connect target, so this fails fast without relying on - // a real port being closed. - let serial = SerialConfig { - device: "tcp://127.0.0.1:0".to_string(), - baudrate: 115_200, - flow_control: FlowControlMode::None, - }; - - assert!( - open_transport(&serial).await.is_err(), - "connecting to port 0 should fail" - ); - } -}