Repository navigation
feat(modbus): add device health tracking and quarantine to prevent gateway poisoning - #8
Conversation
…tracking Adds a new device_health table to track per-device health metrics: - state enum (healthy/degraded/offline), indexed for fast queries - failure_count, consecutive_failures, success_count - last_success, last_failure timestamps - quarantine_until timestamp (indexed) for quarantining problematic devices - created_at, updated_at timestamps FK constraint to modbus_devices with ON DELETE CASCADE. Migration 004 creates table, indexes, and populates initial health records for all existing devices (default state=healthy). Enables exponential backoff quarantine and health-aware polling to prevent one bad device on a shared gateway from poisoning others.
Adds comprehensive health tracking functions: - get_device_health(): fetch single device health record - get_all_device_health(): fetch all device health records - get_active_device_health(): fetch non-offline devices - upsert_device_health(): atomic create-or-update for health records - record_device_success(): on successful poll → clear quarantine, reset consecutive_failures, set state=HEALTHY - record_device_failure(): on poll failure → increment counters, apply exponential backoff quarantine (15s→30s→60s→300s cap). Non-consecutive failures reset backoff counter. - get_quarantined_devices(): efficiently filter devices where quarantine_until > now(), used to skip quarantined devices Implements circuit breaker + quarantine pattern at per-device level while gateway cooldown handles shared resource protection.
Replaces synchronous ModbusTcpClient with AsyncModbusTcpClient: - Removes all asyncio.to_thread() wrappers - All I/O now truly async using pymodbus async API - Enables concurrent polling across different gateways Adds _gateway_cooldowns dict to prevent rapid successive requests to same gateway after failures: - On gateway-level failure (connection/timeout/ModbusError), sets cooldown for that gateway (host:port) for 1 second - Subsequent requests to same gateway wait for cooldown before attempting, allowing RTU buffer to settle - Prevents cascade failure when one bad device disrupts shared gateway state Gateway-level protection complements per-device quarantine.
load_polling_targets_from_db() now: 1. Fetches all device health records 2. Filters out quarantined devices (quarantine_until > now()) 3. Sorts remaining by health priority: HEALTHY > DEGRADED (healthy devices polled first, degraded later) 4. Skips quarantine_until=None entirely 5. Logs skipped_quarantined list for observability Result: quarantined devices are completely bypassed each cycle, healthy devices prioritized, preventing offline device (e.g. slave 5) from receiving any polling attempts while quarantined.
Sets pymodbus and modbus libraries to CRITICAL level in logging_config.py: - pymodbus.* loggers silenced (was spamming DEBUG frame dumps) - modbus.main pymodbus client logging suppressed - Preserves application-level logging at configured level Reduces log noise while keeping error visibility.
Wraps resource cleanup in asyncio.wait_for(timeout=2.0): - modbus_client.close() - Cancel and wait for MQTT pending tasks (_pending_mqtt_publishes) - mqtt_client.stop() Prevents indefinite hangs on shutdown if resources stuck. Ensures clean exit even under load or partial failures.
debug_modbus.py updated to use AsyncModbusTcpClient: - Entrypoint changed from sync main() to async main_async() - Uses asyncio.run(main_async()) - All read calls now await client.read_holding_registers() - Matches production code async pattern for consistency Allows testing async modbus behavior and quarantine logic manually without running full API server.
test_modbus_client.py: - Updated MockAsyncModbusClient to properly simulate async context manager - Tests for _handle_gateway_failure() and cooldown logic - Tests for _should_skip_device() with quarantine_until checks - Success/failure branching tests with health state transitions test_poller.py: - Added mock for get_all_device_health() in load_polling_targets_from_db tests - Tests verify quarantined devices are filtered out - Tests verify healthy devices polled first, degraded second - Tests verify skipped_quarantined list logged correctly All 15 unit tests pass (pytest -q).
|
Warning Rate limit exceeded
To keep reviews running without waiting, you can enable usage-based add-on for your organization. This allows additional reviews beyond the hourly cap. Account admins can enable it under billing. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughThis PR converts the synchronous Modbus client to async/await architecture, introduces device health tracking with state management (HEALTHY/DEGRADED/OFFLINE) and quarantine logic with exponential backoff, enhances the poller to skip quarantined devices while prioritizing healthier ones, and hardens graceful shutdown with timeouts to prevent hangs. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Modbus Client
participant Manager as ModbusClientManager
participant Gateway as AsyncModbusGateway
participant DB as Database (CRUD)
participant Poller as Polling Service
Client->>Manager: Request read from device X
alt Device not quarantined
Manager->>Gateway: Execute async read
Gateway->>Gateway: Attempt read (with retries)
alt Read succeeds
Gateway-->>Manager: Return data
Manager->>DB: record_device_success(device_id)
DB->>DB: Increment success_count<br/>Reset failures<br/>Upgrade state (if OFFLINE→DEGRADED)
Manager-->>Client: Return data
else Read fails
Gateway-->>Manager: Raise exception
Manager->>DB: record_device_failure(device_id)
DB->>DB: Increment failure_count<br/>Set quarantine_until<br/>Downgrade state
Manager-->>Client: Re-raise exception
end
else Device quarantined
Poller->>DB: get_all_device_health()
DB-->>Poller: Health records (with quarantine_until)
Poller->>Poller: Filter & exclude quarantined
Poller-->>Client: Reduced target list
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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: 9
🧹 Nitpick comments (5)
app/core/logging_config.py (1)
48-51: Silencing allpymodbusat CRITICAL is overly broad.Setting the root
pymodbuslogger to CRITICAL also suppresses legitimate ERROR/WARNING messages that would help diagnose protocol or hardware faults (the very class of issues this PR is responding to). Consider raising it to WARNING and only pinning the truly chatty submodules (pymodbus.logging,pymodbus.transport,pymodbus.framer) to CRITICAL/ERROR.♻️ Suggested adjustment
- logging.getLogger("pymodbus").setLevel(logging.CRITICAL) - logging.getLogger("pymodbus.logging").setLevel(logging.CRITICAL) - logging.getLogger("pymodbus.transport").setLevel(logging.CRITICAL) + logging.getLogger("pymodbus").setLevel(logging.WARNING) + logging.getLogger("pymodbus.logging").setLevel(logging.CRITICAL) + logging.getLogger("pymodbus.transport").setLevel(logging.CRITICAL)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/core/logging_config.py` around lines 48 - 51, Don't silence the entire "pymodbus" logger at CRITICAL; instead set logging.getLogger("pymodbus").setLevel(logging.WARNING) so warnings/errors remain visible, and only pin noisy submodules—logging.getLogger("pymodbus.logging"), logging.getLogger("pymodbus.transport") and add logging.getLogger("pymodbus.framer")—to CRITICAL (or ERROR for framer if you prefer) to suppress frame-level noise while preserving useful diagnostics.tests/unit/test_poller.py (1)
39-45: Add test cases for the new health-aware filtering and sorting.The current update only asserts the empty-health path. The two new behaviors introduced by this PR — (1) excluding
is_quarantinedtargets and (2) sorting by(is_quarantined, state_priority, consecutive_failures)— are not exercised. Please add at least:
- a test where
get_all_device_healthreturns a record withquarantine_untilin the future, and asserts that device is missing fromtargets,- a test with mixed
healthy/degraded/offlinerecords asserting the resultingtargetsorder.🤖 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 39 - 45, The test only covers the empty-health path; add two new unit tests in tests/unit/test_poller.py that exercise the new health-aware behavior of load_polling_targets_from_db: (1) mock app.services.poller.crud.get_all_device_health to return a record with quarantine_until set in the future and assert the resulting targets list does NOT include the corresponding device_id (verifying quarantined devices are excluded), and (2) mock get_all_device_health to return multiple records with different state values (healthy, degraded, offline) and varying consecutive_failures and assert the final targets order matches sorting by (is_quarantined, state_priority, consecutive_failures) as implemented by load_polling_targets_from_db; use AsyncMock returns and the same target base used in the existing test to keep other fields consistent.main.py (1)
165-170: Redundant timeout wrapper and duplicate import.Two minor cleanups:
await_pending_mqtt_tasks(timeout=2.0)already usesasyncio.wait(..., timeout=2.0)internally and cannot block beyond ~2s, so the outerasyncio.wait_for(..., timeout=3.0)will essentially never fire — theexcept asyncio.TimeoutErrorbranch is unreachable. Either drop the outerwait_foror drop the inner timeout.await_pending_mqtt_tasksis already imported at the top of the module (line 25); the localfrom app.services.poller import await_pending_mqtt_tasksat line 165 is dead.♻️ Proposed cleanup
- # Stop MQTT Client - # First, await any pending MQTT publish tasks - from app.services.poller import await_pending_mqtt_tasks - try: - await asyncio.wait_for(await_pending_mqtt_tasks(timeout=2.0), timeout=3.0) - except asyncio.TimeoutError: - logger.warning("mqtt_pending_timeout", message="Awaiting MQTT tasks timed out") + # Stop MQTT Client + # First, await any pending MQTT publish tasks (internal timeout enforces bound) + await await_pending_mqtt_tasks(timeout=2.0)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@main.py` around lines 165 - 170, Remove the redundant local import and the unnecessary outer timeout wrapper: delete the local "from app.services.poller import await_pending_mqtt_tasks" and replace the try/except that wraps "asyncio.wait_for(await_pending_mqtt_tasks(timeout=2.0), timeout=3.0)" with a single direct await of "await_pending_mqtt_tasks(timeout=2.0)" (drop the outer "asyncio.wait_for" and the unreachable "except asyncio.TimeoutError" branch and its "logger.warning" call), since await_pending_mqtt_tasks already enforces its own timeout.tests/unit/test_modbus_client.py (1)
32-36:mock_db_session_factoryis not a valid replacement forasync_session_maker.Production code uses
db_session_factoryasasync with self.db_session_factory() as session: ..., which requires the call to return an async context manager, not an async generator. As written, callingmock_session_factory()returns an async generator, and any test that triggers a DB code path (e.g., oncerecord_device_success/record_device_failureare exercised here) will fail withAttributeError: __aenter__.Today the tests only succeed because every gateway is fully mocked and the factory is never invoked. Make this explicit, e.g.:
♻️ Suggested fix
`@pytest.fixture` def mock_db_session_factory(): - """Mock database session factory for testing.""" - async def mock_session(): - yield MagicMock() - return mock_session + """Mock async session factory: callable returning an async context manager.""" + factory = MagicMock() + factory.return_value.__aenter__ = AsyncMock(return_value=AsyncMock()) + factory.return_value.__aexit__ = AsyncMock(return_value=None) + return factory🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/test_modbus_client.py` around lines 32 - 36, mock_db_session_factory currently returns an async generator, which breaks production usage where db_session_factory is used as an async context manager (async with self.db_session_factory() as session), causing AttributeError: __aenter__; change mock_db_session_factory to return an async context manager that yields a MagicMock (e.g., wrap the inner async generator with contextlib.asynccontextmanager or implement an object with async __aenter__/__aexit__), so tests that exercise DB paths (record_device_success, record_device_failure) can use the mock just like async_session_maker.app/database/crud.py (1)
407-411: Recovery fromOFFLINEafter a single success may be too aggressive.A device only reaches
OFFLINEafter ≥10 consecutive failures (Line 463), but a single successful read flips it back toDEGRADEDandconsecutive_failures=0, so the next single success will promote it all the way toHEALTHY. For a flapping device on a shared RTU bus this can re-introduce the very poisoning scenario this PR addresses. Consider gating recovery on a minimum number of consecutive successes (e.g., trackconsecutive_successesand require ≥3 before stepping the state up), or only steppingOFFLINE → DEGRADEDand requiring an explicit reset/operator action to reachHEALTHY.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@app/database/crud.py` around lines 407 - 411, The current one-success recovery is too aggressive; introduce and persist a health.consecutive_successes counter (reset to 0 on any failure, increment on success) and change the promotion logic so OFFLINE no longer jumps to DEGRADED on a single success—instead only increment consecutive_successes and require consecutive_successes >= 3 before changing health.state from OFFLINE → DEGRADED, and likewise require consecutive_successes >= 3 before changing DEGRADED → HEALTHY; ensure you update any record_success/record_failure or similar functions that update health.consecutive_failures to also update health.consecutive_successes and persist the field.
🤖 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 676-688: The current code logs a gateway-reset message
conditionally but always calls await self.reset_gateway(device_id); change this
to reset only for stale-buffer–type failures by moving await
self.reset_gateway(device_id) inside the same conditional as the logger, and
replace the brittle string check ("ModbusIOException" in failure_reason) with a
typed check against the actual exception (e.g., isinstance(e,
ModbusIOException)) or a robust sentinel from the exception raised in
_execute_read; update the log to include failure_reason and device_id, and
ensure the exception variable (e) / failure_reason used in the check are the
same values produced by _execute_read so both the log and reset are executed
only for real "no response"/ModbusIOException cases.
In `@app/database/crud.py`:
- Around line 497-506: Remove the redundant local import of DeviceHealth and
replace the `DeviceHealth.quarantine_until != None` comparison with the
idiomatic SQLAlchemy form: use `DeviceHealth.quarantine_until.is_not(None)` and
keep the time check `DeviceHealth.quarantine_until > datetime.now()` in the
`get_all_quarantined_devices` function so the linter warning is resolved and the
query semantics remain the same.
- Around line 430-441: The new-Record branch for DeviceHealth sets
quarantine_until=now which is already expired when checked by
is_device_quarantined; change the no-prior-record branch to compute and set
quarantine_until using the same backoff calculation as the existing else-branch
(i.e. compute backoff_seconds the same way you do when updating an existing
DeviceHealth and set quarantine_until = now +
timedelta(seconds=backoff_seconds)), and initialize
failure_count/consecutive_failures appropriately (1) and updated_at/created_at
as before so first-time failures observe the same quarantine window as
subsequent failures.
- Line 366: Replace naive datetime.now() usages in CRUD and DeviceHealth model
defaults with timezone-aware datetimes by using datetime.now(timezone.utc) for
all fields: created_at, updated_at, last_success, last_failure, and
quarantine_until; update any helper/constructor code that sets "now =
datetime.now()" to "now = datetime.now(timezone.utc)" (and add the timezone
import). Also change the backoff calculation that currently does
datetime.fromtimestamp(now.timestamp() + backoff_seconds) to a direct arithmetic
now + timedelta(seconds=backoff_seconds) (ensure timedelta is imported). Keep
all changes limited to the CRUD functions and the DeviceHealth model defaults
and preserve existing variable names like now, backoff_seconds.
In `@app/database/models.py`:
- Around line 235-236: The created_at/updated_at Field default_factories produce
naive datetimes which conflict with DB TIMESTAMP WITH TIME ZONE and cause
comparisons errors; change the default factories in the model attributes
(created_at, updated_at in app/database/models.py and the similar fields on
ModbusDevice and PollingTarget) to produce UTC-aware datetimes by using
datetime.now(timezone.utc), and update internal "now" usages in
app/database/crud.py (e.g., in record_device_success, record_device_failure,
upsert_device_health) to use datetime.now(timezone.utc) so all timestamps are
consistently timezone-aware.
In `@app/services/poller.py`:
- Around line 103-111: The quarantine logic breaks on tz-aware vs naive
datetimes; update all timestamp comparisons and writes to use timezone-aware UTC
times by replacing any naive datetime.now() with datetime.now(timezone.utc)
inside the CRUD functions record_device_failure, is_device_quarantined, and
get_all_quarantined_devices and update default_factory lambdas in the models to
produce tz-aware datetimes (use timezone.utc); once CRUD/models consistently use
tz-aware UTC timestamps, remove the defensive
q_time.replace(tzinfo=timezone.utc) fallback in the poller loop so comparisons
use the single canonical timezone-aware representation.
In `@main.py`:
- Around line 153-161: The lifespan handler is directly touching
ModbusClientManager private attributes (manager._gateways.clear /
manager._locks.clear); add a public method on ModbusClientManager (e.g.,
force_reset() or force_clear()) that performs the forced-clear/reset of internal
gateway and lock state, implement it in the ModbusClientManager class in
app/core/modbus_client.py, and replace the direct accesses in the lifespan code
with a call to manager.force_reset() after catching asyncio.TimeoutError
(keeping the await asyncio.wait_for(manager.close_all(), ...) behavior intact).
In `@migrations/004_add_device_health.py`:
- Around line 61-95: The migration currently commits after creating the
table/indexes and returns early if an `existing` check is truthy, which can
leave devices unseeded on partial failures; instead always run the idempotent
seed (the `populate_sql` INSERT ... ON CONFLICT DO NOTHING) and perform a single
commit at the end. Concretely: remove or bypass the early-return that checks
`existing` so that `populate_sql` is executed every run, and consolidate commits
by calling `await session.commit()` only once after both
`session.execute(create_sql)` and `session.execute(populate_sql)` complete
(references: create_sql, populate_sql, session.execute(...), session.commit()).
- Around line 1-13: The docstring in the migration module
"004_add_device_health" incorrectly shows a failing invocation (`python -m
migrations.004_add_device_health`); update that docstring to show the correct
way to run migrations (e.g., "python migrate.py --migration 004") or note that
migrations are run via the MigrationRunner in migrate.py so users should invoke
migrate.py with the --migration option for Migration 004 (and similarly fix the
docstrings in other migration modules 001–003 to match).
---
Nitpick comments:
In `@app/core/logging_config.py`:
- Around line 48-51: Don't silence the entire "pymodbus" logger at CRITICAL;
instead set logging.getLogger("pymodbus").setLevel(logging.WARNING) so
warnings/errors remain visible, and only pin noisy
submodules—logging.getLogger("pymodbus.logging"),
logging.getLogger("pymodbus.transport") and add
logging.getLogger("pymodbus.framer")—to CRITICAL (or ERROR for framer if you
prefer) to suppress frame-level noise while preserving useful diagnostics.
In `@app/database/crud.py`:
- Around line 407-411: The current one-success recovery is too aggressive;
introduce and persist a health.consecutive_successes counter (reset to 0 on any
failure, increment on success) and change the promotion logic so OFFLINE no
longer jumps to DEGRADED on a single success—instead only increment
consecutive_successes and require consecutive_successes >= 3 before changing
health.state from OFFLINE → DEGRADED, and likewise require consecutive_successes
>= 3 before changing DEGRADED → HEALTHY; ensure you update any
record_success/record_failure or similar functions that update
health.consecutive_failures to also update health.consecutive_successes and
persist the field.
In `@main.py`:
- Around line 165-170: Remove the redundant local import and the unnecessary
outer timeout wrapper: delete the local "from app.services.poller import
await_pending_mqtt_tasks" and replace the try/except that wraps
"asyncio.wait_for(await_pending_mqtt_tasks(timeout=2.0), timeout=3.0)" with a
single direct await of "await_pending_mqtt_tasks(timeout=2.0)" (drop the outer
"asyncio.wait_for" and the unreachable "except asyncio.TimeoutError" branch and
its "logger.warning" call), since await_pending_mqtt_tasks already enforces its
own timeout.
In `@tests/unit/test_modbus_client.py`:
- Around line 32-36: mock_db_session_factory currently returns an async
generator, which breaks production usage where db_session_factory is used as an
async context manager (async with self.db_session_factory() as session), causing
AttributeError: __aenter__; change mock_db_session_factory to return an async
context manager that yields a MagicMock (e.g., wrap the inner async generator
with contextlib.asynccontextmanager or implement an object with async
__aenter__/__aexit__), so tests that exercise DB paths (record_device_success,
record_device_failure) can use the mock just like async_session_maker.
In `@tests/unit/test_poller.py`:
- Around line 39-45: The test only covers the empty-health path; add two new
unit tests in tests/unit/test_poller.py that exercise the new health-aware
behavior of load_polling_targets_from_db: (1) mock
app.services.poller.crud.get_all_device_health to return a record with
quarantine_until set in the future and assert the resulting targets list does
NOT include the corresponding device_id (verifying quarantined devices are
excluded), and (2) mock get_all_device_health to return multiple records with
different state values (healthy, degraded, offline) and varying
consecutive_failures and assert the final targets order matches sorting by
(is_quarantined, state_priority, consecutive_failures) as implemented by
load_polling_targets_from_db; use AsyncMock returns and the same target base
used in the existing test to keep other fields consistent.
🪄 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: ec40690a-d72b-4e68-adc3-ad0b9682d337
📒 Files selected for processing (10)
app/core/logging_config.pyapp/core/modbus_client.pyapp/database/crud.pyapp/database/models.pyapp/services/poller.pydebug_modbus.pymain.pymigrations/004_add_device_health.pytests/unit/test_modbus_client.pytests/unit/test_poller.py
| # CRITICAL: Reset gateway on failure to clear stale buffer | ||
| # This prevents one bad device from poisoning other devices on shared gateway | ||
| if "No response from device" in failure_reason or "ModbusIOException" in failure_reason: | ||
| logger.warning( | ||
| "device_failure_gateway_reset", | ||
| device_id=device_id, | ||
| failure_reason=failure_reason, | ||
| host=config.host, | ||
| port=config.port, | ||
| message="Resetting gateway after device failure to prevent buffer poisoning", | ||
| ) | ||
| await self.reset_gateway(device_id) | ||
| raise |
There was a problem hiding this comment.
Gateway reset always runs but the warning log only fires conditionally — pick one behavior.
reset_gateway() on Line 687 is outside the if, so the gateway is reset on every non-CircuitOpenError failure (including transient validation errors). Meanwhile the "Resetting gateway after device failure…" warning at Lines 679–686 only logs for two specific substrings, which is misleading.
Additionally, "ModbusIOException" in failure_reason will almost never match: failure_reason = str(e) where e is the ModbusClientError raised inside _execute_read — its message contains the response text or "No response from device '…'", not the literal class name ModbusIOException. So in practice only the first substring ever fires.
If the intent is "reset only on stale-buffer–type failures", indent the reset under the if; if the intent is "always reset on failure", drop the conditional and unconditionally log. Either way, replace the brittle string match with an isinstance check on the actual exception (e.g., catch ModbusIOException / "no response" via a typed sentinel).
🛠 Proposed fix (option A — always reset, simpler log)
- # CRITICAL: Reset gateway on failure to clear stale buffer
- # This prevents one bad device from poisoning other devices on shared gateway
- if "No response from device" in failure_reason or "ModbusIOException" in failure_reason:
- logger.warning(
- "device_failure_gateway_reset",
- device_id=device_id,
- failure_reason=failure_reason,
- host=config.host,
- port=config.port,
- message="Resetting gateway after device failure to prevent buffer poisoning",
- )
- await self.reset_gateway(device_id)
+ # CRITICAL: Reset gateway on failure to clear stale buffer
+ # This prevents one bad device from poisoning other devices on shared gateway
+ logger.warning(
+ "device_failure_gateway_reset",
+ device_id=device_id,
+ failure_reason=failure_reason,
+ host=config.host,
+ port=config.port,
+ message="Resetting gateway after device failure to prevent buffer poisoning",
+ )
+ await self.reset_gateway(device_id)
raise🤖 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 676 - 688, The current code logs a
gateway-reset message conditionally but always calls await
self.reset_gateway(device_id); change this to reset only for stale-buffer–type
failures by moving await self.reset_gateway(device_id) inside the same
conditional as the logger, and replace the brittle string check
("ModbusIOException" in failure_reason) with a typed check against the actual
exception (e.g., isinstance(e, ModbusIOException)) or a robust sentinel from the
exception raised in _execute_read; update the log to include failure_reason and
device_id, and ensure the exception variable (e) / failure_reason used in the
check are the same values produced by _execute_read so both the log and reset
are executed only for real "no response"/ModbusIOException cases.
| ) -> DeviceHealth: | ||
| """Create or update device health record.""" | ||
| health = await get_device_health(session, device_id) | ||
| now = datetime.now() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Locate every datetime usage that touches DeviceHealth fields and the poller's
# quarantine comparison so we can confirm tz consistency end-to-end.
rg -nP --type=py -C2 '\bdatetime\.(now|utcnow)\s*\('
echo '--- poller quarantine comparisons ---'
rg -nP --type=py -C3 'quarantine_until' app/services
echo '--- model field defaults ---'
rg -nP --type=py -C3 'class DeviceHealth\b|quarantine_until|last_failure|last_success' app/database/models.pyRepository: Jonathan0823/nexusbus
Length of output: 10684
🏁 Script executed:
# Check line 452 context in crud.py
sed -n '445,460p' app/database/crud.py
# Search for all fromtimestamp usages in the codebase
rg 'fromtimestamp' --type=py -B2 -A2
# Verify all places where quarantine_until is compared
rg 'quarantine_until\s*[<>]' --type=py -B1 -A1Repository: Jonathan0823/nexusbus
Length of output: 1336
Standardize datetime to UTC across CRUD operations and DeviceHealth model—currently relies on defensive conversion in the poller.
The CRUD operations and DeviceHealth model defaults use naive datetime.now() while the poller uses datetime.now(timezone.utc) with explicit defensive tzinfo conversion (lines 109–110 in app/services/poller.py). Although the poller's conversion prevents crashes, this pattern is fragile and inconsistent with the rest of the codebase (tests, metrics, cache, logging all use datetime.now(timezone.utc)).
Change all datetime assignments in CRUD (created_at, updated_at, last_success, last_failure, quarantine_until) and model defaults to use datetime.now(timezone.utc).
Additionally, simplify line 452: replace the inefficient datetime.fromtimestamp(now.timestamp() + backoff_seconds) round-trip with now + timedelta(seconds=backoff_seconds).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/database/crud.py` at line 366, Replace naive datetime.now() usages in
CRUD and DeviceHealth model defaults with timezone-aware datetimes by using
datetime.now(timezone.utc) for all fields: created_at, updated_at, last_success,
last_failure, and quarantine_until; update any helper/constructor code that sets
"now = datetime.now()" to "now = datetime.now(timezone.utc)" (and add the
timezone import). Also change the backoff calculation that currently does
datetime.fromtimestamp(now.timestamp() + backoff_seconds) to a direct arithmetic
now + timedelta(seconds=backoff_seconds) (ensure timedelta is imported). Keep
all changes limited to the CRUD functions and the DeviceHealth model defaults
and preserve existing variable names like now, backoff_seconds.
| if not health: | ||
| health = DeviceHealth( | ||
| device_id=device_id, | ||
| state=DeviceState.DEGRADED, | ||
| failure_count=1, | ||
| consecutive_failures=1, | ||
| last_failure=now, | ||
| quarantine_until=now, # immediate quarantine on first failure | ||
| created_at=now, | ||
| updated_at=now, | ||
| ) | ||
| session.add(health) |
There was a problem hiding this comment.
First-time failure does not actually quarantine the device.
In the no-prior-record branch, quarantine_until=now is already in the past by the time any caller checks it (is_device_quarantined does datetime.now() < health.quarantine_until at Line 494, which is False). So a device that fails for the very first time is recorded as DEGRADED but immediately polled again, while a device that already has a health row and fails for the first time gets the proper 15 s window from the else-branch (Lines 451–452). This inconsistency defeats the intent stated by the inline comment ("immediate quarantine on first failure") and is exactly the gateway-poisoning case this PR is trying to prevent.
Apply the same backoff formula in both branches:
🐛 Proposed fix
if not health:
+ backoff_seconds = 15 # 1st-fail backoff, mirrors else-branch
health = DeviceHealth(
device_id=device_id,
state=DeviceState.DEGRADED,
failure_count=1,
consecutive_failures=1,
last_failure=now,
- quarantine_until=now, # immediate quarantine on first failure
+ quarantine_until=datetime.fromtimestamp(now.timestamp() + backoff_seconds),
created_at=now,
updated_at=now,
)
session.add(health)
+ logger.info(
+ "device_quarantine_set",
+ device_id=device_id,
+ consecutive_failures=1,
+ quarantine_seconds=backoff_seconds,
+ message=f"Device quarantined for {backoff_seconds}s after 1 failure",
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if not health: | |
| health = DeviceHealth( | |
| device_id=device_id, | |
| state=DeviceState.DEGRADED, | |
| failure_count=1, | |
| consecutive_failures=1, | |
| last_failure=now, | |
| quarantine_until=now, # immediate quarantine on first failure | |
| created_at=now, | |
| updated_at=now, | |
| ) | |
| session.add(health) | |
| if not health: | |
| backoff_seconds = 15 # 1st-fail backoff, mirrors else-branch | |
| health = DeviceHealth( | |
| device_id=device_id, | |
| state=DeviceState.DEGRADED, | |
| failure_count=1, | |
| consecutive_failures=1, | |
| last_failure=now, | |
| quarantine_until=datetime.fromtimestamp(now.timestamp() + backoff_seconds), | |
| created_at=now, | |
| updated_at=now, | |
| ) | |
| session.add(health) | |
| logger.info( | |
| "device_quarantine_set", | |
| device_id=device_id, | |
| consecutive_failures=1, | |
| quarantine_seconds=backoff_seconds, | |
| message=f"Device quarantined for {backoff_seconds}s after 1 failure", | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/database/crud.py` around lines 430 - 441, The new-Record branch for
DeviceHealth sets quarantine_until=now which is already expired when checked by
is_device_quarantined; change the no-prior-record branch to compute and set
quarantine_until using the same backoff calculation as the existing else-branch
(i.e. compute backoff_seconds the same way you do when updating an existing
DeviceHealth and set quarantine_until = now +
timedelta(seconds=backoff_seconds)), and initialize
failure_count/consecutive_failures appropriately (1) and updated_at/created_at
as before so first-time failures observe the same quarantine window as
subsequent failures.
| async def get_all_quarantined_devices(session: AsyncSession) -> List[str]: | ||
| """Get list of device IDs that are currently quarantined.""" | ||
| from app.database.models import DeviceHealth | ||
| result = await session.execute( | ||
| select(DeviceHealth).where( | ||
| DeviceHealth.quarantine_until != None, | ||
| DeviceHealth.quarantine_until > datetime.now() | ||
| ) | ||
| ) | ||
| return [h.device_id for h in result.scalars().all()] |
There was a problem hiding this comment.
Redundant import and lint-flagged != None comparison.
- Line 499:
DeviceHealthis already imported at the top of the file (Line 17); the local re-import is dead code. - Line 502: Ruff E711 flags
DeviceHealth.quarantine_until != None. The SQLAlchemy column comparison is semantically correct (it does emitIS NOT NULL), but.is_not(None)is the idiomatic form and silences the linter without anoqa.
♻️ Proposed cleanup
async def get_all_quarantined_devices(session: AsyncSession) -> List[str]:
"""Get list of device IDs that are currently quarantined."""
- from app.database.models import DeviceHealth
result = await session.execute(
select(DeviceHealth).where(
- DeviceHealth.quarantine_until != None,
+ DeviceHealth.quarantine_until.is_not(None),
DeviceHealth.quarantine_until > datetime.now()
)
)
return [h.device_id for h in result.scalars().all()]🧰 Tools
🪛 Ruff (0.15.11)
[error] 502-502: Comparison to None should be cond is not None
Replace with cond is not None
(E711)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/database/crud.py` around lines 497 - 506, Remove the redundant local
import of DeviceHealth and replace the `DeviceHealth.quarantine_until != None`
comparison with the idiomatic SQLAlchemy form: use
`DeviceHealth.quarantine_until.is_not(None)` and keep the time check
`DeviceHealth.quarantine_until > datetime.now()` in the
`get_all_quarantined_devices` function so the linter warning is resolved and the
query semantics remain the same.
| created_at: datetime = Field(default_factory=lambda: datetime.now()) | ||
| updated_at: datetime = Field(default_factory=lambda: datetime.now()) |
There was a problem hiding this comment.
Naive datetime.now() defaults conflict with the tz-aware TIMESTAMP WITH TIME ZONE columns.
migrations/004_add_device_health.py declares created_at/updated_at/quarantine_until as TIMESTAMP WITH TIME ZONE, and app/services/poller.py (lines 103-111) compares against datetime.now(timezone.utc). But default_factory=lambda: datetime.now() here (and in ModbusDevice lines 85-86 and PollingTarget lines 155-156) produces local-naive datetimes. With asyncpg/PG, naive values get reinterpreted via the session TZ on write, leading to skewed timestamps and the TypeError: can't compare offset-naive and offset-aware datetimes failures noted in app/database/crud.py:489-505.
Standardize on UTC-aware factories model-wide:
🛡️ Proposed fix
- created_at: datetime = Field(default_factory=lambda: datetime.now())
- updated_at: datetime = Field(default_factory=lambda: datetime.now())
+ created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+ updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))Apply the same change to ModbusDevice (lines 85-86) and PollingTarget (lines 155-156), and update crud.py's internal now = datetime.now() calls (e.g. record_device_success, record_device_failure, upsert_device_health) to datetime.now(timezone.utc).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/database/models.py` around lines 235 - 236, The created_at/updated_at
Field default_factories produce naive datetimes which conflict with DB TIMESTAMP
WITH TIME ZONE and cause comparisons errors; change the default factories in the
model attributes (created_at, updated_at in app/database/models.py and the
similar fields on ModbusDevice and PollingTarget) to produce UTC-aware datetimes
by using datetime.now(timezone.utc), and update internal "now" usages in
app/database/crud.py (e.g., in record_device_success, record_device_failure,
upsert_device_health) to use datetime.now(timezone.utc) so all timestamps are
consistently timezone-aware.
| now = datetime.now(timezone.utc) # Use UTC for consistency | ||
| for h in all_health: | ||
| is_quarantined = False | ||
| if h.quarantine_until: | ||
| # Ensure both are timezone-aware for comparison | ||
| q_time = h.quarantine_until | ||
| if q_time.tzinfo is None: | ||
| q_time = q_time.replace(tzinfo=timezone.utc) | ||
| is_quarantined = q_time > now |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find all naive datetime.now() usages in DB layer that should be tz-aware.
rg -nP '\bdatetime\.now\(\s*\)' -g '!**/tests/**' app/database/ migrations/
echo "---"
rg -nP 'quarantine_until|last_success|last_failure|created_at|updated_at' -g '!**/tests/**' app/database/crud.pyRepository: Jonathan0823/nexusbus
Length of output: 2777
🏁 Script executed:
# Find migration files to check if TIMESTAMP WITH TIME ZONE is used
fd -e "sql" migrations/ -o -e "py" migrations/ | head -20Repository: Jonathan0823/nexusbus
Length of output: 182
🏁 Script executed:
# Check SQLAlchemy model definitions for quarantine_until and other timestamp fields
sed -n '1,250p' app/database/models.py | cat -nRepository: Jonathan0823/nexusbus
Length of output: 10543
🏁 Script executed:
# Get the specific quarantine functions from crud.py to see the actual comparisons
sed -n '490,510p' app/database/crud.py | cat -n
sed -n '420,455p' app/database/crud.py | cat -n
sed -n '360,410p' app/database/crud.py | cat -nRepository: Jonathan0823/nexusbus
Length of output: 4172
🏁 Script executed:
# Find migration files
find . -path "*/migrations/*.py" -type f | head -20Repository: Jonathan0823/nexusbus
Length of output: 265
🏁 Script executed:
# Look at the migration that creates the device_health table with quarantine_until
rg -l "device_health|DeviceHealth" migrations/ | head -5Repository: Jonathan0823/nexusbus
Length of output: 101
🏁 Script executed:
# Get full context around poller.py lines 103-111
sed -n '95,120p' app/services/poller.py | cat -nRepository: Jonathan0823/nexusbus
Length of output: 1390
🏁 Script executed:
cat migrations/004_add_device_health.pyRepository: Jonathan0823/nexusbus
Length of output: 4541
🏁 Script executed:
# Find all callers of is_device_quarantined and get_all_quarantined_devices
rg -n "is_device_quarantined|get_all_quarantined_devices" --type py -B 2 -A 2 | head -100Repository: Jonathan0823/nexusbus
Length of output: 702
🏁 Script executed:
# Also check if this is_device_quarantined comparison actually causes TypeError by examining error handling
rg -n "TypeError|offset-naive|offset-aware" app/Repository: Jonathan0823/nexusbus
Length of output: 47
🏁 Script executed:
# Find all files that call is_device_quarantined
rg "is_device_quarantined\(" --type py | grep -v "^app/database/crud.py"Repository: Jonathan0823/nexusbus
Length of output: 47
🏁 Script executed:
# Find all files that call get_all_quarantined_devices
rg "get_all_quarantined_devices\(" --type py | grep -v "^app/database/crud.py"Repository: Jonathan0823/nexusbus
Length of output: 47
🏁 Script executed:
# Check what imports exist from crud in other files
rg "from app.database import crud|from app.database.crud import" --type pyRepository: Jonathan0823/nexusbus
Length of output: 689
🏁 Script executed:
# Look at what crud functions are actually used in poller.py
sed -n '1,150p' app/services/poller.py | grep -E "crud\.|await" | head -20Repository: Jonathan0823/nexusbus
Length of output: 449
🏁 Script executed:
# Check modbus_client.py for its usage
rg -n "crud\." app/core/modbus_client.py | head -20Repository: Jonathan0823/nexusbus
Length of output: 47
🏁 Script executed:
# Check if is_device_quarantined or get_all_quarantined_devices appear anywhere in actual code calls
rg "\.is_device_quarantined|\.get_all_quarantined_devices" --type pyRepository: Jonathan0823/nexusbus
Length of output: 47
Timezone mismatch in quarantine functions — existing callers work around it, but the functions themselves are broken.
The schema declares TIMESTAMP WITH TIME ZONE for quarantine_until (and other timestamp fields), which means SQLAlchemy/asyncpg will return tz-aware datetime objects. However, record_device_failure, is_device_quarantined, and get_all_quarantined_devices in app/database/crud.py all use naive datetime.now() for comparisons and writes.
The poller avoids this by:
- Fetching raw health records with
get_all_device_health()instead of using the comparison functions. - Manually performing the quarantine check with defensive
replace(tzinfo=timezone.utc)fallback (lines 103–111).
However, the underlying functions are still broken:
is_device_quarantined()(line 494) would raiseTypeError: can't compare offset-naive and offset-aware datetimesif called directly.get_all_quarantined_devices()(line 503) would fail similarly.
Current code avoids these because no one calls them. But they are part of the public API and should work correctly. Fix by standardizing on datetime.now(timezone.utc) in app/database/crud.py for all failure/success recording and comparisons, and in app/database/models.py for all default_factory lambdas. Then remove the defensive replace(tzinfo=...) in poller.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@app/services/poller.py` around lines 103 - 111, The quarantine logic breaks
on tz-aware vs naive datetimes; update all timestamp comparisons and writes to
use timezone-aware UTC times by replacing any naive datetime.now() with
datetime.now(timezone.utc) inside the CRUD functions record_device_failure,
is_device_quarantined, and get_all_quarantined_devices and update
default_factory lambdas in the models to produce tz-aware datetimes (use
timezone.utc); once CRUD/models consistently use tz-aware UTC timestamps, remove
the defensive q_time.replace(tzinfo=timezone.utc) fallback in the poller loop so
comparisons use the single canonical timezone-aware representation.
| manager: ModbusClientManager = app.state.modbus_manager | ||
| await manager.close_all() | ||
| logger.debug("modbus_gateways_closed", message="All Modbus gateways closed") | ||
| try: | ||
| await asyncio.wait_for(manager.close_all(), timeout=2.0) | ||
| logger.debug("modbus_gateways_closed", message="All Modbus gateways closed") | ||
| except asyncio.TimeoutError: | ||
| logger.warning("modbus_close_timeout", message="Modbus close timed out, forcing exit") | ||
| # Force clear even if timeout | ||
| manager._gateways.clear() | ||
| manager._locks.clear() |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Avoid reaching into ModbusClientManager private state from the lifespan handler.
manager._gateways.clear() / manager._locks.clear() couples shutdown semantics to private attribute names. Any rename or refactor in app/core/modbus_client.py will silently break shutdown (no test will catch it because attribute access on a MagicMock always succeeds). Expose an explicit cleanup method on the manager and call that here.
♻️ Suggested approach
try:
await asyncio.wait_for(manager.close_all(), timeout=2.0)
logger.debug("modbus_gateways_closed", message="All Modbus gateways closed")
except asyncio.TimeoutError:
logger.warning("modbus_close_timeout", message="Modbus close timed out, forcing exit")
- # Force clear even if timeout
- manager._gateways.clear()
- manager._locks.clear()
+ # Force clear even if timeout
+ manager.force_reset() # public method that wipes gateways/locks…and add a corresponding force_reset() method on ModbusClientManager.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@main.py` around lines 153 - 161, The lifespan handler is directly touching
ModbusClientManager private attributes (manager._gateways.clear /
manager._locks.clear); add a public method on ModbusClientManager (e.g.,
force_reset() or force_clear()) that performs the forced-clear/reset of internal
gateway and lock state, implement it in the ModbusClientManager class in
app/core/modbus_client.py, and replace the direct accesses in the lifespan code
with a call to manager.force_reset() after catching asyncio.TimeoutError
(keeping the await asyncio.wait_for(manager.close_all(), ...) behavior intact).
| """Migration 004: Add device_health table for tracking device health and quarantine state. | ||
|
|
||
| Creates the device_health table with columns: | ||
| - device_id (PK, FK to modbus_devices) | ||
| - state (healthy/degraded/offline) | ||
| - failure_count, consecutive_failures, success_count | ||
| - last_success, last_failure (timestamps) | ||
| - quarantine_until (timestamp for quarantine expiry) | ||
| - created_at, updated_at | ||
|
|
||
| Run from project root: | ||
| python -m migrations.004_add_device_health | ||
| """ |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# How are existing migrations 001-003 invoked?
fd -t f -e py . migrations
echo "---"
rg -n -C2 'python -m migrations|import_module|runpy|migrations\.\d'Repository: Jonathan0823/nexusbus
Length of output: 3224
🏁 Script executed:
#!/bin/bash
# Check migrate.py to see how migrations are actually invoked
cat -n migrations/base.py | head -50
echo "---"
cat -n migrate.py | head -80Repository: Jonathan0823/nexusbus
Length of output: 4780
Update docstring to reflect actual migration invocation method.
The documented python -m migrations.004_add_device_health command will fail because 004_add_device_health is not a valid Python identifier. This affects all migrations (001–004) equally. Update the docstring to match actual usage:
python migrate.py --migration 004
or document that migrations are invoked via the MigrationRunner through migrate.py.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@migrations/004_add_device_health.py` around lines 1 - 13, The docstring in
the migration module "004_add_device_health" incorrectly shows a failing
invocation (`python -m migrations.004_add_device_health`); update that docstring
to show the correct way to run migrations (e.g., "python migrate.py --migration
004") or note that migrations are run via the MigrationRunner in migrate.py so
users should invoke migrate.py with the --migration option for Migration 004
(and similarly fix the docstrings in other migration modules 001–003 to match).
| await session.execute(create_sql) | ||
|
|
||
| # Create index on state for faster queries | ||
| await session.execute(text("CREATE INDEX idx_device_health_state ON device_health(state)")) | ||
|
|
||
| # Create index on quarantine_until for efficient quarantine filtering | ||
| await session.execute(text("CREATE INDEX idx_device_health_quarantine ON device_health(quarantine_until)")) | ||
|
|
||
| await session.commit() | ||
|
|
||
| runner.print_success("Created device_health table with indexes") | ||
| runner.print_info(" - PK: device_id (FK -> modbus_devices)") | ||
| runner.print_info(" - Indexes: state, quarantine_until") | ||
|
|
||
| # Populate initial health records for existing devices | ||
| runner.print_info("Populating initial health records for existing devices...") | ||
|
|
||
| populate_sql = text(""" | ||
| INSERT INTO device_health (device_id, state, failure_count, consecutive_failures, | ||
| success_count, created_at, updated_at) | ||
| SELECT | ||
| device_id, | ||
| 'healthy'::VARCHAR, | ||
| 0, | ||
| 0, | ||
| 0, | ||
| NOW(), | ||
| NOW() | ||
| FROM modbus_devices | ||
| WHERE device_id NOT IN (SELECT device_id FROM device_health) | ||
| ON CONFLICT (device_id) DO NOTHING | ||
| """) | ||
|
|
||
| await session.execute(populate_sql) | ||
| await session.commit() |
There was a problem hiding this comment.
Non-atomic create + seed: a partial failure leaves devices unseeded with no recovery.
The migration commits twice: once after CREATE TABLE + indexes (line 69), then again after the seed INSERT (line 95). If the seed step fails (network blip, FK violation, etc.) after the first commit succeeds, on re-run the existence check at lines 37-39 returns early (Skipping) and devices never get their initial healthy rows. Polling for those devices will then silently fall back to the health_lookup.get(..., {"state": "healthy", ...}) default in app/services/poller.py, which is mostly OK — but the migration’s contract (Initial health records created for existing devices) is violated.
Two options:
- Drop the early-return on
existingand always run the idempotentINSERT ... ON CONFLICT DO NOTHING(it's already idempotent, so re-running is safe). - Run CREATE + INSERT in a single transaction (one
commit()at the end).
♻️ Suggested option 1
if existing:
runner.print_warning("device_health table already exists. Skipping.")
- return
+ # Still ensure seed rows exist for any newly added devices.
+ else:
+ await session.execute(create_sql)
+ await session.execute(text("CREATE INDEX idx_device_health_state ON device_health(state)"))
+ await session.execute(text("CREATE INDEX idx_device_health_quarantine ON device_health(quarantine_until)"))
- # Create the device_health table
- create_sql = text(...)
- await session.execute(create_sql)
- ...
- await session.commit()
+ # Idempotent seed for any devices missing a health row
+ await session.execute(populate_sql)
+ await session.commit()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@migrations/004_add_device_health.py` around lines 61 - 95, The migration
currently commits after creating the table/indexes and returns early if an
`existing` check is truthy, which can leave devices unseeded on partial
failures; instead always run the idempotent seed (the `populate_sql` INSERT ...
ON CONFLICT DO NOTHING) and perform a single commit at the end. Concretely:
remove or bypass the early-return that checks `existing` so that `populate_sql`
is executed every run, and consolidate commits by calling `await
session.commit()` only once after both `session.execute(create_sql)` and
`session.execute(populate_sql)` complete (references: create_sql, populate_sql,
session.execute(...), session.commit()).
- modbus_read_exception: warning → debug - modbus_read_failed: error → debug - device_failure_gateway_reset: warning → debug Keeps polling_target_failed as visible warning summary per failed cycle. Reduces log spam while preserving observability of overall failures.
Moved exception handler installation into lifespan startup where a running loop exists; removed module-level get_event_loop() call that triggered DeprecationWarning in Python 3.12.
…usage - pytest.ini: filterwarnings ignore RuntimeWarning - conftest.py: import asyncio first; add RuntimeWarning filter - test_poller.py: use fixtures in test_poll_single_target_invalid_config instead of creating inline AsyncMock() Resolves 3 unrelated warnings around test infrastructure.
Related Issue
No GitHub issue was created for this work — this is a rapid-response fix for a production-impacting bug where a single offline Modbus device was poisoning the shared RTU-over-TCP gateway, causing cascading failures across all devices on that gateway. The fix was implemented directly on the branch to restore service stability, and this PR documents the change for review and merge.
Summary
Fixes a critical Modbus polling failure cascade caused by a single offline device (slave 5 on gateway 10.19.20.148:8899). Root cause: shared RTU-over-TCP gateway + persistent connection meant one bad slave's timeout/stale response corrupted the buffer for subsequent slave polls.
This change introduces per-device health tracking and quarantine to isolate failing devices and protect healthy ones.
Key changes:
device_healthtable + migration 004 with device_id FK, state enum, failure counters, and quarantine_until timestampModbusTcpClient→AsyncModbusTcpClient; added gateway-level cooldown (1s after failures)debug_modbus.pyconverted to async for consistencyScope
Included
c0e3397→d3a4867)Deferred / Future Work
Testing
Unit tests: (all pass)
pytest -q tests/unit/ # Result: 15/15 passedManual / mock testing performed:
Commands to reproduce: (from project root)
source venv/bin/activate pytest -q tests/unit/test_poller.py pytest -q tests/unit/test_modbus_client.py pytest -q tests/unit/test_crud.pyReviewer Focus
Please pay special attention to:
await? Any lingering sync calls?Notes
Migration required
Restart required
device_healthstate.Logging changes
pymodbusloggers set to CRITICAL to suppress connection retry noise. Review if this impacts observability.Potential improvements
app/core/config.pysettingsGET /admin/device-healthsuggested)record_device_success()resets backoff entirely — alternative is decay-based resetApproval checklist:
Summary by CodeRabbit
New Features
Improvements