diff --git a/esphome/components/modbus/modbus.cpp b/esphome/components/modbus/modbus.cpp index 92b4dc25ac..2561ee9069 100644 --- a/esphome/components/modbus/modbus.cpp +++ b/esphome/components/modbus/modbus.cpp @@ -313,7 +313,6 @@ void ModbusClientHub::process_modbus_server_frame(uint8_t address, std::spanlast_modbus_byte_ - this->last_send_); if (device) device->on_error(request_pdu, static_cast(exception)); - } else if (device) { // Not an error response device->on_response(request_pdu, pdu); } else { // Not an error response, but no device to respond to @@ -507,16 +506,25 @@ void ModbusClientHub::send_next_frame_() { return; } - ModbusDeviceCommand &command = this->tx_buffer_.front(); - - if (this->send_frame_(command.frame)) { - this->waiting_for_response_ = std::move(command); - } else { - if (command.device) - command.device->on_not_sent(); - } - + // Move the command out and pop BEFORE attempting the send: no callback may run while the frame still + // sits in the queue (the same principle as the clear sweep). A failure callback that sends would + // otherwise queue a new frame and pop_front() could discard the wrong one - and the deque + // reference / PDU span could be invalidated mid-callback. + ModbusDeviceCommand command = std::move(this->tx_buffer_.front()); this->tx_buffer_.pop_front(); + ModbusClientDevice *device = command.device; + const bool sent = this->send_frame_(command.frame); + + if (sent) { + // The frame now lives in the waiting slot; its PDU is the frame without the leading address and + // trailing CRC. + ModbusDeviceCommand &wfr = this->waiting_for_response_.emplace(std::move(command)); + if (device != nullptr) + device->on_sent(wfr.frame.pdu()); + } else { + if (device != nullptr) + device->trigger_not_sent(command.frame.pdu()); + } if (!this->tx_buffer_.empty()) { ESP_LOGV(TAG, "Write queue contains %zu items.", this->tx_buffer_.size()); @@ -574,7 +582,7 @@ void ModbusServerHub::send_exception_(uint8_t address, uint8_t function_code, Ex void ModbusClientHub::notify_no_response_(ModbusDeviceCommand &wfr) { if (wfr.device == nullptr) return; - const bool retry = wfr.device->on_no_response(); + const bool retry = wfr.device->on_no_response(wfr.frame.pdu()); // The callback may have detached the device (e.g. clear_tx_queue_for_device()); honor the detach // over the retry request rather than re-queueing a frame that can no longer be routed. if (retry && wfr.device != nullptr) @@ -588,7 +596,7 @@ void ModbusClientHub::requeue_waiting_frame_(ModbusDeviceCommand &wfr) { if (this->tx_buffer_.size() >= MODBUS_TX_BUFFER_SIZE) { ESP_LOGE(TAG, "Write buffer full, dropped retry for address %" PRIu8, frame.address()); if (wfr.device != nullptr) - wfr.device->on_not_sent(); + wfr.device->trigger_not_sent(frame.pdu()); return; } // Re-queue a copy (not a move): the waiting entry may have to survive as an interrupted shell. @@ -598,16 +606,16 @@ void ModbusClientHub::requeue_waiting_frame_(ModbusDeviceCommand &wfr) { // Raw send for client: pushes to tx queue. Everything except the CRC must be contained in payload. void ModbusClientHub::send_pdu(uint8_t address, std::span pdu, ModbusClientDevice *device) { if (pdu.empty()) { - if (device) - device->on_not_sent(); + if (device != nullptr) + device->trigger_not_sent(pdu); return; } // Bound the PDU so the wire frame (address + pdu + CRC) stays within the Modbus RTU 256-byte limit. if (pdu.size() > MAX_PDU_SIZE) { ESP_LOGE(TAG, "Frame too large, dropped: %" PRIu8 ":%zu bytes", address, pdu.size()); - if (device) - device->on_not_sent(); + if (device != nullptr) + device->trigger_not_sent(pdu); return; } @@ -624,17 +632,36 @@ void ModbusClientHub::send_pdu(uint8_t address, std::span pdu, Mo #endif ESP_LOGE(TAG, "Write buffer full, dropped: %" PRIu8 ":%s", address, format_hex_pretty_to(hex_buf, pdu.data(), pdu.size())); - if (device) - device->on_not_sent(); + if (device != nullptr) + device->trigger_not_sent(pdu); } } void ModbusClientHub::clear_tx_queue_for_address(uint8_t address, bool clear_sent) { - // Remove any pending commands for this address from the tx buffer - auto &tx_buffer = this->tx_buffer_; - tx_buffer.erase(std::remove_if(tx_buffer.begin(), tx_buffer.end(), - [address](const ModbusDeviceCommand &cmd) { return cmd.frame.address() == address; }), - tx_buffer.end()); + // Drop the queued frames for this address, delivering on_not_sent() to each frame's owner: other + // devices talking to the same physical device (e.g. a modbus_client action alongside a controller that + // just went offline) must observe the drop, or their command never resolves. Mark first, then sweep + // only marked frames: anything a callback re-queues is unmarked, so it + // is never swept - or re-notified - by the clear that triggered it. Each marked frame is moved out and erased BEFORE + // its callback runs, so handlers see a consistent queue; termination is guaranteed because only the initially-marked + // frames are ever swept. + for (auto &cmd : this->tx_buffer_) { + if (cmd.frame.address() == address) + cmd.marked_for_deletion = true; + } + for (;;) { + auto it = std::find_if(this->tx_buffer_.begin(), this->tx_buffer_.end(), + [](const ModbusDeviceCommand &cmd) { return cmd.marked_for_deletion; }); + if (it == this->tx_buffer_.end()) + break; + ModbusDeviceCommand dropped = std::move(*it); + this->tx_buffer_.erase(it); + // The sweep delivers through the same per-device guard as refusals: a device clearing from inside + // its own on_not_sent() gets its remaining frames resolved silently (documented in the lifecycle + // contract), other owners are notified normally, and every nested clear stays bounded. + if (dropped.device != nullptr) + dropped.device->trigger_not_sent(dropped.frame.pdu()); + } if (clear_sent && this->waiting_for_response_.has_value() && this->waiting_for_response_.value().device) { if (this->waiting_for_response_.value().frame.address() == address) { @@ -662,8 +689,8 @@ void ModbusClientHub::clear_tx_queue_for_device(ModbusClientDevice *device) { void ModbusClientHub::send_raw(const std::vector &payload, ModbusClientDevice *device) { if (payload.size() < 2) { - if (device) - device->on_not_sent(); + if (device != nullptr) + device->trigger_not_sent({}); // too short to contain a PDU return; } this->send_pdu(payload[0], std::span(payload).subspan(1), device); diff --git a/esphome/components/modbus/modbus.h b/esphome/components/modbus/modbus.h index d83075eda1..2204c0f82b 100644 --- a/esphome/components/modbus/modbus.h +++ b/esphome/components/modbus/modbus.h @@ -94,6 +94,9 @@ struct ModbusDeviceCommand { ModbusClientDevice *device; ModbusFrame frame; bool interrupted{false}; + /// Marked by clear_tx_queue_for_address() before it starts notifying, so frames re-queued by an + /// on_not_sent() callback (which are unmarked) are never swept by the clear that triggered them. + bool marked_for_deletion{false}; ModbusDeviceCommand(ModbusClientDevice *device, uint8_t address, const uint8_t *src, uint16_t len) : device(device), frame(address, src, len) {} @@ -123,6 +126,10 @@ class ModbusClientHub : public Modbus { void send_pdu(uint8_t address, std::span pdu, ModbusClientDevice *device = nullptr); ESPDEPRECATED("Use send_pdu(payload[0], , device) instead. Removed in 2027.2.0", "2026.8.0") void send_raw(const std::vector &payload, ModbusClientDevice *device = nullptr); + // Drop the queued commands for an address; every dropped frame resolves via its owner's on_not_sent(), + // so other devices sharing the address observe the drop. The in-flight frame is only detached (silently) + // when clear_sent is set. clear_tx_queue_for_device() SILENTLY discards the caller's own frames + // (supersede/teardown semantics); see the lifecycle note on ModbusClientDevice. void clear_tx_queue_for_address(uint8_t address, bool clear_sent = true); void clear_tx_queue_for_device(ModbusClientDevice *device); @@ -177,6 +184,25 @@ class ModbusServerHub : public Modbus { // Transaction status: std::nullopt on success, otherwise a Modbus exception code using ResponseStatus = std::optional; +/// Command lifecycle: each accepted command (a send_pdu()/typed-helper call, or a hub re-queue from +/// a retry) ends in exactly ONE terminal callback: on_response() (valid response), on_error() +/// (exception response), on_no_response() (timeout or interrupted transaction), or on_not_sent() +/// (never transmitted: send failure or full queue). on_sent() is additional, not +/// terminal: it fires once per wire transmission, before whichever of data/error/no_response follows, +/// and never for a command that ends in on_not_sent(). +/// The exceptions to "exactly one terminal": +/// - clear_tx_queue_for_device() drops the caller's OWN queued commands SILENTLY (supersede/teardown +/// semantics), and both clear variants detach the in-flight frame silently. +/// clear_tx_queue_for_address() DOES resolve every queued frame it drops via the owner's +/// on_not_sent() (delivered one at a time, after that frame leaves the queue). +/// - while a device's own on_not_sent() is on the stack, further on_not_sent() deliveries to THAT +/// device are dropped (see trigger_not_sent()). In particular, a clear issued from inside your own +/// on_not_sent() resolves your remaining frames silently - treat it like +/// clear_tx_queue_for_device(): you cleared them, you know. Other owners are still notified. +/// Sending from inside on_not_sent() is hazardous: the notification may itself mean the queue is full +/// or refusing, and this device's retry that is refused again is dropped WITHOUT a callback (the +/// guard above, which bounds what would otherwise be unbounded re-entry) - prefer re-sending from a +/// later trigger or the component's update()/loop(). class ModbusClientDevice { public: ModbusClientDevice() = default; @@ -204,21 +230,34 @@ class ModbusClientDevice { virtual void on_error(std::span request_pdu, ExceptionCode exception_code) { this->dispatch_response_(request_pdu, {}, exception_code); } - - /// Called when no request could be sent (e.g. queue full, transmission blocked) + /// Called when no request could be sent (e.g. queue full, transmission blocked). /// Do not attempt to queue a command in this callback. - /// (The on_modbus_* names are signature-identical renames, so the new defaults forward to the old - /// virtuals: external devices overriding the old names keep working through the deprecation window. - /// Remove the forwards together with the deprecated names.) - virtual void on_not_sent() { + /// (The on_modbus_* names below are deprecated pre-rename spellings; the defaults forward so + /// external devices overriding them keep working through the deprecation window.) + virtual void on_not_sent(std::span request_pdu) { #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wdeprecated-declarations" this->on_modbus_not_sent(); #pragma GCC diagnostic pop } + /// Non-virtual entry point the hub uses for EVERY on_not_sent() delivery (refusals and clear-queue + /// sweeps alike). While this device's on_not_sent() is on the stack, further deliveries to it are + /// dropped: this bounds every send->refuse and clear->sweep recursion, including cycles through + /// multiple devices (each device can appear on the stack at most once). The documented cost: a clear + /// issued from inside your own on_not_sent() resolves your remaining frames SILENTLY, while other + /// owners are still notified (their guards are not set) - see the lifecycle contract above. + void trigger_not_sent(std::span request_pdu) { + if (this->notifying_not_sent_) + return; + this->notifying_not_sent_ = true; + this->on_not_sent(request_pdu); + this->notifying_not_sent_ = false; + } + /// Called when this device's frame is actually written to the wire + virtual void on_sent(std::span request_pdu) {} /// Called when no matching, uninterrupted response arrived; return true to have the hub re-queue the frame for a /// retry. The hub does not bound retries: the device is responsible for limiting them. - virtual bool on_no_response() { + virtual bool on_no_response(std::span request_pdu) { #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wdeprecated-declarations" return this->on_modbus_no_response(); @@ -287,7 +326,8 @@ class ModbusClientDevice { ESPDEPRECATED("Use send_pdu() instead (the device address is prepended for you). Removed in 2027.2.0", "2026.8.0") void send_raw(const std::vector &payload) { if (payload.empty()) { - this->on_not_sent(); // match the hub-level send_raw(): a refused send is always signalled + // Through the guard like every other delivery, so a handler calling send_raw({}) cannot recurse. + this->trigger_not_sent({}); return; } this->parent_->send_pdu(payload[0], std::span(payload).subspan(1), this); @@ -345,6 +385,8 @@ class ModbusClientDevice { ResponseStatus status); ModbusClientHub *parent_{nullptr}; + /// True while this device's on_not_sent() is on the stack (see trigger_not_sent()). + bool notifying_not_sent_{false}; uint8_t address_{0}; bool custom_response_warned_{false}; // first unhandled custom response warns; repeats log at VERBOSE }; diff --git a/esphome/components/modbus_controller/modbus_controller.h b/esphome/components/modbus_controller/modbus_controller.h index 9616c1d935..928054f1ab 100644 --- a/esphome/components/modbus_controller/modbus_controller.h +++ b/esphome/components/modbus_controller/modbus_controller.h @@ -306,7 +306,8 @@ class ModbusController final : public PollingComponent, public modbus::ModbusCli /// controller-level raw sends stay supported until the command machinery is replaced. void send_raw(const std::vector &payload) { if (payload.empty()) { - this->on_not_sent(); // match the hub-level send_raw(): a refused send is always signalled + // Through the guard like every other delivery, so a handler calling send_raw({}) cannot recurse. + this->trigger_not_sent({}); return; } this->parent_->send_pdu(payload[0], std::span(payload).subspan(1), this); diff --git a/tests/components/modbus/common.h b/tests/components/modbus/common.h new file mode 100644 index 0000000000..659b72014c --- /dev/null +++ b/tests/components/modbus/common.h @@ -0,0 +1,22 @@ +#pragma once +#include +#include "esphome/components/uart/uart_component.h" + +namespace esphome::modbus::testing { + +// A UART that discards all writes, for tests that never inspect the wire. +class NullUART : public uart::UARTComponent { + public: + NullUART() { this->set_baud_rate(115200); } + void write_array(const uint8_t *data, size_t len) override {} + bool peek_byte(uint8_t *data) override { return false; } + bool read_array(uint8_t *data, size_t len) override { return false; } + size_t available() override { return 0; } + uart::UARTFlushResult flush() override { return uart::UARTFlushResult::UART_FLUSH_RESULT_ASSUMED_SUCCESS; } +#if defined(USE_ESP8266) || defined(USE_ESP32) + void load_settings(bool dump_config) override {} +#endif + void check_logger_conflict() override {} +}; + +} // namespace esphome::modbus::testing diff --git a/tests/components/modbus/modbus_client_hub_test.cpp b/tests/components/modbus/modbus_client_hub_test.cpp index 6335a75775..11bc10200d 100644 --- a/tests/components/modbus/modbus_client_hub_test.cpp +++ b/tests/components/modbus/modbus_client_hub_test.cpp @@ -1,9 +1,13 @@ #include #include +#include #include +#include +#include "common.h" #include "esphome/components/modbus/modbus.h" +#include "esphome/core/hal.h" namespace esphome::modbus::testing { @@ -16,12 +20,14 @@ class NoResponseProbeHub : public ModbusClientHub { public: size_t queued_frames() const { return this->tx_buffer_.size(); } const ModbusDeviceCommand &front() const { return this->tx_buffer_.front(); } + const ModbusDeviceCommand &queued(size_t i) const { return this->tx_buffer_[i]; } bool waiting() const { return this->waiting_for_response_.has_value(); } const ModbusDeviceCommand &waiting_command() const { EXPECT_TRUE(this->waiting_for_response_.has_value()); return *this->waiting_for_response_; // NOLINT(bugprone-unchecked-optional-access) } + void send_next_for_test() { this->send_next_frame_(); } void force_send_front() { this->waiting_for_response_ = std::move(this->tx_buffer_.front()); this->tx_buffer_.pop_front(); @@ -41,7 +47,7 @@ class NoResponseProbeHub : public ModbusClientHub { class RetryingDevice : public ModbusClientDevice { public: RetryingDevice(ModbusClientHub *hub, uint8_t address, bool retry) : ModbusClientDevice(hub, address), retry_(retry) {} - bool on_no_response() override { + bool on_no_response(std::span request_pdu) override { this->no_response_count_++; return this->retry_; } @@ -55,7 +61,7 @@ class RetryingDevice : public ModbusClientDevice { class ClearingRetryDevice : public ModbusClientDevice { public: ClearingRetryDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} - bool on_no_response() override { + bool on_no_response(std::span request_pdu) override { this->no_response_count_++; this->clear_tx_queue_for_device(); // detaches this device from the waiting slot mid-callback return true; // and still requests a retry @@ -176,6 +182,565 @@ TEST(ModbusClientHubNoResponse, MidCallbackClearCancelsRetry) { EXPECT_FALSE(hub.waiting()); } +// A device whose sent/not-sent callbacks are counted. +namespace { +class SentCountingDevice : public ModbusClientDevice { + public: + SentCountingDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_sent(std::span request_pdu) override { + this->sent_count_++; + this->last_sent_pdu_.assign(request_pdu.begin(), request_pdu.end()); + } + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + this->last_not_sent_pdu_.assign(request_pdu.begin(), request_pdu.end()); + } + int sent_count_{0}; + int not_sent_count_{0}; + std::vector last_sent_pdu_; + std::vector last_not_sent_pdu_; +}; +} // namespace + +// on_sent() fires when the frame goes onto the wire, not when it is queued. +TEST(ModbusClientHubSent, FiresOnWireNotOnQueue) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); // frame timing derives from the baud rate + SentCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + EXPECT_EQ(device.sent_count_, 0); // queued only - nothing on the wire yet + + hub.send_next_for_test(); + EXPECT_EQ(device.sent_count_, 1); + EXPECT_EQ(device.not_sent_count_, 0); + // The callback identifies which command transmitted: it carries the request PDU. + EXPECT_EQ(device.last_sent_pdu_, (std::vector(READ_PDU, READ_PDU + sizeof(READ_PDU)))); + EXPECT_TRUE(hub.waiting()); +} + +// Counts response deliveries so requeue semantics can be pinned end to end. +namespace { +class DataCountingDevice : public ModbusClientDevice { + public: + DataCountingDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_response(std::span request_pdu, std::span response_pdu) override { + this->data_count_++; + } + void on_error(std::span request_pdu, ExceptionCode exception_code) override { this->error_count_++; } + bool on_no_response(std::span request_pdu) override { + this->no_response_count_++; + this->last_no_response_pdu_.assign(request_pdu.begin(), request_pdu.end()); + if (this->retries_ == 0) + return false; + this->retries_--; + return true; + } + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + this->last_not_sent_pdu_.assign(request_pdu.begin(), request_pdu.end()); + } + void on_sent(std::span request_pdu) override { this->sent_count_++; } + int terminals() const { + return this->data_count_ + this->error_count_ + this->no_response_count_ + this->not_sent_count_; + } + int data_count_{0}; + int error_count_{0}; + int no_response_count_{0}; + int not_sent_count_{0}; + int sent_count_{0}; + int retries_{0}; + std::vector last_not_sent_pdu_; + std::vector last_no_response_pdu_; +}; + +// Runs full send/respond cycles until the queue drains; returns the number of cycles executed. +int drain_with_responses(NoResponseProbeHub &hub, std::span response_pdu, int max_cycles = 10) { + int cycles = 0; + while (hub.queued_frames() != 0 && cycles < max_cycles) { + hub.force_send_front(); + hub.receive_frame_for_test(0x02, response_pdu); + cycles++; + } + return cycles; +} +} // namespace + +constexpr uint8_t OK_RESPONSE[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + +// One request produces exactly one data callback. +TEST(ModbusClientHubCallbackCount, SingleReadSingleCallback) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + drain_with_responses(hub, OK_RESPONSE); + + EXPECT_EQ(device.data_count_, 1); + EXPECT_EQ(device.not_sent_count_, 0); + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_FALSE(hub.waiting()); +} + +// An exception response is a terminal on its own: exactly one on_error(), no others, +// preceded by exactly one on_sent(). +TEST(ModbusClientHubCallbackCount, ErrorResponseIsSoleTerminal) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.send_next_for_test(); + const uint8_t exception_response[] = {0x83, 0x02}; + hub.receive_frame_for_test(0x02, exception_response); + + EXPECT_EQ(device.error_count_, 1); + EXPECT_EQ(device.terminals(), 1); + EXPECT_EQ(device.sent_count_, 1); +} + +// A timeout is a terminal on its own: exactly one on_no_response(), preceded by one +// on_sent(); a refused duplicate ends in on_not_sent() with NO on_sent(). +TEST(ModbusClientHubCallbackCount, NoResponseIsSoleTerminalAndNotSentHasNoSent) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.send_next_for_test(); + hub.timeout_waiting(); + EXPECT_EQ(device.no_response_count_, 1); + EXPECT_EQ(device.terminals(), 1); + EXPECT_EQ(device.sent_count_, 1); + + // A refused send (empty PDU) is a not_sent terminal, never sent. + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(write_pdu); + device.send_pdu(std::span{}); + EXPECT_EQ(device.not_sent_count_, 1); + EXPECT_EQ(device.terminals(), 2); // the accepted write is still queued - no terminal for it yet + EXPECT_EQ(device.sent_count_, 1); // and it has not transmitted yet + + // Drain it: the write echo response is its data terminal, and the books balance. + hub.send_next_for_test(); + hub.receive_frame_for_test(0x02, write_pdu); + EXPECT_EQ(device.data_count_, 1); + EXPECT_EQ(device.terminals(), 3); // 3 accepted lifecycles, 3 terminals + EXPECT_EQ(device.sent_count_, 2); // 2 transmissions (read + write); the refused send never sent +} + +// A device-requested retry starts a new lifecycle: each transmission gets its own sent + terminal. +TEST(ModbusClientHubCallbackCount, RetryLifecyclesEachGetSentAndTerminal) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + device.retries_ = 1; // ask for exactly one retry + + device.send_pdu(read_pdu()); + hub.send_next_for_test(); + hub.timeout_waiting(); // lifecycle 1: sent + no_response (retry requested -> re-queued) + ASSERT_EQ(hub.queued_frames(), 1u); + hub.send_next_for_test(); + hub.timeout_waiting(); // lifecycle 2: sent + no_response (retry declined -> done) + + EXPECT_EQ(device.no_response_count_, 2); + EXPECT_EQ(device.terminals(), 2); + EXPECT_EQ(device.sent_count_, 2); + EXPECT_EQ(hub.queued_frames(), 0u); + // The retried lifecycle's timeout carries the SAME request PDU as the first attempt. + EXPECT_EQ(device.last_no_response_pdu_, std::vector(READ_PDU, READ_PDU + sizeof(READ_PDU))); +} + +// A retry re-queue that finds the buffer full is refused like any other send: the device gets +// on_not_sent() carrying the request PDU (the previously uncovered requeue_waiting_frame_ branch). +TEST(ModbusClientHubCallbackCount, FullQueueRetryRefusalDeliversNotSentWithPdu) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + device.retries_ = 1; + SentCountingDevice filler(&hub, 0x05); + + device.send_pdu(read_pdu()); + hub.force_send_front(); // in flight + // Fill the queue with distinct frames. + for (uint16_t i = 0; i < MODBUS_TX_BUFFER_SIZE; i++) { + const uint8_t fill[] = {0x03, static_cast(i >> 8), static_cast(i & 0xFF), 0x00, 0x01}; + filler.send_pdu(fill); + } + ASSERT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); + + hub.timeout_waiting(); // retry requested, but the re-queue is refused: not_sent terminal instead + + EXPECT_EQ(device.no_response_count_, 1); + EXPECT_EQ(device.not_sent_count_, 1); + EXPECT_EQ(device.last_not_sent_pdu_, std::vector(READ_PDU, READ_PDU + sizeof(READ_PDU))); + EXPECT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); +} + +// The deprecated device-side send_raw() refusal delivers through the same guard as every other +// path: a handler that reacts to its own refusal with another empty send_raw() stays bounded. +namespace { +class SendRawOnNotSentDevice : public ModbusClientDevice { + public: + SendRawOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" + this->send_raw({}); // refused again; the guard must suppress the nested delivery +#pragma GCC diagnostic pop + } + int not_sent_count_{0}; +}; +} // namespace + +TEST(ModbusClientHubQueue, SendRawRefusalIsGuardedAgainstRecursion) { + NoResponseProbeHub hub; + SendRawOnNotSentDevice device(&hub, 0x02); +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" + device.send_raw({}); // empty payload refused -> on_not_sent -> nested send_raw({}) suppressed +#pragma GCC diagnostic pop + EXPECT_EQ(device.not_sent_count_, 1); +} + +namespace { +// A device that chains a follow-up send from inside on_sent(). +class ChainOnSentDevice : public ModbusClientDevice { + public: + ChainOnSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_sent(std::span request_pdu) override { + if (!this->chained_) { + this->chained_ = true; + const uint8_t follow[] = {0x03, 0x00, 0x09, 0x00, 0x01}; // read holding 0x0009 x1 + this->send_pdu(follow); + } + } + bool chained_{false}; +}; +} // namespace + +// clear_tx_queue_for_address() resolves every dropped frame via its owner's on_not_sent(), so a device +// sharing the address with the clearer (e.g. a modbus_client action alongside an offline controller) +// observes the drop; frames for other addresses are untouched. +TEST(ModbusClientHubQueue, ClearAddressQueueNotifiesEveryOwner) { + NoResponseProbeHub hub; + SentCountingDevice controller_like(&hub, 0x02); + SentCountingDevice bystander_same(&hub, 0x02); + SentCountingDevice bystander_other(&hub, 0x03); + + const uint8_t read_a[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + const uint8_t read_b[] = {0x03, 0x02, 0x00, 0x00, 0x02}; + const uint8_t read_c[] = {0x03, 0x03, 0x00, 0x00, 0x02}; + controller_like.send_pdu(read_a); + bystander_same.send_pdu(read_b); + bystander_other.send_pdu(read_c); + ASSERT_EQ(hub.queued_frames(), 3u); + + controller_like.clear_tx_queue_for_address(false); + + ASSERT_EQ(hub.queued_frames(), 1u); // only the other-address frame remains + EXPECT_EQ(hub.front().frame.address(), 0x03); + EXPECT_EQ(controller_like.not_sent_count_, 1); + EXPECT_EQ(bystander_same.not_sent_count_, 1); + EXPECT_EQ(bystander_other.not_sent_count_, 0); + // each owner saw its own request PDU + EXPECT_EQ(bystander_same.last_not_sent_pdu_, std::vector(std::begin(read_b), std::end(read_b))); +} + +namespace { +// Re-sends its frame once from inside on_not_sent - the re-queued frame must survive the sweep. +class ResendOnNotSentDevice : public ModbusClientDevice { + public: + ResendOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + if (this->not_sent_count_ == 1) { + const uint8_t again[] = {0x06, 0x00, 0x40, 0x00, 0x01}; + this->send_pdu(again); + } + } + int not_sent_count_{0}; +}; +} // namespace + +// A handler that re-sends to the same address from inside on_not_sent() neither corrupts the sweep nor +// loops it: only initially-marked frames are swept, so the re-queued frame stays queued. +TEST(ModbusClientHubQueue, ClearAddressReentrantResendSurvives) { + NoResponseProbeHub hub; + ResendOnNotSentDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + device.send_pdu(read); + ASSERT_EQ(hub.queued_frames(), 1u); + + hub.clear_tx_queue_for_address(0x02, false); + + // The original frame resolved via on_not_sent; the re-send from inside that callback remains queued. + EXPECT_EQ(device.not_sent_count_, 1); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.front().frame.address(), 0x02); +} + +namespace { +// Retries from EVERY on_not_sent - against a full queue this recursed without bound before the guard. +class AlwaysRetryDevice : public ModbusClientDevice { + public: + AlwaysRetryDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + const uint8_t again[] = {0x03, 0x00, 0x50, 0x00, 0x01}; + this->send_pdu(again); + } + int not_sent_count_{0}; +}; + +// From inside on_not_sent, clears ANOTHER address - those victims must still be notified (the per-device +// guard suppresses deliveries only to a device already inside its own on_not_sent()). +class ClearOtherOnNotSentDevice : public ModbusClientDevice { + public: + ClearOtherOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + this->parent_->clear_tx_queue_for_address(0x03, false); + } + int not_sent_count_{0}; +}; +} // namespace + +// A handler that retries from every on_not_sent() against a FULL queue must not recurse: the first +// refusal notifies once, the nested refusal is dropped without a callback (the documented guard). +TEST(ModbusClientHubQueue, FullQueueRetryFromNotSentDoesNotRecurse) { + NoResponseProbeHub hub; + SentCountingDevice filler(&hub, 0x05); + AlwaysRetryDevice retrier(&hub, 0x02); + + // Fill the queue with distinct frames. + for (uint16_t i = 0; i < MODBUS_TX_BUFFER_SIZE; i++) { + const uint8_t fill[] = {0x03, static_cast(i >> 8), static_cast(i & 0xFF), 0x00, 0x01}; + filler.send_pdu(fill); + } + ASSERT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + retrier.send_pdu(read); // refused (full) -> on_not_sent -> retry -> refused under the guard, silently + + EXPECT_EQ(retrier.not_sent_count_, 1); + EXPECT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); +} + +namespace { +// From inside on_not_sent, triggers ANOTHER device's send (which will be refused too). +class SendOtherOnNotSentDevice : public ModbusClientDevice { + public: + SendOtherOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + if (this->other_ != nullptr) { + const uint8_t read[] = {0x03, 0x00, 0x60, 0x00, 0x01}; + this->other_->send_pdu(read); + } + } + ModbusClientDevice *other_{nullptr}; + int not_sent_count_{0}; +}; +} // namespace + +// The refusal recursion guard is per-device: a refusal that lands on a DIFFERENT device while one +// device's notification is on the stack must still deliver - that device did not cause the recursion +// and would otherwise silently lose its terminal callback. +TEST(ModbusClientHubQueue, RefusalForOtherDeviceDeliversDuringNotification) { + NoResponseProbeHub hub; + SentCountingDevice filler(&hub, 0x05); + SendOtherOnNotSentDevice first(&hub, 0x02); + SentCountingDevice second(&hub, 0x03); + first.other_ = &second; + + // Fill the queue with distinct frames. + for (uint16_t i = 0; i < MODBUS_TX_BUFFER_SIZE; i++) { + const uint8_t fill[] = {0x03, static_cast(i >> 8), static_cast(i & 0xFF), 0x00, 0x01}; + filler.send_pdu(fill); + } + ASSERT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + first.send_pdu(read); // refused -> first.on_not_sent -> second's send refused -> second notified + + EXPECT_EQ(first.not_sent_count_, 1); + EXPECT_EQ(second.not_sent_count_, 1); +} + +// Two devices whose handlers each trigger the other's send cannot recurse without bound: each device +// can be on the notification stack at most once, so the cycle dies as soon as it returns to a device +// whose own on_not_sent() is still running. +TEST(ModbusClientHubQueue, TwoDeviceRefusalCycleTerminates) { + NoResponseProbeHub hub; + SentCountingDevice filler(&hub, 0x05); + SendOtherOnNotSentDevice first(&hub, 0x02); + SendOtherOnNotSentDevice second(&hub, 0x03); + first.other_ = &second; + second.other_ = &first; + + // Fill the queue with distinct frames. + for (uint16_t i = 0; i < MODBUS_TX_BUFFER_SIZE; i++) { + const uint8_t fill[] = {0x03, static_cast(i >> 8), static_cast(i & 0xFF), 0x00, 0x01}; + filler.send_pdu(fill); + } + ASSERT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + first.send_pdu(read); // refuse -> first -> second refused -> second -> first suppressed -> unwind + + EXPECT_EQ(first.not_sent_count_, 1); + EXPECT_EQ(second.not_sent_count_, 1); +} + +namespace { +// From inside on_not_sent, clears its OWN address - its remaining queued frames resolve silently +// (the guard suppresses self-deliveries), while other owners on the address are still notified. +class ClearOwnAddressOnNotSentDevice : public ModbusClientDevice { + public: + ClearOwnAddressOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + this->clear_tx_queue_for_address(/*clear_sent=*/false); + } + int not_sent_count_{0}; +}; +} // namespace + +// The documented cost of the per-device guard: a clear issued from inside your own on_not_sent() +// resolves your remaining frames silently (like clear_tx_queue_for_device() - you cleared them, you +// know), while other owners sharing the address are still notified. +TEST(ModbusClientHubQueue, SelfClearFromNotSentSilentForClearerNotifiesOthers) { + NoResponseProbeHub hub; + ClearOwnAddressOnNotSentDevice clearer(&hub, 0x02); + SentCountingDevice bystander(&hub, 0x02); + + const uint8_t read_a[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t read_b[] = {0x03, 0x00, 0x20, 0x00, 0x01}; + const uint8_t read_c[] = {0x03, 0x00, 0x30, 0x00, 0x01}; + clearer.send_pdu(read_a); + clearer.send_pdu(read_b); + bystander.send_pdu(read_c); + ASSERT_EQ(hub.queued_frames(), 3u); + + clearer.send_pdu(std::span{}); // refused (empty) -> the handler clears the shared address + + EXPECT_EQ(clearer.not_sent_count_, 1); // only the refusal; the two swept frames resolve silently + EXPECT_EQ(bystander.not_sent_count_, 1); // the bystander's swept frame is still notified + EXPECT_EQ(hub.queued_frames(), 0u); +} + +// The guard must not over-suppress: a sweep started from inside on_not_sent() still delivers its +// victims' notifications (only nested refusals are silenced). +TEST(ModbusClientHubQueue, NestedClearFromNotSentStillNotifiesVictims) { + NoResponseProbeHub hub; + ClearOtherOnNotSentDevice clearer(&hub, 0x02); + SentCountingDevice victim(&hub, 0x03); + + const uint8_t read_a[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t read_b[] = {0x03, 0x00, 0x20, 0x00, 0x01}; + clearer.send_pdu(read_a); + victim.send_pdu(read_b); + ASSERT_EQ(hub.queued_frames(), 2u); + + hub.clear_tx_queue_for_address(0x02, false); // clearer's on_not_sent clears address 0x03 in turn + + EXPECT_EQ(clearer.not_sent_count_, 1); + EXPECT_EQ(victim.not_sent_count_, 1); // delivered despite arriving from a nested sweep + EXPECT_EQ(hub.queued_frames(), 0u); +} + +namespace { +// tx_blocked() flips to blocked after the first check, so send_next_frame_() passes its own gate but +// send_frame_() refuses - a deterministic transmit failure. +class FlakyBlockHub : public NoResponseProbeHub { + public: + bool tx_blocked() override { + this->tx_blocked_calls_++; + return this->tx_blocked_calls_ > 1; + } + int tx_blocked_calls_{0}; +}; + +// Reacts to a transmit failure by sending another frame from inside the failure callback. +class WriteOnNotSentDevice : public ModbusClientDevice { + public: + WriteOnNotSentDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { + this->not_sent_count_++; + const uint8_t write[] = {0x06, 0x00, 0x40, 0x01, 0x02}; + this->send_pdu(write); + } + int not_sent_count_{0}; +}; + +} // namespace + +// A transmit failure must resolve with the failed frame OUT of the queue before its on_not_sent runs: a +// handler that reacts by sending a new frame must not have that frame discarded by the pop that +// follows - the failed frame is popped first, the new frame survives. +TEST(ModbusClientHubQueue, TransmitFailurePopsBeforeNotify) { + FlakyBlockHub hub; + WriteOnNotSentDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + device.send_pdu(read); + ASSERT_EQ(hub.queued_frames(), 1u); + + hub.send_next_for_test(); // tx_blocked gate passes, send_frame_ refuses -> failure path + + EXPECT_EQ(device.not_sent_count_, 1); + ASSERT_EQ(hub.queued_frames(), 1u); // the handler's write survives... + EXPECT_EQ(hub.front().frame.pdu()[0], 0x06); // ...and it is the write, not the failed read +} + +// clear_tx_queue_for_device() drops queued frames SILENTLY - no terminal callback (the documented +// exception to the exactly-one-terminal contract; used during teardown/offline handling). +TEST(ModbusClientHubQueue, ClearDeviceQueueDropsSilently) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t read_a[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + const uint8_t read_b[] = {0x03, 0x02, 0x00, 0x00, 0x02}; + device.send_pdu(read_a); + device.send_pdu(read_b); + ASSERT_EQ(hub.queued_frames(), 2u); + + device.clear_tx_queue_for_device(); + + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(device.not_sent_count_, 0); // silent drop: no terminal callback +} + +// A send_pdu() from inside on_sent() enqueues behind the in-flight frame rather than sending +// immediately or corrupting the in-flight transaction. +TEST(ModbusClientHubSent, ReentrantSendFromOnSentQueues) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + ChainOnSentDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.send_next_for_test(); // first frame goes on the wire -> on_sent chains a follow-up + + EXPECT_TRUE(hub.waiting()); // first frame is in flight + ASSERT_EQ(hub.queued_frames(), 1u); // the follow-up queued behind it, not sent + EXPECT_EQ(hub.queued(0).frame.pdu()[2], 0x09); // it is the chained read (start address 0x0009) +} + namespace { // Overrides only the DEPRECATED on_modbus_* names: the new-name default implementations must forward, so // external devices written against the old names keep working through the deprecation window. @@ -336,7 +901,7 @@ namespace { class NotSentCountingDevice : public ModbusClientDevice { public: NotSentCountingDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} - void on_not_sent() override { this->not_sent_++; } + void on_not_sent(std::span request_pdu) override { this->not_sent_++; } int not_sent_{0}; }; } // namespace