From ecdbe4517174dddd5c4b641ee9064e61077a484d Mon Sep 17 00:00:00 2001 From: ashaffah Date: Tue, 21 Jul 2026 15:49:58 +0700 Subject: [PATCH] fix(edge-client): serialize Modbus TCP read/write over one shared connection Small PLCs / Modbus-TCP gateways (e.g. vakum2t) accept only one concurrent TCP socket. The old 2-client pattern opened a separate reader (telemetry) and writer (control) socket to the same host:port, so the writer's socket never became usable and every control write failed with EINPROGRESS (os error 115), even when the control gate granted access. Route every TCP transport through the existing actor pattern (previously only RTU): one connection per endpoint, read+write serialized over an mpsc channel. Actors are shared via ActorCache keyed by connection identity (tcp:host:port / rtu-tcp:host:port / serial:path): - Same host:port (vakum2t :502 read+write) -> one shared connection, fixed. - Split-port (vakum750 :502 read / :503 write) -> two independent actors, unchanged behavior. Remove the now-orphaned direct-socket path (ModbusReader/ModbusWriter, Backend::Tcp, for_tcp, with_unit_id). --- src/control_subscriber.rs | 4 +- src/main.rs | 8 +- src/modbus_actor.rs | 63 ++-- src/modbus_client.rs | 619 ++++++++++++-------------------------- src/telemetry.rs | 12 +- 5 files changed, 228 insertions(+), 478 deletions(-) diff --git a/src/control_subscriber.rs b/src/control_subscriber.rs index b54d623..c9a5d08 100644 --- a/src/control_subscriber.rs +++ b/src/control_subscriber.rs @@ -117,7 +117,7 @@ impl ControlSubscriberConfig { pub async fn from_settings( settings: &Settings, plc_writer_factory: WriterFactory, - serial_cache: &crate::modbus_client::SerialActorCache, + actor_cache: &crate::modbus_client::ActorCache, ) -> Result { settings.modbus.as_ref().context( "ControlSubscriberConfig::from_settings called without a PLC in the mapping", @@ -143,7 +143,7 @@ impl ControlSubscriberConfig { } // Additional PLC or slave — build the factory from the connection block. let factory = - crate::modbus_client::build_source_writer(&device.connection, serial_cache) + crate::modbus_client::build_source_writer(&device.connection, actor_cache) .await .with_context(|| format!("build writer for device '{}'", device.device_id))?; writer_factories diff --git a/src/main.rs b/src/main.rs index 801009e..a367959 100644 --- a/src/main.rs +++ b/src/main.rs @@ -155,10 +155,10 @@ async fn main() -> Result<()> { // If there isn't, the telemetry/control/plc_status tasks use pending() // (idle). // --------------------------------------------------------------------------- - let serial_cache = modbus_client::SerialActorCache::new(); + let actor_cache = modbus_client::ActorCache::new(); let (reader_factory_opt, writer_factory_opt) = if let Some(modbus) = &settings.modbus { - let (rf, wf) = modbus_client::build_io(modbus, &serial_cache) + let (rf, wf) = modbus_client::build_io(modbus, &actor_cache) .await .context("build modbus I/O")?; (Some(rf), Some(wf)) @@ -170,10 +170,10 @@ async fn main() -> Result<()> { let (plc_status_tx, plc_status_rx) = tokio::sync::watch::channel(false); let tc = - edge_client::telemetry::TelemetryConfig::from_settings(&settings, rf, &serial_cache) + edge_client::telemetry::TelemetryConfig::from_settings(&settings, rf, &actor_cache) .await .context("build telemetry config")?; - let cc = ControlSubscriberConfig::from_settings(&settings, wf, &serial_cache) + let cc = ControlSubscriberConfig::from_settings(&settings, wf, &actor_cache) .await .context("build control subscriber config")?; diff --git a/src/modbus_actor.rs b/src/modbus_actor.rs index 7346ac7..46502ae 100644 --- a/src/modbus_actor.rs +++ b/src/modbus_actor.rs @@ -1,8 +1,11 @@ -//! Actor task for RTU mode (RTU-over-TCP & RTU-Serial). +//! Actor task for every Modbus transport (TCP, RTU-over-TCP, RTU-Serial). //! -//! Modbus RTU = half-duplex, one physical port. It can't be opened twice -//! (serial port: OS lock; RTU-over-TCP: the gateway converter usually doesn't -//! handle multi-client well). So we use the actor pattern: +//! A single connection that serializes all requests. Needed because the physical +//! link can't be opened twice safely: serial port = OS exclusive lock; RTU-over-TCP +//! = the gateway converter usually mishandles multi-client; plain TCP = many small +//! PLCs / Modbus-TCP gateways accept only ONE concurrent socket, so a separate +//! reader + writer socket to the same host:port would make the second one fail +//! (write returns EINPROGRESS / os error 115). So we use the actor pattern: //! //! - **One task** that owns the connection (serial port or TcpStream). //! - **An mpsc channel** for inbound requests from telemetry + control + bridge. @@ -38,7 +41,7 @@ use async_trait::async_trait; use std::time::Duration; use tokio::sync::{mpsc, oneshot}; use tokio_modbus::client::Context as ModbusContext; -use tokio_modbus::prelude::{Reader, Slave, SlaveContext, Writer, rtu}; +use tokio_modbus::prelude::{Reader, Slave, SlaveContext, Writer, rtu, tcp}; use tracing::{debug, info, warn}; use crate::modbus_client::{ReadConn, WriteConn}; @@ -133,22 +136,12 @@ impl ActorHandle { /// Requests that arrive before the connect finishes are queued and executed once /// the connection is up. /// -/// An error here means the transport is fundamentally invalid (e.g. an -/// unsupported variant). Network errors don't surface — the actor handles them -/// internally. +/// Returns `Result` for call-site ergonomics, but currently never fails: every +/// transport is supported and network errors don't surface here — the actor +/// handles connect/reconnect internally. pub async fn spawn(transport: Transport, unit_id: u8, timings: RtuTimings) -> Result { - // Sanity check at startup so misconfig is detected quickly. The real connect - // happens inside the actor loop (for automatic retry-on-fail). - match &transport { - Transport::RtuOverTcp { .. } | Transport::RtuSerial { .. } => {} - Transport::Tcp { .. } => { - return Err(anyhow!( - "modbus_actor::spawn called with Transport::Tcp; this is a bug — \ - TCP should use ModbusReader/Writer direct" - )); - } - } - + // The real connect happens inside the actor loop (for automatic + // retry-on-fail); nothing to validate up front — all transports are handled. let (tx, rx) = mpsc::channel(INBOX_CAP); tokio::spawn(run_actor(transport, timings, rx)); Ok(ActorHandle { @@ -535,9 +528,17 @@ async fn connect(transport: &Transport) -> Result { tokio::time::sleep(Duration::from_millis(200)).await; Ok(rtu::attach_slave(stream, DUMMY_SLAVE)) } - Transport::Tcp { .. } => Err(anyhow!( - "actor connect should not be called with Tcp transport — bug in factory" - )), + Transport::Tcp { host, port } => { + let addr = format!("{host}:{port}"); + let stream = tokio::net::TcpStream::connect(&addr) + .await + .with_context(|| format!("tcp connect {addr}"))?; + // Standard Modbus TCP framing (MBAP). One shared connection, + // serialized by this actor, so a single-session PLC/gateway isn't hit + // by two concurrent sockets (reader + writer). The slave/unit is set + // per request via `ctx.set_slave()`. + Ok(tcp::attach_slave(stream, DUMMY_SLAVE)) + } } } @@ -571,22 +572,18 @@ fn map_data_bits(d: SerialDataBits) -> tokio_serial::DataBits { mod tests { use super::*; - /// Sanity: spawn() explicitly rejects the Tcp transport. Defensive — - /// without this, a factory bug could slip through into an actor task that - /// immediately crashes in connect(). Better to fail in spawn() at startup. + /// spawn() accepts every transport, including plain Modbus TCP. The actor + /// task connects (and reconnects) in the background; spawn itself returns Ok + /// without touching the network. #[tokio::test] - async fn spawn_rejects_tcp_transport() { + async fn spawn_accepts_tcp_transport() { let t = Transport::Tcp { host: "127.0.0.1".into(), port: 502, }; let res = spawn(t, 1, RtuTimings::default()).await; - assert!(res.is_err(), "spawn must reject Tcp transport"); - let msg = res.unwrap_err().to_string(); - assert!( - msg.contains("Tcp"), - "error message should mention Tcp, got: {msg}" - ); + assert!(res.is_ok(), "spawn must accept Tcp transport"); + // Drop the handle → channel closes → the background actor exits. } /// Reply type assertion: if the actor reply variant mismatches (a bug in diff --git a/src/modbus_client.rs b/src/modbus_client.rs index 3624309..0464601 100644 --- a/src/modbus_client.rs +++ b/src/modbus_client.rs @@ -1,11 +1,19 @@ //! Modbus client abstraction: `ReadConn`/`WriteConn` traits + a factory for the //! 3 transports (TCP / RTU-over-TCP / RTU-Serial). //! -//! The TCP path uses `ModbusReader` + `ModbusWriter` as two separate -//! connections (a 2-client pattern). RTU half-duplex cannot be emulated with two -//! separate connections to the same physical port → it needs a single -//! connection that serializes requests. Solution: an actor task that owns the -//! connection while Reader/Writer tasks send requests over an mpsc channel. +//! **Every transport goes through a single actor connection** ([`crate::modbus_actor`]). +//! One task owns the link; reader + writer + bridge send requests over an mpsc +//! channel and the actor serializes them. This is required for RTU (half-duplex, +//! one physical port) and also for plain Modbus TCP, because many small PLCs / +//! Modbus-TCP gateways accept only ONE concurrent socket — a separate reader and +//! writer socket to the same host:port makes the second one fail (write returns +//! EINPROGRESS / os error 115). +//! +//! Actors are shared per endpoint via [`ActorCache`], keyed by connection +//! identity (`tcp:host:port`, `rtu-tcp:host:port`, `serial:path`). So a reader +//! and a writer to the **same** host:port share one connection, while a +//! split-port setup (e.g. `:502` read / `:503` write) keeps two independent +//! connections — one per key. //! //! Two layers: //! @@ -14,35 +22,17 @@ //! which transport is behind it. //! //! 2. **Factories** [`ReaderFactory`] + [`WriterFactory`] = built once at -//! startup ([`build_io`]). The `.connect()` method returns a -//! `Box` or `Box`. Per transport: -//! - **TCP**: each `.connect()` opens a new socket (preserves the 2-client -//! pattern). -//! - **RTU**: the factory holds an actor handle; `.connect()` clones the -//! handle (cheap). Reconnecting the physical link is handled inside the -//! actor. -//! -//! TCP lifecycle: -//! ```text -//! loop { -//! let reader = factory.connect().await?; // open TCP -//! loop { reader.read_holdings(...) } // until error -//! // drop reader → socket close → outer loop reconnects -//! } -//! ``` -//! -//! The RTU lifecycle is identical from the consumer's point of view — but -//! `.connect()` just clones the handle, and a read error = a per-request error -//! (the actor restarts in the background without the caller needing to know). +//! startup ([`build_io`]). Each holds an actor handle; `.connect()` just +//! clones the handle (cheap), returning a `Box` / +//! `Box`. Reconnecting the physical link is handled inside the +//! actor, so a read/write error is a per-request error the caller can retry +//! without managing sockets itself. -use anyhow::{Context as _, Result, anyhow}; +use anyhow::{Context as _, Result}; use async_trait::async_trait; use std::collections::HashMap; -use std::net::SocketAddr; use std::sync::Arc; use tokio::sync::Mutex; -use tokio_modbus::client::Context; -use tokio_modbus::prelude::{Reader, Slave, Writer, tcp}; use tracing::debug; use crate::modbus_actor::ActorHandle; @@ -91,328 +81,118 @@ pub trait WriteConn: Send { async fn write_multiple_registers(&mut self, address: u16, values: Vec) -> Result<()>; } -// =========================================================================== -// TCP: ModbusReader + ModbusWriter (preserve existing 2-client pattern) -// =========================================================================== - -/// Stateful Modbus TCP reader. Connect-per-session: on disconnect/error, drop -/// and recreate it via [`ModbusReader::connect`]. -pub struct ModbusReader { - ctx: Context, -} - -impl ModbusReader { - /// Connect to the PLC via Modbus TCP with a given unit ID. - pub async fn connect(addr: SocketAddr, unit_id: u8) -> Result { - let ctx = tcp::connect_slave(addr, Slave(unit_id)) - .await - .with_context(|| format!("modbus tcp connect {addr} (unit={unit_id})"))?; - debug!(%addr, unit_id, "modbus tcp reader connected"); - Ok(Self { ctx }) - } -} - -#[async_trait] -impl ReadConn for ModbusReader { - async fn read_holdings(&mut self, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - // tokio-modbus 0.16+ separates a transport error (outer Result) from a - // Modbus protocol exception (inner Result, e.g. ILLEGAL_ADDRESS from the - // PLC). Both = fatal in the poll cycle, the handler reconnects. - let res = self - .ctx - .read_holding_registers(0, count) - .await - .context("modbus read_holding_registers transport error")?; - let regs = res.map_err(|e| anyhow!("modbus protocol exception (holdings): {e:?}"))?; - Ok(regs) - } - - async fn read_coils(&mut self, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_coils(0, count) - .await - .context("modbus read_coils transport error")?; - let coils = res.map_err(|e| anyhow!("modbus protocol exception (coils): {e:?}"))?; - Ok(coils) - } - - async fn read_input_registers(&mut self, address: u16, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_input_registers(address, count) - .await - .context("modbus read_input_registers transport error")?; - let regs = - res.map_err(|e| anyhow!("modbus protocol exception (input_registers): {e:?}"))?; - Ok(regs) - } - - async fn read_discrete_inputs(&mut self, address: u16, count: u16) -> Result> { - if count == 0 { - return Ok(Vec::new()); - } - let res = self - .ctx - .read_discrete_inputs(address, count) - .await - .context("modbus read_discrete_inputs transport error")?; - let bits = - res.map_err(|e| anyhow!("modbus protocol exception (discrete_inputs): {e:?}"))?; - Ok(bits) - } -} - -/// Stateful Modbus TCP writer. A connection separate from [`ModbusReader`] — -/// the 2-client pattern so telemetry reads aren't disturbed by control writes. -/// For RTU this lifecycle collapses to a single actor (see -/// [`crate::modbus_actor`]). -pub struct ModbusWriter { - ctx: Context, -} - -impl ModbusWriter { - pub async fn connect(addr: SocketAddr, unit_id: u8) -> Result { - let ctx = tcp::connect_slave(addr, Slave(unit_id)) - .await - .with_context(|| format!("modbus tcp connect (writer) {addr} (unit={unit_id})"))?; - debug!(%addr, unit_id, "modbus tcp writer connected"); - Ok(Self { ctx }) - } -} - -#[async_trait] -impl WriteConn for ModbusWriter { - async fn write_single_register(&mut self, address: u16, value: u16) -> Result<()> { - let res = self - .ctx - .write_single_register(address, value) - .await - .with_context(|| { - format!("modbus write_single_register(addr={address}) transport error") - })?; - res.map_err(|e| { - anyhow!("modbus protocol exception (write register {address}={value}): {e:?}") - })?; - debug!(address, value, "modbus register written"); - Ok(()) - } - - async fn write_single_coil(&mut self, address: u16, value: bool) -> Result<()> { - let res = self - .ctx - .write_single_coil(address, value) - .await - .with_context(|| format!("modbus write_single_coil(addr={address}) transport error"))?; - res.map_err(|e| anyhow!("modbus protocol exception (write coil {address}): {e:?}"))?; - debug!(address, value, "modbus coil written"); - Ok(()) - } - - async fn write_multiple_registers(&mut self, address: u16, values: Vec) -> Result<()> { - let count = values.len(); - let res = self - .ctx - .write_multiple_registers(address, &values) - .await - .with_context(|| { - format!( - "modbus write_multiple_registers(addr={address}, count={count}) transport error" - ) - })?; - res.map_err(|e| { - anyhow!("modbus protocol exception (write multi {address}, {count} regs): {e:?}") - })?; - debug!(address, count, "modbus multi registers written"); - Ok(()) - } -} - // =========================================================================== // Factory layer: ReaderFactory / WriterFactory // =========================================================================== -/// Factory for creating a `ReadConn` session. For TCP, each `connect()` opens a -/// new socket. For RTU, it returns a clone of the actor handle (shared with the -/// `WriterFactory` on the same transport). -#[derive(Clone)] -pub struct ReaderFactory(Backend); - -/// Factory for creating a `WriteConn` session. +/// Factory for creating a `ReadConn` session. Holds an actor handle; `connect()` +/// clones it (cheap). The actor owns the single shared connection. #[derive(Clone)] -pub struct WriterFactory(Backend); +pub struct ReaderFactory(ActorHandle); +/// Factory for creating a `WriteConn` session. Holds an actor handle; `connect()` +/// clones it (cheap). Shares the same actor as the [`ReaderFactory`] when both +/// target the same endpoint (see [`ActorCache`]). #[derive(Clone)] -enum Backend { - /// TCP: descriptor only, the connection happens on each `.connect()` call. - Tcp { addr: SocketAddr, unit_id: u8 }, - /// RTU (over-TCP or serial): a handle to the actor task. Cheap to clone. - Actor(ActorHandle), -} +pub struct WriterFactory(ActorHandle); impl ReaderFactory { pub async fn connect(&self) -> Result> { - match &self.0 { - Backend::Tcp { addr, unit_id } => { - let r = ModbusReader::connect(*addr, *unit_id).await?; - Ok(Box::new(r)) - } - Backend::Actor(h) => Ok(Box::new(h.clone())), - } - } - - /// Derive a new factory for a different slave/unit ID (e.g. PLC unit 1 - /// → chiller unit 2). For TCP: clone the descriptor with the new unit_id - /// (each `.connect()` opens a separate TCP socket with the MBAP unit field - /// = this unit_id). For RTU/Actor: clone the handle with the new slave_id - /// (sharing the same actor task). Cheap in both cases. - pub fn with_unit_id(&self, unit_id: u8) -> Self { - let backend = match &self.0 { - Backend::Tcp { addr, .. } => Backend::Tcp { - addr: *addr, - unit_id, - }, - Backend::Actor(h) => Backend::Actor(h.with_slave_id(unit_id)), - }; - Self(backend) - } - - /// Create a new TCP factory to a different host:port — for a source device - /// that has its own IP (a second PLC, etc.). Independent of the main - /// transport. - pub fn for_tcp(addr: std::net::SocketAddr, unit_id: u8) -> Self { - Self(Backend::Tcp { addr, unit_id }) + Ok(Box::new(self.0.clone())) } } impl WriterFactory { pub async fn connect(&self) -> Result> { - match &self.0 { - Backend::Tcp { addr, unit_id } => { - let w = ModbusWriter::connect(*addr, *unit_id).await?; - Ok(Box::new(w)) - } - Backend::Actor(h) => Ok(Box::new(h.clone())), - } - } - - /// See [`ReaderFactory::with_unit_id`]. - pub fn with_unit_id(&self, unit_id: u8) -> Self { - let backend = match &self.0 { - Backend::Tcp { addr, .. } => Backend::Tcp { - addr: *addr, - unit_id, - }, - Backend::Actor(h) => Backend::Actor(h.with_slave_id(unit_id)), - }; - Self(backend) - } - - /// See [`ReaderFactory::for_tcp`]. - pub fn for_tcp(addr: std::net::SocketAddr, unit_id: u8) -> Self { - Self(Backend::Tcp { addr, unit_id }) + Ok(Box::new(self.0.clone())) } } // =========================================================================== -// RTU-serial actor cache — shared across reader + writer + slaves on the same port +// Actor cache — one actor per endpoint, shared across reader + writer + slaves // =========================================================================== -/// A cache of RTU-serial actors keyed by the serial port `path`. +/// A cache of Modbus actors keyed by connection identity (`tcp:host:port`, +/// `rtu-tcp:host:port`, `serial:path`). /// -/// **Why it's needed:** `tokio_serial::SerialStream::open` opens the serial port -/// with `TIOCEXCL` (Linux). A second open to the same path is immediately -/// rejected by the OS with `"Unable to acquire exclusive lock on serial port"`. -/// Edge-client -/// has several code paths that each need Modbus access to a slave on a serial -/// port: (a) `build_io` for the main PLC if its transport is RTU-serial, -/// (b) `build_source_reader` per non-PLC slave for telemetry, (c) -/// `build_source_writer` per non-PLC slave for control. Without a cache, -/// (a) + (b) + (c) each `spawn` their own actor trying to open the same port → -/// only one wins the exclusive lock, the rest are stuck in a retry loop forever. +/// **Why it's needed:** several code paths each need Modbus access to the same +/// endpoint — (a) `build_io` for the main PLC, (b) `build_source_reader` per +/// device for telemetry, (c) `build_source_writer` per device for control. +/// Without a cache each would spawn its own actor and open its own connection. +/// That breaks two ways: +/// - **Serial**: `tokio_serial::SerialStream::open` takes `TIOCEXCL` (Linux) — +/// a second open to the same path is rejected ("Unable to acquire exclusive +/// lock on serial port"). +/// - **TCP / RTU-over-TCP**: many PLCs / gateways accept only ONE concurrent +/// socket — a second connection to the same host:port makes writes fail with +/// EINPROGRESS (os error 115). /// -/// This cache ensures one `path` = one actor task = one `open()` call. -/// Different slave-ids on the same RS-485 bus are handled via -/// [`ActorHandle::with_slave_id`] — a cheap sender clone + changing the ID per -/// request. The sequential FIFO of the actor's mpsc channel already guarantees -/// no two Modbus frames overlap on the bus (see the `modbus_actor` module docs). +/// The cache guarantees one endpoint = one actor task = one connection. Reader +/// and writer to the same endpoint share it; different endpoints (e.g. a +/// split-port `:502` read / `:503` write setup) get independent actors. Multiple +/// slave-ids on one RS-485 bus are handled via [`ActorHandle::with_slave_id`] — +/// a cheap sender clone + per-request ID. The actor's FIFO mpsc channel +/// serializes frames so none overlap (see the `modbus_actor` module docs). /// -/// Cheap to clone (internal `Arc`). Usage pattern: create one instance at -/// startup in `main`, then pass `&SerialActorCache` to every call site that -/// needs RTU-serial access. +/// Cheap to clone (internal `Arc`). Create one instance at startup in `main` and +/// pass `&ActorCache` to every call site that needs Modbus access. #[derive(Default, Clone)] -pub struct SerialActorCache { +pub struct ActorCache { inner: Arc>>, } -impl SerialActorCache { +/// Cache key for an endpoint. Reader + writer to the same endpoint collapse to +/// one key (= one shared actor); a different host:port or path stays separate. +fn actor_cache_key(transport: &Transport) -> String { + match transport { + Transport::Tcp { host, port } => format!("tcp:{host}:{port}"), + Transport::RtuOverTcp { host, port } => format!("rtu-tcp:{host}:{port}"), + Transport::RtuSerial { path, .. } => format!("serial:{path}"), + } +} + +impl ActorCache { pub fn new() -> Self { Self::default() } - /// Look up or spawn an actor for `Transport::RtuSerial`. Cache key = `path`. - /// A repeated call with the same path reuses the actor; only the slave_id is - /// updated in the returned handle. + /// Look up or spawn the actor for `transport`. A repeated call with the same + /// endpoint reuses the actor; only the slave_id is updated in the returned + /// handle. /// - /// `timings` is used only on the first spawn for that path. Subsequent calls - /// for the same path reuse the existing actor regardless of the new - /// `timings`. All slaves currently use the fixed default from - /// `device_conn_to_transport`, so there is no inconsistency to warn about. + /// `timings` is used only on the first spawn for that endpoint. Subsequent + /// calls reuse the existing actor regardless of the new `timings` — all + /// callers currently pass the fixed default from `device_conn_to_transport`. /// - /// Error: `Err` only if `transport` is not `RtuSerial` (a programmer bug, - /// not a runtime fault) or if `modbus_actor::spawn` rejects the transport. - /// Serial I/O itself is not attempted here — `spawn` returns quickly and the - /// actor task connects in the background with retry-on-fail. - async fn get_or_spawn_serial( + /// I/O is not attempted here — `spawn` returns quickly and the actor task + /// connects in the background with retry-on-fail. + async fn get_or_spawn( &self, transport: Transport, slave_id: u8, timings: crate::modbus_actor::RtuTimings, ) -> Result { - let path = match &transport { - Transport::RtuSerial { path, .. } => path.clone(), - other => { - return Err(anyhow!( - "SerialActorCache::get_or_spawn_serial called with a non-serial transport \ - ({}) — programmer bug", - other.display() - )); - } - }; + let key = actor_cache_key(&transport); let mut lock = self.inner.lock().await; - if let Some(existing) = lock.get(&path) { - debug!( - path = %path, - slave_id, - "RTU serial actor reused from cache" - ); + if let Some(existing) = lock.get(&key) { + debug!(key = %key, slave_id, "modbus actor reused from cache"); return Ok(existing.with_slave_id(slave_id)); } debug!( - path = %path, + key = %key, slave_id, transport = %transport.display(), - "RTU serial actor: cache miss, spawning new actor" + "modbus actor: cache miss, spawning new actor" ); let handle = crate::modbus_actor::spawn(transport, slave_id, timings) .await - .context("spawn RTU serial actor (via SerialActorCache)")?; - lock.insert(path, handle.clone()); + .context("spawn modbus actor (via ActorCache)")?; + lock.insert(key, handle.clone()); Ok(handle) } - /// Number of RTU-serial actor entries in the cache. Used by tests + can be - /// used for observability in the startup log (number of physical serial - /// ports in use). + /// Number of distinct endpoints (actor tasks) in the cache. Used by tests + + /// can be used for observability in the startup log. #[allow(dead_code)] pub(crate) async fn len(&self) -> usize { self.inner.lock().await.len() @@ -423,134 +203,71 @@ impl SerialActorCache { // Main builder // =========================================================================== -/// Build a factory pair from [`ModbusSettings`]. For RTU mode, this -/// **spawns an actor task** — the RTU process keeps running in the background -/// until both factories drop (all handles drop). For TCP, no I/O happens here, -/// it just packages the descriptor. -/// -/// `serial_cache` is only used if `settings.transport` is `RtuSerial`; it is -/// registered so `build_source_reader` / `build_source_writer` for slaves on the -/// same port can reuse the actor. +/// Build a factory pair from [`ModbusSettings`] for the main PLC. Gets (or +/// spawns) the shared actor for this endpoint via `cache`, so the reader and +/// writer share one connection, and any source device on the same endpoint +/// reuses it too. No I/O happens here — the actor connects in the background. pub async fn build_io( settings: &crate::settings::ModbusSettings, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> Result<(ReaderFactory, WriterFactory)> { - match &settings.transport { - Transport::Tcp { host, port } => { - let addr: SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow!("TCP address invalid: {e}"))?; - let backend = Backend::Tcp { - addr, - unit_id: settings.unit_id, - }; - Ok((ReaderFactory(backend.clone()), WriterFactory(backend))) - } - Transport::RtuOverTcp { .. } => { - // RTU-over-TCP is not cached via SerialActorCache: it's not a - // serial port, so the exclusive lock is irrelevant. Spawn one actor - // + share it via a cloned handle (reader & writer use the same - // handle). RTU-serial slaves added via `build_source_reader` / - // `_writer` still go through the cache — different transport, - // different actor. - let handle = crate::modbus_actor::spawn( - settings.transport.clone(), - settings.unit_id, - settings.rtu_timings, - ) - .await?; - Ok(( - ReaderFactory(Backend::Actor(handle.clone())), - WriterFactory(Backend::Actor(handle)), - )) - } - Transport::RtuSerial { .. } => { - let handle = serial_cache - .get_or_spawn_serial( - settings.transport.clone(), - settings.unit_id, - settings.rtu_timings, - ) - .await - .context("build_io: spawn RTU serial actor for the main PLC")?; - Ok(( - ReaderFactory(Backend::Actor(handle.clone())), - WriterFactory(Backend::Actor(handle)), - )) - } - } + let handle = cache + .get_or_spawn( + settings.transport.clone(), + settings.unit_id, + settings.rtu_timings, + ) + .await + .context("build_io: spawn modbus actor for the main PLC")?; + Ok((ReaderFactory(handle.clone()), WriterFactory(handle))) } -/// Build a ReaderFactory for one slave device. Used by telemetry polling. +/// Build a ReaderFactory for one source device. Used by telemetry polling. /// -/// 2 cases: -/// - TCP slave → a TCP factory to the slave host (independent of the PLC). -/// - RTU-serial slave → look up the actor in `serial_cache` (keyed by path). -/// On a cache hit (e.g. the main PLC uses the same serial port, or another -/// slave was registered earlier, or this slave's writer counterpart is already -/// registered), reuse the handle with `with_slave_id`. On a miss, spawn a new -/// actor and cache it so other call sites (the same slave's writer, or other -/// slaves on the same RS-485 bus) reuse it too. +/// Gets (or spawns) the shared actor for the device's endpoint via `cache`, so +/// it's shared with the main PLC / other devices on the same endpoint and with +/// this device's writer counterpart. A weigher (SerialAscii) has no Modbus +/// reader → error. pub(crate) async fn build_source_reader( conn: &crate::shared::DeviceConnection, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> anyhow::Result { use crate::shared::DeviceConnection; - match conn { - DeviceConnection::Tcp { host, port, .. } => { - let addr: std::net::SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow::anyhow!("slave tcp addr invalid: {e}"))?; - Ok(ReaderFactory::for_tcp(addr, conn.unit_id())) - } - DeviceConnection::RtuSerial { .. } => { - let (transport, timings) = crate::settings::device_conn_to_transport(conn) - .context("slave device_conn_to_transport (reader)")?; - let handle = serial_cache - .get_or_spawn_serial(transport, conn.unit_id(), timings) - .await - .context("get_or_spawn RTU serial actor for slave reader")?; - Ok(ReaderFactory(Backend::Actor(handle))) - } - DeviceConnection::SerialAscii { .. } => { - anyhow::bail!( - "build_source_reader called for SerialAscii — a weigher has no Modbus reader" - ) - } + if let DeviceConnection::SerialAscii { .. } = conn { + anyhow::bail!( + "build_source_reader called for SerialAscii — a weigher has no Modbus reader" + ); } + let (transport, timings) = crate::settings::device_conn_to_transport(conn) + .context("slave device_conn_to_transport (reader)")?; + let handle = cache + .get_or_spawn(transport, conn.unit_id(), timings) + .await + .context("get_or_spawn modbus actor for source reader")?; + Ok(ReaderFactory(handle)) } -/// Used by control_subscriber. -/// Logic identical to [`build_source_reader`] — look up the actor in -/// `serial_cache` so it's shared with the reader counterpart (if the same slave -/// also has a monitoring entry) and with other slaves on the same port. +/// Build a WriterFactory for one source device. Used by control_subscriber. +/// Logic identical to [`build_source_reader`] — shares the actor with the reader +/// counterpart (if the device also has a monitoring entry) and with other +/// devices on the same endpoint. pub(crate) async fn build_source_writer( conn: &crate::shared::DeviceConnection, - serial_cache: &SerialActorCache, + cache: &ActorCache, ) -> anyhow::Result { use crate::shared::DeviceConnection; - match conn { - DeviceConnection::Tcp { host, port, .. } => { - let addr: std::net::SocketAddr = format!("{host}:{port}") - .parse() - .map_err(|e| anyhow::anyhow!("slave tcp addr invalid: {e}"))?; - Ok(WriterFactory::for_tcp(addr, conn.unit_id())) - } - DeviceConnection::RtuSerial { .. } => { - let (transport, timings) = crate::settings::device_conn_to_transport(conn) - .context("slave device_conn_to_transport (writer)")?; - let handle = serial_cache - .get_or_spawn_serial(transport, conn.unit_id(), timings) - .await - .context("get_or_spawn RTU serial actor for slave writer")?; - Ok(WriterFactory(Backend::Actor(handle))) - } - DeviceConnection::SerialAscii { .. } => { - anyhow::bail!( - "build_source_writer called for SerialAscii — a weigher has no Modbus writer" - ) - } + if let DeviceConnection::SerialAscii { .. } = conn { + anyhow::bail!( + "build_source_writer called for SerialAscii — a weigher has no Modbus writer" + ); } + let (transport, timings) = crate::settings::device_conn_to_transport(conn) + .context("slave device_conn_to_transport (writer)")?; + let handle = cache + .get_or_spawn(transport, conn.unit_id(), timings) + .await + .context("get_or_spawn modbus actor for source writer")?; + Ok(WriterFactory(handle)) } // =========================================================================== @@ -573,23 +290,30 @@ mod tests { } } + fn tcp_transport(host: &str, port: u16) -> Transport { + Transport::Tcp { + host: host.to_string(), + port, + } + } + /// The cache returns the same actor handle (identical channel sender) for /// two calls with the same serial path. The slave_id may differ; the actor /// task stays single. This is the key invariant that prevents the "Unable to /// acquire exclusive lock on serial port" bug — with the cache, one path = /// one `tokio_serial::SerialStream::open`. #[tokio::test] - async fn serial_cache_reuses_actor_for_same_path() { - let cache = SerialActorCache::new(); + async fn cache_reuses_actor_for_same_serial_path() { + let cache = ActorCache::new(); let transport = serial_transport("/tmp/fake-ttyUSB-cache-test-a"); let timings = RtuTimings::default(); let reader_handle = cache - .get_or_spawn_serial(transport.clone(), 1, timings) + .get_or_spawn(transport.clone(), 1, timings) .await .expect("first call should spawn"); let writer_handle = cache - .get_or_spawn_serial(transport.clone(), 2, timings) + .get_or_spawn(transport.clone(), 2, timings) .await .expect("second call should reuse"); @@ -604,12 +328,12 @@ mod tests { /// different physical ports into one actor — that would send frames onto the /// wrong bus. #[tokio::test] - async fn serial_cache_separates_actors_per_path() { - let cache = SerialActorCache::new(); + async fn cache_separates_actors_per_serial_path() { + let cache = ActorCache::new(); let timings = RtuTimings::default(); let h_a = cache - .get_or_spawn_serial( + .get_or_spawn( serial_transport("/tmp/fake-ttyUSB-cache-test-b"), 1, timings, @@ -617,7 +341,7 @@ mod tests { .await .expect("spawn A"); let h_b = cache - .get_or_spawn_serial( + .get_or_spawn( serial_transport("/tmp/fake-ttyUSB-cache-test-c"), 1, timings, @@ -632,21 +356,50 @@ mod tests { assert_eq!(cache.len().await, 2); } - /// Calling the cache with a non-serial transport = programmer bug, it must - /// return Err (not panic, not silently spawn the wrong thing). + /// TCP reader + writer to the same host:port share ONE actor (= one socket). + /// This is the fix for single-session PLCs: a separate reader + writer socket + /// would make the writer fail with EINPROGRESS (os error 115). #[tokio::test] - async fn serial_cache_rejects_non_serial_transport() { - let cache = SerialActorCache::new(); - let result = cache - .get_or_spawn_serial( - Transport::Tcp { - host: "127.0.0.1".to_string(), - port: 502, - }, - 1, - RtuTimings::default(), - ) - .await; - assert!(result.is_err(), "TCP transport must be rejected"); + async fn cache_shares_actor_for_same_tcp_endpoint() { + let cache = ActorCache::new(); + let timings = RtuTimings::default(); + + let reader = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("reader spawn"); + let writer = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("writer reuse"); + + assert!( + reader.same_channel(&writer), + "reader & writer on the same host:port must share one actor/connection" + ); + assert_eq!(cache.len().await, 1); + } + + /// Split-port setup (`:502` read / `:503` write, same host) keeps TWO + /// independent actors — matches PLCs that expose a separate write port. + #[tokio::test] + async fn cache_separates_actors_per_tcp_port() { + let cache = ActorCache::new(); + let timings = RtuTimings::default(); + + let read = cache + .get_or_spawn(tcp_transport("192.168.10.10", 502), 1, timings) + .await + .expect("read spawn"); + let write = cache + .get_or_spawn(tcp_transport("192.168.10.10", 503), 1, timings) + .await + .expect("write spawn"); + + assert!( + !read.same_channel(&write), + "different ports on the same host must have separate actors" + ); + assert_eq!(cache.len().await, 2); } } diff --git a/src/telemetry.rs b/src/telemetry.rs index acd4e1c..200a44c 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -35,9 +35,9 @@ const RECONNECT_BACKOFF: Duration = Duration::from_secs(5); /// The PLC and source devices are polled separately: /// - PLC: `entries` + `holding_count` + `coil_count`, polled via /// `reader_factory` (unit_id from the mapping's primary PLC). -/// - Source (chiller, etc.): each entry in `source_sessions`, polled via a -/// `ReaderFactory` already `.with_unit_id(source.unit_id)`. An error on one -/// source does not affect the PLC poll loop or the other sources. +/// - Source (chiller, etc.): each entry in `source_sessions`, polled via its own +/// `ReaderFactory` (built with the device's unit_id). An error on one source +/// does not affect the PLC poll loop or the other sources. #[derive(Clone)] pub struct TelemetryConfig { pub reader_factory: ReaderFactory, @@ -64,7 +64,7 @@ pub struct SourcePollSession { /// This device's machine id `{location}/{name}` — used for the per-machine /// plc_status topic (`{base}/{machine_id}/plc/status`). pub machine_id: String, - /// ReaderFactory with this source's unit_id (derived via `with_unit_id`). + /// ReaderFactory for this source (built with the device's unit_id). pub reader_factory: ReaderFactory, /// Holding + coil entries — bulk read from address 0. pub entries: Vec, @@ -82,7 +82,7 @@ impl TelemetryConfig { pub async fn from_settings( settings: &Settings, reader_factory: ReaderFactory, - serial_cache: &crate::modbus_client::SerialActorCache, + actor_cache: &crate::modbus_client::ActorCache, ) -> Result { let modbus = settings .modbus @@ -139,7 +139,7 @@ impl TelemetryConfig { continue; } let source_reader_factory = - crate::modbus_client::build_source_reader(&device.connection, serial_cache) + crate::modbus_client::build_source_reader(&device.connection, actor_cache) .await .with_context(|| format!("build source reader for slave '{}'", device_id))?; let input_entries: Vec = source_entries