Repository navigation
fix(modbus): serialize gateway access to prevent slave ID mismatch - #6
Merged
Merged
Conversation
- Update pymodbus from 3.9.2 to 3.13.0 - Fix breaking API: slave= -> device_id= parameter
- Change gateway key from (host,port,slave_id) to (host,port) - shared connection per gateway - Add per-gateway locks to serialize access across devices sharing same host:port - Add _validate_gateway_config() to warn on shared gateways at startup - Update poller to group targets by gateway and poll sequentially - Poll different gateways in parallel for performance
📝 WalkthroughWalkthroughEnforces shared Modbus gateway semantics (per-(host,port) connection and serialization), updates Modbus calls to use Changes
Sequence DiagramsequenceDiagram
participant Poller as Polling Service
participant Mgr as ModbusClientManager
participant GW as ModbusGateway\r\n(host, port)
participant Dev as Device\r\n(slave_id)
Poller->>Poller: Group targets by (host, port)
Poller->>Poller: Spawn one task per gateway group
Poller->>Mgr: poll targets for gateway (host, port)
Mgr->>GW: get_gateway_and_lock((host, port))
GW->>GW: acquire lock (serialize)
Mgr->>Dev: read_registers(device_id=slave_id)
Dev-->>Mgr: response
Mgr->>GW: release lock
Mgr-->>Poller: return result
note over Poller,Mgr: Other gateway tasks run in parallel
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
- Change slave= to device_id= in test assertions
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
app/core/modbus_client.py (3)
647-667:⚠️ Potential issue | 🟠 MajorHandle devices that move gateways during config reloads.
reload_configs()only resets gateways for removeddevice_ids. If a device keeps the samedevice_idbut changeshostorport,removedis empty and the old gateway/lock entry is left behind. That leaks stale connections and makesget_gateways_status()report gateways that no longer belong to any active config.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 647 - 667, reload_configs currently only resets gateways for removed device_ids, missing devices that kept the same device_id but changed host/port; detect changed-host devices by comparing the old config objects in self._configs to the incoming new_configs (e.g., compare host and port on each cfg), call await self.reset_gateway(device_id) for each device_id whose host/port changed (catch DeviceNotFoundError as you already do), then update self._configs and call self._validate_gateway_config(); reference reload_configs, reset_gateway, _configs, _validate_gateway_config and get_gateways_status when making the change.
225-317:⚠️ Potential issue | 🔴 CriticalDon't return a failed validation frame after the last retry.
last_responseis assigned before_is_valid_response(), so Line 317 can still return a slave-ID mismatch or other invalid frame after retries are exhausted.read_registers()only checksisError()/payload shape, so the wrong device's data can be accepted, cached, and published. The same pattern exists inwrite_holding_register().Proposed fix
try: response = read_method( address=address, count=count, device_id=slave_id ) - last_response = response if self._is_valid_response(response, op_name, slave_id): + last_response = response if attempt > 0: logger.info( "modbus_read_success_after_retry", operation=op_name, slave_id=slave_id, @@ - else: + else: + last_response = None # Invalid response (slave_id mismatch, error response, etc.) # Close connection to flush buffer and prevent stale responses logger.debug( "modbus_invalid_response_flush", @@ - return last_response + return None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 225 - 317, The code assigns last_response before validating and may return an invalid frame after retries; update the logic in the read loop (the block using read_method, last_response, and _is_valid_response) so last_response is only set when _is_valid_response(response, op_name, slave_id) returns True (or otherwise clear last_response on invalid), and ensure after all retries the function returns None (or an explicit failure) instead of returning a stale/invalid last_response; apply the same fix pattern to write_holding_register where the same last_response / validation-before-assign pattern exists.
420-433:⚠️ Potential issue | 🟠 MajorValidate transport settings for shared gateways, not just shared host/port.
The shared gateway is created from whichever
DeviceConfighits_create_gateway()first. If two devices on the samehost:portdisagree ontimeout,framer,max_retries, orretry_delay, one of them silently runs with the other's transport settings.frameris especially risky here because a mismatch can make the whole shared connection unusable.Also applies to: 435-492
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 420 - 433, The current gateway reuse logic must validate transport settings beyond host:port to avoid silent mismatch; when retrieving or creating gateways (the code paths that call _create_gateway and the shared gateway lookup), compare the incoming DeviceConfig.transport fields (timeout, framer, max_retries, retry_delay) against the existing ModbusGateway's settings and if any differ, fail fast (raise/return an error or explicit exception) or log and refuse to reuse the gateway; leave _create_gateway to only construct new ModbusGateway instances when no gateway exists for the host:port and ensure the gateway-lookup code (where host:port keys are used) performs this comparison using the unique symbols DeviceConfig, _create_gateway, and ModbusGateway.
🧹 Nitpick comments (1)
app/services/poller.py (1)
427-433: Consider per-gateway timing metrics while this rolls out.The new bottleneck is gateway-local contention, but this log/metric surface still only exposes cycle-wide totals. Recording per-gateway duration or largest group size here would make hot gateways obvious before they start missing intervals.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/services/poller.py` around lines 427 - 433, Add per-gateway timing and sizing to the polling_cycle_completed log: compute per-gateway durations (e.g., a map of gateway id/name to its cycle time or last-start-to-end delta) and the largest gateway group size (max(len(group) for group in gateway_groups)), then include them as additional fields in the logger.debug call (e.g., per_gateway_durations and max_gateway_group_size) alongside existing fields like success_count, failure_count, total_targets, gateway_count, and duration_seconds so hot gateways are visible.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@app/core/modbus_client.py`:
- Around line 402-406: The shared-gateway design is being broken by calls to
ModbusGateway.close() inside ModbusGateway._read_registers() and
write_holding_register(), which tear down the single (host, port) socket for all
slaves; change the error handling so that protocol/frame-level errors for a
specific slave do NOT call self.close() on the gateway—only close the gateway on
transport-level failures (socket/connection exceptions). Update
ModbusGateway._read_registers and write_holding_register to distinguish
ModbusResponse/CRC/frame errors (raise or return a per-request error without
closing the gateway) from actual connection errors (where calling self.close()
is appropriate), and ensure caller code that uses self._gateways and self._locks
treats per-request failures as isolated to that slave rather than triggering
gateway removal.
- Around line 633-638: reset_gateway() currently only uses _manager_lock when
deleting entries, which allows a racing task holding the old per-gateway lock to
overlap with a new gateway creation; fix by coordinating with the per-gateway
lock: lookup and acquire the existing per-gateway lock
(self._locks[gateway_key]) before closing/deleting the gateway, then while
holding that per-gateway lock also hold _manager_lock to remove entries (call
await lock.acquire() on the existing lock, await
asyncio.to_thread(gateway.close), then acquire _manager_lock and del
self._gateways[gateway_key] and del self._locks[gateway_key]), and finally
release the per-gateway lock; ensure you handle the case where no per-gateway
lock exists (skip) and avoid deadlock by always acquiring the per-gateway lock
before _manager_lock in this code path.
---
Outside diff comments:
In `@app/core/modbus_client.py`:
- Around line 647-667: reload_configs currently only resets gateways for removed
device_ids, missing devices that kept the same device_id but changed host/port;
detect changed-host devices by comparing the old config objects in self._configs
to the incoming new_configs (e.g., compare host and port on each cfg), call
await self.reset_gateway(device_id) for each device_id whose host/port changed
(catch DeviceNotFoundError as you already do), then update self._configs and
call self._validate_gateway_config(); reference reload_configs, reset_gateway,
_configs, _validate_gateway_config and get_gateways_status when making the
change.
- Around line 225-317: The code assigns last_response before validating and may
return an invalid frame after retries; update the logic in the read loop (the
block using read_method, last_response, and _is_valid_response) so last_response
is only set when _is_valid_response(response, op_name, slave_id) returns True
(or otherwise clear last_response on invalid), and ensure after all retries the
function returns None (or an explicit failure) instead of returning a
stale/invalid last_response; apply the same fix pattern to
write_holding_register where the same last_response / validation-before-assign
pattern exists.
- Around line 420-433: The current gateway reuse logic must validate transport
settings beyond host:port to avoid silent mismatch; when retrieving or creating
gateways (the code paths that call _create_gateway and the shared gateway
lookup), compare the incoming DeviceConfig.transport fields (timeout, framer,
max_retries, retry_delay) against the existing ModbusGateway's settings and if
any differ, fail fast (raise/return an error or explicit exception) or log and
refuse to reuse the gateway; leave _create_gateway to only construct new
ModbusGateway instances when no gateway exists for the host:port and ensure the
gateway-lookup code (where host:port keys are used) performs this comparison
using the unique symbols DeviceConfig, _create_gateway, and ModbusGateway.
---
Nitpick comments:
In `@app/services/poller.py`:
- Around line 427-433: Add per-gateway timing and sizing to the
polling_cycle_completed log: compute per-gateway durations (e.g., a map of
gateway id/name to its cycle time or last-start-to-end delta) and the largest
gateway group size (max(len(group) for group in gateway_groups)), then include
them as additional fields in the logger.debug call (e.g., per_gateway_durations
and max_gateway_group_size) alongside existing fields like success_count,
failure_count, total_targets, gateway_count, and duration_seconds so hot
gateways are visible.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d197acad-1f57-48a7-9527-67951c7a5a4a
📒 Files selected for processing (3)
app/core/modbus_client.pyapp/services/poller.pyrequirements.txt
Comment on lines
+402
to
406
| # IMPORTANT: One gateway per host:port, NOT per slave_id | ||
| # All devices sharing the same host:port share ONE connection | ||
| self._gateways: Dict[Tuple[str, int], ModbusGateway] = {} | ||
| # Map (host, port) -> Lock | ||
| # Map (host, port) -> Lock - ensures serialized access per gateway | ||
| self._locks: Dict[Tuple[str, int], asyncio.Lock] = {} |
There was a problem hiding this comment.
Avoid tearing down the shared socket for one device's Modbus-level failure.
Once (host, port) is the shared key, the existing self.close() paths in ModbusGateway._read_registers() / write_holding_register() become global to every slave behind that gateway. A bad frame from one device now drops the shared connection for all queued peers, which reintroduces the connection churn this PR is trying to remove.
Also applies to: 498-505
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/core/modbus_client.py` around lines 402 - 406, The shared-gateway design
is being broken by calls to ModbusGateway.close() inside
ModbusGateway._read_registers() and write_holding_register(), which tear down
the single (host, port) socket for all slaves; change the error handling so that
protocol/frame-level errors for a specific slave do NOT call self.close() on the
gateway—only close the gateway on transport-level failures (socket/connection
exceptions). Update ModbusGateway._read_registers and write_holding_register to
distinguish ModbusResponse/CRC/frame errors (raise or return a per-request error
without closing the gateway) from actual connection errors (where calling
self.close() is appropriate), and ensure caller code that uses self._gateways
and self._locks treats per-request failures as isolated to that slave rather
than triggering gateway removal.
Comment on lines
633
to
+638
| async with self._manager_lock: | ||
| if key in self._gateways: | ||
| gateway = self._gateways[key] | ||
| if gateway_key in self._gateways: | ||
| gateway = self._gateways[gateway_key] | ||
| await asyncio.to_thread(gateway.close) | ||
| del self._gateways[key] | ||
| del self._locks[key] | ||
| del self._gateways[gateway_key] | ||
| del self._locks[gateway_key] |
There was a problem hiding this comment.
Coordinate reset_gateway() with the existing per-gateway lock.
Deleting _gateways[gateway_key] and _locks[gateway_key] under _manager_lock alone is not enough. A task that already captured the old (gateway, lock) can keep running while the next task creates a fresh pair for the same host:port, so two operations can overlap during a reset/reload window and defeat the serialization guarantee.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/core/modbus_client.py` around lines 633 - 638, reset_gateway() currently
only uses _manager_lock when deleting entries, which allows a racing task
holding the old per-gateway lock to overlap with a new gateway creation; fix by
coordinating with the per-gateway lock: lookup and acquire the existing
per-gateway lock (self._locks[gateway_key]) before closing/deleting the gateway,
then while holding that per-gateway lock also hold _manager_lock to remove
entries (call await lock.acquire() on the existing lock, await
asyncio.to_thread(gateway.close), then acquire _manager_lock and del
self._gateways[gateway_key] and del self._locks[gateway_key]), and finally
release the per-gateway lock; ensure you handle the case where no per-gateway
lock exists (skip) and avoid deadlock by always acquiring the per-gateway lock
before _manager_lock in this code path.
- reload_configs(): detect when device changes host/port - Close old gateway when device moves to different endpoint - Prevents stale TCP connections and incorrect gateway status
- _read_registers(): only assign last_response after validation succeeds - write_holding_register(): same fix for write operations - Return None instead of potentially invalid last_response - tests: add slave_id to mock responses for validation
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
app/core/modbus_client.py (1)
661-663:⚠️ Potential issue | 🟠 MajorMaterialize
new_configsbefore multiple passes.
reload_configs()treatsnew_configsas reusable, but it is typed asIterable. If a generator is passed, later passes can be empty and wipe config state unexpectedly.Proposed fix
async def reload_configs(self, new_configs: Iterable[DeviceConfig]) -> None: + new_configs = list(new_configs) """Reload device configurations dynamically. ... old_device_ids = set(self._configs.keys()) new_device_ids = {cfg.device_id for cfg in new_configs}Also applies to: 672-693, 695-699
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 661 - 663, In reload_configs(), new_configs is consumed multiple times but is typed as Iterable and could be a one-time generator; materialize it immediately (e.g., assign list(new_configs)) so subsequent uses (the comprehension that builds new_device_ids, the loops that iterate to add/update configs, and the later passes referenced around the sections handling removed/updated devices) operate on the same concrete sequence; update references to use the newly created list variable (new_configs_list or similar) in all places where new_configs is iterated to prevent accidental state wipes.
♻️ Duplicate comments (2)
app/core/modbus_client.py (2)
240-257:⚠️ Potential issue | 🟠 MajorKeep the shared gateway alive on per-request Modbus failures.
self.close()on invalid/Modbus-level responses (Line 250, Line 366) still tears down the shared(host, port)connection for all devices. That reintroduces gateway churn the PR is trying to eliminate. Limit connection close to transport/socket failures only.Proposed direction
- except (ModbusException, ModbusIOException, OSError) as exc: + except (ModbusIOException, OSError) as exc: last_exception = exc ... self.close() if attempt < num_attempts - 1: self.ensure_connection() self._apply_temp_timeout(timeout) + except ModbusException as exc: + last_exception = exc + logger.warning( + "modbus_read_exception", + ... + message="Protocol-level Modbus error, retrying without closing shared gateway", + ) + # keep shared gateway open; retry request only- self.close() + # keep shared gateway open for protocol-level failures + # (close only on transport/socket errors)Also applies to: 257-273, 351-367
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 240 - 257, The invalid-response branch should not call self.close() because that tears down the shared (host, port) gateway; remove the self.close() call inside the "Invalid response detected" path and instead perform the brief sleep, call self.ensure_connection(), and re-apply the temp timeout (use the existing ensure_connection() and _apply_temp_timeout(timeout) calls), leaving connection teardown to the transport/socket error handling; modify the exception handler around the try/except that catches (ModbusException, ModbusIOException, OSError) so that self.close() is only invoked for transport-level failures (ModbusIOException or OSError) and not for ModbusException/application-level errors.
641-646:⚠️ Potential issue | 🔴 CriticalCoordinate
reset_gateway()with the existing per-gateway lock.Current reset removes gateway/lock under
_manager_lockonly. A task already holding the old lock can overlap with a newly created gateway for the same endpoint, breaking serialization guarantees.Proposed fix
- async with self._manager_lock: - if gateway_key in self._gateways: - gateway = self._gateways[gateway_key] - await asyncio.to_thread(gateway.close) - del self._gateways[gateway_key] - del self._locks[gateway_key] + async with self._manager_lock: + gateway = self._gateways.get(gateway_key) + lock = self._locks.get(gateway_key) + + if gateway is None: + return + + if lock is not None: + async with lock: + await asyncio.to_thread(gateway.close) + async with self._manager_lock: + if self._gateways.get(gateway_key) is gateway: + self._gateways.pop(gateway_key, None) + self._locks.pop(gateway_key, None) + else: + await asyncio.to_thread(gateway.close) + async with self._manager_lock: + self._gateways.pop(gateway_key, None) + self._locks.pop(gateway_key, None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/modbus_client.py` around lines 641 - 646, reset_gateway currently deletes the gateway and its per-gateway lock while only holding _manager_lock, allowing a racing task to acquire the old lock -> operate on a new gateway; fix by coordinating with the per-gateway lock: inside reset_gateway, while holding self._manager_lock fetch the current per-gateway lock (old_lock) and replace it with a new lock placeholder in self._locks[gateway_key] to block new acquirers, then release _manager_lock, acquire old_lock (await old_lock.acquire()) and perform gateway.close() in that critical section (use asyncio.to_thread for close if necessary), then re-acquire self._manager_lock to delete self._gateways[gateway_key] and remove the placeholder lock; reference symbols: reset_gateway, self._manager_lock, self._gateways, self._locks, gateway.close.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@app/core/modbus_client.py`:
- Around line 661-663: In reload_configs(), new_configs is consumed multiple
times but is typed as Iterable and could be a one-time generator; materialize it
immediately (e.g., assign list(new_configs)) so subsequent uses (the
comprehension that builds new_device_ids, the loops that iterate to add/update
configs, and the later passes referenced around the sections handling
removed/updated devices) operate on the same concrete sequence; update
references to use the newly created list variable (new_configs_list or similar)
in all places where new_configs is iterated to prevent accidental state wipes.
---
Duplicate comments:
In `@app/core/modbus_client.py`:
- Around line 240-257: The invalid-response branch should not call self.close()
because that tears down the shared (host, port) gateway; remove the self.close()
call inside the "Invalid response detected" path and instead perform the brief
sleep, call self.ensure_connection(), and re-apply the temp timeout (use the
existing ensure_connection() and _apply_temp_timeout(timeout) calls), leaving
connection teardown to the transport/socket error handling; modify the exception
handler around the try/except that catches (ModbusException, ModbusIOException,
OSError) so that self.close() is only invoked for transport-level failures
(ModbusIOException or OSError) and not for ModbusException/application-level
errors.
- Around line 641-646: reset_gateway currently deletes the gateway and its
per-gateway lock while only holding _manager_lock, allowing a racing task to
acquire the old lock -> operate on a new gateway; fix by coordinating with the
per-gateway lock: inside reset_gateway, while holding self._manager_lock fetch
the current per-gateway lock (old_lock) and replace it with a new lock
placeholder in self._locks[gateway_key] to block new acquirers, then release
_manager_lock, acquire old_lock (await old_lock.acquire()) and perform
gateway.close() in that critical section (use asyncio.to_thread for close if
necessary), then re-acquire self._manager_lock to delete
self._gateways[gateway_key] and remove the placeholder lock; reference symbols:
reset_gateway, self._manager_lock, self._gateways, self._locks, gateway.close.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 89304620-101b-4320-a58c-4b303b318311
📒 Files selected for processing (2)
app/core/modbus_client.pytests/unit/test_modbus_client.py
5 tasks done
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Related issue
Summary
(host,port,slave_id)to(host,port)- shared connection per gateway_validate_gateway_config()to warn about shared gateways at startupScope
Testing
Test result
Reviewer focus
app/core/modbus_client.py,app/services/poller.pyNotes
Summary by CodeRabbit
New Features
Bug Fixes
Chores
Tests