fix(clients/rust): harden client resilience for v0.1.0 release
- Track serial timeout in field instead of returning hardcoded 2s - Always restore timeout in drain_rx, even on read errors - Handle EINTR/Interrupted in read_frame (retry instead of error) - Use connect_timeout for TCP mux (no more 2-min hangs) - Set write timeout on mux sockets (unix + tcp) - Log unexpected unsolicited frames in recv via tracing::warn - Add unix socket timeout test for WouldBlock handling - Add CHANGELOG.md, remove resolved CLAUDE_TODO.md
This commit is contained in:
6 files changed
+150
-15
No files matched your search
@@ -0,0 +1,27 @@
|
|||||||
|
# Changelog
|
||||||
|
|
||||||
|
## 0.1.0 — 2026-04-06
|
||||||
|
|
||||||
|
Initial release.
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
- High-level `Client<T>` with send/recv and command helpers (ping, set_config,
|
||||||
|
start_rx, stop_rx, transmit, get_config, get_mac, display_on, display_off)
|
||||||
|
- COBS wire framing via `ucobs` (matches firmware implementation)
|
||||||
|
- Auto-detection connection: TCP mux, Unix socket mux, direct USB serial
|
||||||
|
- Bounded RX packet buffering (256 packets, FIFO eviction)
|
||||||
|
- `FrameReader` accumulator for streaming byte sources
|
||||||
|
- USB device discovery by VID:PID with blocking wait
|
||||||
|
|
||||||
|
### Resilience
|
||||||
|
|
||||||
|
- Cross-platform timeout handling: `TimedOut` (Windows) and `WouldBlock`
|
||||||
|
(Linux/macOS) both treated as clean timeouts in `read_frame`
|
||||||
|
- `EINTR`/`Interrupted` signals retried automatically in `read_frame`
|
||||||
|
- `drain_rx` always restores the original timeout, even on I/O errors
|
||||||
|
- TCP mux connections use `connect_timeout` (bounded by caller's timeout)
|
||||||
|
- Mux sockets set both read and write timeouts
|
||||||
|
- `SerialTransport` tracks timeout accurately for save/restore
|
||||||
|
- Unexpected unsolicited frames logged via `tracing::warn` instead of silently
|
||||||
|
discarded
|
||||||
@@ -46,7 +46,10 @@ impl<T: Transport> Client<T> {
|
|||||||
let data = read_frame(&mut self.transport)?
|
let data = read_frame(&mut self.transport)?
|
||||||
.ok_or_else(|| anyhow::anyhow!("timeout waiting for response"))?;
|
.ok_or_else(|| anyhow::anyhow!("timeout waiting for response"))?;
|
||||||
let resp = Response::from_bytes(&data)
|
let resp = Response::from_bytes(&data)
|
||||||
.ok_or_else(|| anyhow::anyhow!("malformed response ({} bytes)", data.len()))?;
|
.ok_or_else(|| {
|
||||||
|
let hex: Vec<String> = data.iter().map(|b| format!("{b:02x}")).collect();
|
||||||
|
anyhow::anyhow!("malformed response ({} bytes): [{}]", data.len(), hex.join(" "))
|
||||||
|
})?;
|
||||||
if resp.is_rx_packet() {
|
if resp.is_rx_packet() {
|
||||||
self.buffer_rx(resp);
|
self.buffer_rx(resp);
|
||||||
continue;
|
continue;
|
||||||
@@ -71,7 +74,7 @@ impl<T: Transport> Client<T> {
|
|||||||
if resp.is_rx_packet() {
|
if resp.is_rx_packet() {
|
||||||
Ok(Some(resp))
|
Ok(Some(resp))
|
||||||
} else {
|
} else {
|
||||||
// Non-RxPacket unsolicited response — shouldn't happen, discard
|
tracing::warn!("recv: discarding unexpected unsolicited response: {resp:?}");
|
||||||
Ok(None)
|
Ok(None)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -87,9 +90,10 @@ impl<T: Transport> Client<T> {
|
|||||||
self.transport
|
self.transport
|
||||||
.set_timeout(std::time::Duration::from_millis(10))?;
|
.set_timeout(std::time::Duration::from_millis(10))?;
|
||||||
|
|
||||||
|
let result: anyhow::Result<()> = (|| {
|
||||||
loop {
|
loop {
|
||||||
let Some(data) = read_frame(&mut self.transport)? else {
|
let Some(data) = read_frame(&mut self.transport)? else {
|
||||||
break;
|
return Ok(());
|
||||||
};
|
};
|
||||||
if let Some(resp) = Response::from_bytes(&data)
|
if let Some(resp) = Response::from_bytes(&data)
|
||||||
&& resp.is_rx_packet()
|
&& resp.is_rx_packet()
|
||||||
@@ -97,8 +101,11 @@ impl<T: Transport> Client<T> {
|
|||||||
packets.push(resp);
|
packets.push(resp);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
})();
|
||||||
|
|
||||||
|
// Always restore timeout, even if the loop errored
|
||||||
self.transport.set_timeout(old_timeout)?;
|
self.transport.set_timeout(old_timeout)?;
|
||||||
|
result?;
|
||||||
Ok(packets)
|
Ok(packets)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -107,9 +107,15 @@ pub fn read_frame(reader: &mut dyn std::io::Read) -> anyhow::Result<Option<Vec<u
|
|||||||
}
|
}
|
||||||
buf.push(byte[0]);
|
buf.push(byte[0]);
|
||||||
}
|
}
|
||||||
Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {
|
Err(e)
|
||||||
|
if matches!(
|
||||||
|
e.kind(),
|
||||||
|
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
|
||||||
|
) =>
|
||||||
|
{
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
}
|
}
|
||||||
|
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
|
||||||
Err(e) => return Err(e.into()),
|
Err(e) => return Err(e.into()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -274,4 +280,51 @@ mod tests {
|
|||||||
let decoded = decode_frame(&encoded[..encoded.len() - 1]);
|
let decoded = decode_frame(&encoded[..encoded.len() - 1]);
|
||||||
assert_eq!(decoded, Some(data));
|
assert_eq!(decoded, Some(data));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_response_cobs_roundtrip() {
|
||||||
|
// Exact Config response bytes: tag=1, freq=910525000, bw=62.5k, sf=7,
|
||||||
|
// cr=5, sync=0x3444, tx=22dBm, preamble=16, cad=1
|
||||||
|
// Note: preamble high byte is 0x00 — tests COBS with embedded zero.
|
||||||
|
let data = vec![
|
||||||
|
0x01, 0x48, 0x82, 0x45, 0x36, 0x06, 0x07, 0x05,
|
||||||
|
0x44, 0x34, 0x16, 0x10, 0x00, 0x01,
|
||||||
|
];
|
||||||
|
assert_eq!(data.len(), 14);
|
||||||
|
let encoded = encode_frame(&data);
|
||||||
|
let decoded = decode_frame(&encoded[..encoded.len() - 1]);
|
||||||
|
assert_eq!(decoded, Some(data));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_response_cobs_roundtrip_cad_zero() {
|
||||||
|
// Same but with cad=0 — two consecutive 0x00 bytes at end.
|
||||||
|
let data = vec![
|
||||||
|
0x01, 0x48, 0x82, 0x45, 0x36, 0x06, 0x07, 0x05,
|
||||||
|
0x44, 0x34, 0x16, 0x10, 0x00, 0x00,
|
||||||
|
];
|
||||||
|
assert_eq!(data.len(), 14);
|
||||||
|
let encoded = encode_frame(&data);
|
||||||
|
let decoded = decode_frame(&encoded[..encoded.len() - 1]);
|
||||||
|
assert_eq!(decoded, Some(data));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
#[test]
|
||||||
|
fn read_frame_socket_timeout_returns_none() {
|
||||||
|
use std::os::unix::net::UnixStream;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
let (mut a, _b) = UnixStream::pair().unwrap();
|
||||||
|
a.set_read_timeout(Some(Duration::from_millis(10)))
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// No data written to _b, so read on `a` will time out with WouldBlock on Linux.
|
||||||
|
let result = read_frame(&mut a);
|
||||||
|
assert!(
|
||||||
|
result.is_ok(),
|
||||||
|
"socket timeout should be Ok(None), got: {result:?}"
|
||||||
|
);
|
||||||
|
assert!(result.unwrap().is_none());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -131,6 +131,36 @@ pub fn connect_default() -> anyhow::Result<Client<AnyTransport>> {
|
|||||||
connect(None, DEFAULT_TIMEOUT)
|
connect(None, DEFAULT_TIMEOUT)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Connect to a mux daemon only (TCP via env var, then Unix socket).
|
||||||
|
///
|
||||||
|
/// Unlike [`connect`], this **never** falls back to direct USB serial.
|
||||||
|
/// Returns an error if no mux is reachable — the caller can retry with backoff.
|
||||||
|
///
|
||||||
|
/// This is the Rust equivalent of the Python client's "sticky mux" behavior:
|
||||||
|
/// once you decide to use the mux, you stay on the mux.
|
||||||
|
pub fn connect_mux_auto(timeout: Duration) -> anyhow::Result<Client<AnyTransport>> {
|
||||||
|
// Try TCP mux via environment variable.
|
||||||
|
if let Ok(tcp) = std::env::var("DONGLORA_MUX_TCP")
|
||||||
|
&& let Some(transport) = try_tcp_mux(&tcp, timeout)
|
||||||
|
{
|
||||||
|
debug!("connected to TCP mux at {tcp}");
|
||||||
|
return Ok(Client::new(AnyTransport::Mux(transport)));
|
||||||
|
}
|
||||||
|
|
||||||
|
// Try Unix socket mux.
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
let path = find_mux_socket()
|
||||||
|
.ok_or_else(|| anyhow::anyhow!("no mux socket found"))?;
|
||||||
|
let transport = MuxTransport::unix(&path, timeout)?;
|
||||||
|
debug!("connected to mux socket at {path}");
|
||||||
|
Ok(Client::new(AnyTransport::Mux(transport)))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(not(unix))]
|
||||||
|
anyhow::bail!("mux-only mode requires Unix socket support or DONGLORA_MUX_TCP")
|
||||||
|
}
|
||||||
|
|
||||||
fn try_tcp_mux(addr: &str, timeout: Duration) -> Option<MuxTransport> {
|
fn try_tcp_mux(addr: &str, timeout: Duration) -> Option<MuxTransport> {
|
||||||
let (host, port) = if let Some((h, p)) = addr.rsplit_once(':') {
|
let (host, port) = if let Some((h, p)) = addr.rsplit_once(':') {
|
||||||
let host = if h.is_empty() { "localhost" } else { h };
|
let host = if h.is_empty() { "localhost" } else { h };
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ pub mod transport;
|
|||||||
// Flat re-exports for convenience
|
// Flat re-exports for convenience
|
||||||
pub use client::Client;
|
pub use client::Client;
|
||||||
pub use codec::{decode_frame, encode_frame, read_frame, FrameReader};
|
pub use codec::{decode_frame, encode_frame, read_frame, FrameReader};
|
||||||
pub use connect::{connect, connect_default, default_socket_path};
|
pub use connect::{connect, connect_default, connect_mux_auto, default_socket_path};
|
||||||
pub use discovery::{find_port, wait_for_device, USB_PID, USB_VID};
|
pub use discovery::{find_port, wait_for_device, USB_PID, USB_VID};
|
||||||
pub use protocol::{
|
pub use protocol::{
|
||||||
Bandwidth, Command, ErrorCode, RadioConfig, Response, CMD_TAG_SET_CONFIG, CMD_TAG_START_RX,
|
Bandwidth, Command, ErrorCode, RadioConfig, Response, CMD_TAG_SET_CONFIG, CMD_TAG_START_RX,
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ pub trait Transport: Read + Write + Send {
|
|||||||
/// Direct USB serial port connection to a DongLoRa dongle.
|
/// Direct USB serial port connection to a DongLoRa dongle.
|
||||||
pub struct SerialTransport {
|
pub struct SerialTransport {
|
||||||
port: Box<dyn serialport::SerialPort>,
|
port: Box<dyn serialport::SerialPort>,
|
||||||
|
timeout: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SerialTransport {
|
impl SerialTransport {
|
||||||
@@ -31,7 +32,7 @@ impl SerialTransport {
|
|||||||
.timeout(timeout)
|
.timeout(timeout)
|
||||||
.open()
|
.open()
|
||||||
.map_err(|e| anyhow::anyhow!("failed to open serial port {path}: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("failed to open serial port {path}: {e}"))?;
|
||||||
Ok(Self { port })
|
Ok(Self { port, timeout })
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Clear the serial input buffer.
|
/// Clear the serial input buffer.
|
||||||
@@ -62,13 +63,13 @@ impl Transport for SerialTransport {
|
|||||||
fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()> {
|
fn set_timeout(&mut self, timeout: Duration) -> anyhow::Result<()> {
|
||||||
self.port
|
self.port
|
||||||
.set_timeout(timeout)
|
.set_timeout(timeout)
|
||||||
.map_err(|e| anyhow::anyhow!("failed to set serial timeout: {e}"))
|
.map_err(|e| anyhow::anyhow!("failed to set serial timeout: {e}"))?;
|
||||||
|
self.timeout = timeout;
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn timeout(&self) -> Duration {
|
fn timeout(&self) -> Duration {
|
||||||
// serialport doesn't have a getter, so we'd need to track it ourselves.
|
self.timeout
|
||||||
// For now, return a sensible default. The actual timeout is set on the port.
|
|
||||||
Duration::from_secs(2)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -94,7 +95,10 @@ impl MuxTransport {
|
|||||||
.map_err(|e| anyhow::anyhow!("failed to connect to mux socket {path}: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("failed to connect to mux socket {path}: {e}"))?;
|
||||||
stream
|
stream
|
||||||
.set_read_timeout(Some(timeout))
|
.set_read_timeout(Some(timeout))
|
||||||
.map_err(|e| anyhow::anyhow!("failed to set socket timeout: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("failed to set socket read timeout: {e}"))?;
|
||||||
|
stream
|
||||||
|
.set_write_timeout(Some(timeout))
|
||||||
|
.map_err(|e| anyhow::anyhow!("failed to set socket write timeout: {e}"))?;
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
stream: MuxStream::Unix(stream),
|
stream: MuxStream::Unix(stream),
|
||||||
timeout,
|
timeout,
|
||||||
@@ -103,12 +107,22 @@ impl MuxTransport {
|
|||||||
|
|
||||||
/// Connect to the mux daemon via TCP.
|
/// Connect to the mux daemon via TCP.
|
||||||
pub fn tcp(host: &str, port: u16, timeout: Duration) -> anyhow::Result<Self> {
|
pub fn tcp(host: &str, port: u16, timeout: Duration) -> anyhow::Result<Self> {
|
||||||
|
use std::net::ToSocketAddrs;
|
||||||
|
|
||||||
let addr = format!("{host}:{port}");
|
let addr = format!("{host}:{port}");
|
||||||
let stream = std::net::TcpStream::connect(&addr)
|
let sock_addr = addr
|
||||||
|
.to_socket_addrs()
|
||||||
|
.map_err(|e| anyhow::anyhow!("failed to resolve mux address {addr}: {e}"))?
|
||||||
|
.next()
|
||||||
|
.ok_or_else(|| anyhow::anyhow!("no addresses found for {addr}"))?;
|
||||||
|
let stream = std::net::TcpStream::connect_timeout(&sock_addr, timeout)
|
||||||
.map_err(|e| anyhow::anyhow!("failed to connect to mux at {addr}: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("failed to connect to mux at {addr}: {e}"))?;
|
||||||
stream
|
stream
|
||||||
.set_read_timeout(Some(timeout))
|
.set_read_timeout(Some(timeout))
|
||||||
.map_err(|e| anyhow::anyhow!("failed to set TCP timeout: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("failed to set TCP read timeout: {e}"))?;
|
||||||
|
stream
|
||||||
|
.set_write_timeout(Some(timeout))
|
||||||
|
.map_err(|e| anyhow::anyhow!("failed to set TCP write timeout: {e}"))?;
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
stream: MuxStream::Tcp(stream),
|
stream: MuxStream::Tcp(stream),
|
||||||
timeout,
|
timeout,
|
||||||
@@ -149,8 +163,12 @@ impl Transport for MuxTransport {
|
|||||||
self.timeout = timeout;
|
self.timeout = timeout;
|
||||||
let result = match &self.stream {
|
let result = match &self.stream {
|
||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
MuxStream::Unix(s) => s.set_read_timeout(Some(timeout)),
|
MuxStream::Unix(s) => s
|
||||||
MuxStream::Tcp(s) => s.set_read_timeout(Some(timeout)),
|
.set_read_timeout(Some(timeout))
|
||||||
|
.and_then(|()| s.set_write_timeout(Some(timeout))),
|
||||||
|
MuxStream::Tcp(s) => s
|
||||||
|
.set_read_timeout(Some(timeout))
|
||||||
|
.and_then(|()| s.set_write_timeout(Some(timeout))),
|
||||||
};
|
};
|
||||||
result.map_err(|e| anyhow::anyhow!("failed to set timeout: {e}"))
|
result.map_err(|e| anyhow::anyhow!("failed to set timeout: {e}"))
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user