diff --git a/Cargo.lock b/Cargo.lock index fbd301e3d..1b7a08002 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3866,8 +3866,10 @@ dependencies = [ name = "fw-esp32-common" version = "40.0.0" dependencies = [ + "embassy-futures", "embassy-sync 0.8.0", "embassy-time", + "embedded-io-async 0.6.1", "fw-core", "hashbrown 0.15.5", "littlefs-rust", @@ -3880,6 +3882,7 @@ dependencies = [ "lpc-shared", "lpc-wire", "lpfs", + "ser-write-json", "smart-leds", ] @@ -3926,7 +3929,6 @@ dependencies = [ "lps-shared", "lpvm", "lpvm-native", - "ser-write-json", "serde", "smart-leds", "unwinding", @@ -3967,7 +3969,6 @@ dependencies = [ "lps-shared", "lpvm", "lpvm-native", - "ser-write-json", ] [[package]] diff --git a/lp-fw/fw-esp32-common/Cargo.toml b/lp-fw/fw-esp32-common/Cargo.toml index 717dfc4ee..a99206db6 100644 --- a/lp-fw/fw-esp32-common/Cargo.toml +++ b/lp-fw/fw-esp32-common/Cargo.toml @@ -21,6 +21,11 @@ lpc-hardware = { path = "../../lp-core/lpc-hardware", default-features = false } lpc-shared = { path = "../../lp-core/lpc-shared", default-features = false } embassy-time = "0.5.0" embassy-sync = "0.8.0" +# `serial::chunked_write`: the select-with-timeout and the async byte sink. Both +# are `no_std` and chip-free, and both are already on every bin crate's line — +# this is the seam moving down, not a new dependency for the images. +embassy-futures = "0.1" +embedded-io-async = "0.6" smart-leds = "0.3" log = { workspace = true, default-features = false } hashbrown = { workspace = true } @@ -39,7 +44,12 @@ lpa-server = { path = "../../lp-app/lpa-server", default-features = false, optio lpc-model = { path = "../../lp-core/lpc-model", default-features = false, optional = true } lpc-wire = { path = "../../lp-core/lpc-wire", default-features = false, optional = true, features = ["ser-write-json"] } lpfs = { path = "../../lp-base/lpfs", default-features = false, optional = true } +# `serial::server_msg`'s `StackJsonWriter` implements this crate's `SerWrite` +# trait, so it must be a direct dependency and not just `lpc-wire`'s. Same +# version/features as the bin crates so all three resolve to one instance — +# two would make the impl invisible to `lpc_wire::ser_write_json_to`. +ser-write-json = { version = "0.3", optional = true, default-features = false, features = ["alloc"] } [features] default = [] -server = ["dep:littlefs-rust", "dep:lp-recovery", "dep:lpa-server", "dep:lpc-model", "dep:lpc-wire", "dep:lpfs"] +server = ["dep:littlefs-rust", "dep:lp-recovery", "dep:lpa-server", "dep:lpc-model", "dep:lpc-wire", "dep:lpfs", "dep:ser-write-json"] diff --git a/lp-fw/fw-esp32-common/src/serial/chunked_write.rs b/lp-fw/fw-esp32-common/src/serial/chunked_write.rs new file mode 100644 index 000000000..42c1a0cbc --- /dev/null +++ b/lp-fw/fw-esp32-common/src/serial/chunked_write.rs @@ -0,0 +1,113 @@ +//! Chunked, timeout-bounded writes to a host link's TX half. +//! +//! Every ESP32 firmware writes to its host link the same way: split the buffer +//! into chunks, bound each chunk with a timeout, and do something small between +//! chunks. Only that last part is a chip fact, so it arrives as a hook rather +//! than a dependency — see [`ChunkedWriter::new`]. + +use embassy_time::{Duration, Timer}; +use embedded_io_async::Write; + +/// How a link chunks and bounds its writes. +/// +/// A chip fact, passed in rather than hard-coded, because the two numbers mean +/// different things per transport: USB-Serial-JTAG sizes its chunk for syscall +/// overhead, a UART sizes it for *line time* (the RX FIFO overflows if the TX +/// loop does not yield often enough at 115200 baud). +#[derive(Debug, Clone, Copy)] +pub struct WritePolicy { + /// Per-chunk timeout. A chunk that does not complete inside it fails the + /// whole write. + pub timeout: Duration, + /// Bytes per chunk. + pub chunk_size: usize, + /// What to call the link in error text (`"USB"`, `"UART"`). Surfaces to the + /// host inside `TransportError::Other`. + pub link_name: &'static str, +} + +impl WritePolicy { + /// The USB-Serial-JTAG policy used by `fw-esp32c6` and `fw-esp32s3`. + /// + /// Timeout: if a chunk doesn't complete in this time, the host is not + /// draining. A healthy USB full-speed host drains a chunk in well under a + /// millisecond, so this is still very generous — but short enough that the + /// frame loop's inline sends stall briefly, not for seconds, in the window + /// before the not-draining latch kicks in. + /// + /// Chunk size: small enough to avoid timeout on slow USB, large enough to + /// avoid excessive syscalls. Resource snapshots can be 10KB+. + pub const USB_SERIAL_JTAG: Self = Self { + timeout: Duration::from_millis(250), + chunk_size: 256, + link_name: "USB", + }; +} + +/// A link's TX half plus the policy and per-chunk hook that make writing to it +/// chip-correct. +/// +/// Construct one per write; it borrows the transmitter and holds no state of +/// its own, so the cost is nothing and the borrow stays as short as the write. +pub struct ChunkedWriter<'a, W, F> { + pub(crate) tx: &'a mut W, + pub(crate) policy: WritePolicy, + on_chunk: F, +} + +impl<'a, W: Write, F: FnMut()> ChunkedWriter<'a, W, F> { + /// Wrap `tx` with a write policy and a per-chunk hook. + /// + /// `on_chunk` runs **before every chunk**, including the first, and is the + /// single crate-specific seam in this module. It exists because a bounded + /// in-flight write is healthy, not silence, and each firmware has its own + /// thing to do about that: + /// + /// * `fw-esp32c6` / `fw-esp32s3` tick `recovery::watchdog::note_io_alive`, + /// so a slow host cannot starve the watchdog feeder into resetting the + /// device. The watchdog is a chip fact (it is an `esp-hal` `Rwdt`), so + /// this crate may not reach for it — hence the hook. + /// * `fw-esp32v3` drains its UART RX FIFO, which holds only ~1.4 ms of + /// incoming line and would overflow during a multi-second write. + /// + /// A `FnMut` rather than a `fn()` precisely so the second kind — which + /// needs the RX half and the line buffer — fits without a second writer. + pub fn new(tx: &'a mut W, policy: WritePolicy, on_chunk: F) -> Self { + Self { + tx, + policy, + on_chunk, + } + } + + /// Write all of `data` in chunks with the policy's per-chunk timeout. + /// + /// Prevents large messages (e.g. resource snapshots) from timing out + /// mid-write and corrupting the stream by concatenating with the next + /// message. Uses `write_all` per chunk to handle partial writes. + /// + /// Returns false on the first chunk that times out or errors. + pub async fn write_all(&mut self, data: &[u8]) -> bool { + self.write_all_with(data, self.policy.timeout).await + } + + /// [`write_all`](Self::write_all) with an explicit per-chunk timeout, for + /// callers that want a shorter bound than the policy's (the C6/S3 + /// not-draining probe writes a single byte and will not wait long for it). + pub async fn write_all_with(&mut self, data: &[u8], timeout: Duration) -> bool { + use embassy_futures::select::{Either, select}; + let mut offset = 0; + while offset < data.len() { + (self.on_chunk)(); + let chunk_end = (offset + self.policy.chunk_size).min(data.len()); + let chunk = &data[offset..chunk_end]; + match select(Timer::after(timeout), self.tx.write_all(chunk)).await { + Either::First(_) => return false, + Either::Second(Err(_)) => return false, + Either::Second(Ok(())) => {} + } + offset = chunk_end; + } + true + } +} diff --git a/lp-fw/fw-esp32-common/src/serial/mod.rs b/lp-fw/fw-esp32-common/src/serial/mod.rs index bb1c71b3f..19cd8b196 100644 --- a/lp-fw/fw-esp32-common/src/serial/mod.rs +++ b/lp-fw/fw-esp32-common/src/serial/mod.rs @@ -1,3 +1,9 @@ //! Chip-generic serial helpers. +pub mod chunked_write; pub mod shared_serial; + +/// Wire-protocol serialization for the host link — the chip-agnostic half of +/// every firmware's `serial::io_task`. +#[cfg(feature = "server")] +pub mod server_msg; diff --git a/lp-fw/fw-esp32-common/src/serial/server_msg.rs b/lp-fw/fw-esp32-common/src/serial/server_msg.rs new file mode 100644 index 000000000..56171b5c8 --- /dev/null +++ b/lp-fw/fw-esp32-common/src/serial/server_msg.rs @@ -0,0 +1,230 @@ +//! Wire-protocol server messages, serialized to JSON and written to a host link. +//! +//! This is the chip-agnostic half of every firmware's `serial::io_task`: take a +//! [`lpc_wire::WireServerMessage`], serialize it into a stack buffer with the +//! `\nM!` framing, and hand the bytes to a [`ChunkedWriter`]. The io loop around +//! it — connection monitoring, RX handling, the request/result channels — stays +//! in the bin crate, because that is where the transport differs. + +use alloc::{format, string::String}; +use ser_write_json::SerWrite; + +use super::chunked_write::ChunkedWriter; + +/// A [`SerWrite`] sink over a caller-provided stack buffer. +/// +/// Fails rather than allocating when the buffer fills, which is what lets the +/// serialization budget be a checked constant instead of a heap surprise. +pub struct StackJsonWriter<'a> { + buf: &'a mut [u8], + len: usize, +} + +/// The only way [`StackJsonWriter`] can fail: out of buffer. +#[derive(Debug)] +pub struct StackJsonError; + +impl core::fmt::Display for StackJsonError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str("stack JSON buffer full") + } +} + +impl<'a> StackJsonWriter<'a> { + /// Wrap `buf`, writing from its start. + pub fn new(buf: &'a mut [u8]) -> Self { + Self { buf, len: 0 } + } + + /// The bytes written so far. + pub fn bytes(&self) -> &[u8] { + &self.buf[..self.len] + } +} + +impl ser_write_json::SerWrite for StackJsonWriter<'_> { + type Error = StackJsonError; + + fn write(&mut self, buf: &[u8]) -> Result<(), StackJsonError> { + let end = self.len.checked_add(buf.len()).ok_or(StackJsonError)?; + if end > self.buf.len() { + return Err(StackJsonError); + } + self.buf[self.len..end].copy_from_slice(buf); + self.len = end; + Ok(()) + } +} + +impl ChunkedWriter<'_, W, F> { + /// Serialize `msg` and write it, framed, to the link. + /// + /// `connected` is the caller's connection verdict — the USB-Serial-JTAG + /// connection monitor is a chip fact and stays in the bin crate; links with + /// no such signal (a UART) pass `true`. + pub async fn write_server_msg( + &mut self, + msg: lpc_wire::WireServerMessage, + connected: bool, + ) -> Result<(), lpc_wire::TransportError> { + if !connected { + return Err(lpc_wire::TransportError::ConnectionLost); + } + + let result = self.write_full_server_msg(msg).await; + if result.is_err() { + // If a timeout interrupts a JSON frame before the trailing newline, separate the + // next frame so host parsers can recover instead of concatenating two `M!` messages. + let _ = self.write_all(b"\n").await; + } + result + } + + async fn write_full_server_msg( + &mut self, + msg: lpc_wire::WireServerMessage, + ) -> Result<(), lpc_wire::TransportError> { + // TODO(M3 stretch): this ~16.7 KiB stack buffer lives as async-fn state and + // is paid even for tiny acks. The preferred fix — a `SerWrite` that streams + // straight to the chunked+timeout USB writer — is blocked because + // `SerWrite::write` is synchronous while `timed_write_all` is `async`, so the + // streaming impl cannot `.await` the USB write without an internal buffer. + // The StaticCell fallback needs an aliasing/RAM measurement first (io_task is + // the sole writer, but drain paths interleave), so it is deferred rather than + // forced here. Revisit once an async-capable streaming writer exists. + // + // Derived from the shared budget: the serial buffer already reserves the + // frame budget plus `PROJECT_READ_FRAME_SERIAL_MARGIN_BYTES`; this only adds + // room for the `\nM!` framing prefix and trailing `\n` written around the + // message (4 bytes, padded to 16 for alignment slack). + const SERVER_MSG_FRAMING_BYTES: usize = 16; + const SERVER_MSG_JSON_BUFFER_SIZE: usize = + lpc_wire::PROJECT_READ_FRAME_SERIAL_BUFFER_BYTES + SERVER_MSG_FRAMING_BYTES; + let mut buf = [0u8; SERVER_MSG_JSON_BUFFER_SIZE]; + let mut writer = StackJsonWriter::new(&mut buf); + if writer.write(b"\nM!").is_err() { + log::warn!("[io_task] server message prefix exceeded JSON buffer"); + return Err(lpc_wire::TransportError::Serialization( + "server message prefix exceeded JSON buffer".into(), + )); + } + // Erased writer: shares one serializer instantiation per wire type with the + // frame-budget measurement path (lpc_wire::ser_write_json_len), instead of + // emitting a second copy of every type's serializer for this sink. + if lpc_wire::ser_write_json_to(&mut writer, &msg).is_err() { + let detail = server_message_detail(&msg); + log::warn!( + "[io_task] server message id={} {} exceeded JSON buffer size={} frame_budget={}; write failed", + msg.id, + detail, + SERVER_MSG_JSON_BUFFER_SIZE, + lpc_wire::PROJECT_READ_FRAME_MAX_BYTES + ); + return Err(lpc_wire::TransportError::Serialization(format!( + "server message id={} {} exceeded JSON buffer", + msg.id, detail + ))); + } + if writer.write(b"\n").is_err() { + let detail = server_message_detail(&msg); + log::warn!( + "[io_task] server message id={} {} suffix exceeded JSON buffer size={}; write failed", + msg.id, + detail, + SERVER_MSG_JSON_BUFFER_SIZE + ); + return Err(lpc_wire::TransportError::Serialization(format!( + "server message id={} {} suffix exceeded JSON buffer", + msg.id, detail + ))); + } + let id = msg.id; + let link_name = self.policy.link_name; + // `writer` borrows `buf`, which lives in this frame; the write must be + // driven from here rather than returned. + if self.write_all(writer.bytes()).await { + Ok(()) + } else { + Err(lpc_wire::TransportError::Other(format!( + "server message id={id} {link_name} write timed out or failed" + ))) + } + } +} + +/// One-line human description of a server message, for the buffer-overflow +/// warnings above. +pub fn server_message_detail(msg: &lpc_wire::WireServerMessage) -> String { + match &msg.msg { + lpc_wire::server::ServerMsgBody::Hello(hello) => { + format!("Hello proto={}", hello.proto) + } + lpc_wire::server::ServerMsgBody::Filesystem(_) => "Filesystem".into(), + lpc_wire::server::ServerMsgBody::LoadProject { .. } => "LoadProject".into(), + lpc_wire::server::ServerMsgBody::UnloadProject => "UnloadProject".into(), + lpc_wire::server::ServerMsgBody::ProjectRead { events } => format!( + "ProjectRead seq={} fin={} events={} [{}]", + msg.seq, + msg.fin, + events.len(), + project_read_event_summary(events) + ), + lpc_wire::server::ServerMsgBody::ProjectCommand { .. } => "ProjectCommand".into(), + lpc_wire::server::ServerMsgBody::ListAvailableProjects { projects } => { + format!("ListAvailableProjects projects={}", projects.len()) + } + lpc_wire::server::ServerMsgBody::ListLoadedProjects { projects } => { + format!("ListLoadedProjects projects={}", projects.len()) + } + lpc_wire::server::ServerMsgBody::StopAllProjects => "StopAllProjects".into(), + lpc_wire::server::ServerMsgBody::SetLogLevel => "SetLogLevel".into(), + lpc_wire::server::ServerMsgBody::Log { level, .. } => { + format!("Log level={level:?}") + } + lpc_wire::server::ServerMsgBody::Heartbeat { + frame_count, + loaded_projects, + .. + } => format!( + "Heartbeat frame_count={frame_count} loaded_projects={}", + loaded_projects.len() + ), + lpc_wire::server::ServerMsgBody::Error { .. } => "Error".into(), + } +} + +/// The first eight event kinds in a `ProjectRead` frame, comma-separated. +pub fn project_read_event_summary(events: &[lpc_wire::ProjectReadEvent]) -> String { + let mut summary = String::new(); + for (index, event) in events.iter().take(8).enumerate() { + if index > 0 { + summary.push_str(", "); + } + summary.push_str(project_read_event_kind(event)); + } + if events.len() > 8 { + summary.push_str(", ..."); + } + summary +} + +/// The static name of one `ProjectRead` event kind. +pub fn project_read_event_kind(event: &lpc_wire::ProjectReadEvent) -> &'static str { + match event { + lpc_wire::ProjectReadEvent::Begin { .. } => "begin", + lpc_wire::ProjectReadEvent::Query { event, .. } => match event { + lpc_wire::ProjectReadQueryEvent::Shapes(_) => "query.shapes", + lpc_wire::ProjectReadQueryEvent::Nodes(_) => "query.nodes", + lpc_wire::ProjectReadQueryEvent::Resources(_) => "query.resources", + lpc_wire::ProjectReadQueryEvent::Runtime(_) => "query.runtime", + }, + lpc_wire::ProjectReadEvent::Probe { event, .. } => match event { + lpc_wire::ProjectReadProbeEvent::Result(_) => "probe.result", + lpc_wire::ProjectReadProbeEvent::ResultBegin { .. } => "probe.result_begin", + lpc_wire::ProjectReadProbeEvent::ResultBytes { .. } => "probe.result_bytes", + lpc_wire::ProjectReadProbeEvent::ResultEnd => "probe.result_end", + }, + lpc_wire::ProjectReadEvent::End { .. } => "end", + lpc_wire::ProjectReadEvent::Error { .. } => "error", + } +} diff --git a/lp-fw/fw-esp32c6/Cargo.toml b/lp-fw/fw-esp32c6/Cargo.toml index 6c936bba2..96c7693f2 100644 --- a/lp-fw/fw-esp32c6/Cargo.toml +++ b/lp-fw/fw-esp32c6/Cargo.toml @@ -24,7 +24,6 @@ server = [ "lpa-server", "lpc-shared", "lpfs", - "ser-write-json", ] # Enable server dependencies (lpa-server, lpc-shared, lpc-model, lpc-wire, lpfs) radio = [ "lpc-hardware", @@ -116,7 +115,6 @@ lpc-model = { path = "../../lp-core/lpc-model", default-features = false, option lpc-wire = { path = "../../lp-core/lpc-wire", default-features = false, optional = true, features = ["ser-write-json"] } lp-perf = { path = "../../lp-base/lp-perf", default-features = false } lp-recovery = { path = "../../lp-base/lp-recovery", default-features = false } -ser-write-json = { version = "0.3", optional = true, default-features = false, features = ["alloc"] } serde = { workspace = true, default-features = false, features = ["alloc", "derive"] } hashbrown = { workspace = true } diff --git a/lp-fw/fw-esp32c6/src/serial/io_task.rs b/lp-fw/fw-esp32c6/src/serial/io_task.rs index 37c8f1bde..b2f8a8ada 100644 --- a/lp-fw/fw-esp32c6/src/serial/io_task.rs +++ b/lp-fw/fw-esp32c6/src/serial/io_task.rs @@ -6,19 +6,25 @@ //! - Read from serial and push to incoming queue (filter M! prefix) //! - Monitor USB host connection; skip writes when disconnected to prevent blocking //! - All serial writes use timeouts to prevent blocking if host disconnects mid-write +//! +//! The JSON-serialization half — the stack JSON writer, the framed +//! server-message write, and the chunked-with-timeout byte writer under both — +//! is chip-agnostic and lives in `fw_esp32_common::serial`, shared with +//! fw-esp32s3. What stays here is the transport: the USB-Serial-JTAG +//! peripheral, the connection monitor, the not-draining probe, and the +//! channels. extern crate alloc; -use alloc::{format, string::String, vec::Vec}; +use alloc::{string::String, vec::Vec}; use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; use embassy_sync::channel::Channel; use embassy_time::{Duration, Timer}; use embedded_io_async::{Read, Write}; use esp_hal::usb_serial_jtag::UsbSerialJtag; use fw_core::message_router::MessageRouter; +use fw_esp32_common::serial::chunked_write::{ChunkedWriter, WritePolicy}; use log; -#[cfg(feature = "server")] -use ser_write_json::SerWrite; use crate::board::esp32c6::usb_connection::UsbConnectionMonitor; @@ -51,13 +57,6 @@ static SERVER_WRITE_RESULT: Channel< 1, > = Channel::new(); -/// Write timeout per chunk: if a chunk doesn't complete in this time, the host -/// is not draining. A healthy USB full-speed host drains a chunk in well under -/// a millisecond, so this is still very generous — but short enough that the -/// frame loop's inline sends stall briefly, not for seconds, in the window -/// before the not-draining latch kicks in. -const WRITE_TIMEOUT: Duration = Duration::from_millis(250); - /// Probe timeout while latched not-draining: a single byte either leaves /// immediately (host is back) or it doesn't — no reason to wait long. const PROBE_TIMEOUT: Duration = Duration::from_millis(100); @@ -65,79 +64,26 @@ const PROBE_TIMEOUT: Duration = Duration::from_millis(100); /// Minimum spacing between probe writes while latched not-draining. const PROBE_INTERVAL: Duration = Duration::from_millis(2000); -/// Chunk size for large writes. Small enough to avoid timeout on slow USB, -/// large enough to avoid excessive syscalls. Resource snapshots can be 10KB+. -const WRITE_CHUNK_SIZE: usize = 256; - -/// Write all data in chunks with per-chunk timeout. Prevents large messages -/// (e.g. resource snapshots) from timing out mid-write and corrupting the stream -/// by concatenating with the next message. Uses write_all per chunk to -/// handle partial writes. -async fn timed_write_all(tx: &mut W, data: &[u8]) -> bool { - timed_write_all_with(tx, data, WRITE_TIMEOUT).await -} - -async fn timed_write_all_with(tx: &mut W, data: &[u8], timeout: Duration) -> bool { - use embassy_futures::select::{Either, select}; - let mut offset = 0; - while offset < data.len() { - // A bounded in-flight write is healthy, not silence: tick liveness - // per chunk so a slow host cannot starve the watchdog feeder into - // resetting the device. - crate::recovery::watchdog::note_io_alive(); - let chunk_end = (offset + WRITE_CHUNK_SIZE).min(data.len()); - let chunk = &data[offset..chunk_end]; - match select(Timer::after(timeout), tx.write_all(chunk)).await { - Either::First(_) => return false, - Either::Second(Err(_)) => return false, - Either::Second(Ok(())) => {} - } - offset = chunk_end; - } - true -} - -#[cfg(feature = "server")] -struct StackJsonWriter<'a> { - buf: &'a mut [u8], - len: usize, -} - -#[cfg(feature = "server")] -#[derive(Debug)] -struct StackJsonError; - -#[cfg(feature = "server")] -impl core::fmt::Display for StackJsonError { - fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { - f.write_str("stack JSON buffer full") - } -} - -#[cfg(feature = "server")] -impl<'a> StackJsonWriter<'a> { - fn new(buf: &'a mut [u8]) -> Self { - Self { buf, len: 0 } - } - - fn bytes(&self) -> &[u8] { - &self.buf[..self.len] - } -} - -#[cfg(feature = "server")] -impl ser_write_json::SerWrite for StackJsonWriter<'_> { - type Error = StackJsonError; - - fn write(&mut self, buf: &[u8]) -> Result<(), StackJsonError> { - let end = self.len.checked_add(buf.len()).ok_or(StackJsonError)?; - if end > self.buf.len() { - return Err(StackJsonError); - } - self.buf[self.len..end].copy_from_slice(buf); - self.len = end; - Ok(()) - } +/// Wrap this board's TX half in the shared chunked writer. +/// +/// The per-chunk hook is the one crate-specific seam in that writer. A bounded +/// in-flight write is healthy, not silence, so liveness is ticked once per +/// chunk and a slow host cannot starve the watchdog feeder into resetting the +/// device. The RWDT is an `esp-hal` peripheral and therefore a chip fact, which +/// fw-esp32-common is forbidden to hold — so it takes the tick as a hook rather +/// than calling `note_io_alive` itself. Timeout and chunk size come from +/// [`WritePolicy::USB_SERIAL_JTAG`]. +/// +/// The hook is returned as an opaque `impl FnMut()` — the zero-sized fn *item* +/// — rather than a `fn()` pointer. Naming it `fn()` compiles and reads a little +/// plainer, and costs 256 B of un-inlinable indirect calls in this image +/// (measured with `just fw-esp32c6-size-check`). +fn link_writer(tx: &mut W) -> ChunkedWriter<'_, W, impl FnMut()> { + ChunkedWriter::new( + tx, + WritePolicy::USB_SERIAL_JTAG, + crate::recovery::watchdog::note_io_alive, + ) } /// I/O task for handling serial communication @@ -177,7 +123,10 @@ pub async fn io_task(usb_device: esp_hal::peripherals::USB_DEVICE<'static>) { // sending anything. if conn.needs_probe() && last_probe.elapsed() >= PROBE_INTERVAL { last_probe = embassy_time::Instant::now(); - if timed_write_all_with(&mut tx, b"\n", PROBE_TIMEOUT).await { + if link_writer(&mut tx) + .write_all_with(b"\n", PROBE_TIMEOUT) + .await + { conn.note_host_active(); } } @@ -203,14 +152,15 @@ async fn drain_outgoing_messages( conn: &mut UsbConnectionMonitor, ) { let receiver = router.outgoing().receiver(); + let mut writer = link_writer(tx); loop { match receiver.try_receive() { Ok(msg) if conn.is_connected() => { - if !timed_write_all(tx, b"\n").await { + if !writer.write_all(b"\n").await { conn.note_write_timeout(); break; } - if !timed_write_all(tx, msg.as_bytes()).await { + if !writer.write_all(msg.as_bytes()).await { conn.note_write_timeout(); break; } @@ -254,7 +204,9 @@ async fn drain_server_write_request(tx: &mut W, conn: &mut UsbConnecti return; }; - let result = timed_write_server_msg(tx, msg, conn.is_connected()).await; + let result = link_writer(tx) + .write_server_msg(msg, conn.is_connected()) + .await; match &result { Ok(()) => conn.note_host_active(), // Only a USB write timeout/failure is draining evidence; fail-fast @@ -270,170 +222,6 @@ async fn drain_server_write_request(tx: &mut W, conn: &mut UsbConnecti .await; } -#[cfg(feature = "server")] -async fn timed_write_server_msg( - tx: &mut W, - msg: lpc_wire::WireServerMessage, - connected: bool, -) -> Result<(), lpc_wire::TransportError> { - if !connected { - return Err(lpc_wire::TransportError::ConnectionLost); - } - - let result = timed_write_full_server_msg(tx, msg).await; - if result.is_err() { - // If a timeout interrupts a JSON frame before the trailing newline, separate the - // next frame so host parsers can recover instead of concatenating two `M!` messages. - let _ = timed_write_all(tx, b"\n").await; - } - result -} - -#[cfg(feature = "server")] -async fn timed_write_full_server_msg( - tx: &mut W, - msg: lpc_wire::WireServerMessage, -) -> Result<(), lpc_wire::TransportError> { - // TODO(M3 stretch): this ~16.7 KiB stack buffer lives as async-fn state and - // is paid even for tiny acks. The preferred fix — a `SerWrite` that streams - // straight to the chunked+timeout USB writer — is blocked because - // `SerWrite::write` is synchronous while `timed_write_all` is `async`, so the - // streaming impl cannot `.await` the USB write without an internal buffer. - // The StaticCell fallback needs an aliasing/RAM measurement first (io_task is - // the sole writer, but drain paths interleave), so it is deferred rather than - // forced here. Revisit once an async-capable streaming writer exists. - // - // Derived from the shared budget: the serial buffer already reserves the - // frame budget plus `PROJECT_READ_FRAME_SERIAL_MARGIN_BYTES`; this only adds - // room for the `\nM!` framing prefix and trailing `\n` written around the - // message (4 bytes, padded to 16 for alignment slack). - const SERVER_MSG_FRAMING_BYTES: usize = 16; - const SERVER_MSG_JSON_BUFFER_SIZE: usize = - lpc_wire::PROJECT_READ_FRAME_SERIAL_BUFFER_BYTES + SERVER_MSG_FRAMING_BYTES; - let mut buf = [0u8; SERVER_MSG_JSON_BUFFER_SIZE]; - let mut writer = StackJsonWriter::new(&mut buf); - if writer.write(b"\nM!").is_err() { - log::warn!("[io_task] server message prefix exceeded JSON buffer"); - return Err(lpc_wire::TransportError::Serialization( - "server message prefix exceeded JSON buffer".into(), - )); - } - // Erased writer: shares one serializer instantiation per wire type with the - // frame-budget measurement path (lpc_wire::ser_write_json_len), instead of - // emitting a second copy of every type's serializer for this sink. - if lpc_wire::ser_write_json_to(&mut writer, &msg).is_err() { - let detail = server_message_detail(&msg); - log::warn!( - "[io_task] server message id={} {} exceeded JSON buffer size={} frame_budget={}; write failed", - msg.id, - detail, - SERVER_MSG_JSON_BUFFER_SIZE, - lpc_wire::PROJECT_READ_FRAME_MAX_BYTES - ); - return Err(lpc_wire::TransportError::Serialization(format!( - "server message id={} {} exceeded JSON buffer", - msg.id, detail - ))); - } - if writer.write(b"\n").is_err() { - let detail = server_message_detail(&msg); - log::warn!( - "[io_task] server message id={} {} suffix exceeded JSON buffer size={}; write failed", - msg.id, - detail, - SERVER_MSG_JSON_BUFFER_SIZE - ); - return Err(lpc_wire::TransportError::Serialization(format!( - "server message id={} {} suffix exceeded JSON buffer", - msg.id, detail - ))); - } - if timed_write_all(tx, writer.bytes()).await { - Ok(()) - } else { - Err(lpc_wire::TransportError::Other(format!( - "server message id={} USB write timed out or failed", - msg.id - ))) - } -} - -#[cfg(feature = "server")] -fn server_message_detail(msg: &lpc_wire::WireServerMessage) -> String { - match &msg.msg { - lpc_wire::server::ServerMsgBody::Hello(hello) => { - format!("Hello proto={}", hello.proto) - } - lpc_wire::server::ServerMsgBody::Filesystem(_) => "Filesystem".into(), - lpc_wire::server::ServerMsgBody::LoadProject { .. } => "LoadProject".into(), - lpc_wire::server::ServerMsgBody::UnloadProject => "UnloadProject".into(), - lpc_wire::server::ServerMsgBody::ProjectRead { events } => format!( - "ProjectRead seq={} fin={} events={} [{}]", - msg.seq, - msg.fin, - events.len(), - project_read_event_summary(events) - ), - lpc_wire::server::ServerMsgBody::ProjectCommand { .. } => "ProjectCommand".into(), - lpc_wire::server::ServerMsgBody::ListAvailableProjects { projects } => { - format!("ListAvailableProjects projects={}", projects.len()) - } - lpc_wire::server::ServerMsgBody::ListLoadedProjects { projects } => { - format!("ListLoadedProjects projects={}", projects.len()) - } - lpc_wire::server::ServerMsgBody::StopAllProjects => "StopAllProjects".into(), - lpc_wire::server::ServerMsgBody::SetLogLevel => "SetLogLevel".into(), - lpc_wire::server::ServerMsgBody::Log { level, .. } => { - format!("Log level={level:?}") - } - lpc_wire::server::ServerMsgBody::Heartbeat { - frame_count, - loaded_projects, - .. - } => format!( - "Heartbeat frame_count={frame_count} loaded_projects={}", - loaded_projects.len() - ), - lpc_wire::server::ServerMsgBody::Error { .. } => "Error".into(), - } -} - -#[cfg(feature = "server")] -fn project_read_event_summary(events: &[lpc_wire::ProjectReadEvent]) -> String { - let mut summary = String::new(); - for (index, event) in events.iter().take(8).enumerate() { - if index > 0 { - summary.push_str(", "); - } - summary.push_str(project_read_event_kind(event)); - } - if events.len() > 8 { - summary.push_str(", ..."); - } - summary -} - -#[cfg(feature = "server")] -fn project_read_event_kind(event: &lpc_wire::ProjectReadEvent) -> &'static str { - match event { - lpc_wire::ProjectReadEvent::Begin { .. } => "begin", - lpc_wire::ProjectReadEvent::Query { event, .. } => match event { - lpc_wire::ProjectReadQueryEvent::Shapes(_) => "query.shapes", - lpc_wire::ProjectReadQueryEvent::Nodes(_) => "query.nodes", - lpc_wire::ProjectReadQueryEvent::Resources(_) => "query.resources", - lpc_wire::ProjectReadQueryEvent::Runtime(_) => "query.runtime", - }, - lpc_wire::ProjectReadEvent::Probe { event, .. } => match event { - lpc_wire::ProjectReadProbeEvent::Result(_) => "probe.result", - lpc_wire::ProjectReadProbeEvent::ResultBegin { .. } => "probe.result_begin", - lpc_wire::ProjectReadProbeEvent::ResultBytes { .. } => "probe.result_bytes", - lpc_wire::ProjectReadProbeEvent::ResultEnd => "probe.result_end", - }, - lpc_wire::ProjectReadEvent::End { .. } => "end", - lpc_wire::ProjectReadEvent::Error { .. } => "error", - } -} - /// Process read buffer and extract complete lines /// /// Looks for newlines, extracts lines starting with `M!`, and pushes to incoming queue. diff --git a/lp-fw/fw-esp32s3/Cargo.toml b/lp-fw/fw-esp32s3/Cargo.toml index b20d1451c..a36bc2e56 100644 --- a/lp-fw/fw-esp32s3/Cargo.toml +++ b/lp-fw/fw-esp32s3/Cargo.toml @@ -18,7 +18,8 @@ esp32s3 = [ ] # The LightPlayer app: `LpServer` over the USB-Serial-JTAG transport, plus the -# wire-protocol serialization inside `serial::io_task`. +# wire-protocol serialization `serial::io_task` drives (which lives in +# `fw_esp32_common::serial::server_msg`, hence the forward below). # # ⚠️ Still NARROWER than fw-esp32c6's feature of the same name: this one takes # `lpa-server` with **two** of the eight node gates rather than all of them. @@ -32,7 +33,6 @@ server = [ "dep:lpc-hardware", "dep:lpfs", "dep:lpc-wire", - "dep:ser-write-json", ] # Print every transmitted WS281x frame to the serial log: one full hex dump per @@ -130,13 +130,15 @@ embedded-storage = "0.3" esp-storage = { version = "0.9", default-features = false, features = ["critical-section", "esp32s3", "panic-unaligned-buffer"] } littlefs-rust = { version = "0.1.0", default-features = false, features = ["alloc"] } -# io_task's wire-serialization path plus the boot-time `ServerHello`. +# The boot-time `ServerHello` plus the wire types `io_task` hands to +# `fw_esp32_common::serial::server_msg`. The `ser-write-json` sink itself moved +# down with that module, so this crate no longer names the serializer. lpc-wire = { path = "../../lp-core/lpc-wire", default-features = false, optional = true, features = ["ser-write-json"] } -ser-write-json = { version = "0.3", optional = true, default-features = false, features = ["alloc"] } -# Chip-generic firmware layer: logger, time provider, output provider, and — -# behind `server` — the server loop, transport, littlefs `LpFs`, project -# auto-load, and hardware-manifest loader. Consumed, never reimplemented; this +# Chip-generic firmware layer: logger, time provider, output provider, the +# chunked serial writer, and — behind `server` — the server loop, transport, +# the io_task JSON-serialization half, littlefs `LpFs`, project auto-load, and +# hardware-manifest loader. Consumed, never reimplemented; this # crate must stay free of chip code (ADR 2026-07-29-per-chip-fw-toolchains). fw-esp32-common = { path = "../fw-esp32-common", default-features = false } diff --git a/lp-fw/fw-esp32s3/src/serial/io_task.rs b/lp-fw/fw-esp32s3/src/serial/io_task.rs index fdf90b51e..9611dc03b 100644 --- a/lp-fw/fw-esp32s3/src/serial/io_task.rs +++ b/lp-fw/fw-esp32s3/src/serial/io_task.rs @@ -6,31 +6,25 @@ //! - Read from serial and push to incoming queue (filter M! prefix) //! - Monitor USB host connection; skip writes when disconnected to prevent blocking //! - All serial writes use timeouts to prevent blocking if host disconnects mid-write +//! +//! The JSON-serialization half — the stack JSON writer, the framed +//! server-message write, and the chunked-with-timeout byte writer under both — +//! is chip-agnostic and lives in `fw_esp32_common::serial`, shared with +//! fw-esp32c6. What stays here is the transport: the USB-Serial-JTAG +//! peripheral, the connection monitor, the not-draining probe, and the +//! channels. extern crate alloc; -// Sole divergence from fw-esp32c6's copy beyond the `board::esp32s3` import: -// `format!` is used only by the server-gated write paths. `server` is in the -// default features now (M3 P5), same as the C6, so the `not(server)` arm is not -// reachable from any build recipe — the attribute stays only so the two copies -// of this file remain diffable. -#[cfg_attr( - not(feature = "server"), - allow( - unused_imports, - reason = "`format!` is used only by the server-gated write paths" - ) -)] -use alloc::{format, string::String, vec::Vec}; +use alloc::{string::String, vec::Vec}; use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; use embassy_sync::channel::Channel; use embassy_time::{Duration, Timer}; use embedded_io_async::{Read, Write}; use esp_hal::usb_serial_jtag::UsbSerialJtag; use fw_core::message_router::MessageRouter; +use fw_esp32_common::serial::chunked_write::{ChunkedWriter, WritePolicy}; use log; -#[cfg(feature = "server")] -use ser_write_json::SerWrite; use crate::board::esp32s3::usb_connection::UsbConnectionMonitor; @@ -63,13 +57,6 @@ static SERVER_WRITE_RESULT: Channel< 1, > = Channel::new(); -/// Write timeout per chunk: if a chunk doesn't complete in this time, the host -/// is not draining. A healthy USB full-speed host drains a chunk in well under -/// a millisecond, so this is still very generous — but short enough that the -/// frame loop's inline sends stall briefly, not for seconds, in the window -/// before the not-draining latch kicks in. -const WRITE_TIMEOUT: Duration = Duration::from_millis(250); - /// Probe timeout while latched not-draining: a single byte either leaves /// immediately (host is back) or it doesn't — no reason to wait long. const PROBE_TIMEOUT: Duration = Duration::from_millis(100); @@ -77,79 +64,26 @@ const PROBE_TIMEOUT: Duration = Duration::from_millis(100); /// Minimum spacing between probe writes while latched not-draining. const PROBE_INTERVAL: Duration = Duration::from_millis(2000); -/// Chunk size for large writes. Small enough to avoid timeout on slow USB, -/// large enough to avoid excessive syscalls. Resource snapshots can be 10KB+. -const WRITE_CHUNK_SIZE: usize = 256; - -/// Write all data in chunks with per-chunk timeout. Prevents large messages -/// (e.g. resource snapshots) from timing out mid-write and corrupting the stream -/// by concatenating with the next message. Uses write_all per chunk to -/// handle partial writes. -async fn timed_write_all(tx: &mut W, data: &[u8]) -> bool { - timed_write_all_with(tx, data, WRITE_TIMEOUT).await -} - -async fn timed_write_all_with(tx: &mut W, data: &[u8], timeout: Duration) -> bool { - use embassy_futures::select::{Either, select}; - let mut offset = 0; - while offset < data.len() { - // A bounded in-flight write is healthy, not silence: tick liveness - // per chunk so a slow host cannot starve the watchdog feeder into - // resetting the device. - crate::recovery::watchdog::note_io_alive(); - let chunk_end = (offset + WRITE_CHUNK_SIZE).min(data.len()); - let chunk = &data[offset..chunk_end]; - match select(Timer::after(timeout), tx.write_all(chunk)).await { - Either::First(_) => return false, - Either::Second(Err(_)) => return false, - Either::Second(Ok(())) => {} - } - offset = chunk_end; - } - true -} - -#[cfg(feature = "server")] -struct StackJsonWriter<'a> { - buf: &'a mut [u8], - len: usize, -} - -#[cfg(feature = "server")] -#[derive(Debug)] -struct StackJsonError; - -#[cfg(feature = "server")] -impl core::fmt::Display for StackJsonError { - fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { - f.write_str("stack JSON buffer full") - } -} - -#[cfg(feature = "server")] -impl<'a> StackJsonWriter<'a> { - fn new(buf: &'a mut [u8]) -> Self { - Self { buf, len: 0 } - } - - fn bytes(&self) -> &[u8] { - &self.buf[..self.len] - } -} - -#[cfg(feature = "server")] -impl ser_write_json::SerWrite for StackJsonWriter<'_> { - type Error = StackJsonError; - - fn write(&mut self, buf: &[u8]) -> Result<(), StackJsonError> { - let end = self.len.checked_add(buf.len()).ok_or(StackJsonError)?; - if end > self.buf.len() { - return Err(StackJsonError); - } - self.buf[self.len..end].copy_from_slice(buf); - self.len = end; - Ok(()) - } +/// Wrap this board's TX half in the shared chunked writer. +/// +/// The per-chunk hook is the one crate-specific seam in that writer. A bounded +/// in-flight write is healthy, not silence, so liveness is ticked once per +/// chunk and a slow host cannot starve the watchdog feeder into resetting the +/// device. The RWDT is an `esp-hal` peripheral and therefore a chip fact, which +/// fw-esp32-common is forbidden to hold — so it takes the tick as a hook rather +/// than calling `note_io_alive` itself. Timeout and chunk size come from +/// [`WritePolicy::USB_SERIAL_JTAG`]. +/// +/// The hook is returned as an opaque `impl FnMut()` — the zero-sized fn *item* +/// — rather than a `fn()` pointer. Naming it `fn()` compiles and reads a little +/// plainer, and costs 256 B of un-inlinable indirect calls in this image +/// (measured with `just fw-esp32c6-size-check`). +fn link_writer(tx: &mut W) -> ChunkedWriter<'_, W, impl FnMut()> { + ChunkedWriter::new( + tx, + WritePolicy::USB_SERIAL_JTAG, + crate::recovery::watchdog::note_io_alive, + ) } /// I/O task for handling serial communication @@ -189,7 +123,10 @@ pub async fn io_task(usb_device: esp_hal::peripherals::USB_DEVICE<'static>) { // sending anything. if conn.needs_probe() && last_probe.elapsed() >= PROBE_INTERVAL { last_probe = embassy_time::Instant::now(); - if timed_write_all_with(&mut tx, b"\n", PROBE_TIMEOUT).await { + if link_writer(&mut tx) + .write_all_with(b"\n", PROBE_TIMEOUT) + .await + { conn.note_host_active(); } } @@ -215,14 +152,15 @@ async fn drain_outgoing_messages( conn: &mut UsbConnectionMonitor, ) { let receiver = router.outgoing().receiver(); + let mut writer = link_writer(tx); loop { match receiver.try_receive() { Ok(msg) if conn.is_connected() => { - if !timed_write_all(tx, b"\n").await { + if !writer.write_all(b"\n").await { conn.note_write_timeout(); break; } - if !timed_write_all(tx, msg.as_bytes()).await { + if !writer.write_all(msg.as_bytes()).await { conn.note_write_timeout(); break; } @@ -266,7 +204,9 @@ async fn drain_server_write_request(tx: &mut W, conn: &mut UsbConnecti return; }; - let result = timed_write_server_msg(tx, msg, conn.is_connected()).await; + let result = link_writer(tx) + .write_server_msg(msg, conn.is_connected()) + .await; match &result { Ok(()) => conn.note_host_active(), // Only a USB write timeout/failure is draining evidence; fail-fast @@ -282,170 +222,6 @@ async fn drain_server_write_request(tx: &mut W, conn: &mut UsbConnecti .await; } -#[cfg(feature = "server")] -async fn timed_write_server_msg( - tx: &mut W, - msg: lpc_wire::WireServerMessage, - connected: bool, -) -> Result<(), lpc_wire::TransportError> { - if !connected { - return Err(lpc_wire::TransportError::ConnectionLost); - } - - let result = timed_write_full_server_msg(tx, msg).await; - if result.is_err() { - // If a timeout interrupts a JSON frame before the trailing newline, separate the - // next frame so host parsers can recover instead of concatenating two `M!` messages. - let _ = timed_write_all(tx, b"\n").await; - } - result -} - -#[cfg(feature = "server")] -async fn timed_write_full_server_msg( - tx: &mut W, - msg: lpc_wire::WireServerMessage, -) -> Result<(), lpc_wire::TransportError> { - // TODO(M3 stretch): this ~16.7 KiB stack buffer lives as async-fn state and - // is paid even for tiny acks. The preferred fix — a `SerWrite` that streams - // straight to the chunked+timeout USB writer — is blocked because - // `SerWrite::write` is synchronous while `timed_write_all` is `async`, so the - // streaming impl cannot `.await` the USB write without an internal buffer. - // The StaticCell fallback needs an aliasing/RAM measurement first (io_task is - // the sole writer, but drain paths interleave), so it is deferred rather than - // forced here. Revisit once an async-capable streaming writer exists. - // - // Derived from the shared budget: the serial buffer already reserves the - // frame budget plus `PROJECT_READ_FRAME_SERIAL_MARGIN_BYTES`; this only adds - // room for the `\nM!` framing prefix and trailing `\n` written around the - // message (4 bytes, padded to 16 for alignment slack). - const SERVER_MSG_FRAMING_BYTES: usize = 16; - const SERVER_MSG_JSON_BUFFER_SIZE: usize = - lpc_wire::PROJECT_READ_FRAME_SERIAL_BUFFER_BYTES + SERVER_MSG_FRAMING_BYTES; - let mut buf = [0u8; SERVER_MSG_JSON_BUFFER_SIZE]; - let mut writer = StackJsonWriter::new(&mut buf); - if writer.write(b"\nM!").is_err() { - log::warn!("[io_task] server message prefix exceeded JSON buffer"); - return Err(lpc_wire::TransportError::Serialization( - "server message prefix exceeded JSON buffer".into(), - )); - } - // Erased writer: shares one serializer instantiation per wire type with the - // frame-budget measurement path (lpc_wire::ser_write_json_len), instead of - // emitting a second copy of every type's serializer for this sink. - if lpc_wire::ser_write_json_to(&mut writer, &msg).is_err() { - let detail = server_message_detail(&msg); - log::warn!( - "[io_task] server message id={} {} exceeded JSON buffer size={} frame_budget={}; write failed", - msg.id, - detail, - SERVER_MSG_JSON_BUFFER_SIZE, - lpc_wire::PROJECT_READ_FRAME_MAX_BYTES - ); - return Err(lpc_wire::TransportError::Serialization(format!( - "server message id={} {} exceeded JSON buffer", - msg.id, detail - ))); - } - if writer.write(b"\n").is_err() { - let detail = server_message_detail(&msg); - log::warn!( - "[io_task] server message id={} {} suffix exceeded JSON buffer size={}; write failed", - msg.id, - detail, - SERVER_MSG_JSON_BUFFER_SIZE - ); - return Err(lpc_wire::TransportError::Serialization(format!( - "server message id={} {} suffix exceeded JSON buffer", - msg.id, detail - ))); - } - if timed_write_all(tx, writer.bytes()).await { - Ok(()) - } else { - Err(lpc_wire::TransportError::Other(format!( - "server message id={} USB write timed out or failed", - msg.id - ))) - } -} - -#[cfg(feature = "server")] -fn server_message_detail(msg: &lpc_wire::WireServerMessage) -> String { - match &msg.msg { - lpc_wire::server::ServerMsgBody::Hello(hello) => { - format!("Hello proto={}", hello.proto) - } - lpc_wire::server::ServerMsgBody::Filesystem(_) => "Filesystem".into(), - lpc_wire::server::ServerMsgBody::LoadProject { .. } => "LoadProject".into(), - lpc_wire::server::ServerMsgBody::UnloadProject => "UnloadProject".into(), - lpc_wire::server::ServerMsgBody::ProjectRead { events } => format!( - "ProjectRead seq={} fin={} events={} [{}]", - msg.seq, - msg.fin, - events.len(), - project_read_event_summary(events) - ), - lpc_wire::server::ServerMsgBody::ProjectCommand { .. } => "ProjectCommand".into(), - lpc_wire::server::ServerMsgBody::ListAvailableProjects { projects } => { - format!("ListAvailableProjects projects={}", projects.len()) - } - lpc_wire::server::ServerMsgBody::ListLoadedProjects { projects } => { - format!("ListLoadedProjects projects={}", projects.len()) - } - lpc_wire::server::ServerMsgBody::StopAllProjects => "StopAllProjects".into(), - lpc_wire::server::ServerMsgBody::SetLogLevel => "SetLogLevel".into(), - lpc_wire::server::ServerMsgBody::Log { level, .. } => { - format!("Log level={level:?}") - } - lpc_wire::server::ServerMsgBody::Heartbeat { - frame_count, - loaded_projects, - .. - } => format!( - "Heartbeat frame_count={frame_count} loaded_projects={}", - loaded_projects.len() - ), - lpc_wire::server::ServerMsgBody::Error { .. } => "Error".into(), - } -} - -#[cfg(feature = "server")] -fn project_read_event_summary(events: &[lpc_wire::ProjectReadEvent]) -> String { - let mut summary = String::new(); - for (index, event) in events.iter().take(8).enumerate() { - if index > 0 { - summary.push_str(", "); - } - summary.push_str(project_read_event_kind(event)); - } - if events.len() > 8 { - summary.push_str(", ..."); - } - summary -} - -#[cfg(feature = "server")] -fn project_read_event_kind(event: &lpc_wire::ProjectReadEvent) -> &'static str { - match event { - lpc_wire::ProjectReadEvent::Begin { .. } => "begin", - lpc_wire::ProjectReadEvent::Query { event, .. } => match event { - lpc_wire::ProjectReadQueryEvent::Shapes(_) => "query.shapes", - lpc_wire::ProjectReadQueryEvent::Nodes(_) => "query.nodes", - lpc_wire::ProjectReadQueryEvent::Resources(_) => "query.resources", - lpc_wire::ProjectReadQueryEvent::Runtime(_) => "query.runtime", - }, - lpc_wire::ProjectReadEvent::Probe { event, .. } => match event { - lpc_wire::ProjectReadProbeEvent::Result(_) => "probe.result", - lpc_wire::ProjectReadProbeEvent::ResultBegin { .. } => "probe.result_begin", - lpc_wire::ProjectReadProbeEvent::ResultBytes { .. } => "probe.result_bytes", - lpc_wire::ProjectReadProbeEvent::ResultEnd => "probe.result_end", - }, - lpc_wire::ProjectReadEvent::End { .. } => "end", - lpc_wire::ProjectReadEvent::Error { .. } => "error", - } -} - /// Process read buffer and extract complete lines /// /// Looks for newlines, extracts lines starting with `M!`, and pushes to incoming queue.