From 9ddeb88ced2d59aa97efb2fae2173d8e9a3f0ab4 Mon Sep 17 00:00:00 2001
From: ashaffah <wagamama.morph@gmail.com>
Date: Mon, 29 Jun 2026 13:44:59 +0700
Subject: [PATCH] fix(modbus-read): chunked reads + read timeout

Some PLCs accept the TCP connection but reject a large read request,
leaving the node hung with no error (verified: qty 2 works, qty 18 hangs
on the same device). Add a `max_read` field: `quantity` is read in chunks
of that size on the same connection and concatenated, so the whole block
comes back without exceeding the device's per-request limit (0 = one
request, unchanged default). Also bound each read with a 3s timeout so an
unresponsive device surfaces an io_error and reconnects instead of
hanging forever.
---
 src/catalog.rs              |  3 +-
 src/runtime/nodes/modbus.rs | 74 +++++++++++++++++++++++++++++--------
 2 files changed, 60 insertions(+), 17 deletions(-)

diff --git a/src/catalog.rs b/src/catalog.rs
index b7338cd..76ac801 100644
--- a/src/catalog.rs
+++ b/src/catalog.rs
@@ -270,7 +270,8 @@ pub fn catalog() -> Value {
                 { "key": "function", "label": "Function", "type": "select",
                   "options": ["holding", "input", "coils", "discrete"], "default": "holding" },
                 { "key": "address", "label": "Address", "type": "number", "default": 0 },
-                { "key": "quantity", "label": "Quantity", "type": "number", "default": 1 }
+                { "key": "quantity", "label": "Quantity", "type": "number", "default": 1 },
+                { "key": "max_read", "label": "Max per request (0 = one)", "type": "number", "default": 0 }
             ]
         },
         {
diff --git a/src/runtime/nodes/modbus.rs b/src/runtime/nodes/modbus.rs
index b248cac..7cb82bd 100644
--- a/src/runtime/nodes/modbus.rs
+++ b/src/runtime/nodes/modbus.rs
@@ -43,6 +43,10 @@ pub struct ModbusReadNode {
     address: u16,
     #[serde(default = "default_quantity")]
     quantity: u16,
+    /// Max items per Modbus request — `quantity` is read in chunks of this on
+    /// the same connection (some PLCs reject large reads). 0 = one request.
+    #[serde(default)]
+    max_read: u16,
 }
 
 impl ModbusReadNode {
@@ -59,6 +63,52 @@ enum ReadOutcome {
     Io(String),
 }
 
+/// How long to wait for one Modbus read reply before treating it as an I/O
+/// error (so an unresponsive device doesn't hang the node forever).
+const READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+
+/// Read `quantity` items starting at `address`, in chunks of `chunk` (0 = one
+/// request) on the same connection. Concatenates the chunks into one array so a
+/// PLC that rejects large reads still returns the whole block. Each chunk is
+/// time-bounded.
+async fn read_chunked(
+    c: &mut ModbusCtx,
+    function: &str,
+    address: u16,
+    quantity: u16,
+    chunk: u16,
+) -> ReadOutcome {
+    let step = if chunk == 0 { quantity.max(1) } else { chunk };
+    let bits = matches!(function, "coils" | "discrete");
+    let mut words: Vec<u16> = Vec::new();
+    let mut flags: Vec<bool> = Vec::new();
+    let mut addr = address;
+    let mut left = quantity;
+    // timeout(Elapsed) wraps Result(transport error) wraps Result(exception).
+    macro_rules! do_read {
+        ($call:expr, $sink:expr) => {
+            match tokio::time::timeout(READ_TIMEOUT, $call).await {
+                Err(_) => return ReadOutcome::Io("read timeout".to_string()),
+                Ok(Err(e)) => return ReadOutcome::Io(format!("{e}")),
+                Ok(Ok(Err(ex))) => return ReadOutcome::Exception(format!("{ex:?}")),
+                Ok(Ok(Ok(v))) => $sink.extend(v),
+            }
+        };
+    }
+    while left > 0 {
+        let n = left.min(step);
+        match function {
+            "input" => do_read!(c.read_input_registers(addr, n), words),
+            "coils" => do_read!(c.read_coils(addr, n), flags),
+            "discrete" => do_read!(c.read_discrete_inputs(addr, n), flags),
+            _ => do_read!(c.read_holding_registers(addr, n), words),
+        }
+        addr = addr.wrapping_add(n);
+        left -= n;
+    }
+    ReadOutcome::Ok(if bits { json!(flags) } else { json!(words) })
+}
+
 #[async_trait]
 impl Node for ModbusReadNode {
     async fn run(self: Box<Self>, mut ctx: NodeCtx) -> anyhow::Result<()> {
@@ -83,22 +133,14 @@ impl Node for ModbusReadNode {
             }
             let c = conn.as_mut().expect("just connected");
 
-            // Outer Result = transport error, inner = protocol exception.
-            macro_rules! decode {
-                ($call:expr) => {
-                    match $call.await {
-                        Ok(Ok(v)) => ReadOutcome::Ok(json!(v)),
-                        Ok(Err(ex)) => ReadOutcome::Exception(format!("{ex:?}")),
-                        Err(e) => ReadOutcome::Io(format!("{e}")),
-                    }
-                };
-            }
-            let outcome = match self.function.as_str() {
-                "input" => decode!(c.read_input_registers(self.address, self.quantity)),
-                "coils" => decode!(c.read_coils(self.address, self.quantity)),
-                "discrete" => decode!(c.read_discrete_inputs(self.address, self.quantity)),
-                _ => decode!(c.read_holding_registers(self.address, self.quantity)),
-            };
+            let outcome = read_chunked(
+                c,
+                &self.function,
+                self.address,
+                self.quantity,
+                self.max_read,
+            )
+            .await;
             match outcome {
                 ReadOutcome::Ok(v) => ctx.emit(Msg::new(v)).await,
                 ReadOutcome::Exception(ex) => {