diff --git a/esphome/components/modbus/modbus.cpp b/esphome/components/modbus/modbus.cpp index 2561ee9069..57371b9e79 100644 --- a/esphome/components/modbus/modbus.cpp +++ b/esphome/components/modbus/modbus.cpp @@ -45,26 +45,46 @@ void Modbus::loop() { } void ModbusClientHub::loop() { - // Call base class to receive bytes and parse frames - this->Modbus::loop(); + // Drain anything owed since the last loop (e.g. an external clear) before the watchdog runs, so it + // never times out an entry whose pending count has not been drained. No-op when nothing is owed. + this->sweep_(); - // If we're past the send_wait_time timeout and response buffer doesn't have the start of the expected response - if (this->waiting_for_response_.has_value()) { - ModbusDeviceCommand &wfr = this->waiting_for_response_.value(); - uint8_t expected_address = wfr.frame.address(); - if (this->last_receive_check_ - this->last_send_ > this->last_send_tx_offset_ + this->send_wait_time_ && - (this->rx_buffer_.empty() || this->rx_buffer_[0] != expected_address)) { - ESP_LOGW(TAG, "Stop waiting for response from %" PRIu8 " %" PRIu32 "ms after last send", expected_address, - this->last_receive_check_ - this->last_send_); - this->notify_no_response_(wfr); - this->waiting_for_response_.reset(); - } + this->Modbus::loop(); // receive bytes and parse frames + + // Send-wait watchdog: only the cheap time check runs at loop rate; expire_waiting_() looks the + // entry up and holds off if the response has started arriving. + if (this->waiting_for_response_ && + this->last_receive_check_ - this->last_send_ > this->last_send_tx_offset_ + this->send_wait_time_) { + this->expire_waiting_(); } - // If there's no response pending and there's commands in the buffer + this->sweep_(); // deliver owed callbacks with the hub quiescent this->send_next_frame_(); } +void ModbusClientHub::expire_waiting_() { + ModbusDeviceCommand *cmd = this->find_waiting_(); + if (cmd == nullptr) { + this->waiting_for_response_ = false; + return; + } + if (!this->rx_buffer_.empty() && this->rx_buffer_[0] == cmd->frame.address()) { + // The start of the response is in the buffer: let the frame finish arriving. + return; + } + // Only a genuine WAITING entry warrants the log (a cleared or interrupted shell timing out is expected). + if (cmd->state == FrameState::WAITING) { + ESP_LOGW(TAG, "Stop waiting for response from %" PRIu8 " %" PRIu32 "ms after last send", cmd->frame.address(), + this->last_receive_check_ - this->last_send_); + } + // Deliver on_no_response directly, the way the parse path delivers response()/error(): the entry + // lands in TIMED_OUT and the following sweep reschedules a retry or erases it. Free the + // wire first so a resend from inside the callback sees it available. + this->waiting_for_response_ = false; + this->sweep_needed_ = true; + cmd->timed_out(); +} + bool Modbus::timeout_() { // If the response frame is finished (including interframe delay) - we timeout. // The long_rx_buffer_delay accounts for long responses (larger than the UART rx_full_threshold) to avoid timeouts @@ -110,12 +130,21 @@ bool Modbus::tx_blocked() { bool ModbusClientHub::tx_blocked() { // We block transmission in any of these case: - // 1. We're waiting for a response + // 1. We're waiting for a response (a waiting entry: WAITING/INTERRUPTED/WAITING_RETIRED/INTERRUPTED_RETIRED) // 2. Any of the base class tx_blocked conditions - return (this->waiting_for_response_.has_value()) || this->Modbus::tx_blocked(); + return this->waiting_for_response_ || this->Modbus::tx_blocked(); } -bool ModbusClientHub::tx_buffer_empty() { return this->tx_buffer_.empty(); } +bool ModbusClientHub::tx_buffer_empty() { + // "Empty" for ready_for_immediate_send(): no one-shot is queued ahead of the caller. Entries in + // other states are mid-transaction or owed bookkeeping, not queued sends - and a READY continuous + // poll does not count either, since it ranks below every one-shot, so a new send goes out first. + for (const auto &cmd : this->tx_buffer_) { + if (cmd.state == FrameState::READY && !cmd.continuous) + return false; + } + return true; +} void Modbus::receive_bytes_() { this->last_receive_check_ = millis(); @@ -257,69 +286,57 @@ bool ModbusServerHub::parse_modbus_client_frame_() { return true; } -// Bounds contract, enforced by the parser (parse_modbus_server_frame_) rather than locally: -// - pdu is never empty: helpers::server_pdu_length() returns at least MIN_PDU_SIZE (1) on every -// branch, and find_custom_frame_end_() only ever lengthens the frame, so the PDU always holds -// at least the function code. -// - When the exception bit is set, pdu has at least 2 bytes: server_pdu_length() checks the -// exception bit before anything else and pins those PDUs to 2 bytes, so the exception code -// read below is always present. -// Keep those guarantees in mind when changing server_pdu_length() or adding callers. +// The parser (parse_modbus_server_frame_) guarantees the bounds relied on here: pdu is never empty, +// and an exception-flagged pdu is at least 2 bytes. Keep that in mind when changing server_pdu_length(). void ModbusClientHub::process_modbus_server_frame(uint8_t address, std::span pdu) { const uint8_t function_code = pdu[0]; - if (!this->waiting_for_response_.has_value()) { + ModbusDeviceCommand *cmd = this->waiting_for_response_ ? this->find_waiting_() : nullptr; + if (cmd == nullptr) { ESP_LOGW(TAG, "Received unexpected frame from address %" PRIu8 ", function code 0x%X, %" PRIu32 "ms after last send", address, function_code, this->last_modbus_byte_ - this->last_send_); return; - } else { // We are waiting for a response - // Check if the response matches the expected address and function code + } - ModbusDeviceCommand &wfr = this->waiting_for_response_.value(); - uint8_t expected_address = wfr.frame.address(); - uint8_t expected_function_code = wfr.frame.pdu()[0]; - if (expected_address != address || expected_function_code != (function_code & FUNCTION_CODE_MASK)) { - ESP_LOGW(TAG, - "Received incorrect frame address %" PRIu8 " <> %" PRIu8 " or function code 0x%X <> 0x%X, %" PRIu32 - "ms after last send", - address, expected_address, (function_code & FUNCTION_CODE_MASK), expected_function_code, - this->last_modbus_byte_ - this->last_send_); - // Invalidate the device; the entry survives as an interrupted shell so the late response is ignored. - // A retry requested here stays queued behind the shell until the send-wait timeout clears it. - this->notify_no_response_(wfr); - wfr.interrupted = true; - return; - } + // Check if the response matches the expected address and function code + const uint8_t expected_address = cmd->frame.address(); + const uint8_t expected_function_code = cmd->frame.pdu()[0]; + if (expected_address != address || expected_function_code != (function_code & FUNCTION_CODE_MASK)) { + ESP_LOGW(TAG, + "Received incorrect frame address %" PRIu8 " <> %" PRIu8 " or function code 0x%X <> 0x%X, %" PRIu32 + "ms after last send", + address, expected_address, (function_code & FUNCTION_CODE_MASK), expected_function_code, + this->last_modbus_byte_ - this->last_send_); + // Unexpected frame: flip a WAITING entry to an INTERRUPTED shell that ignores the rest of this + // transaction and blocks tx until the send-wait timeout, where it gets its on_no_response. + cmd->interrupt(); + return; + } - if (wfr.interrupted) { - ESP_LOGW(TAG, - "Ignoring response from %" PRIu8 " - transmission interrupted by previous unexpected response, %" PRIu32 - "ms after last send", - address, this->last_modbus_byte_ - this->last_send_); - return; - } else { // We have a valid device waiting for this response + if (cmd->state == FrameState::INTERRUPTED || cmd->state == FrameState::INTERRUPTED_RETIRED) { + // An interrupted shell keeps blocking until the send-wait timeout; a late response for it is + // ignored and does NOT free the wire. The distrust survives a clear (INTERRUPTED_RETIRED), so a + // cleared-interrupted frame still ends in on_no_response rather than delivering a late response. + ESP_LOGW(TAG, + "Ignoring response from %" PRIu8 " - transmission interrupted by previous unexpected response, %" PRIu32 + "ms after last send", + address, this->last_modbus_byte_ - this->last_send_); + return; + } - // Move the command out of the waiting slot so the request PDU stays alive for the callback. - ModbusDeviceCommand command = std::move(this->waiting_for_response_.value()); - this->waiting_for_response_.reset(); - ModbusClientDevice *device = command.device; - // The request PDU is the sent frame without the leading address and the trailing CRC. - std::span request_pdu = command.frame.pdu(); - // Is it an error response? - if (helpers::is_function_code_exception(function_code)) { - uint8_t exception = pdu[1]; // exception frames are fixed-length, so the code is always present - ESP_LOGW(TAG, - "Error function code: 0x%X exception: %" PRIu8 ", address: %" PRIu8 ", %" PRIu32 "ms after last send", - function_code, exception, address, this->last_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 - ESP_LOGV(TAG, "Ignoring response from %" PRIu8 " - no callback device set, %" PRIu32 "ms after last send", - address, this->last_modbus_byte_ - this->last_send_); - } - } + // Deliver at parse time so the response span can point into the rx buffer (zero copy). error()/ + // response() set the state and consume the request BEFORE the callback, so a clear from inside it + // ("stop polling now") wins. A device-less shell runs no callback and the sweep erases it. + this->waiting_for_response_ = false; + this->sweep_needed_ = true; + if (helpers::is_function_code_exception(function_code)) { + uint8_t exception = pdu[1]; // exception frames are fixed-length, so the code is always present + ESP_LOGW(TAG, "Error function code: 0x%X exception: %" PRIu8 ", address: %" PRIu8 ", %" PRIu32 "ms after last send", + function_code, exception, address, this->last_modbus_byte_ - this->last_send_); + cmd->error(static_cast(exception)); + } else if (!cmd->response(pdu)) { + ESP_LOGV(TAG, "Ignoring response from %" PRIu8 " - no callback device set, %" PRIu32 "ms after last send", address, + this->last_modbus_byte_ - this->last_send_); } } @@ -460,21 +477,20 @@ void ModbusServerHub::process_modbus_client_frame_(uint8_t address, uint8_t func } } +// Callers gate on tx_blocked() first, but the pre-send delay below can span several ms, so re-check +// after it and refuse (return false) if a byte arrived in that window rather than transmit over it. bool Modbus::send_frame_(const ModbusFrame &frame) { - if (this->tx_blocked()) { - ESP_LOGE(TAG, "Attempted to send while transmission blocked"); - return false; - } - if (frame.size() > MAX_FRAME_SIZE) { - ESP_LOGE(TAG, "Attempted to send frame larger than max frame size of %" PRIu16 " bytes", MAX_FRAME_SIZE); - return false; - } - const int32_t tx_delay_remaining = this->tx_delay_remaining(); if (tx_delay_remaining > 0) { delay(tx_delay_remaining); } + // The delay above can span several ms; a byte arriving in that window blocks transmission after the + // caller's gate already passed. Don't collide with the incoming frame - leave the entry to retry. + if (this->tx_blocked()) { + return false; + } + if (this->flow_control_pin_ != nullptr) { this->flow_control_pin_->digital_write(true); this->write_array(frame.data.data(), frame.size()); @@ -498,37 +514,21 @@ bool Modbus::send_frame_(const ModbusFrame &frame) { } void ModbusClientHub::send_next_frame_() { - if (this->tx_buffer_.empty()) { + if (this->tx_blocked()) + return; + + ModbusDeviceCommand *cmd = this->select_next_ready_(); + if (cmd == nullptr) + return; + + if (!this->send_frame_(cmd->frame)) { + ESP_LOGV(TAG, "Send deferred for %" PRIu8 ": a frame arrived during the send delay, will retry", + cmd->frame.address()); return; } - if (this->tx_blocked()) { - return; - } - - // 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()); - } + cmd->sent(); + this->waiting_for_response_ = true; } void ModbusClientHub::dump_config() { @@ -579,118 +579,258 @@ void ModbusServerHub::send_exception_(uint8_t address, uint8_t function_code, Ex this->send_raw_(raw_frame, 3); } -void ModbusClientHub::notify_no_response_(ModbusDeviceCommand &wfr) { - if (wfr.device == nullptr) - return; - 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) - this->requeue_waiting_frame_(wfr); - // The old transaction is over either way; never deliver anything else to the device through it. - wfr.device = nullptr; +ModbusDeviceCommand *ModbusClientHub::find_waiting_() { + for (auto &cmd : this->tx_buffer_) { + if (cmd.waiting_state()) + return &cmd; + } + return nullptr; } -void ModbusClientHub::requeue_waiting_frame_(ModbusDeviceCommand &wfr) { - const ModbusFrame &frame = wfr.frame; - 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->trigger_not_sent(frame.pdu()); +ModbusDeviceCommand *ModbusClientHub::select_next_ready_() { + // Class first (WRITE, then one-shot READ, then CONTINUOUS), oldest within a class. seq is a + // free-running counter, so compare each entry's AGE against it (correct across the full range). + const uint16_t now = this->next_seq_; + const auto age = [now](const ModbusDeviceCommand &cmd) -> uint16_t { return now - cmd.seq; }; + const auto older = [&age](const ModbusDeviceCommand &a, const ModbusDeviceCommand &b) { return age(a) > age(b); }; + ModbusDeviceCommand *best = nullptr; + for (auto &cmd : this->tx_buffer_) { + if (cmd.state != FrameState::READY) + continue; + if (best == nullptr || cmd.priority() > best->priority() || + (cmd.priority() == best->priority() && older(cmd, *best))) { + best = &cmd; + } + } + return best; +} + +bool ModbusDeviceCommand::sent() { + this->state = FrameState::WAITING; + // on_sent() is not a terminal, so nothing is consumed. + if (this->device == nullptr) + return false; + this->device->on_sent(this->frame.pdu()); + return true; +} + +bool ModbusDeviceCommand::notify_retired() { + if (!this->decrement_pending()) + return false; // nothing owed - stop the sweep draining this entry + if (this->device != nullptr) + this->device->on_not_sent(this->frame.pdu()); + return true; // consumed one debt (delivered, or silent when device-less) - keep draining to zero +} + +bool ModbusDeviceCommand::response(std::span response_pdu) { + this->state = this->state == FrameState::WAITING_RETIRED ? FrameState::RETIRED : FrameState::RECEIVED_RESPONSE; + // A continuous poll is never consumed by its own response; a one-shot consumes one request here. + if (!this->continuous) + this->decrement_pending(); + if (this->device == nullptr) + return false; + this->device->on_response(this->frame.pdu(), response_pdu); + return true; +} + +bool ModbusDeviceCommand::error(ExceptionCode exception_code) { + this->state = this->state == FrameState::WAITING_RETIRED ? FrameState::RETIRED : FrameState::RECEIVED_EXCEPTION; + // An exception ends a continuous poll too, so decrement unconditionally. + this->decrement_pending(); + if (this->device == nullptr) + return false; + this->device->on_error(this->frame.pdu(), exception_code); + return true; +} + +bool ModbusDeviceCommand::interrupt() { + // An unexpected frame distrusts the transaction. A cleared-but-still-waiting shell distrusts too, so + // the interrupt survives the clear in either order (WAITING_RETIRED -> INTERRUPTED_RETIRED). + if (this->state == FrameState::WAITING) { + this->state = FrameState::INTERRUPTED; + return true; + } + if (this->state == FrameState::WAITING_RETIRED) { + this->state = FrameState::INTERRUPTED_RETIRED; + return true; + } + return false; +} + +bool ModbusDeviceCommand::timed_out() { + this->state = FrameState::TIMED_OUT; // advance BEFORE the callback so a clear from inside it wins + this->decrement_pending(); // resolve this request (WAITING-origin, so pending >= 1) + if (this->device == nullptr) + return false; // resolved, no one to tell + if (this->device->on_no_response(this->frame.pdu())) + this->increment_pending(); // granted retry = re-request (capped) + return true; +} + +void ModbusClientHub::sweep_() { + if (!this->sweep_needed_) return; + this->sweep_needed_ = false; + // Serve only the entries present now: a callback may append (a re-send), but those sit beyond + // work_set and are left for the next sweep, which bounds the work and is the termination argument. + // Entries leave the container only in the erase pass below, so indices/references stay valid. + const size_t work_set = this->tx_buffer_.size(); + // Restart the walk after every callback: a handler may have moved any entry to any state. + bool callback_ran = true; + while (callback_ran) { + callback_ran = false; + for (size_t i = 0; i != work_set && !callback_ran; i++) { + ModbusDeviceCommand &cmd = this->tx_buffer_[i]; + switch (cmd.state) { + case FrameState::RECEIVED_RESPONSE: + case FrameState::RECEIVED_EXCEPTION: + case FrameState::TIMED_OUT: + // Off the wire, callback already delivered: reschedule what is still pending, else erase. + if (cmd.pending) + cmd.requeue(this->next_seq_++); + break; + case FrameState::RETIRED: + // Owes one on_not_sent() per accepted request; notify_retired() consumes one and reports + // whether a debt remained, so the restart loop drains the entry to zero - even a device-less + // shell with pending > 1 (no callback fires, but it still drains rather than stranding). + callback_ran = cmd.notify_retired(); + break; + case FrameState::WAITING_RETIRED: + case FrameState::INTERRUPTED_RETIRED: + // Cleared shell: drain only the un-run duplicates; the request in flight keeps pending 1 + // and gets its usual callback when it resolves. + if (cmd.pending > 1) + callback_ran = cmd.notify_retired(); + break; + default: // READY / WAITING / INTERRUPTED: idle or waiting for a response, nothing owed until the timeout + break; + } + } + } + // Erase pass: the only place entries leave the container. Storage order carries no meaning, so a + // finished entry is swap-and-popped; walking backwards means a moved-down entry is already seen. + for (size_t i = this->tx_buffer_.size(); i-- > 0;) { + const ModbusDeviceCommand &cmd = this->tx_buffer_[i]; + // pending == 0 is erasable, but shells still waiting for a response are exempt until it resolves. + if (cmd.pending != 0 || cmd.waiting_state()) + continue; + if (i + 1 != this->tx_buffer_.size()) + this->tx_buffer_[i] = std::move(this->tx_buffer_.back()); + this->tx_buffer_.pop_back(); } - // Re-queue a copy (not a move): the waiting entry may have to survive as an interrupted shell. - this->tx_buffer_.emplace_back(wfr.device, frame.address(), frame.pdu()); } // 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) { +bool ModbusClientHub::send_pdu(uint8_t address, std::span pdu, ModbusClientDevice *device, + CommandOptions options) { + // Requests refused here never enter the machine and get no callback - the false return is it. if (pdu.empty()) { - if (device != nullptr) - device->trigger_not_sent(pdu); - return; + ESP_LOGW(TAG, "Empty PDU refused for address %" PRIu8, address); + return false; } - // 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 != nullptr) - device->trigger_not_sent(pdu); - return; + ESP_LOGE(TAG, "Frame too large, refused: %" PRIu8 ":%zu bytes", address, pdu.size()); + return false; } - if (this->tx_buffer_.size() < MODBUS_TX_BUFFER_SIZE) { -#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE - char hex_buf[format_hex_pretty_size(MODBUS_MAX_LOG_BYTES)]; -#endif - ESP_LOGV(TAG, "Adding frame to tx queue: %" PRIu8 ":%s", address, - format_hex_pretty_to(hex_buf, pdu.data(), pdu.size())); - this->tx_buffer_.emplace_back(device, address, pdu); - } else { + // continuous is ignored for every mutating code (re-writing a value forever is never intended). + const bool mutates = ModbusDeviceCommand::classify(pdu[0]) == CommandPriority::WRITE; + bool continuous = false; + if (options.continuous) { + if (mutates) { + ESP_LOGV(TAG, "continuous is ignored for a mutating function (0x%X, address %" PRIu8 ")", pdu[0], address); + } else { + continuous = true; + } + } + + // A duplicate of a live entry with the same owner is not queued twice; it resolves against that + // entry: anonymous -> dropped; continuous incoming -> convert the entry to a poll; one-shot onto a + // poll -> downgrade the poll to one-shot; both one-shots -> pending++ below the cap, else refused. + for (auto &item : this->tx_buffer_) { + if (item.state == FrameState::RETIRED || item.state == FrameState::WAITING_RETIRED || + item.state == FrameState::INTERRUPTED_RETIRED) + continue; // cleared, on their way out: a new identical send queues fresh, never absorbs + if (item.device != device || !item.same_frame(address, pdu)) + continue; + if (device == nullptr) { + // A dropped read is routine (DEBUG); a dropped write/custom warns (unobservable without a device). + const bool requeueable = !helpers::is_function_code_exception(pdu[0]) && helpers::is_function_code_read(pdu[0]); + if (requeueable) { + ESP_LOGD(TAG, "Anonymous duplicate of active frame for %" PRIu8 " (function 0x%X), dropped", address, pdu[0]); + } else { + ESP_LOGW(TAG, + "Anonymous duplicate of active frame for %" PRIu8 " (function 0x%X), dropped - register a " + "device for delivery accounting", + address, pdu[0]); + } + return false; // dropped: no entry, no callbacks - the refusal is the return value + } + if (continuous) { + item.make_continuous(true); + ESP_LOGV(TAG, "Frame already active for %" PRIu8 ", now polled continuously", address); + } else if (item.continuous) { + // A one-shot duplicate downgrades the poll to a one-shot: it runs one more cycle to serve this + // request, then stops (mirrors continuous incoming converting a one-shot the other way). + item.make_continuous(false); + ESP_LOGV(TAG, "Frame already active for %" PRIu8 ", downgraded from continuous to one-shot", address); + } else if (!item.increment_pending()) { + // At the servable cap, so refused. (An absorbed duplicate leaves seq alone - the entry keeps + // its place in line, held by its oldest outstanding request.) + ESP_LOGD(TAG, "Frame already active for %" PRIu8 " with %" PRIu8 " requests pending, refused", address, + item.pending); + return false; + } else { + ESP_LOGV(TAG, "Frame already active for %" PRIu8 ", request absorbed (pending %" PRIu8 ")", address, + item.pending); + } + return true; + } + + // Backstop counts every entry; dead ones are gone by the sweep's end, so at worst they cost one + // refusal at the very cap for one loop. + if (this->tx_buffer_.size() >= MODBUS_TX_BUFFER_SIZE) { #if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_ERROR char hex_buf[format_hex_pretty_size(MODBUS_MAX_LOG_BYTES)]; #endif - ESP_LOGE(TAG, "Write buffer full, dropped: %" PRIu8 ":%s", address, + ESP_LOGE(TAG, "Write buffer full, refused: %" PRIu8 ":%s", address, format_hex_pretty_to(hex_buf, pdu.data(), pdu.size())); - if (device != nullptr) - device->trigger_not_sent(pdu); + return false; } +#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE + char hex_buf[format_hex_pretty_size(MODBUS_MAX_LOG_BYTES)]; +#endif + ESP_LOGV(TAG, "Adding frame to tx queue: %" PRIu8 ":%s", address, + format_hex_pretty_to(hex_buf, pdu.data(), pdu.size())); + this->tx_buffer_.emplace_back(device, address, pdu, continuous, this->next_seq_++); + return true; } -void ModbusClientHub::clear_tx_queue_for_address(uint8_t address, bool clear_sent) { - // 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. +void ModbusClientHub::clear_tx_queue_for_address(uint8_t address) { + // A clear is a pure state flip; the sweep delivers every owed on_not_sent() from a quiescent hub. 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) { - ESP_LOGV(TAG, "Clearing waiting for response for address %" PRIu8, address); - // Invalidate the waiting device so it won't process a response. - this->waiting_for_response_.value().device = nullptr; - } + if (cmd.frame.address() != address) + continue; + cmd.retire(); + this->sweep_needed_ = true; } } -void ModbusClientHub::clear_tx_queue_for_device(ModbusClientDevice *device) { - // 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(), - [device](const ModbusDeviceCommand &cmd) { return cmd.device == device; }), - tx_buffer.end()); - if (this->waiting_for_response_.has_value() && this->waiting_for_response_.value().device) { - if (this->waiting_for_response_.value().device == device) { - ESP_LOGV(TAG, "Clearing waiting for response"); - // Invalidate the waiting device so it won't process a response. - this->waiting_for_response_.value().device = nullptr; - } +void ModbusClientHub::clear_tx_queue_for_device(ModbusClientDevice *device) { + // Silent teardown (supersede semantics): the caller's own frames vanish without callbacks; see + // the lifecycle note on ModbusClientDevice. + for (auto &cmd : this->tx_buffer_) { + if (cmd.device != device) + continue; + cmd.silent_retire(); + this->sweep_needed_ = true; } } void ModbusClientHub::send_raw(const std::vector &payload, ModbusClientDevice *device) { if (payload.size() < 2) { - if (device != nullptr) - device->trigger_not_sent({}); // too short to contain a PDU + ESP_LOGW(TAG, "send_raw() payload too short to contain a PDU, refused"); return; } this->send_pdu(payload[0], std::span(payload).subspan(1), device); @@ -706,23 +846,26 @@ void ModbusServerHub::send_raw_(const uint8_t *payload, uint16_t len) { return; } - // In the rare case that the server is blocked (frame delay has not elapsed), we delay the send. - // This should only happen at low baud rates with long frame delays. + // If blocked now (frame delay not elapsed at low baud, or a frame arriving), defer rather than + // busy-waiting the loop; send_frame_ itself re-checks after its delay, so the deferred callback + // just reports whatever it returns. if (this->tx_blocked()) { // Stash the raw payload in a single member buffer so the deferred callback can rebuild the frame - // without a heap allocation. Only one server reply is ever in flight, and the named timeout ensures - // only one deferred send is pending, so a single buffer is sufficient. + // without a heap allocation. Only one server reply is ever waiting, so a single buffer suffices. std::memcpy(this->deferred_payload_.data(), payload, len); this->deferred_payload_len_ = len; this->set_timeout("deferred_send", this->tx_delay_remaining(), [this]() { ModbusFrame frame(this->deferred_payload_[0], this->deferred_payload_.data() + 1, this->deferred_payload_len_ - 1); - this->send_frame_(frame); + if (!this->send_frame_(frame)) + ESP_LOGE(TAG, "Deferred server reply dropped: transmission still blocked"); }); - } else { - ModbusFrame frame(payload[0], payload + 1, len - 1); - this->send_frame_(frame); + return; } + + ModbusFrame frame(payload[0], payload + 1, len - 1); + if (!this->send_frame_(frame)) + ESP_LOGE(TAG, "Server reply dropped: a frame arrived during the send delay"); } void Modbus::clear_rx_buffer_(const LogString *reason, bool warn, size_t bytes_to_clear) { @@ -778,7 +921,7 @@ void ModbusClientDevice::dispatch_response_(std::span request_pdu const bool bits = function_code == FunctionCode::READ_COILS || function_code == FunctionCode::READ_DISCRETE_INPUTS; const size_t expected_data_size = - bits ? (static_cast(count_or_value) + 7) / 8 : static_cast(count_or_value) * 2; + bits ? packed_bit_bytes(count_or_value) : static_cast(count_or_value) * 2; if (response_pdu.size() != expected_data_size + 2) { ESP_LOGD(TAG, "Response length %zu does not match request (expected %zu) for function code 0x%X", response_pdu.size(), expected_data_size + 2, static_cast(function_code)); diff --git a/esphome/components/modbus/modbus.h b/esphome/components/modbus/modbus.h index 2204c0f82b..c49f52df55 100644 --- a/esphome/components/modbus/modbus.h +++ b/esphome/components/modbus/modbus.h @@ -16,7 +16,13 @@ namespace esphome::modbus { -static constexpr uint16_t MODBUS_TX_BUFFER_SIZE = 15; +// Tx queue backstop. Duplicate frames dedup into one entry, so reads can never approach this in a +// sane config - it exists to stop a runaway generator of distinct frames (e.g. a loop writing a +// changing value) from growing the heap unboundedly. The deque grows on demand; this reserves nothing. +// Worst case the cap permits: 128 distinct max-size frames = ~32 kB of spilled frame data plus +// ~3 kB of deque node storage (typical 8-byte frames stay inline; large PDUs spill to one +// allocation each) - pathological configs only, but the numbers matter when tuning for ESP8266. +static constexpr uint16_t MODBUS_TX_BUFFER_SIZE = 128; static constexpr uint16_t MODBUS_TX_MAX_DELAY_MS = 5; // Typical frames -- reads and single-register/coil writes -- are exactly 8 bytes @@ -70,6 +76,9 @@ class Modbus : public uart::UARTDevice, public Component { // pdu is the whole PDU (function code + payload, no address/CRC); pdu[0] is the (standard or custom) function code. virtual void process_modbus_server_frame(uint8_t address, std::span pdu) = 0; void clear_rx_buffer_(const LogString *reason, bool warn = false, size_t bytes_to_clear = 0); + // Transmit a frame. Callers gate on tx_blocked() first, but the pre-send delay can span several ms, + // so this re-checks after the delay and returns false without transmitting if a byte arrived in that + // window (the caller then leaves its entry to retry). Returns true once the frame has been transmitted. bool send_frame_(const ModbusFrame &frame); // Scans forward from min_length to find a frame boundary by CRC match for custom function codes. // Returns the matched frame length, or 0 if no valid CRC was found within MAX_FRAME_SIZE. @@ -90,20 +99,158 @@ class Modbus : public uart::UARTDevice, public Component { class ModbusClientDevice; class ModbusServerDevice; +// Transmit ordering, highest first: writes before one-shot reads before continuous polls. Derived +// at selection time, never caller-chosen or stored. +enum class CommandPriority : uint8_t { CONTINUOUS = 0, READ, WRITE }; + +// Per-entry lifecycle state. Waiting states (see waiting_state()) hold the bus; the sweep delivers owed +// callbacks from a quiescent hub, and an entry is erased once pending == 0 && !waiting_state(). +enum class FrameState : uint8_t { + READY = 0, + WAITING, + RECEIVED_RESPONSE, + RECEIVED_EXCEPTION, + TIMED_OUT, // on_no_response delivered at the send-wait timeout; awaiting reschedule/erase + INTERRUPTED, // unexpected frame arrived; ignores this transaction, waits out the timeout + WAITING_RETIRED, // cleared while WAITING: a late response is still delivered as its usual terminal + INTERRUPTED_RETIRED, // cleared while INTERRUPTED: still distrusts late frames, ends in on_no_response + RETIRED, // cleared, off the wire +}; + +// Per-command send options. Append-only; pass via designated initializers ({.continuous = true}). +struct CommandOptions { + // A continuous poll lives in the queue until cancelled or failed; ignored for mutating codes. + bool continuous{false}; +}; + 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}; + FrameState state{FrameState::READY}; + // A continuous poll is a subscription: pending fixed at 1, removed only by cancellation or failure. + bool continuous{false}; + // Accepted requests this entry stands for, capped at max_pending(); drains one terminal each. + uint8_t pending{1}; + // Place-in-line stamp (hub's free-running counter); selection takes the oldest for round-robin + // fairness within a class. Meant to wrap. + uint16_t seq{0}; - ModbusDeviceCommand(ModbusClientDevice *device, uint8_t address, const uint8_t *src, uint16_t len) - : device(device), frame(address, src, len) {} - /// Build a command from a PDU span: a caller-supplied PDU, or an existing frame's own pdu() when re-queueing - /// Callers must bound the PDU to MAX_PDU_SIZE - ModbusDeviceCommand(ModbusClientDevice *device, uint8_t address, std::span pdu) - : device(device), frame(address, pdu.data(), static_cast(pdu.size())) {} + // Build a command from a PDU span (caller bounds it to MAX_PDU_SIZE); fully initialized here. + ModbusDeviceCommand(ModbusClientDevice *device, uint8_t address, std::span pdu, + bool continuous = false, uint16_t seq = 0) + : device(device), + frame(address, pdu.data(), static_cast(pdu.size())), + continuous(continuous), + seq(seq) {} + + // Transmit ordering class, derived (never stored): a continuous poll ranks below every one-shot. + CommandPriority priority() const { + return this->continuous ? CommandPriority::CONTINUOUS : classify(this->frame.pdu()[0]); + } + // Wire-derived class: mutating codes rank WRITE; exception-flagged codes are excluded. + static CommandPriority classify(uint8_t function_code) { + if (helpers::is_function_code_exception(function_code)) + return CommandPriority::READ; + const auto code = static_cast(function_code); + if (helpers::is_function_code_write(function_code) || code == FunctionCode::MASK_WRITE_REGISTER || + code == FunctionCode::READ_WRITE_MULTIPLE_REGISTERS) { + return CommandPriority::WRITE; + } + return CommandPriority::READ; + } + + // Requests this entry can serve: a standard read twice (run plus one re-run), everything else once. + uint8_t max_pending() const { + const uint8_t fc = this->frame.pdu()[0]; + const bool requeueable = !helpers::is_function_code_exception(fc) && helpers::is_function_code_read(fc); + return (requeueable && !this->continuous) ? 2 : 1; + } + // Device-scoped clear: detach with no callback (device-less, pending 0). An entry still waiting for + // a response keeps its state as a reply-ignoring shell that resolves silently; any other goes RETIRED. + void silent_retire() { + if (!this->waiting_state()) + this->state = FrameState::RETIRED; + this->pending = 0; + this->device = nullptr; + } + // Re-ready for another transmission, restamped to the tail of its class (hub passes next_seq_++). + void requeue(uint16_t seq) { + this->state = FrameState::READY; + this->seq = seq; + } + // Re-task a frame that lives on: upgrade a one-shot to a continuous poll, or downgrade a poll back to + // a one-shot. Either way the entry keeps running and owes a request, so this is not a plain setter - + // to tear an entry down instead, use retire()/silent_retire(), which leave pending as the count owed. + // On: the entry becomes a continuous poll, superseding any absorbed requests (pending resets to the + // single subscription). Off: a one-shot duplicate has cancelled the poll, but the entry must still run + // once to serve that request - so restore one first. While the flag is still set max_pending() is 1, + // so the restore lifts a terminated poll (pending 0, after an error/timeout) back to 1 and is a no-op + // on a live poll already at 1; the flag drops afterwards, when a read's cap can widen to 2 without + // retroactively inflating that no-op. + void make_continuous(bool continuous) { + if (continuous) { + this->continuous = true; + this->pending = 1; + } else { + this->increment_pending(); + this->continuous = false; + } + } + // Address-scoped clear: keep pending and device so the sweep delivers one on_not_sent() per un-run + // request. An entry still waiting for a response keeps its in-flight request (whose usual terminal is + // still coming) and drains only its duplicates: WAITING -> WAITING_RETIRED, and INTERRUPTED -> + // INTERRUPTED_RETIRED which keeps distrusting late frames (they were already interrupted). Any other + // state -> RETIRED, draining everything. A cleared frame that then times out still honors a retry: + // the clear is address-scoped (any device may call it) while the retry is the owning device's call + // via on_no_response - the bus obeys the owner. + void retire() { + if (this->state == FrameState::WAITING) { + this->state = FrameState::WAITING_RETIRED; + } else if (this->state == FrameState::INTERRUPTED) { + this->state = FrameState::INTERRUPTED_RETIRED; + } else if (!this->waiting_state()) { // an already-retired shell stays put; off the wire -> RETIRED + this->state = FrameState::RETIRED; + } + this->continuous = false; + } + + // True while the entry is still waiting for a response; the erase pass exempts these even at pending 0. + bool waiting_state() const { + return this->state == FrameState::WAITING || this->state == FrameState::INTERRUPTED || + this->state == FrameState::WAITING_RETIRED || this->state == FrameState::INTERRUPTED_RETIRED; + } + + bool decrement_pending() { + if (this->pending > 0) { + this->pending--; + return true; + } + return false; + } + // Add one request, honouring the cap; false = already at cap (absorb a duplicate, restore a retry). + bool increment_pending() { + if (this->pending < this->max_pending()) { + this->pending++; + return true; + } + return false; + } + + // Terminal/lifecycle methods: each owns its transition, callback, and pending accounting and + // returns whether a callback ran. Out-of-line: ModbusClientDevice is incomplete here. + bool sent(); + bool response(std::span response_pdu); + bool error(ExceptionCode exception_code); + bool interrupt(); + bool timed_out(); + bool notify_retired(); + + /// True if this command carries the same wire frame (address + PDU) as the given one. + bool same_frame(uint8_t address, std::span pdu) const { + const auto own_pdu = this->frame.pdu(); + return own_pdu.size() == pdu.size() && this->frame.address() == address && + memcmp(own_pdu.data(), pdu.data(), pdu.size()) == 0; + } }; class ModbusClientHub : public Modbus { @@ -123,14 +270,15 @@ class ModbusClientHub : public Modbus { payload_len), device); }; - void send_pdu(uint8_t address, std::span pdu, ModbusClientDevice *device = nullptr); + // Queue a request; true once it is a live entry (resolving in one terminal), false if it never + // entered the machine (empty/oversize PDU, full queue, anonymous or over-cap duplicate) - no callback. + bool send_pdu(uint8_t address, std::span pdu, ModbusClientDevice *device = nullptr, + CommandOptions options = {}); 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); + // Clear an address's commands; each un-run request resolves via on_not_sent(), but a frame on the + // wire still runs to its usual terminal. clear_tx_queue_for_device() instead discards silently. + void clear_tx_queue_for_address(uint8_t address); void clear_tx_queue_for_device(ModbusClientDevice *device); protected: @@ -138,18 +286,29 @@ class ModbusClientHub : public Modbus { void parse_modbus_frames() override; void process_modbus_server_frame(uint8_t address, std::span pdu) override; void send_next_frame_(); - // Notify the waiting device of no response; re-queues the frame if on_no_response() returns true. - // wfr is the caller's checked reference to waiting_for_response_. - void notify_no_response_(ModbusDeviceCommand &wfr); - void requeue_waiting_frame_(ModbusDeviceCommand &wfr); + // Deliver owed callbacks from a quiescent hub and apply lifecycle bookkeeping; see FrameState. + void sweep_(); + // The selection function: best READY entry (WRITE class first, then one-shot reads, then the + // least-recently-served continuous; FIFO by seq within each group), or nullptr. + ModbusDeviceCommand *select_next_ready_(); + // Locate the single entry waiting for a response (WAITING/INTERRUPTED/WAITING_RETIRED/INTERRUPTED_RETIRED). + ModbusDeviceCommand *find_waiting_(); + // End the wait for a response on send-wait timeout (the loop() watchdog body); see FrameState. + void expire_waiting_(); uint16_t send_wait_time_{2000}; uint16_t turnaround_delay_ms_{0}; - std::optional waiting_for_response_; - // std::deque is appropriate here since we need a FIFO buffer, and we can't know ahead of time how many - // requests will be queued. Each modbus component may queue multiple requests, and the sequence of scheduling - // may change at run time. + // Set on transmit, cleared on the transaction-ending transition; send_next_frame_ won't select + // while it is set, so at most one frame is awaiting a response. + bool waiting_for_response_{false}; + + // Set whenever a transition leaves owed callbacks behind; quiet loop() passes skip the sweep. + bool sweep_needed_{false}; + // Monotonic stamp source for ModbusDeviceCommand::seq. + uint16_t next_seq_{0}; + + // Plain append-order container; ordering lives in select_next_ready_(), lifecycle in FrameState. std::deque tx_buffer_; }; @@ -176,7 +335,7 @@ class ModbusServerHub : public Modbus { std::vector devices_; // Holds the raw payload of a single reply deferred for sending when tx was blocked at send time. - // Only one server reply can be in flight at once, so a single fixed buffer avoids heap allocation. + // Only one server reply can be waiting at once, so a single fixed buffer avoids heap allocation. std::array deferred_payload_; uint16_t deferred_payload_len_{0}; }; @@ -184,25 +343,24 @@ 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(). +/// Callback contract. Each accepted request ends in exactly ONE terminal: on_response() (data), +/// on_error() (exception), on_no_response() (timeout/interruption), or on_not_sent() (dropped by +/// clear_tx_queue_for_address before transmission). A request refused at send_pdu() (false return) +/// gets none. on_sent() is additional, once per transmission, never for an on_not_sent() request. +/// on_response()/on_error() fire at parse time and on_no_response() at the send-wait watchdog, all +/// from a quiescent hub; only on_not_sent() is delivered by the sweep. Sending or clearing from +/// inside a callback is safe (picked up by the next sweep). Exceptions to "exactly one terminal": +/// clear_tx_queue_for_device() drops the caller's own frames silently; a continuous poll's cycles are +/// its own accounting (a one-shot duplicate downgrades the poll to a one-shot; a continuous duplicate +/// merges into it). +/// +/// Invariants: +/// - Public entry points (send_pdu/clear_tx_queue_*) only append to the queue or mutate an existing +/// entry through its callback-free transition methods. +/// - Public entry points can never trigger a callback synchronously. +/// - Callbacks are delivered only from within loop(). +/// - At most one callback is ever issued between calls to sweep_(): +/// sweep_ -> parse (response OR error) OR timeout (no_response) -> sweep_ -> send (sent) -> sweep_ (next loop) class ModbusClientDevice { public: ModbusClientDevice() = default; @@ -230,29 +388,14 @@ 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). - /// Do not attempt to queue a command in this callback. - /// (The on_modbus_* names below are deprecated pre-rename spellings; the defaults forward so - /// external devices overriding them keep working through the deprecation window.) + /// Called when an accepted request was dropped before transmission by clear_tx_queue_for_address(). + /// (on_modbus_* below are deprecated pre-rename spellings; the defaults forward during deprecation.) 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 @@ -322,56 +465,60 @@ class ModbusClientDevice { helpers::create_client_pdu((FunctionCode) function, start_address, number_of_entities, payload, payload_len), this); } - void send_pdu(std::span pdu) { this->parent_->send_pdu(this->address_, pdu, this); } + /// See ModbusClientHub::send_pdu(): true = accepted (a terminal callback will follow), + /// false = refused at the door (no callback). + bool send_pdu(std::span pdu, CommandOptions options = {}) { + return this->parent_->send_pdu(this->address_, pdu, this, options); + } 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()) { - // 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); + bool send_raw(const std::vector &payload) { + if (payload.empty()) + return false; // too short to contain a PDU; refused at the door like any invalid send + return this->parent_->send_pdu(payload[0], std::span(payload).subspan(1), this); } // Reads via the table-appropriate function code; an unreadable entity type maps to INVALID, which - // create_read_pdu() rejects into an empty PDU and send_pdu() signals via on_not_sent(). - void read_entities(EntityType entity_type, uint16_t start_address, uint16_t number_of_entities) { - this->send_pdu(helpers::create_read_pdu(helpers::modbus_register_read_function(entity_type), start_address, - number_of_entities)); + // create_read_pdu() rejects into an empty PDU and send_pdu() refuses with a false return. + bool read_entities(EntityType entity_type, uint16_t start_address, uint16_t number_of_entities, + CommandOptions options = {}) { + return this->send_pdu(helpers::create_read_pdu(helpers::modbus_register_read_function(entity_type), start_address, + number_of_entities), + options); } - void read_input_registers(uint16_t start_address, uint16_t number_of_registers) { - this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_INPUT_REGISTERS, start_address, number_of_registers)); + bool read_input_registers(uint16_t start_address, uint16_t number_of_registers, CommandOptions options = {}) { + return this->send_pdu( + helpers::create_read_pdu(FunctionCode::READ_INPUT_REGISTERS, start_address, number_of_registers), options); } - void read_holding_registers(uint16_t start_address, uint16_t number_of_registers) { - this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_HOLDING_REGISTERS, start_address, number_of_registers)); + bool read_holding_registers(uint16_t start_address, uint16_t number_of_registers, CommandOptions options = {}) { + return this->send_pdu( + helpers::create_read_pdu(FunctionCode::READ_HOLDING_REGISTERS, start_address, number_of_registers), options); } - void read_coils(uint16_t start_address, uint16_t number_of_coils) { - this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_COILS, start_address, number_of_coils)); + bool read_coils(uint16_t start_address, uint16_t number_of_coils, CommandOptions options = {}) { + return this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_COILS, start_address, number_of_coils), options); } - void read_discrete_inputs(uint16_t start_address, uint16_t number_of_inputs) { - this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_DISCRETE_INPUTS, start_address, number_of_inputs)); + bool read_discrete_inputs(uint16_t start_address, uint16_t number_of_inputs, CommandOptions options = {}) { + return this->send_pdu(helpers::create_read_pdu(FunctionCode::READ_DISCRETE_INPUTS, start_address, number_of_inputs), + options); } - void write_single_register(uint16_t start_address, uint16_t value) { - this->send_pdu(helpers::create_write_single_register_pdu(start_address, value)); + bool write_single_register(uint16_t start_address, uint16_t value) { + return this->send_pdu(helpers::create_write_single_register_pdu(start_address, value)); } - void write_single_coil(uint16_t address, bool value) { - this->send_pdu(helpers::create_write_single_coil_pdu(address, value)); + bool write_single_coil(uint16_t address, bool value) { + return this->send_pdu(helpers::create_write_single_coil_pdu(address, value)); } - void write_multiple_registers(uint16_t start_address, std::span values) { - this->send_pdu(helpers::create_write_registers_pdu(start_address, values)); + bool write_multiple_registers(uint16_t start_address, std::span values) { + return this->send_pdu(helpers::create_write_registers_pdu(start_address, values)); } /// Note: std::vector cannot bind to std::span; use a contiguous bool container or the packed /// overload. - void write_multiple_coils(uint16_t start_address, std::span values) { - this->send_pdu(helpers::create_write_coils_pdu(start_address, values)); + bool write_multiple_coils(uint16_t start_address, std::span values) { + return this->send_pdu(helpers::create_write_coils_pdu(start_address, values)); } /// Packed variant: a PackedBits view (the same layout on_read_coils() delivers), so /// read-modify-write needs no unpack/repack. - void write_multiple_coils(uint16_t start_address, PackedBits bits) { - this->send_pdu(helpers::create_write_coils_pdu(start_address, bits)); - } - inline void clear_tx_queue_for_address(bool clear_sent = true) { - this->parent_->clear_tx_queue_for_address(this->address_, clear_sent); + bool write_multiple_coils(uint16_t start_address, PackedBits bits) { + return this->send_pdu(helpers::create_write_coils_pdu(start_address, bits)); } + inline void clear_tx_queue_for_address() { this->parent_->clear_tx_queue_for_address(this->address_); } inline void clear_tx_queue_for_device() { this->parent_->clear_tx_queue_for_device(this); } // If more than one device is connected block sending a new command before a response is received @@ -385,8 +532,6 @@ 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/modbus_definitions.h b/esphome/components/modbus/modbus_definitions.h index fd99055ca2..f883bfff30 100644 --- a/esphome/components/modbus/modbus_definitions.h +++ b/esphome/components/modbus/modbus_definitions.h @@ -116,6 +116,14 @@ static constexpr uint16_t READ_PDU_SIZE = 5; // A single-write PDU is always function code(1) + address(2) + value(2) static constexpr uint16_t WRITE_SINGLE_PDU_SIZE = 5; static constexpr uint16_t MAX_FRAME_SIZE = 256; +// Both send paths bound their payload so the framed result lands exactly on the RTU limit: a client +// PDU gains an address byte and a CRC, a raw server frame gains a CRC. send_frame_() therefore never +// has to check the framed size - it cannot be exceeded. +static_assert(MAX_PDU_SIZE + 3 == MAX_FRAME_SIZE, "a framed client PDU must fill the RTU frame limit"); +static_assert(MAX_RAW_SIZE + 2 == MAX_FRAME_SIZE, "a framed raw server payload must fill the RTU frame limit"); +/// Bits pack 8 per data byte, rounded up to whole bytes. +constexpr size_t packed_bit_bytes(size_t bits) { return (bits + 7) / 8; } + /** Read-only view of Modbus-packed bits: bit 0 of byte 0 is the first bit (LSB first), the layout * coil/discrete-input values use on the wire. Bundles the bit count with the packed bytes so the * two cannot desynchronize. The view does not own the bytes - it is only valid while they are. @@ -123,9 +131,6 @@ static constexpr uint16_t MAX_FRAME_SIZE = 256; * with any subscript. Writes and forwarding are defensive: set() drops out-of-range bits and * bytes() clamps to the real span, because those paths touch buffers and the wire directly. */ -/// Bits pack 8 per data byte, rounded up to whole bytes. -constexpr size_t packed_bit_bytes(size_t bits) { return (bits + 7) / 8; } - class PackedBits { public: PackedBits(std::span data, uint16_t count) : data_(data), count_(count) {} diff --git a/esphome/components/modbus/modbus_helpers.cpp b/esphome/components/modbus/modbus_helpers.cpp index 85c5d3e882..8428ea27ea 100644 --- a/esphome/components/modbus/modbus_helpers.cpp +++ b/esphome/components/modbus/modbus_helpers.cpp @@ -114,8 +114,10 @@ bool is_server_pdu_standard(const uint8_t *pdu, size_t size) { case FunctionCode::WRITE_MULTIPLE_REGISTERS: { // The response echoes start address and quantity: bound them like the request side does. const bool bits = function_code == FunctionCode::WRITE_MULTIPLE_COILS; + const uint16_t start_address = get_data(pdu, 1); + const uint16_t quantity = get_data(pdu, 3); const uint16_t max_quantity = bits ? MAX_NUM_OF_COILS_TO_WRITE : MAX_NUM_OF_REGISTERS_TO_WRITE; - return quantity_in_range(get_data(pdu, 1), get_data(pdu, 3), max_quantity); + return quantity_in_range(start_address, quantity, max_quantity); } case FunctionCode::WRITE_SINGLE_COIL: // The response echoes the request, so the same ON/OFF constraint applies. @@ -137,25 +139,31 @@ bool is_client_pdu_standard(const uint8_t *pdu, size_t size) { case FunctionCode::READ_INPUT_REGISTERS: { const bool bits = function_code == FunctionCode::READ_COILS || function_code == FunctionCode::READ_DISCRETE_INPUTS; + const uint16_t start_address = get_data(pdu, 1); + const uint16_t quantity = get_data(pdu, 3); const uint16_t max_quantity = bits ? MAX_NUM_OF_COILS_TO_READ : MAX_NUM_OF_REGISTERS_TO_READ; - return quantity_in_range(get_data(pdu, 1), get_data(pdu, 3), max_quantity); + return quantity_in_range(start_address, quantity, max_quantity); } case FunctionCode::WRITE_MULTIPLE_COILS: case FunctionCode::WRITE_MULTIPLE_REGISTERS: { const bool bits = function_code == FunctionCode::WRITE_MULTIPLE_COILS; + const uint16_t start_address = get_data(pdu, 1); const uint16_t quantity = get_data(pdu, 3); const uint16_t max_quantity = bits ? MAX_NUM_OF_COILS_TO_WRITE : MAX_NUM_OF_REGISTERS_TO_WRITE; // Coils are packed 8 per data byte; registers are 2 bytes each. const size_t expected_data_bytes = bits ? packed_bit_bytes(quantity) : quantity * 2; - return quantity_in_range(get_data(pdu, 1), quantity, max_quantity) && pdu[5] == expected_data_bytes; + return quantity_in_range(start_address, quantity, max_quantity) && pdu[5] == expected_data_bytes; } case FunctionCode::READ_FILE_RECORD: case FunctionCode::WRITE_FILE_RECORD: return pdu[1] <= MAX_PDU_SIZE - 2; case FunctionCode::READ_WRITE_MULTIPLE_REGISTERS: { + const uint16_t start_address_read = get_data(pdu, 1); + const uint16_t quantity_read = get_data(pdu, 3); + const uint16_t start_address_write = get_data(pdu, 5); const uint16_t quantity_write = get_data(pdu, 7); - return quantity_in_range(get_data(pdu, 1), get_data(pdu, 3), MAX_NUM_OF_REGISTERS_TO_READ) && - quantity_in_range(get_data(pdu, 5), quantity_write, MAX_NUM_OF_REGISTERS_TO_WRITE_RW) && + return quantity_in_range(start_address_read, quantity_read, MAX_NUM_OF_REGISTERS_TO_READ) && + quantity_in_range(start_address_write, quantity_write, MAX_NUM_OF_REGISTERS_TO_WRITE_RW) && pdu[9] == quantity_write * 2; } case FunctionCode::WRITE_SINGLE_COIL: @@ -299,6 +307,8 @@ static void append_pdu_header(StaticVector &pdu, FunctionCode func // Zero the unused bits of a multi-coil write's final data byte, as the spec requires. Kept in one // place so the generic and typed coil builders produce identical wire bytes for the same write. +// The caller must pass a span whose LAST byte is the final packed-bit byte - both builders pass the +// whole PDU, which qualifies because the coil data is always the PDU's tail. static void mask_trailing_pad_bits(std::span data, uint16_t bit_count) { if (data.empty() || bit_count % 8 == 0) return; diff --git a/esphome/components/modbus_controller/modbus_controller.h b/esphome/components/modbus_controller/modbus_controller.h index 928054f1ab..3c789936af 100644 --- a/esphome/components/modbus_controller/modbus_controller.h +++ b/esphome/components/modbus_controller/modbus_controller.h @@ -305,11 +305,8 @@ class ModbusController final : public PollingComponent, public modbus::ModbusCli /// Deliberately shadows the deprecated ModbusClientDevice::send_raw() with identical semantics: /// controller-level raw sends stay supported until the command machinery is replaced. void send_raw(const std::vector &payload) { - if (payload.empty()) { - // Through the guard like every other delivery, so a handler calling send_raw({}) cannot recurse. - this->trigger_not_sent({}); - return; - } + if (payload.empty()) + return; // refused at the door, like every invalid send; no callback follows this->parent_->send_pdu(payload[0], std::span(payload).subspan(1), this); } /// Registers a sensor with the controller. Called by esphomes code generator diff --git a/tests/components/modbus/heap_probe_test.cpp b/tests/components/modbus/heap_probe_test.cpp index 2c7d9747bd..ddf905a8df 100644 --- a/tests/components/modbus/heap_probe_test.cpp +++ b/tests/components/modbus/heap_probe_test.cpp @@ -117,9 +117,10 @@ TEST(HeapProbe, TypicalFrameConstructionIsAllocationFree) { EXPECT_EQ(large.count, 1u); } -// Queueing typical commands is fully allocation-free: the frame fits the inline buffer and the tx -// deque's first block is already allocated when the hub is constructed. (A queue deeper than one -// deque block - roughly a dozen commands - would allocate further blocks.) +// Queueing typical commands is allocation-free within the deque's first block: the frame fits the +// inline buffer, every entry is a plain append (ordering lives in selection, not storage), and the +// first block is already allocated when the hub is constructed. A 512-byte deque block holds +// 512 / sizeof(ModbusDeviceCommand) entries (16 on the 64-bit host); a deeper queue allocates more. TEST(HeapProbe, QueueingTypicalCommandsIsAllocationFree) { ModbusClientHub hub; ModbusClientDevice device(&hub, 0x02); @@ -129,14 +130,36 @@ TEST(HeapProbe, QueueingTypicalCommandsIsAllocationFree) { req.assign(read_pdu, read_pdu + sizeof(read_pdu)); constexpr int n = 12; + static_assert(n * sizeof(ModbusDeviceCommand) < 512, "keep n within one deque block so the probe stays meaningful"); size_t total = 0; for (int i = 0; i != n; i++) { + req[2] = static_cast(i); // distinct start addresses: identical frames would dedup, not enqueue total += sample([&] { device.send_pdu(req); }).count; } printf("HEAPPROBE queue_%d_typical_commands total_allocs=%zu\n", n, total); EXPECT_EQ(total, 0u); } +// A WRITE arriving behind queued reads is a plain append too - the old priority front-insert (and +// its possible front-block allocation) is gone; the write wins transmit SELECTION instead. +TEST(HeapProbe, WriteBehindQueuedReadsAppendsAllocationFree) { + ModbusClientHub hub; + ModbusClientDevice device(&hub, 0x02); + + StaticVector req; + const uint8_t read_pdu[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + req.assign(read_pdu, read_pdu + sizeof(read_pdu)); + for (int i = 0; i != 3; i++) { + req[2] = static_cast(i); // distinct start addresses: identical frames would dedup, not enqueue + device.send_pdu(req); + } + + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + Sample append = sample([&] { device.send_pdu(write_pdu); }); + printf("HEAPPROBE write_append count=%zu bytes=%zu\n", append.count, append.bytes); + EXPECT_EQ(append.count, 0u); +} + // End to end: bytes injected at the UART travel through receive, frame parsing, response matching and // device dispatch. The first response may grow the hub's rx buffer once; after that warm-up, handling a // response performs zero heap allocations all the way to the device callback. @@ -189,6 +212,9 @@ TEST(HeapProbe, TypicalFrameConstructionIsAllocationFree) { TEST(HeapProbe, QueueingTypicalCommandsIsAllocationFree) { GTEST_SKIP() << "allocation counting requires an AddressSanitizer build"; } +TEST(HeapProbe, WriteBehindQueuedReadsAppendsAllocationFree) { + GTEST_SKIP() << "allocation counting requires an AddressSanitizer build"; +} TEST(HeapProbe, ResponseHandlingIsAllocationFreeAfterWarmup) { GTEST_SKIP() << "allocation counting requires an AddressSanitizer build"; } diff --git a/tests/components/modbus/modbus_client_hub_test.cpp b/tests/components/modbus/modbus_client_hub_test.cpp index 11bc10200d..4d5b4e7ee8 100644 --- a/tests/components/modbus/modbus_client_hub_test.cpp +++ b/tests/components/modbus/modbus_client_hub_test.cpp @@ -13,33 +13,64 @@ namespace esphome::modbus::testing { namespace { -// Exposes the protected tx queue and waiting-for-response slot so tests can drive the -// no-response path without a UART: force_send_front() mimics send_next_frame_() moving the -// front frame in flight, timeout_waiting() mimics the loop() no-response timeout handling. +// Exposes the frame state machine so tests can drive it without a UART (force_send_next(), +// timeout_waiting(), sweep_for_test() stand in for the loop() transmit/watchdog/sweep steps). 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) + // The old "queue" view: entries awaiting transmission, in STORAGE order (selection order is + // what the engine transmits by; use next_ready() for that). + size_t queued_frames() const { + size_t count = 0; + for (const auto &cmd : this->tx_buffer_) { + if (cmd.state == FrameState::READY) + count++; + } + return count; + } + // A never-null placeholder to return when a lookup fails, so a tripped EXPECT/ADD_FAILURE reports + // the assertion instead of dereferencing null / an empty deque and segfaulting the whole suite. + static const ModbusDeviceCommand &dummy_command() { + static const uint8_t DUMMY_PDU[1] = {0x00}; + static ModbusDeviceCommand cmd(nullptr, 0, std::span(DUMMY_PDU, 1)); + return cmd; + } + const ModbusDeviceCommand &queued(size_t i) const { + for (const auto &cmd : this->tx_buffer_) { + if (cmd.state == FrameState::READY && i-- == 0) + return cmd; + } + ADD_FAILURE() << "no READY entry at that index"; + return dummy_command(); + } + size_t entries() const { return this->tx_buffer_.size(); } + const ModbusDeviceCommand *next_ready() { return this->select_next_ready_(); } + bool waiting() const { return this->waiting_for_response_; } + const ModbusDeviceCommand &waiting_command() { + ModbusDeviceCommand *cmd = this->find_waiting_(); + EXPECT_NE(cmd, nullptr); + return cmd != nullptr ? *cmd : dummy_command(); } - 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(); + void sweep_for_test() { this->sweep_(); } + void send_next_for_test() { + this->send_next_frame_(); + this->sweep_(); // a transmit failure's on_not_sent() is delivered by the loop's sweep } - // Drives the real unexpected-frame branch in process_modbus_server_frame(). + void force_send_next() { + ModbusDeviceCommand *cmd = this->select_next_ready_(); + ASSERT_NE(cmd, nullptr) << "no READY entry to send"; + cmd->state = FrameState::WAITING; + this->waiting_for_response_ = true; + } + // Drives the real response/interruption branches, followed by the loop's sweep. void receive_frame_for_test(uint8_t address, std::span pdu) { this->process_modbus_server_frame(address, pdu); + this->sweep_(); } void timeout_waiting() { - if (this->waiting_for_response_.has_value()) - this->notify_no_response_(*this->waiting_for_response_); - this->waiting_for_response_.reset(); + this->sweep_(); // deliver anything already owed (e.g. an interruption's on_no_response) + this->expire_waiting_(); + this->sweep_(); } }; @@ -87,7 +118,7 @@ TEST(ModbusClientHubNoResponse, RetryRequeuesWaitingFrame) { device.send_pdu(read_pdu()); ASSERT_EQ(hub.queued_frames(), 1u); - hub.force_send_front(); + hub.force_send_next(); ASSERT_EQ(hub.queued_frames(), 0u); ASSERT_TRUE(hub.waiting()); @@ -96,7 +127,7 @@ TEST(ModbusClientHubNoResponse, RetryRequeuesWaitingFrame) { EXPECT_EQ(device.no_response_count_, 1); EXPECT_FALSE(hub.waiting()); ASSERT_EQ(hub.queued_frames(), 1u); - const ModbusDeviceCommand &requeued = hub.front(); + const ModbusDeviceCommand &requeued = hub.queued(0); EXPECT_EQ(requeued.device, &device); // address + PDU + CRC ASSERT_EQ(requeued.frame.size(), sizeof(READ_PDU) + 3); @@ -111,7 +142,7 @@ TEST(ModbusClientHubNoResponse, NoRetryDropsWaitingFrame) { RetryingDevice device(&hub, 0x02, /*retry=*/false); device.send_pdu(read_pdu()); - hub.force_send_front(); + hub.force_send_next(); hub.timeout_waiting(); @@ -127,7 +158,7 @@ TEST(ModbusClientHubNoResponse, DetachedDeviceIsNotNotified) { { RetryingDevice device(&hub, 0x02, /*retry=*/true); device.send_pdu(read_pdu()); - hub.force_send_front(); + hub.force_send_next(); // device destructor clears its queue entries, including the waiting frame's device pointer } ASSERT_TRUE(hub.waiting()); @@ -139,32 +170,57 @@ TEST(ModbusClientHubNoResponse, DetachedDeviceIsNotNotified) { EXPECT_EQ(hub.queued_frames(), 0u); } -// An unexpected frame interrupts the transaction: the retry is re-queued immediately, but the -// waiting entry survives as an interrupted shell (device detached) that keeps tx blocked until the -// send-wait timeout clears it - without a second no-response callback or a duplicate requeue. +// An unexpected frame interrupts the transaction: the entry becomes an INTERRUPTED shell that +// ignores this transaction and blocks tx until the send-wait timeout, where it gets its single +// on_no_response() - a granted retry is requeued there, like any other timeout. TEST(ModbusClientHubNoResponse, RetryBehindInterruptedShell) { NoResponseProbeHub hub; RetryingDevice device(&hub, 0x02, /*retry=*/true); device.send_pdu(read_pdu()); - hub.force_send_front(); + hub.force_send_next(); // A frame from the wrong address (0x07, expected 0x02) hits the unexpected-frame branch. const uint8_t stray_pdu[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; hub.receive_frame_for_test(0x07, stray_pdu); + hub.sweep_for_test(); - EXPECT_EQ(device.no_response_count_, 1); - ASSERT_EQ(hub.queued_frames(), 1u); // exactly one requeue... - EXPECT_EQ(hub.front().device, &device); - ASSERT_TRUE(hub.waiting()); // ...while the shell stays in the waiting slot - EXPECT_TRUE(hub.waiting_command().interrupted); - EXPECT_EQ(hub.waiting_command().device, nullptr); + EXPECT_EQ(device.no_response_count_, 0); // not notified early: it waits out the timeout + ASSERT_TRUE(hub.waiting()); // and keeps blocking the bus + EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED); + EXPECT_EQ(hub.waiting_command().pending, 1u); - // The send-wait timeout clears the shell without a second callback or another requeue. + // The send-wait timeout delivers on_no_response and requeues the granted retry. hub.timeout_waiting(); EXPECT_FALSE(hub.waiting()); EXPECT_EQ(device.no_response_count_, 1); - EXPECT_EQ(hub.queued_frames(), 1u); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).device, &device); +} + +// The declined-retry interrupted shell blocks until the send-wait timeout, then gets its single +// on_no_response() there and retires with nothing left to send. +TEST(ModbusClientHubNoResponse, InterruptedShellDeclinedRetryRetiresOnRelease) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + + const uint8_t stray_pdu[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x07, stray_pdu); // wrong address: interrupts the transaction + hub.sweep_for_test(); + + EXPECT_EQ(device.no_response_count_, 0); // not notified early + ASSERT_TRUE(hub.waiting()); // the shell still blocks the wire + EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED); + EXPECT_EQ(hub.waiting_command().pending, 1u); + + hub.timeout_waiting(); // on_no_response (declined), then the shell retires + + EXPECT_FALSE(hub.waiting()); + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(device.no_response_count_, 1); } // A callback that detaches the device (clear_tx_queue_for_device()) wins over its own retry request: @@ -174,7 +230,7 @@ TEST(ModbusClientHubNoResponse, MidCallbackClearCancelsRetry) { ClearingRetryDevice device(&hub, 0x02); device.send_pdu(read_pdu()); - hub.force_send_front(); + hub.force_send_next(); hub.timeout_waiting(); EXPECT_EQ(device.no_response_count_, 1); @@ -182,6 +238,271 @@ TEST(ModbusClientHubNoResponse, MidCallbackClearCancelsRetry) { EXPECT_FALSE(hub.waiting()); } +// Writes jump ahead of queued reads; reads keep FIFO order among themselves. +TEST(ModbusClientHubPriority, WritesSendBeforeQueuedReads) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + const uint8_t read_a[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + const uint8_t read_b[] = {0x03, 0x02, 0x00, 0x00, 0x02}; + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(read_a); + device.send_pdu(read_b); + device.send_pdu(write_pdu); + + ASSERT_EQ(hub.queued_frames(), 3u); + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[0], 0x06); // the write transmits first + hub.timeout_waiting(); + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[1], 0x01); // reads follow in FIFO order + hub.timeout_waiting(); + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[1], 0x02); +} + +// Re-requesting a queued frame is absorbed into the existing entry instead of queueing a duplicate. +TEST(ModbusClientHubPriority, DuplicateQueuedFrameAbsorbedNotDuplicated) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.send_pdu(read_pdu()); + device.send_pdu(read_pdu()); + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 2u); // one entry standing for two accepted requests +} + +// Re-requesting the frame currently waiting is absorbed into the waiting entry; after a +// no-response timeout the absorbed request still gets its run even though the device declines a +// retry, and a second timeout does not run it again. +TEST(ModbusClientHubPriority, InFlightDuplicateRunsOnceMore) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + device.send_pdu(read_pdu()); // duplicate of the waiting frame + + EXPECT_EQ(hub.queued_frames(), 0u); // not queued twice + EXPECT_EQ(hub.waiting_command().pending, 2u); + + hub.timeout_waiting(); + ASSERT_EQ(hub.queued_frames(), 1u); // the timeout resolved one request; the absorbed one runs + EXPECT_EQ(hub.queued(0).pending, 1u); + + hub.force_send_next(); + hub.timeout_waiting(); + EXPECT_EQ(hub.queued_frames(), 0u); // the last request resolved; nothing left to run +} + +// An entry with an absorbed extra request that times out while the device asks to retry: the +// retry is not a resolution, so BOTH requests remain pending rather than one being dropped - +// which would leave that caller without a resolution. +TEST(ModbusClientHubPriority, AbsorbedRequestSurvivesDeviceRetry) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/true); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + device.send_pdu(read_pdu()); // duplicate of the waiting frame -> absorbed + ASSERT_EQ(hub.waiting_command().pending, 2u); + + hub.timeout_waiting(); // no response; the device requests a retry + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 2u); // preserved: the retry resolved nothing +} + +// A continuous read re-queues itself (at the lowest priority) after each successful response, +// but not after an exception response. +TEST(ModbusClientHubPriority, ContinuousReadRequeuesOnSuccessOnly) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_TRUE(hub.queued(0).continuous); + hub.force_send_next(); + + // A matching successful response cycles the continuous entry back to READY. + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_TRUE(hub.queued(0).continuous); + + // An exception response ends the poll. + hub.force_send_next(); + const uint8_t exception_response[] = {0x83, 0x02}; + hub.receive_frame_for_test(0x02, exception_response); + EXPECT_EQ(hub.queued_frames(), 0u); +} + +// A continuous read that gets no response and is retried stays continuous: an explicit retry of a +// continuous poll is assumed to still want continuous polling (the entry stays continuous). +TEST(ModbusClientHubPriority, RetriedContinuousReadStaysContinuous) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/true); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + ASSERT_TRUE(hub.queued(0).continuous); + hub.force_send_next(); + + hub.timeout_waiting(); // no response -> device requests retry + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_TRUE(hub.queued(0).continuous); // the retried poll stays continuous +} + +// A one-shot duplicate downgrades a continuous poll to a one-shot (the mirror of a continuous +// duplicate upgrading a one-shot): the entry runs one more cycle to serve the request, then stops. +TEST(ModbusClientHubPriority, DuplicateSendDowngradesContinuous) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + ASSERT_TRUE(hub.queued(0).continuous); + + device.read_holding_registers(0x100, 2); // one-shot duplicate downgrades the poll + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_FALSE(hub.queued(0).continuous); + EXPECT_EQ(hub.queued(0).pending, 1u); + + // It runs one more cycle to serve the request, then stops - not re-queued as a poll. + hub.force_send_next(); + EXPECT_FALSE(hub.waiting_command().continuous); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(hub.entries(), 0u); +} + +namespace { +// Re-sends its frame once as a one-shot from inside on_error(), to exercise the downgrade branch +// when the poll it duplicates has already reached a terminal (pending drained to 0). +class ResendOnErrorDevice : public ModbusClientDevice { + public: + ResendOnErrorDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_error(std::span request_pdu, ExceptionCode exception_code) override { + this->error_count_++; + if (this->resend_) { + this->resend_ = false; + this->read_holding_registers(0x100, 2); // one-shot re-send from inside the failure callback + } + } + int error_count_{0}; + bool resend_{true}; +}; +} // namespace + +// A one-shot re-send issued from inside a continuous poll's failure callback must still run. The +// poll's exception terminal has already drained pending to 0, so the re-send absorbs into that entry +// via the downgrade branch - which must restore the debt, or the sweep erases the entry with the +// request never sent and no callback delivered. +TEST(ModbusClientHubPriority, DowngradeAfterTerminalKeepsRequestAlive) { + NoResponseProbeHub hub; + ResendOnErrorDevice device(&hub, 0x02); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + ASSERT_TRUE(hub.queued(0).continuous); + + hub.force_send_next(); + const uint8_t exception_response[] = {0x83, 0x02}; + hub.receive_frame_for_test(0x02, exception_response); // exception ends the poll; on_error re-sends + + EXPECT_EQ(device.error_count_, 1); // one terminal delivered so far + ASSERT_EQ(hub.queued_frames(), 1u); // the re-send survived the sweep instead of being erased + EXPECT_FALSE(hub.queued(0).continuous); // downgraded to a one-shot + EXPECT_EQ(hub.queued(0).pending, 1u); // debt restored so the request runs + + // And it runs to its own terminal - a good response this time - then the entry is gone. + hub.force_send_next(); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(hub.entries(), 0u); +} + +// Requesting continuous polling for a frame that is already queued as a one-shot turns that entry +// into the continuous poll instead of leaving a promotion that never polls. +TEST(ModbusClientHubPriority, ContinuousRequestUpgradesQueuedDuplicate) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + device.read_holding_registers(0x100, 2); + ASSERT_EQ(hub.queued_frames(), 1u); + ASSERT_FALSE(hub.queued(0).continuous); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_TRUE(hub.queued(0).continuous); + + // And it behaves as a poll from here: success cycles it back to READY. + hub.force_send_next(); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_TRUE(hub.queued(0).continuous); +} + +// The transmit order is one key with three levels: writes, then one-shot reads, then continuous +// polls - a poll only gets the bus when nothing else wants it. +TEST(ModbusClientHubPriority, WritesThenOneShotReadsThenContinuousPolls) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + // Queued oldest-first in the opposite order to the one they must transmit in, so age cannot be + // what produces the expected sequence. + device.read_holding_registers(0x100, 2, {.continuous = true}); + const uint8_t one_shot[] = {0x03, 0x02, 0x00, 0x00, 0x01}; + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(one_shot); + device.send_pdu(write_pdu); + ASSERT_EQ(hub.queued_frames(), 3u); + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::CONTINUOUS); + EXPECT_EQ(hub.queued(1).priority(), CommandPriority::READ); + EXPECT_EQ(hub.queued(2).priority(), CommandPriority::WRITE); + + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[0], 0x06); // the write goes first + hub.timeout_waiting(); + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[1], 0x02); // then the one-shot read + hub.timeout_waiting(); + hub.force_send_next(); + EXPECT_TRUE(hub.waiting_command().continuous); // and the poll takes what is left +} + +// continuous is ignored for writes: the frame still sends at WRITE priority, once. +TEST(ModbusClientHubPriority, ContinuousIgnoredForWrites) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(write_pdu, {.continuous = true}); + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::WRITE); + EXPECT_FALSE(hub.queued(0).continuous); +} + +// A queued continuous poll does not count against immediate-send readiness: it ranks below every +// one-shot, so a new one-shot goes out ahead of it. A queued one-shot does count. +TEST(ModbusClientHubPriority, ContinuousPollDoesNotBlockImmediateSend) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + EXPECT_TRUE(hub.tx_buffer_empty()); // nothing queued + device.read_holding_registers(0x100, 2, {.continuous = true}); + ASSERT_TRUE(hub.queued(0).continuous); + EXPECT_TRUE(hub.tx_buffer_empty()); // a READY continuous poll still leaves room to send now + + device.read_holding_registers(0x200, 2); // a one-shot does count + EXPECT_FALSE(hub.tx_buffer_empty()); +} + // A device whose sent/not-sent callbacks are counted. namespace { class SentCountingDevice : public ModbusClientDevice { @@ -202,6 +523,148 @@ class SentCountingDevice : public ModbusClientDevice { }; } // namespace +// A write is never requeueable, so its entry can serve exactly one request: a duplicate of a +// queued write is refused at the door rather than earning the write an extra transmission. +TEST(ModbusClientHubPriority, DuplicateQueuedWriteRefused) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + EXPECT_TRUE(device.send_pdu(write_pdu)); + EXPECT_FALSE(device.send_pdu(write_pdu)); // duplicate write: refused + hub.sweep_for_test(); + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::WRITE); + EXPECT_EQ(hub.queued(0).pending, 1u); // a write's cap + EXPECT_EQ(device.not_sent_count_, 0); // refusals are returned, never delivered +} + +// Requeueability is an allow-list of the standard reads: a custom function code's idempotency is +// unknown, so its duplicate is refused like a write's instead of earning a silent re-send. +TEST(ModbusClientHubPriority, DuplicateCustomFunctionCodeRefused) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t custom_pdu[] = {0x41, 0x01, 0x02}; // user-defined function code + EXPECT_TRUE(device.send_pdu(custom_pdu)); + EXPECT_FALSE(device.send_pdu(custom_pdu)); // duplicate custom command: refused + hub.sweep_for_test(); + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 1u); // the non-requeueable cap of one run + EXPECT_EQ(device.not_sent_count_, 0); +} + +// An anonymous duplicate (no device - the YAML-lambda path) is always dropped, never promoted: +// with no callback there is no lifecycle to absorb into and no owner to route a re-run to. +TEST(ModbusClientHubPriority, AnonymousDuplicateDroppedNotPromoted) { + NoResponseProbeHub hub; + + const uint8_t read[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + hub.send_pdu(0x02, read); + hub.send_pdu(0x02, read); // anonymous duplicate: dropped + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 1u); // never absorbed for a null owner +} + +// A retried entry is re-stamped to the queue tail: reads that arrived while it was waiting get +// their turn before the retry, so a frame that keeps timing out cannot starve the rest of the bus. +TEST(ModbusClientHubPriority, RetriedReadGoesBehindFreshReads) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/true); + + device.send_pdu(read_pdu()); + hub.force_send_next(); // the frame that will time out and retry + device.send_pdu(read_pdu()); // waiting duplicate: absorbed into the waiting entry + const uint8_t fresh_a[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t fresh_b[] = {0x03, 0x00, 0x20, 0x00, 0x01}; + device.send_pdu(fresh_a); + device.send_pdu(fresh_b); + ASSERT_EQ(hub.queued_frames(), 2u); + + hub.timeout_waiting(); // device retries; the entry returns to READY behind the fresh reads + + ASSERT_EQ(hub.queued_frames(), 3u); + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->frame.pdu()[2], 0x10); // fresh reads keep FIFO order ahead of the retry + hub.force_send_next(); + hub.timeout_waiting(); + hub.force_send_next(); + EXPECT_EQ(hub.waiting_command().frame.pdu()[2], 0x20); + hub.timeout_waiting(); + hub.force_send_next(); // the retry gets its turn last, both requests still on the entry + EXPECT_TRUE(std::equal(hub.waiting_command().frame.pdu().begin(), hub.waiting_command().frame.pdu().end(), READ_PDU)); + EXPECT_EQ(hub.waiting_command().pending, 2u); +} + +// An absorbed duplicate does not move the entry back in line: seq belongs to the entry, and only +// re-entering the line (retry, resolved request, continuous cycle) re-stamps it. +TEST(ModbusClientHubPriority, AbsorbedDuplicateKeepsPlaceInLine) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/false); + + const uint8_t read_a[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t read_b[] = {0x03, 0x00, 0x20, 0x00, 0x01}; + device.send_pdu(read_a); + device.send_pdu(read_b); + device.send_pdu(read_a); // duplicate of the older entry: absorbed, place unchanged + ASSERT_EQ(hub.queued_frames(), 2u); + + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->frame.pdu()[2], 0x10); // read_a still transmits first + EXPECT_EQ(next->pending, 2u); +} + +// A write that is retried after a no-response keeps the WRITE class, so it stays ahead of reads, +// and a later duplicate still resolves against it instead of queueing twice. +TEST(ModbusClientHubPriority, RetriedWriteKeepsWritePriorityAndStaysNonRequeueable) { + NoResponseProbeHub hub; + RetryingDevice device(&hub, 0x02, /*retry=*/true); + + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(write_pdu); + hub.force_send_next(); + hub.timeout_waiting(); // no response -> device requests retry -> back to READY + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::WRITE); // retry preserves the WRITE class + + device.send_pdu(write_pdu); // duplicate of the retried write + ASSERT_EQ(hub.queued_frames(), 1u); // still not queued twice... + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::WRITE); + hub.sweep_for_test(); + EXPECT_EQ(hub.queued(0).pending, 1u); // ...the duplicate was refused at the door (write cap is 1) +} + +namespace { +// A hub that is never free to transmit. +class AlwaysBlockedHub : public NoResponseProbeHub { + public: + bool tx_blocked() override { return true; } +}; +} // namespace + +// Transmitting cannot fail, so a hub that is busy simply does not transmit: the frame keeps its +// place in the queue and goes out on a later loop, with no callback and no lifecycle change. (The +// caller owns the tx_blocked() check; send_frame_() has no gate of its own to refuse at.) +TEST(ModbusClientHubSent, BlockedHubDefersInsteadOfFailing) { + AlwaysBlockedHub hub; + SentCountingDevice device(&hub, 0x02); + + EXPECT_TRUE(device.send_pdu(read_pdu())); + hub.send_next_for_test(); + + EXPECT_EQ(device.sent_count_, 0); + EXPECT_EQ(device.not_sent_count_, 0); // nothing failed - it has not been attempted + ASSERT_EQ(hub.queued_frames(), 1u); // still queued, still owed exactly one terminal + EXPECT_EQ(hub.queued(0).pending, 1u); + EXPECT_FALSE(hub.waiting()); +} + // on_sent() fires when the frame goes onto the wire, not when it is queued. TEST(ModbusClientHubSent, FiresOnWireNotOnQueue) { NullUART uart; @@ -211,7 +674,7 @@ TEST(ModbusClientHubSent, FiresOnWireNotOnQueue) { SentCountingDevice device(&hub, 0x02); device.send_pdu(read_pdu()); - EXPECT_EQ(device.sent_count_, 0); // queued only - nothing on the wire yet + EXPECT_EQ(device.sent_count_, 0); // queued only - nothing sent yet hub.send_next_for_test(); EXPECT_EQ(device.sent_count_, 1); @@ -221,6 +684,34 @@ TEST(ModbusClientHubSent, FiresOnWireNotOnQueue) { EXPECT_TRUE(hub.waiting()); } +namespace { +// tx_blocked() clear for send_next_frame_'s gate, then blocked for send_frame_'s post-delay re-check. +class RejectPostDelayHub : public NoResponseProbeHub { + public: + bool tx_blocked() override { return this->tx_blocked_calls_++ > 0; } + int tx_blocked_calls_{0}; +}; +} // namespace + +// A byte arriving during send_frame_'s pre-send delay blocks transmission after the caller's gate +// already passed. send_frame_ rejects, and send_next_frame_ leaves the frame READY to retry - it is +// not marked WAITING and the bus is not claimed. +TEST(ModbusClientHubSent, SendRejectedAfterDelayLeavesFrameReady) { + NullUART uart; + RejectPostDelayHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + SentCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.send_next_for_test(); // gate passes, send_frame_ rejects on the post-delay re-check + + EXPECT_EQ(device.sent_count_, 0); // nothing transmitted + EXPECT_FALSE(hub.waiting()); // the frame was left untouched, bus not claimed + ASSERT_EQ(hub.entries(), 1u); + EXPECT_EQ(hub.queued(0).state, FrameState::READY); // still selectable next loop +} + // Counts response deliveries so requeue semantics can be pinned end to end. namespace { class DataCountingDevice : public ModbusClientDevice { @@ -260,7 +751,7 @@ class DataCountingDevice : public ModbusClientDevice { 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.force_send_next(); hub.receive_frame_for_test(0x02, response_pdu); cycles++; } @@ -284,6 +775,89 @@ TEST(ModbusClientHubCallbackCount, SingleReadSingleCallback) { EXPECT_FALSE(hub.waiting()); } +// Requesting the same read twice while queued yields exactly two callbacks: +// the promoted entry completes, re-queues once (demoted), completes again, and stops. +TEST(ModbusClientHubCallbackCount, DuplicateReadExactlyTwoCallbacks) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + device.send_pdu(read_pdu()); + int cycles = drain_with_responses(hub, OK_RESPONSE); + + EXPECT_EQ(cycles, 2); + EXPECT_EQ(device.data_count_, 2); + EXPECT_EQ(device.not_sent_count_, 0); // both requests were served + EXPECT_EQ(hub.queued_frames(), 0u); +} + +namespace { +// Clears the address queue from inside its first response callback, so a duplicate still owed on the +// same entry has to be resolved (or, as things stand, is dropped) by that clear. +class ClearOnFirstResponseDevice : public DataCountingDevice { + public: + ClearOnFirstResponseDevice(ModbusClientHub *hub, uint8_t address) : DataCountingDevice(hub, address) {} + void on_response(std::span request_pdu, std::span response_pdu) override { + this->data_count_++; + if (this->data_count_ == 1) + this->clear_tx_queue_for_address(); // clear mid-completion, from inside the first response + } +}; +} // namespace + +// A duplicate read absorbs into one entry (pending 2). The first response resolves one request, and +// its callback clears the address queue mid-completion. The still-owed duplicate is a second accepted +// request, so it must get its own terminal - on_not_sent() - not be dropped silently. +TEST(ModbusClientHubCallbackCount, ClearFromResponseResolvesDuplicateWithNotSent) { + NoResponseProbeHub hub; + ClearOnFirstResponseDevice device(&hub, 0x02); + + EXPECT_TRUE(device.send_pdu(read_pdu())); + EXPECT_TRUE(device.send_pdu(read_pdu())); // absorbed: one entry, pending 2 + hub.force_send_next(); + hub.receive_frame_for_test(0x02, OK_RESPONSE); // response -> on_response -> clear, then sweep + + EXPECT_EQ(device.data_count_, 1); // exactly one response delivered + EXPECT_EQ(device.not_sent_count_, 1); // the duplicate resolved with a terminal, not dropped + EXPECT_EQ(device.terminals(), 2); // one terminal per accepted request + EXPECT_EQ(hub.queued_frames(), 0u); +} + +// A read entry serves two requests (this run plus one re-run), so the third identical request is +// refused at the door: two data callbacks, and no terminal for the request that was never taken. +TEST(ModbusClientHubCallbackCount, TripleReadRefusesTheThird) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + EXPECT_TRUE(device.send_pdu(read_pdu())); + EXPECT_TRUE(device.send_pdu(read_pdu())); + EXPECT_FALSE(device.send_pdu(read_pdu())); // the entry is already at its cap + hub.sweep_for_test(); + EXPECT_EQ(device.not_sent_count_, 0); // refused synchronously, nothing owed + int cycles = drain_with_responses(hub, OK_RESPONSE); + + EXPECT_EQ(cycles, 2); + EXPECT_EQ(device.data_count_, 2); + EXPECT_EQ(device.terminals(), 2); // exactly one per accepted request + EXPECT_EQ(hub.queued_frames(), 0u); +} + +// A duplicate write is refused at the door and the original write sends once - the caller learns +// immediately, and no lifecycle is created for the request that was never taken. +TEST(ModbusClientHubCallbackCount, DuplicateWriteRefusedWithoutLifecycle) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + const uint8_t write_pdu[] = {0x06, 0x00, 0x10, 0xBE, 0xEF}; + EXPECT_TRUE(device.send_pdu(write_pdu)); + EXPECT_FALSE(device.send_pdu(write_pdu)); + hub.sweep_for_test(); + + EXPECT_EQ(device.terminals(), 0); // the accepted write has not resolved; the other never existed + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).priority(), CommandPriority::WRITE); +} + // An exception response is a terminal on its own: exactly one on_error(), no others, // preceded by exactly one on_sent(). TEST(ModbusClientHubCallbackCount, ErrorResponseIsSoleTerminal) { @@ -319,20 +893,22 @@ TEST(ModbusClientHubCallbackCount, NoResponseIsSoleTerminalAndNotSentHasNoSent) EXPECT_EQ(device.terminals(), 1); EXPECT_EQ(device.sent_count_, 1); - // A refused send (empty PDU) is a not_sent terminal, never sent. + // Unabsorbable duplicate: the second identical write is refused at the door - no lifecycle, no + // terminal, nothing 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 + EXPECT_TRUE(device.send_pdu(write_pdu)); + EXPECT_FALSE(device.send_pdu(write_pdu)); + hub.sweep_for_test(); + EXPECT_EQ(device.not_sent_count_, 0); + EXPECT_EQ(device.terminals(), 1); // still just the read's timeout + EXPECT_EQ(device.sent_count_, 1); - // Drain it: the write echo response is its data terminal, and the books balance. + // Drain the accepted write: its echo response is the 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 + EXPECT_EQ(device.terminals(), 2); // 2 accepted lifecycles, 2 terminals + EXPECT_EQ(device.sent_count_, 2); // 2 transmissions; the refused duplicate never sent } // A device-requested retry starts a new lifecycle: each transmission gets its own sent + terminal. @@ -359,9 +935,9 @@ TEST(ModbusClientHubCallbackCount, RetryLifecyclesEachGetSentAndTerminal) { 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) { +// A retry is a state flip on an existing entry, never a new insertion, so a full queue can't refuse +// it: fill the queue, time out the waiting frame with a retry, and it survives as READY. +TEST(ModbusClientHubCallbackCount, RetryIsNeverRefusedByFullQueue) { NullUART uart; NoResponseProbeHub hub; hub.set_uart_parent(&uart); @@ -371,50 +947,91 @@ TEST(ModbusClientHubCallbackCount, FullQueueRetryRefusalDeliversNotSentWithPdu) 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++) { + hub.force_send_next(); // waiting + device.send_pdu(read_pdu()); // absorbed: two requests pending + // Fill the remaining live capacity with distinct frames. + for (uint16_t i = 0; hub.entries() < 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 + hub.timeout_waiting(); // retry requested; the entry flips back to READY regardless of capacity 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); + EXPECT_EQ(device.not_sent_count_, 0); // nothing was refused + ASSERT_EQ(hub.queued_frames(), MODBUS_TX_BUFFER_SIZE); + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->device, &filler); // round-robin: the retry re-stamped behind the fillers + // The retried entry survives as READY with both absorbed requests intact. + bool found = false; + for (size_t i = 0; i < hub.queued_frames(); i++) { + const ModbusDeviceCommand &cmd = hub.queued(i); + if (cmd.device == &device) { + EXPECT_EQ(cmd.pending, 2u); + found = true; + } + } + EXPECT_TRUE(found); } -// 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. +// The deprecated device-side send_raw() reports an unusable payload the same way every other +// refused send does: false at the call site, with no queue entry and no callback. namespace { -class SendRawOnNotSentDevice : public ModbusClientDevice { +class NotSentCountingRawDevice : 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 - } + NotSentCountingRawDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_not_sent(std::span request_pdu) override { this->not_sent_count_++; } int not_sent_count_{0}; }; } // namespace -TEST(ModbusClientHubQueue, SendRawRefusalIsGuardedAgainstRecursion) { +TEST(ModbusClientHubQueue, SendRawTooShortIsRefusedAtTheDoor) { NoResponseProbeHub hub; - SendRawOnNotSentDevice device(&hub, 0x02); + NotSentCountingRawDevice 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 + EXPECT_FALSE(device.send_raw({})); // too short to contain a PDU #pragma GCC diagnostic pop - EXPECT_EQ(device.not_sent_count_, 1); + EXPECT_EQ(device.not_sent_count_, 0); // refusals are returned, never delivered + EXPECT_TRUE(hub.tx_buffer_empty()); +} + +// A continuous read: every wire transmission pairs one sent with one terminal, ending on the error. +TEST(ModbusClientHubCallbackCount, ContinuousLifecyclesBalance) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + const uint8_t exception_response[] = {0x83, 0x02}; + hub.send_next_for_test(); + hub.receive_frame_for_test(0x02, ok_response); // lifecycle 1 -> requeued + hub.send_next_for_test(); + hub.receive_frame_for_test(0x02, ok_response); // lifecycle 2 -> requeued + hub.send_next_for_test(); + hub.receive_frame_for_test(0x02, exception_response); // lifecycle 3 -> stops + + EXPECT_EQ(device.data_count_, 2); + EXPECT_EQ(device.error_count_, 1); + EXPECT_EQ(device.terminals(), 3); + EXPECT_EQ(device.sent_count_, 3); + EXPECT_EQ(hub.queued_frames(), 0u); } namespace { +// A device that stops itself (clears its own queue) from inside on_response(). +class ClearOnDataDevice : public ModbusClientDevice { + public: + ClearOnDataDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_response(std::span request_pdu, std::span response_pdu) override { + this->clear_tx_queue_for_device(); + } +}; + // A device that chains a follow-up send from inside on_sent(). class ChainOnSentDevice : public ModbusClientDevice { public: @@ -447,10 +1064,11 @@ TEST(ModbusClientHubQueue, ClearAddressQueueNotifiesEveryOwner) { bystander_other.send_pdu(read_c); ASSERT_EQ(hub.queued_frames(), 3u); - controller_like.clear_tx_queue_for_address(false); + controller_like.clear_tx_queue_for_address(); + hub.sweep_for_test(); // the loop's sweep delivers the owed terminals and erases the entries ASSERT_EQ(hub.queued_frames(), 1u); // only the other-address frame remains - EXPECT_EQ(hub.front().frame.address(), 0x03); + EXPECT_EQ(hub.queued(0).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); @@ -458,6 +1076,72 @@ TEST(ModbusClientHubQueue, ClearAddressQueueNotifiesEveryOwner) { EXPECT_EQ(bystander_same.last_not_sent_pdu_, std::vector(std::begin(read_b), std::end(read_b))); } +// A cleared entry resolves with one on_not_sent() per accepted request it stood for, so the +// books balance for owners counting outstanding requests - all within the one sweep. +TEST(ModbusClientHubQueue, ClearAddressDeliversOneTerminalPerAcceptedRequest) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x01, 0x00, 0x00, 0x02}; + device.send_pdu(read); + device.send_pdu(read); // duplicate: absorbed into the queued entry + ASSERT_EQ(hub.queued_frames(), 1u); + ASSERT_EQ(hub.queued(0).pending, 2u); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); + + EXPECT_EQ(device.not_sent_count_, 2); // one terminal per accepted request + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(hub.entries(), 0u); // fully drained and erased +} + +// A duplicate read absorbs into one waiting entry (pending 2 once the first is sent). A clear with +// clear_sent detaches the in-flight frame as a silent shell, but the duplicate - a second accepted +// request that would have re-run - was never transmitted, so it must still get its on_not_sent(). +TEST(ModbusClientHubQueue, ClearSentOnInFlightDuplicateStillNotifiesTheDuplicate) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + device.send_pdu(read_pdu()); // absorbed: one entry, pending 2 + ASSERT_EQ(hub.queued(0).pending, 2u); + hub.force_send_next(); // the frame is sent (WAITING); pending still 2 + ASSERT_TRUE(hub.waiting()); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); + + EXPECT_EQ(device.not_sent_count_, 1); // the un-transmitted duplicate is resolved, not dropped +} + +// A clear does not abandon the in-flight frame: it becomes a WAITING_RETIRED shell that keeps the +// bus and still delivers the in-flight request's usual callback (here on_response) when the reply +// arrives. Only un-run duplicates are turned into on_not_sent(); a lone in-flight frame has none. +TEST(ModbusClientHubQueue, ClearWhileInFlightStillDeliversTheResponse) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); // sent, now WAITING + ASSERT_TRUE(hub.waiting()); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); + + EXPECT_EQ(device.not_sent_count_, 0); // no un-run duplicate to resolve + ASSERT_TRUE(hub.waiting()); // still waiting for a response, holding the bus + ASSERT_EQ(hub.entries(), 1u); // entry preserved as a cleared shell + EXPECT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED); + + // the in-flight request still gets its usual callback when the response finally arrives + hub.receive_frame_for_test(0x02, OK_RESPONSE); + EXPECT_EQ(device.data_count_, 1); + EXPECT_EQ(device.not_sent_count_, 0); + EXPECT_FALSE(hub.waiting()); + EXPECT_EQ(hub.entries(), 0u); +} + namespace { // Re-sends its frame once from inside on_not_sent - the re-queued frame must survive the sweep. class ResendOnNotSentDevice : public ModbusClientDevice { @@ -466,7 +1150,7 @@ class ResendOnNotSentDevice : public ModbusClientDevice { 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}; + const uint8_t again[] = {0x06, 0x00, 0x40, 0x00, 0x01}; // a write: ranked first at selection, not by position this->send_pdu(again); } } @@ -474,8 +1158,8 @@ class ResendOnNotSentDevice : public ModbusClientDevice { }; } // 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. +// A handler that re-sends to the same address from inside on_not_sent() neither corrupts the sweep +// nor loops it: the fresh entry starts within its cap, so the sweep never touches it. TEST(ModbusClientHubQueue, ClearAddressReentrantResendSurvives) { NoResponseProbeHub hub; ResendOnNotSentDevice device(&hub, 0x02); @@ -484,19 +1168,47 @@ TEST(ModbusClientHubQueue, ClearAddressReentrantResendSurvives) { device.send_pdu(read); ASSERT_EQ(hub.queued_frames(), 1u); - hub.clear_tx_queue_for_address(0x02, false); + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); // 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); + EXPECT_EQ(hub.queued(0).frame.address(), 0x02); +} + +// The hard case for the sweep: the notified handler re-queues a WRITE to the cleared address. The +// fresh entry must be neither dropped nor re-notified - and the bystander's frame at the other +// address survives untouched, while the write still wins transmit selection. +TEST(ModbusClientHubQueue, ClearAddressReentrantResendNotSwept) { + NoResponseProbeHub hub; + ResendOnNotSentDevice resender(&hub, 0x02); + SentCountingDevice bystander_other(&hub, 0x03); + + const uint8_t read_victim[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t read_other[] = {0x03, 0x00, 0x20, 0x00, 0x01}; + resender.send_pdu(read_victim); + bystander_other.send_pdu(read_other); + ASSERT_EQ(hub.queued_frames(), 2u); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); + + EXPECT_EQ(resender.not_sent_count_, 1); // notified once, never re-notified for the re-send + ASSERT_EQ(hub.queued_frames(), 2u); // the re-queued write AND the other-address read survive + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->frame.address(), 0x02); // the WRITE class wins selection over the older read + EXPECT_EQ(next->frame.pdu()[0], 0x06); } namespace { -// Retries from EVERY on_not_sent - against a full queue this recursed without bound before the guard. -class AlwaysRetryDevice : public ModbusClientDevice { +// Re-sends its own frame from EVERY on_not_sent. There is no serve/absorb treadmill: a duplicate at +// the servable cap is refused at the door, and a re-send issued while the entry is retiring queues a +// fresh entry beyond the sweep's captured work_set (served next sweep), never re-absorbing the one draining. +class AlwaysResendDevice : public ModbusClientDevice { public: - AlwaysRetryDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + AlwaysResendDevice(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}; @@ -512,20 +1224,40 @@ class ClearOtherOnNotSentDevice : public ModbusClientDevice { 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); + this->parent_->clear_tx_queue_for_address(0x03); } 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) { +// pending can never exceed what the entry can serve, so the old serve/absorb treadmill is +// impossible by construction: the surplus request is refused at the door instead of being absorbed +// and resolved later, and a handler that re-sends gets false rather than another lifecycle. +TEST(ModbusClientHubQueue, PendingNeverExceedsTheServableCap) { + NoResponseProbeHub hub; + AlwaysResendDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x00, 0x50, 0x00, 0x01}; + EXPECT_TRUE(device.send_pdu(read)); + EXPECT_TRUE(device.send_pdu(read)); + EXPECT_FALSE(device.send_pdu(read)); // at the cap: refused + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 2u); + + hub.sweep_for_test(); // nothing is owed, so the handler never runs + + EXPECT_EQ(device.not_sent_count_, 0); + EXPECT_EQ(hub.queued(0).pending, 2u); +} + +// A full queue refuses at the door: false at the call site, no entry, no callback - so the +// refusal cannot re-enter the hub at all and needs no recursion bound of its own. +TEST(ModbusClientHubQueue, FullQueueRefusesWithoutCallbacks) { NoResponseProbeHub hub; SentCountingDevice filler(&hub, 0x05); - AlwaysRetryDevice retrier(&hub, 0x02); + SentCountingDevice device(&hub, 0x02); - // Fill the queue with distinct frames. + // Fill the queue with distinct frames (distinct start addresses keep the dedup from absorbing them). 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); @@ -533,76 +1265,12 @@ TEST(ModbusClientHubQueue, FullQueueRetryFromNotSentDoesNotRecurse) { 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_FALSE(device.send_pdu(read)); // refused synchronously + hub.sweep_for_test(); - EXPECT_EQ(retrier.not_sent_count_, 1); + EXPECT_EQ(device.not_sent_count_, 0); // nothing was accepted, so nothing is owed 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); + EXPECT_EQ(hub.entries(), MODBUS_TX_BUFFER_SIZE); // and no refusal bookkeeping was stored } namespace { @@ -613,16 +1281,16 @@ class ClearOwnAddressOnNotSentDevice : public ModbusClientDevice { 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); + this->clear_tx_queue_for_address(); } 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) { +// An address clear issued from inside on_not_sent() resolves EVERY dropped request with its own +// terminal at the sweep - including the clearer's (the sweep delivers from a quiescent hub, so the +// old stack-nesting silence no longer applies; use clear_tx_queue_for_device() for silent teardown). +TEST(ModbusClientHubQueue, SelfClearFromNotSentResolvesEveryRequest) { NoResponseProbeHub hub; ClearOwnAddressOnNotSentDevice clearer(&hub, 0x02); SentCountingDevice bystander(&hub, 0x02); @@ -635,15 +1303,19 @@ TEST(ModbusClientHubQueue, SelfClearFromNotSentSilentForClearerNotifiesOthers) { 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_FALSE(clearer.send_pdu(std::span{})); // empty: refused, no callback + clearer.clear_tx_queue_for_address(); // the clear the handler used to make - 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 + hub.sweep_for_test(); + + EXPECT_EQ(clearer.not_sent_count_, 2); // one per cleared request of its own + EXPECT_EQ(bystander.not_sent_count_, 1); // the bystander's cleared frame is notified too EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_EQ(hub.entries(), 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). +// A clear issued from inside on_not_sent() still delivers its victims' notifications in the same sweep: +// the newly-retired entries set sweep_needed_ and the sweep's restart loop drains them before it ends. TEST(ModbusClientHubQueue, NestedClearFromNotSentStillNotifiesVictims) { NoResponseProbeHub hub; ClearOtherOnNotSentDevice clearer(&hub, 0x02); @@ -655,55 +1327,90 @@ TEST(ModbusClientHubQueue, NestedClearFromNotSentStillNotifiesVictims) { 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 + hub.clear_tx_queue_for_address(0x02); // clearer's on_not_sent clears address 0x03 in turn + hub.sweep_for_test(); EXPECT_EQ(clearer.not_sent_count_, 1); - EXPECT_EQ(victim.not_sent_count_, 1); // delivered despite arriving from a nested sweep + EXPECT_EQ(victim.not_sent_count_, 1); // the nested clear's victim resolves in the same 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 { +// From on_not_sent (delivered by the sweep), re-sends a frame identical to ANOTHER doomed queued +// frame; the dedup must not absorb into the doomed entry. +class ResendSecondFrameDevice : public ModbusClientDevice { 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) {} + ResendSecondFrameDevice(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); + if (this->not_sent_count_ == 1) { + const uint8_t same_as_r2[] = {0x03, 0x00, 0x22, 0x00, 0x01}; + this->send_pdu(same_as_r2); + } } 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); +// A send during a sweep that matches a DELETED (doomed) frame must queue fresh, not absorb into +// the doomed entry - absorption would tie the new request to a frame the sweep is draining. +TEST(ModbusClientHubQueue, SweepDedupSkipsDeletedFrames) { + NoResponseProbeHub hub; + ResendSecondFrameDevice device(&hub, 0x02); - const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t r1[] = {0x03, 0x00, 0x21, 0x00, 0x01}; + const uint8_t r2[] = {0x03, 0x00, 0x22, 0x00, 0x01}; + device.send_pdu(r1); + device.send_pdu(r2); + ASSERT_EQ(hub.queued_frames(), 2u); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); // r1 and r2 both resolve; r1's handler re-sends a frame identical to r2 + // Without the dedup's dead-state skip the re-send would be absorbed into r2 and drained with it; + // with the skip it queues fresh and survives. + + EXPECT_EQ(device.not_sent_count_, 2); // r1 and r2 both resolved + ASSERT_EQ(hub.queued_frames(), 1u); // the re-send survives + EXPECT_EQ(hub.queued(0).frame.pdu()[2], 0x22); +} + +namespace { +// The worst-case handler: from every on_not_sent() it both re-sends and clears its own address, so +// each delivery manufactures a fresh entry AND a fresh terminal debt. +class ResendAndClearOnNotSentDevice : public ModbusClientDevice { + public: + ResendAndClearOnNotSentDevice(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, 0x70, 0x00, 0x01}; + this->send_pdu(again); + this->clear_tx_queue_for_address(); + } + int not_sent_count_{0}; +}; +} // namespace + +// Sweep-termination worst case: a handler re-sending AND clearing from every on_not_sent() still +// can't extend the sweep, since it serves only the entries it started with (new debt waits). +TEST(ModbusClientHubQueue, ResendAndClearFromNotSentCannotExtendTheSweep) { + NoResponseProbeHub hub; + ResendAndClearOnNotSentDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x00, 0x70, 0x00, 0x01}; device.send_pdu(read); - ASSERT_EQ(hub.queued_frames(), 1u); + hub.clear_tx_queue_for_address(0x02); - hub.send_next_for_test(); // tx_blocked gate passes, send_frame_ refuses -> failure path + hub.sweep_for_test(); + EXPECT_EQ(device.not_sent_count_, 1); // exactly the one terminal that was owed on entry + EXPECT_EQ(hub.entries(), 1u); // the frame the handler queued (and then cleared itself) - 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 + hub.sweep_for_test(); + EXPECT_EQ(device.not_sent_count_, 2); // its terminal comes on the next loop, not this sweep + EXPECT_EQ(hub.entries(), 1u); // and the container is still not growing + + hub.sweep_for_test(); + EXPECT_EQ(device.not_sent_count_, 3); + EXPECT_EQ(hub.entries(), 1u); } // clear_tx_queue_for_device() drops queued frames SILENTLY - no terminal callback (the documented @@ -724,8 +1431,8 @@ TEST(ModbusClientHubQueue, ClearDeviceQueueDropsSilently) { 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. +// A send_pdu() from inside on_sent() enqueues behind the waiting frame rather than sending +// immediately or corrupting the waiting transaction. TEST(ModbusClientHubSent, ReentrantSendFromOnSentQueues) { NullUART uart; NoResponseProbeHub hub; @@ -734,13 +1441,53 @@ TEST(ModbusClientHubSent, ReentrantSendFromOnSentQueues) { 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 + hub.send_next_for_test(); // first frame is sent -> on_sent chains a follow-up - EXPECT_TRUE(hub.waiting()); // first frame is in flight + EXPECT_TRUE(hub.waiting()); // first frame is waiting 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) } +// "Stop polling now" from inside on_response() works: the completing command is exposed to the +// clear routines, which detach it, cancelling the pending continuous re-queue. +TEST(ModbusClientHubPriority, ClearDeviceDuringDataCancelsContinuousRequeue) { + NoResponseProbeHub hub; + ClearOnDataDevice device(&hub, 0x02); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + hub.force_send_next(); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + + // "Stop polling now" from inside on_response() works: the completing command is detached, so the + // continuous re-queue is cancelled. + EXPECT_EQ(hub.queued_frames(), 0u); +} + +namespace { +// A device that stops polling for its address (clear by address) from inside on_response(). +class ClearAddressOnDataDevice : public ModbusClientDevice { + public: + ClearAddressOnDataDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_response(std::span request_pdu, std::span response_pdu) override { + this->clear_tx_queue_for_address(); + } +}; +} // namespace + +// The address-scoped clear cancels the mid-completion re-queue the same way the device-scoped one does. +TEST(ModbusClientHubPriority, ClearAddressDuringDataCancelsContinuousRequeue) { + NoResponseProbeHub hub; + ClearAddressOnDataDevice device(&hub, 0x02); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + hub.force_send_next(); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + + EXPECT_EQ(hub.queued_frames(), 0u); +} + 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. @@ -766,23 +1513,31 @@ TEST(ModbusClientHubCompat, LegacyCallbackNamesStillForward) { const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; device.send_pdu(read); - hub.force_send_front(); + hub.force_send_next(); hub.timeout_waiting(); // no reply -> on_no_response -> forwards to on_modbus_no_response EXPECT_EQ(device.legacy_no_response_, 1); - device.send_pdu(std::span()); // empty PDU refused -> on_not_sent -> forwards + // A refused send returns false with no callback, so exercise the forward through an accepted + // request instead: a cleared queue entry delivers on_not_sent(), which forwards to the old name. + EXPECT_FALSE(device.send_pdu(std::span())); // empty PDU: refused at the door + EXPECT_EQ(device.legacy_not_sent_, 0); + const uint8_t queued[] = {0x03, 0x00, 0x11, 0x00, 0x01}; + EXPECT_TRUE(device.send_pdu(queued)); + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); EXPECT_EQ(device.legacy_not_sent_, 1); } // The send_pdu() capacity bound: a PDU larger than MAX_PDU_SIZE would build a frame past the RTU -// 256-byte limit, so it is refused up front and signalled like any other failed send. -TEST(ModbusClientHub, OversizedPduIsRefusedWithNotSent) { +// 256-byte limit, so it is refused up front - false at the call site, no entry, no callback. +TEST(ModbusClientHub, OversizedPduIsRefusedAtTheDoor) { NoResponseProbeHub hub; LegacyNameDevice device(&hub, 0x02); std::vector big(MAX_PDU_SIZE + 1, 0x41); - device.send_pdu(big); - EXPECT_EQ(device.legacy_not_sent_, 1); // on_not_sent, observed via the legacy forward + EXPECT_FALSE(device.send_pdu(big)); + EXPECT_EQ(device.legacy_not_sent_, 0); // refusals are returned, never delivered EXPECT_TRUE(hub.tx_buffer_empty()); + EXPECT_EQ(hub.entries(), 0u); } // --- ModbusDevice compatibility shim ------------------------------------------------------------ @@ -814,7 +1569,7 @@ TEST(ModbusDeviceShim, LegacyCallbacksReceiveTheOldShapes) { // the byte-count byte, as an owning vector. const uint8_t read_req[] = {0x03, 0x00, 0x10, 0x00, 0x02}; device.send_pdu(read_req); - hub.force_send_front(); + hub.force_send_next(); const uint8_t response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; hub.receive_frame_for_test(0x02, response); const std::vector expected{0x00, 0x2A, 0x01, 0x00}; @@ -823,14 +1578,14 @@ TEST(ModbusDeviceShim, LegacyCallbacksReceiveTheOldShapes) { // Write echo: no byte-count byte, so the payload is everything after the function code. const uint8_t write_req[] = {0x06, 0x00, 0x10, 0x00, 0x2A}; device.send_pdu(write_req); - hub.force_send_front(); + hub.force_send_next(); hub.receive_frame_for_test(0x02, write_req); // single-write responses echo the request const std::vector expected_echo{0x00, 0x10, 0x00, 0x2A}; EXPECT_EQ(device.last_data_, expected_echo); // Exception response: on_modbus_error() received the masked function code and the exception code. device.send_pdu(read_req); - hub.force_send_front(); + hub.force_send_next(); const uint8_t error[] = {0x83, 0x02}; hub.receive_frame_for_test(0x02, error); EXPECT_EQ(device.last_error_fc_, 0x03); @@ -845,9 +1600,9 @@ TEST(ModbusTypedSendHelpers, HelpersQueueExpectedPdus) { ModbusClientDevice device(&hub, 0x02); auto check = [&](const std::vector &expected) { ASSERT_EQ(hub.queued_frames(), 1u); - auto pdu = hub.front().frame.pdu(); + auto pdu = hub.queued(0).frame.pdu(); EXPECT_EQ(std::vector(pdu.begin(), pdu.end()), expected); - hub.force_send_front(); + hub.force_send_next(); hub.timeout_waiting(); // default on_no_response() declines the retry, dropping the frame }; @@ -882,21 +1637,21 @@ TEST(ModbusTypedSendHelpers, ReadEntitiesDispatchesByTypeAndRejectsInvalid) { device.read_entities(EntityType::HOLDING, 0x0001, 1); ASSERT_EQ(hub.queued_frames(), 1u); - EXPECT_EQ(hub.front().frame.pdu()[0], 0x03); - hub.force_send_front(); + EXPECT_EQ(hub.queued(0).frame.pdu()[0], 0x03); + hub.force_send_next(); hub.timeout_waiting(); device.read_entities(EntityType::DISCRETE_INPUT, 0x0001, 1); ASSERT_EQ(hub.queued_frames(), 1u); - EXPECT_EQ(hub.front().frame.pdu()[0], 0x02); - hub.force_send_front(); + EXPECT_EQ(hub.queued(0).frame.pdu()[0], 0x02); + hub.force_send_next(); hub.timeout_waiting(); device.read_entities(EntityType::CUSTOM, 0x0001, 1); // no read function: logged and not queued EXPECT_EQ(hub.queued_frames(), 0u); } -// A rejected read_entities() signals on_not_sent() like every other refused send. +// A rejected read_entities() returns false like every other refused send. namespace { class NotSentCountingDevice : public ModbusClientDevice { public: @@ -906,12 +1661,306 @@ class NotSentCountingDevice : public ModbusClientDevice { }; } // namespace -TEST(ModbusTypedSendHelpers, InvalidReadEntitiesSignalsNotSent) { +TEST(ModbusTypedSendHelpers, InvalidReadEntitiesIsRefusedAtTheDoor) { NoResponseProbeHub hub; NotSentCountingDevice device(&hub, 0x02); - device.read_entities(EntityType::CUSTOM, 0x0001, 1); - EXPECT_EQ(device.not_sent_, 1); + EXPECT_FALSE(device.read_entities(EntityType::CUSTOM, 0x0001, 1)); + EXPECT_EQ(device.not_sent_, 0); // refused sends report through the return value EXPECT_EQ(hub.queued_frames(), 0u); } +namespace { +// Re-sends its own frame from inside on_response() - matching the command mid-completion. +class ResendOnDataDevice : public ModbusClientDevice { + public: + ResendOnDataDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + void on_response(std::span request_pdu, std::span response_pdu) override { + this->send_pdu(std::vector(request_pdu.begin(), request_pdu.end())); + } + void send_pdu(const std::vector &pdu) { ModbusClientDevice::send_pdu(pdu); } +}; +} // namespace + +// A send from inside on_response() that matches the RECEIVED (completing) entry is absorbed into it, +// never a fresh twin. Here it is a one-shot re-send of a continuous poll, so it also downgrades the +// poll to a one-shot: one entry on the queue afterwards, now non-continuous. +TEST(ModbusClientHubPriority, ResendFromOnResponseAbsorbsIntoCompletingCommand) { + NoResponseProbeHub hub; + ResendOnDataDevice device(&hub, 0x02); + + device.read_holding_registers(0x100, 2, {.continuous = true}); + hub.force_send_next(); + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); // handler re-sends the identical frame mid-completion + + ASSERT_EQ(hub.queued_frames(), 1u); // absorbed into the same entry, not a fresh twin + EXPECT_FALSE(hub.queued(0).continuous); // the one-shot re-send downgraded the poll +} + +// An exception-flagged function code is never silently re-sendable, even though the read check +// masks the exception bit: its duplicate takes the drop path like any other non-read. +TEST(ModbusClientHubPriority, ExceptionFlaggedDuplicateDroppedNotPromoted) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t weird[] = {0x83, 0x01, 0x00, 0x00, 0x02}; // read-shaped but exception-flagged + EXPECT_TRUE(device.send_pdu(weird)); + EXPECT_FALSE(device.send_pdu(weird)); // non-requeueable: cap of one, so the duplicate is refused + hub.sweep_for_test(); + + ASSERT_EQ(hub.queued_frames(), 1u); + EXPECT_EQ(hub.queued(0).pending, 1u); + EXPECT_EQ(device.not_sent_count_, 0); + + // The write-shaped twin (0x86 masks to WRITE_SINGLE_REGISTER) must not take WRITE-class + // ordering either: exception-flagged codes are excluded from the mutates classification. + const uint8_t weird_write[] = {0x86, 0x00, 0x10, 0xBE, 0xEF}; + device.send_pdu(weird_write); + ASSERT_EQ(hub.queued_frames(), 2u); + EXPECT_EQ(hub.queued(1).priority(), CommandPriority::READ); // not WRITE + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->frame.pdu()[0], 0x83); // FIFO by age: it did not jump the older entry +} + +namespace { +// From inside the sweep's on_not_sent, re-sends the frame that is currently WAITING. +class ResendInFlightOnNotSentDevice : public ModbusClientDevice { + public: + ResendInFlightOnNotSentDevice(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 same_as_waiting[] = {0x03, 0x01, 0x00, 0x00, 0x02}; // == READ_PDU + this->send_pdu(same_as_waiting); + } + } + int not_sent_count_{0}; +}; +} // namespace + +// A clear turns the waiting entry into a WAITING_RETIRED shell. The shell keeps its device (so the +// in-flight request still gets its callback), but the dedup skips it, so a sweep handler re-sending +// that frame queues fresh instead of being absorbed into the cleared shell and drained as on_not_sent. +TEST(ModbusClientHubQueue, SweepResendAfterClearQueuesFreshNotAbsorbedIntoShell) { + NoResponseProbeHub hub; + ResendInFlightOnNotSentDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); // READ_PDU now waiting + const uint8_t queued_read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + device.send_pdu(queued_read); // a queued frame for the sweep to notify + ASSERT_EQ(hub.queued_frames(), 1u); + + hub.clear_tx_queue_for_address(0x02); + hub.sweep_for_test(); + + EXPECT_EQ(device.not_sent_count_, 1); // only the cleared queued frame, not the re-send + ASSERT_EQ(hub.queued_frames(), 1u); // the handler's re-send queued fresh... + EXPECT_EQ(hub.queued(0).pending, 1u); // ...not absorbed into the cleared shell + EXPECT_TRUE(std::equal(hub.queued(0).frame.pdu().begin(), hub.queued(0).frame.pdu().end(), READ_PDU)); + EXPECT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED); // in-flight one still awaiting a reply +} + +namespace { +// Gives up after a timeout by clearing its address from inside on_no_response() - the natural +// "device is dead, drop my traffic" pattern, and the reentrant case the address clear must handle. +class ClearAddressOnNoResponseDevice : public ModbusClientDevice { + public: + ClearAddressOnNoResponseDevice(ModbusClientHub *hub, uint8_t address) : ModbusClientDevice(hub, address) {} + bool on_no_response(std::span request_pdu) override { + this->no_response_count_++; + this->clear_tx_queue_for_address(); + return false; // gave up + } + void on_not_sent(std::span request_pdu) override { this->not_sent_count_++; } + int terminals() const { return this->no_response_count_ + this->not_sent_count_; } + int no_response_count_{0}; + int not_sent_count_{0}; +}; + +} // namespace + +// A clear issued from inside on_no_response() must not cause the request to be resolved twice: +// that callback already was its terminal, so the entry it hijacks owes nothing more. +TEST(ModbusClientHubNoResponse, SelfClearFromNoResponseDoesNotDoubleResolve) { + NoResponseProbeHub hub; + ClearAddressOnNoResponseDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + hub.timeout_waiting(); + + EXPECT_EQ(device.no_response_count_, 1); + EXPECT_EQ(device.terminals(), 1); // exactly one terminal for the one accepted request + EXPECT_EQ(hub.entries(), 0u); +} + +// The same entry standing for two accepted requests: the timeout resolves one, and the clear that +// cancels the re-run must resolve exactly the other. +TEST(ModbusClientHubNoResponse, SelfClearFromNoResponseResolvesTheAbsorbedRequestOnce) { + NoResponseProbeHub hub; + ClearAddressOnNoResponseDevice device(&hub, 0x02); + + EXPECT_TRUE(device.send_pdu(read_pdu())); + EXPECT_TRUE(device.send_pdu(read_pdu())); // absorbed: one entry, two requests + hub.force_send_next(); + hub.timeout_waiting(); + + EXPECT_EQ(device.no_response_count_, 1); + EXPECT_EQ(device.terminals(), 2); // one per accepted request, no more + EXPECT_EQ(hub.entries(), 0u); +} + +// A cleared in-flight frame must release the bus by both exits and still deliver the in-flight +// request's usual callback (on_response here, on_no_response on timeout); no on_not_sent, no duplicate. +TEST(ModbusClientHubQueue, ClearedShellReleasesTheBusOnLateResponse) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + hub.clear_tx_queue_for_address(0x02); + ASSERT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED); + + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); // the late reply for the cleared frame + + EXPECT_FALSE(hub.waiting()); // the bus is free again + EXPECT_EQ(hub.entries(), 0u); // the shell is gone + EXPECT_EQ(device.data_count_, 1); // the in-flight request still got its response callback + EXPECT_EQ(device.not_sent_count_, 0); // no un-run duplicate +} + +TEST(ModbusClientHubQueue, ClearedShellReleasesTheBusOnTimeout) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + hub.clear_tx_queue_for_address(0x02); + ASSERT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED); + + hub.timeout_waiting(); // no reply ever arrives; the watchdog releases the shell + + EXPECT_FALSE(hub.waiting()); + EXPECT_EQ(hub.entries(), 0u); + EXPECT_EQ(device.no_response_count_, 1); // the in-flight request got its on_no_response + EXPECT_EQ(device.not_sent_count_, 0); // no un-run duplicate +} + +// Clearing an interrupted (not-yet-notified) frame keeps its distrust: it becomes an +// INTERRUPTED_RETIRED shell that still ends in on_no_response at the timeout - never delivering a +// late response as on_response. No duplicate here, so no on_not_sent. +TEST(ModbusClientHubQueue, ClearInterruptedFrameGetsNoResponseAtTimeout) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); // declines the retry (retries_ == 0) + + device.send_pdu(read_pdu()); + hub.force_send_next(); + const uint8_t stray_pdu[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x07, stray_pdu); // wrong address: interrupts the transaction + hub.sweep_for_test(); + ASSERT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED); + + hub.clear_tx_queue_for_address(0x02); + ASSERT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED_RETIRED); + + // A late MATCHING response is ignored (distrust survives the clear), not delivered as on_response. + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); + EXPECT_EQ(device.data_count_, 0); + ASSERT_TRUE(hub.waiting()); // still held; the ignored response did not free the wire + + hub.timeout_waiting(); + + EXPECT_EQ(device.no_response_count_, 1); // the interrupted request's usual terminal, at the timeout + EXPECT_EQ(device.not_sent_count_, 0); // no un-run duplicate + EXPECT_EQ(hub.queued_frames(), 0u); + EXPECT_FALSE(hub.waiting()); + EXPECT_EQ(hub.entries(), 0u); +} + +// The other order: clear a WAITING frame, THEN an unexpected frame arrives. The distrust must still +// take hold - the cleared shell becomes INTERRUPTED_RETIRED and a later matching frame is ignored. +TEST(ModbusClientHubQueue, InterruptAfterClearStillDistrusts) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + hub.clear_tx_queue_for_address(0x02); + ASSERT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED); + + const uint8_t stray_pdu[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x07, stray_pdu); // unexpected frame interrupts the cleared shell + ASSERT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED_RETIRED); + + const uint8_t ok_response[] = {0x03, 0x04, 0x00, 0x2A, 0x01, 0x00}; + hub.receive_frame_for_test(0x02, ok_response); // now-distrusted late response is ignored + EXPECT_EQ(device.data_count_, 0); + + hub.timeout_waiting(); + EXPECT_EQ(device.no_response_count_, 1); + EXPECT_FALSE(hub.waiting()); + EXPECT_EQ(hub.entries(), 0u); +} + +// A cleared waiting duplicate (pending 2) that times out: the duplicate drains to on_not_sent and +// the in-flight request gets on_no_response, with nothing re-transmitted (sweep runs before timeout). +TEST(ModbusClientHubQueue, ClearedInFlightDuplicateTimesOutWithoutRerunning) { + NoResponseProbeHub hub; + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + device.send_pdu(read_pdu()); // absorbed: one entry, pending 2 + ASSERT_EQ(hub.queued(0).pending, 2u); + hub.force_send_next(); // sent, pending still 2 + hub.clear_tx_queue_for_address(0x02); + + hub.timeout_waiting(); + + EXPECT_EQ(device.not_sent_count_, 1); // the un-run duplicate + EXPECT_EQ(device.no_response_count_, 1); // the in-flight request's usual terminal + EXPECT_EQ(hub.queued_frames(), 0u); // nothing re-transmitted + EXPECT_EQ(hub.entries(), 0u); // fully drained and erased + EXPECT_FALSE(hub.waiting()); +} + +// An absorbed extra request also gets its run after an error response - the re-request was +// explicit, so it runs once more whether this attempt succeeded or not. +TEST(ModbusClientHubCallbackCount, AbsorbedRequestRunsAfterErrorResponse) { + NullUART uart; + NoResponseProbeHub hub; + hub.set_uart_parent(&uart); + hub.setup(); + DataCountingDevice device(&hub, 0x02); + + device.send_pdu(read_pdu()); + hub.force_send_next(); + device.send_pdu(read_pdu()); // waiting duplicate: absorbed + const uint8_t exception_response[] = {0x83, 0x02}; + hub.receive_frame_for_test(0x02, exception_response); // error terminal for request 1 + + EXPECT_EQ(device.error_count_, 1); + ASSERT_EQ(hub.queued_frames(), 1u); // request 2's run still queued + EXPECT_EQ(hub.queued(0).pending, 1u); +} + +// Read-modify-write function codes mutate registers, so they rank as WRITE for transmit ordering. +TEST(ModbusClientHubPriority, ReadModifyWritesRankAsWrites) { + NoResponseProbeHub hub; + SentCountingDevice device(&hub, 0x02); + + const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x01}; + const uint8_t mask_write[] = {0x16, 0x00, 0x10, 0x00, 0xFF, 0x00, 0x01}; + device.send_pdu(read); + device.send_pdu(mask_write); + + ASSERT_EQ(hub.queued_frames(), 2u); + const ModbusDeviceCommand *next = hub.next_ready(); + ASSERT_NE(next, nullptr); + EXPECT_EQ(next->priority(), CommandPriority::WRITE); // 0x16 wins selection over the queued read + EXPECT_EQ(next->frame.pdu()[0], 0x16); +} } // namespace esphome::modbus::testing