Repository navigation
feat(poller): add failure isolation for offline devices - #7
Jonathan0823 wants to merge 3 commits into
Conversation
- Add POLLING_TIMEOUT_SECONDS (2.0s) - faster than API but not too aggressive - Add POLLING_RETRIES (1) - at least 1 for reliability - Add POLLING_RETRY_DELAY (0.1s) - delay between retries - Add POLLING_TARGET_COOLDOWN (60s) - skip offline device after threshold failures
- Use configurable timeout/retry from settings instead of hardcoded values - Reset gateway after failure before polling next device - Add pacing delay between sequential reads on same gateway - Enhanced logging: device_id, slave_id, host_port, timeout, retries - Circuit breaker logging includes config values
- Test: one offline device doesn't block others - Test: gateway reset after timeout - Test: circuit breaker skips offline device - Test: polling uses configurable settings
📝 WalkthroughWalkthroughConfiguration settings for polling behavior (timeout, retries, delays, cooldown) are added to the Settings model. The poller service is refactored to use these configurations instead of hardcoded values and includes enhanced error handling with device metadata logging and gateway reset on failures. Changes
Sequence DiagramsequenceDiagram
participant Client as Caller
participant Poller as Poller Service
participant Config as Settings
participant Manager as Device Manager
participant Modbus as Modbus/Connection
participant CB as Circuit Breaker
Client->>Poller: poll_gateway_group(targets)
loop For each target with delay
Poller->>Config: Read POLLING_TIMEOUT_SECONDS<br/>POLLING_RETRIES
Config-->>Poller: timeout, retries
Poller->>Poller: _poll_single_target(device_id)
alt Success Path
Poller->>Manager: get_config(device_id)
Manager-->>Poller: device metadata
Poller->>Modbus: read(timeout, retries)
Modbus-->>Poller: Success
Poller->>Poller: Log with metadata
else Connection Failure
Poller->>Modbus: read(timeout, retries)
Modbus-->>Poller: Timeout/Error
Poller->>Manager: get_config(device_id)
Manager-->>Poller: device metadata
Poller->>Poller: Log failure with metadata
Poller->>Manager: reset_gateway(device_id)
Manager-->>Poller: Reset Complete
else Circuit Breaker Open
Poller->>CB: Check state
CB-->>Poller: OPEN
Poller->>Poller: Log CircuitOpenError
end
Poller->>Config: Read POLLING_RETRY_DELAY
Config-->>Poller: delay_seconds
Poller->>Poller: asyncio.sleep(delay)
end
Poller-->>Client: Results
Estimated Code Review Effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
app/services/poller.py (2)
429-433: ReusingPOLLING_RETRY_DELAYfor inter-read pacing conflates two knobs.This sleep is a pacing delay between different targets on the same gateway — unrelated to retry backoff inside
manager.read_registers. Binding them to one setting means tuning retry latency (e.g., lowering to0.01sfor faster retries) unintentionally tightens inter-read pacing, which can actually stress the gateway.Consider introducing a dedicated setting (e.g.,
POLLING_INTER_READ_DELAY) defaulting to the same0.1sso operators can tune them independently.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/services/poller.py` around lines 429 - 433, The inter-read sleep currently uses settings.POLLING_RETRY_DELAY (used for retry backoff) which couples retry tuning to gateway pacing; create a new setting settings.POLLING_INTER_READ_DELAY (default 0.1s) and replace the sleep call in the poll loop that references group_targets/i with await asyncio.sleep(settings.POLLING_INTER_READ_DELAY); also add the new constant to the settings/config with the default value so operators can tune retry backoff (used by manager.read_registers) and inter-read pacing independently.
322-323: Operational: cycle duration can exceedPOLL_INTERVAL_SECONDSunder failures.With defaults
POLLING_TIMEOUT_SECONDS=2.0,POLLING_RETRIES=1, a failing device costs ~4s (2 attempts × 2s) plus a gateway-reset round-trip plus pacing.POLL_INTERVAL_SECONDSdefaults to 5s, so even two offline devices on the same gateway will push a single gateway group past the interval;await asyncio.sleep(interval_seconds)at the bottom then means the next cycle starts immediately with no idle time, and effective interval drifts.Consider either:
- computing
sleep = max(0, interval_seconds - cycle_duration)so the cadence is stable, or- emitting a warning metric/log when
cycle_duration > interval_secondsso operators can detect sustained overruns.Also applies to: 492-492
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/services/poller.py` around lines 322 - 323, The loop currently does await asyncio.sleep(interval_seconds) which lets failures lengthen the effective cadence; record cycle start with time.monotonic() at the top of the polling loop (before work is done), compute cycle_duration = time.monotonic() - start after the work, then replace the sleep with await asyncio.sleep(max(0, interval_seconds - cycle_duration)); additionally emit a warning/log/metric (using the existing logger or metrics helper) when cycle_duration > interval_seconds so operators can detect sustained overruns; update uses of interval_seconds and asyncio.sleep in the function containing that sleep to implement these changes.tests/unit/test_poller.py (1)
346-355: Test is tautological.This only asserts that attributes exist on the Pydantic
Settingsmodel with reasonable bounds — essentially testing Pydantic itself. It doesn't verify the poller actually passes these values tomanager.read_registers, which is the real contract the PR changes introduce.Consider replacing/augmenting with a test that patches
settings.POLLING_TIMEOUT_SECONDS/POLLING_RETRIES, invokes_poll_single_target, and assertsread_registerswas called with matchingtimeout=andretries=kwargs.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_poller.py` around lines 346 - 355, The current test only checks that the Settings model has attributes; instead modify this test to patch settings.POLLING_TIMEOUT_SECONDS and settings.POLLING_RETRIES to specific values, mock or monkeypatch the manager.read_registers method, call/await the poller function _poll_single_target (the function under test), and assert that manager.read_registers was invoked with timeout=<patched timeout> and retries=<patched retries> as kwargs; use pytest-mock or monkeypatch to set the settings and to spy/assert the call on manager.read_registers so the test verifies the poller actually passes the configured values.
🤖 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/config.py`:
- Line 38: POLLING_TARGET_COOLDOWN is an unused config constant in
app/core/config.py; remove the declaration of POLLING_TARGET_COOLDOWN and any
related comments/docs, and verify there are no remaining references (search for
POLLING_TARGET_COOLDOWN) to avoid breakage; if you prefer to keep the feature
instead, wire this constant into the device polling logic (where offline-device
skip/thresholds are handled—e.g., the polling/retry code path that currently
tracks failures) so the cooldown value is actually used, but do one or the other
(remove the unused constant or implement its use) so no dead config is shipped.
In `@app/services/poller.py`:
- Around line 241-257: The current handler unconditionally calls
manager.reset_gateway (ModbusClientManager.reset_gateway) when catching
ModbusClientError, which resets the entire (host,port) gateway shared by devices
polled in poll_gateway_group and therefore punishes healthy devices; change the
exception handling so only transport/connection failures trigger await
manager.reset_gateway(device_id) — e.g., catch only ConnectionError,
ConnectionResetError, OSError (or inspect the wrapped exception inside
ModbusClientError and reset only if it is one of those connection types), and
for other ModbusClientError cases (protocol-level ModbusException/bad register)
just log a warning without resetting the gateway.
In `@tests/unit/test_poller.py`:
- Around line 224-274: The test currently only polls the online device; update
it to exercise isolation by making mock_manager.read_registers.side_effect raise
an exception when called for the offline device's slave_id (slave_id==2) and
return a successful list for the online slave_id, then call _poll_single_target
first with a target for the offline device (assert success is False and error is
non-empty) and then call _poll_single_target for the online target (assert
success is True and error == ""), and finally assert mock_manager.reset_gateway
was invoked for the failing/offline call (e.g., called once) while the online
call succeeds despite the prior failure; reference _poll_single_target,
DeviceConfig, mock_manager.read_registers, and mock_manager.reset_gateway when
making the changes.
---
Nitpick comments:
In `@app/services/poller.py`:
- Around line 429-433: The inter-read sleep currently uses
settings.POLLING_RETRY_DELAY (used for retry backoff) which couples retry tuning
to gateway pacing; create a new setting settings.POLLING_INTER_READ_DELAY
(default 0.1s) and replace the sleep call in the poll loop that references
group_targets/i with await asyncio.sleep(settings.POLLING_INTER_READ_DELAY);
also add the new constant to the settings/config with the default value so
operators can tune retry backoff (used by manager.read_registers) and inter-read
pacing independently.
- Around line 322-323: The loop currently does await
asyncio.sleep(interval_seconds) which lets failures lengthen the effective
cadence; record cycle start with time.monotonic() at the top of the polling loop
(before work is done), compute cycle_duration = time.monotonic() - start after
the work, then replace the sleep with await asyncio.sleep(max(0,
interval_seconds - cycle_duration)); additionally emit a warning/log/metric
(using the existing logger or metrics helper) when cycle_duration >
interval_seconds so operators can detect sustained overruns; update uses of
interval_seconds and asyncio.sleep in the function containing that sleep to
implement these changes.
In `@tests/unit/test_poller.py`:
- Around line 346-355: The current test only checks that the Settings model has
attributes; instead modify this test to patch settings.POLLING_TIMEOUT_SECONDS
and settings.POLLING_RETRIES to specific values, mock or monkeypatch the
manager.read_registers method, call/await the poller function
_poll_single_target (the function under test), and assert that
manager.read_registers was invoked with timeout=<patched timeout> and
retries=<patched retries> as kwargs; use pytest-mock or monkeypatch to set the
settings and to spy/assert the call on manager.read_registers so the test
verifies the poller actually passes the configured values.
🪄 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: 8f31de31-4d91-4fb4-9a25-2c58cb3f2b2f
📒 Files selected for processing (3)
app/core/config.pyapp/services/poller.pytests/unit/test_poller.py
| POLLING_TIMEOUT_SECONDS: float = 2.0 # Timeout for polling reads (faster than API's 5s but not too aggressive) | ||
| POLLING_RETRIES: int = 1 # Retries for polling (at least 1 for reliability) | ||
| POLLING_RETRY_DELAY: float = 0.1 # Delay between retries in polling (seconds) | ||
| POLLING_TARGET_COOLDOWN: int = 60 # Skip offline device for 60 seconds after threshold failures |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether POLLING_TARGET_COOLDOWN is referenced anywhere in the codebase.
rg -nP '\bPOLLING_TARGET_COOLDOWN\b' -C2Repository: Jonathan0823/nexusbus
Length of output: 472
Remove unused POLLING_TARGET_COOLDOWN config setting.
This setting is defined at line 38 but not referenced anywhere in the codebase. The PR mentions skipping offline devices after threshold failures, but this setting was never wired into the implementation. Either implement the feature using this config or remove it to avoid shipping dead code.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/core/config.py` at line 38, POLLING_TARGET_COOLDOWN is an unused config
constant in app/core/config.py; remove the declaration of
POLLING_TARGET_COOLDOWN and any related comments/docs, and verify there are no
remaining references (search for POLLING_TARGET_COOLDOWN) to avoid breakage; if
you prefer to keep the feature instead, wire this constant into the device
polling logic (where offline-device skip/thresholds are handled—e.g., the
polling/retry code path that currently tracks failures) so the cooldown value is
actually used, but do one or the other (remove the unused constant or implement
its use) so no dead config is shipped.
| # Reset gateway after failure to prevent stale connection from affecting next device | ||
| try: | ||
| await manager.reset_gateway(device_id) | ||
| logger.debug( | ||
| "polling_gateway_reset_after_failure", | ||
| device_id=device_id, | ||
| host_port=host_port, | ||
| message="Gateway reset after poll failure", | ||
| ) | ||
| except Exception as reset_err: | ||
| logger.warning( | ||
| "polling_gateway_reset_failed", | ||
| device_id=device_id, | ||
| host_port=host_port, | ||
| error=str(reset_err), | ||
| message="Failed to reset gateway after failure", | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm reset_gateway contract and any other callers that may rely on its semantics.
rg -nP '\breset_gateway\s*\(' -C3Repository: Jonathan0823/nexusbus
Length of output: 3140
🏁 Script executed:
# 1. Read the reset_gateway implementation (lines 626-653)
sed -n '626,653p' app/core/modbus_client.py | cat -n
# 2. Read poll_gateway_group to understand device iteration (review mentions lines 419-433)
sed -n '419,433p' app/services/poller.py | cat -n
# 3. Check the error handling context around line 243 in poller.py
sed -n '210,260p' app/services/poller.py | cat -nRepository: Jonathan0823/nexusbus
Length of output: 4701
🏁 Script executed:
# Find ModbusClientError definition and see what it covers
rg -n 'class ModbusClientError' -A 10
# Also check what exceptions are typically raised by modbus operations
rg -n 'raise ModbusClientError' -B 2 -A 2 | head -50Repository: Jonathan0823/nexusbus
Length of output: 3304
🏁 Script executed:
# Confirm the exact catch clause and what precedes it
sed -n '213,225p' app/services/poller.py | cat -nRepository: Jonathan0823/nexusbus
Length of output: 826
Unconditional gateway reset on ModbusClientError punishes other devices sharing the same gateway.
Per ModbusClientManager.reset_gateway (app/core/modbus_client.py:626-653), the reset closes the gateway keyed by (host, port) — not per-device — and the docstring explicitly states "Resets the entire gateway (host:port), not just one device." In poll_gateway_group (lines 419-433) devices sharing a gateway are polled sequentially in a single cycle, so a failure on device at position i resets the shared gateway that device at i+1..N is about to use.
The problem is compounded by the catch clause (line 213): ModbusClientError wraps both connection failures and protocol-level errors. When a device returns a bad register address or other exception response (lines 567, 610 in modbus_client.py), it raises ModbusClientError — triggering an unconditional gateway reset that disrupts healthy devices on that gateway. This amplifies transient issues instead of isolating them.
Restrict resets to actual connection failures:
- Replace the catch of
ModbusClientErrorwith the underlying connection-specific types only (ConnectionError,ConnectionResetError,OSError), or - Check the error type within the handler and only reset on transport failures, not protocol-level
ModbusExceptionresponses.
Example: restrict to connection-shaped errors
except (ConnectionError, ConnectionResetError, OSError) as exc:
# Connection error - reset gateway
await manager.reset_gateway(device_id)
except ModbusClientError as exc:
# Protocol error (bad register, device exception, etc.) - no reset
logger.warning(...)🧰 Tools
🪛 Ruff (0.15.10)
[warning] 250-250: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/services/poller.py` around lines 241 - 257, The current handler
unconditionally calls manager.reset_gateway (ModbusClientManager.reset_gateway)
when catching ModbusClientError, which resets the entire (host,port) gateway
shared by devices polled in poll_gateway_group and therefore punishes healthy
devices; change the exception handling so only transport/connection failures
trigger await manager.reset_gateway(device_id) — e.g., catch only
ConnectionError, ConnectionResetError, OSError (or inspect the wrapped exception
inside ModbusClientError and reset only if it is one of those connection types),
and for other ModbusClientError cases (protocol-level ModbusException/bad
register) just log a warning without resetting the gateway.
| @pytest.mark.asyncio | ||
| async def test_one_offline_device_does_not_block_others(): | ||
| """Test that one offline device doesn't prevent others from being polled.""" | ||
| _pending_mqtt_tasks.clear() | ||
|
|
||
| # Setup mock manager | ||
| mock_manager = AsyncMock(spec=ModbusClientManager) | ||
| mock_cache = AsyncMock(spec=RegisterCache) | ||
|
|
||
| # Device configs for testing | ||
| from app.core.modbus_client import DeviceConfig | ||
|
|
||
| online_device = DeviceConfig( | ||
| device_id="online-device", | ||
| host="192.168.1.100", | ||
| port=502, | ||
| slave_id=1, | ||
| timeout=1, | ||
| max_retries=1, | ||
| retry_delay=0.01, | ||
| ) | ||
| offline_device = DeviceConfig( | ||
| device_id="offline-device", | ||
| host="192.168.1.100", | ||
| port=502, | ||
| slave_id=2, | ||
| timeout=1, | ||
| max_retries=1, | ||
| retry_delay=0.01, | ||
| ) | ||
|
|
||
| # Configure mock manager to return config | ||
| mock_manager.get_config.side_effect = lambda did: online_device if did == "online-device" else offline_device | ||
|
|
||
| # First device succeeds | ||
| mock_manager.read_registers.return_value = [100, 200, 300] | ||
|
|
||
| # Target for online device | ||
| target_online = { | ||
| "device_id": "online-device", | ||
| "register_type": "holding", | ||
| "address": 0, | ||
| "count": 3, | ||
| } | ||
|
|
||
| # Poll the online device - should succeed | ||
| success, error = await _poll_single_target(target_online, mock_manager, mock_cache) | ||
|
|
||
| assert success is True | ||
| assert error == "" | ||
| mock_manager.reset_gateway.assert_not_called() # No failure, no reset needed |
There was a problem hiding this comment.
Test doesn't validate what its name claims.
The test name says an offline device "does not block others," but the body only polls target_online once. The offline DeviceConfig is constructed and never used, read_registers only ever returns success, and there's no sequential call that would exercise the isolation path. As written, this is effectively a duplicate of test_poll_single_target_success.
To actually prove isolation, poll an offline target first (with read_registers.side_effect raising on that slave_id) and then an online target, asserting the second call still succeeds.
🧪 Proposed test that exercises the isolation path
- # Configure mock manager to return config
- mock_manager.get_config.side_effect = lambda did: online_device if did == "online-device" else offline_device
-
- # First device succeeds
- mock_manager.read_registers.return_value = [100, 200, 300]
-
- # Target for online device
- target_online = {
- "device_id": "online-device",
- "register_type": "holding",
- "address": 0,
- "count": 3,
- }
-
- # Poll the online device - should succeed
- success, error = await _poll_single_target(target_online, mock_manager, mock_cache)
-
- assert success is True
- assert error == ""
- mock_manager.reset_gateway.assert_not_called() # No failure, no reset needed
+ mock_manager.get_config.side_effect = lambda did: (
+ online_device if did == "online-device" else offline_device
+ )
+
+ # Offline device raises; online device succeeds.
+ async def read_side_effect(device_id, **_):
+ if device_id == "offline-device":
+ raise ModbusClientError("No response")
+ return [100, 200, 300]
+
+ mock_manager.read_registers.side_effect = read_side_effect
+
+ target_offline = {"device_id": "offline-device", "register_type": "holding", "address": 0, "count": 3}
+ target_online = {"device_id": "online-device", "register_type": "holding", "address": 0, "count": 3}
+
+ off_ok, _ = await _poll_single_target(target_offline, mock_manager, mock_cache)
+ on_ok, on_err = await _poll_single_target(target_online, mock_manager, mock_cache)
+
+ assert off_ok is False
+ assert on_ok is True and on_err == ""
+ mock_manager.reset_gateway.assert_called_once_with("offline-device")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/unit/test_poller.py` around lines 224 - 274, The test currently only
polls the online device; update it to exercise isolation by making
mock_manager.read_registers.side_effect raise an exception when called for the
offline device's slave_id (slave_id==2) and return a successful list for the
online slave_id, then call _poll_single_target first with a target for the
offline device (assert success is False and error is non-empty) and then call
_poll_single_target for the online target (assert success is True and error ==
""), and finally assert mock_manager.reset_gateway was invoked for the
failing/offline call (e.g., called once) while the online call succeeds despite
the prior failure; reference _poll_single_target, DeviceConfig,
mock_manager.read_registers, and mock_manager.reset_gateway when making the
changes.
Related issue
Summary
Scope
In scope:
app/core/config.py- new polling settingsapp/services/poller.py- timeout/retry usage, gateway reset, pacing delay, enhanced loggingtests/unit/test_poller.py- failure isolation testsOut of scope:
Testing
Test details
pytest tests/unit/test_poller.py -v(14 tests passed)pytest tests/ -v(80 tests passed)Reviewer focus
_poll_single_targetexception handlerNotes
Summary by CodeRabbit
New Features
Improvements
Tests