Use timeout-triggered reconnect with command queue and replay - #2001
Use timeout-triggered reconnect with command queue and replay#2001Mark-Hetherington wants to merge 1 commit into
Conversation
GitHub Actions
Failure
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a FIFO OCPP command queue that captures timed-out calls, coalesces selected commands, triggers reconnects, and replays queued commands. OCPP 1.6 and 2.0.1 request paths now use shared timeout handling with coverage for queue and reconnect behavior. ChangesTimeout queue and replay
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ChargePoint
participant CommandQueue
participant ChargerConnection
ChargePoint->>ChargerConnection: call(request)
ChargerConnection-->>ChargePoint: asyncio.TimeoutError
ChargePoint->>CommandQueue: enqueue(QueuedCommand)
ChargePoint->>ChargerConnection: reconnect()
ChargePoint->>CommandQueue: dequeue_all()
ChargePoint->>CommandQueue: QueuedCommand.execute()
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ 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 |
6a0c036 to
4d3aa2a
Compare
GitHub Actions
Failure
4d3aa2a to
d96b48e
Compare
GitHub Actions
Failure
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (1)
tests/test_ocpp_timeout_and_queue.py (1)
139-171: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftExercise the reconnect-and-replay path.
This test verifies queueing only; it never calls
reconnect()or executes the queued command. Add coverage for a first-call timeout followed by a successful replay, plus a replay timeout that remains queued for the next reconnect.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_ocpp_timeout_and_queue.py` around lines 139 - 171, Extend test_timeout_triggers_reconnect to invoke reconnect() after the initial timeout and execute the queued command with a successful mocked call, asserting the replay completes and the queue is emptied; then simulate a timeout during replay, assert it raises and remains queued for the next reconnect. Preserve the existing stop assertion and use the ChargePoint reconnect-and-queue symbols already exercised by the test.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@custom_components/ocpp/chargepoint.py`:
- Around line 614-621: Update the reconnect handling around
_reconnect_in_progress so a timeout for a command queued for replay does not
propagate as the normal terminal failure. Return or raise the established
distinct queued/pending outcome, and ensure OCPP 1.6 callers handle that outcome
without returning False or sending a permanent-failure notification.
- Around line 574-580: Remove the final _command_queue.clear() call from the
command replay flow after dequeue_all() and command execution. Preserve commands
requeued by _call_with_timeout_handling() or added concurrently while replaying,
while retaining the existing dequeue and execution behavior.
- Line 631: Update reconnect() so super().start() activates the OCPP receiver
before _replay_queue() processes queued commands. Run the replay only after the
receiver is active, using a separate task or equivalent sequencing, while
preserving queued command execution through self.call().
In `@custom_components/ocpp/command_queue.py`:
- Around line 41-47: Update the coalescing logic in the command queue’s
call_type handling to distinguish charging-profile commands by their profile
purpose, preserving separate TxProfile and TxDefaultProfile entries for the same
connector. Extend the coalescing key with the relevant purpose field while
retaining coalescing for identical command types, connectors, and purposes.
In `@custom_components/ocpp/ocppv16.py`:
- Around line 423-425: Update start_transaction() and stop_transaction() so
their RemoteStartTransaction and RemoteStopTransaction requests use
_call_with_timeout_handling() instead of self.call(), passing the appropriate
connector_id metadata as done for SetChargingProfile. Preserve each method’s
existing request and response handling.
---
Nitpick comments:
In `@tests/test_ocpp_timeout_and_queue.py`:
- Around line 139-171: Extend test_timeout_triggers_reconnect to invoke
reconnect() after the initial timeout and execute the queued command with a
successful mocked call, asserting the replay completes and the queue is emptied;
then simulate a timeout during replay, assert it raises and remains queued for
the next reconnect. Preserve the existing stop assertion and use the ChargePoint
reconnect-and-queue symbols already exercised by the test.
🪄 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 Plus
Run ID: 53fbfb98-1a82-421c-acda-4539bcb160f2
📒 Files selected for processing (5)
custom_components/ocpp/chargepoint.pycustom_components/ocpp/command_queue.pycustom_components/ocpp/ocppv16.pycustom_components/ocpp/ocppv201.pytests/test_ocpp_timeout_and_queue.py
| if not self._reconnect_in_progress: | ||
| self._reconnect_in_progress = True | ||
| try: | ||
| await self.stop() | ||
| finally: | ||
| self._reconnect_in_progress = False | ||
|
|
||
| raise |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Do not expose queued timeouts as terminal failures.
After queueing, Line 621 re-raises the timeout. OCPP 1.6 callers then return False and may notify a failure, even though the command is pending replay. Return or raise a distinct queued/pending outcome and handle it without a permanent-failure notification.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@custom_components/ocpp/chargepoint.py` around lines 614 - 621, Update the
reconnect handling around _reconnect_in_progress so a timeout for a command
queued for replay does not propagate as the normal terminal failure. Return or
raise the established distinct queued/pending outcome, and ensure OCPP 1.6
callers handle that outcome without returning False or sending a
permanent-failure notification.
d96b48e to
af1d5ed
Compare
GitHub Actions
Failure
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
custom_components/ocpp/command_queue.py (1)
41-54: 🎯 Functional Correctness | 🟠 MajorPreserve charging-profile purpose when coalescing queued commands.
The queue currently treats
TxProfileandTxDefaultProfileas interchangeable when they target the same connector, so replay can silently lose an active-session charging update.
custom_components/ocpp/command_queue.py#L41-L54: include charging-profile purpose in the coalescing key.custom_components/ocpp/ocppv16.py#L423-L425: pass purpose metadata for custom profiles.custom_components/ocpp/ocppv16.py#L544-L546: identify this request asTxProfile.custom_components/ocpp/ocppv16.py#L571-L573: identify this request asTxDefaultProfile.tests/test_ocpp_timeout_and_queue.py#L44-L74: test same-purpose replacement and different-purpose retention.tests/test_ocpp_timeout_and_queue.py#L175-L209: update current-value coalescing fixtures with explicit purpose metadata.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/command_queue.py` around lines 41 - 54, The command queue must coalesce charging-profile commands by both connector and profile purpose, so update the filtering logic around the queue coalescing block in custom_components/ocpp/command_queue.py. In custom_components/ocpp/ocppv16.py, pass purpose metadata for custom profiles and mark the relevant requests as TxProfile or TxDefaultProfile. Update tests/test_ocpp_timeout_and_queue.py to verify same-purpose replacement, different-purpose retention, and explicit purpose metadata in current-value fixtures.
🧹 Nitpick comments (4)
custom_components/ocpp/command_queue.py (1)
5-5: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
collections.abc.Callablefor the annotation.Ruff UP035 flags the
typing.Callableimport; update it, or verify that the repository’s target/lint configuration intentionally permits the alias.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/command_queue.py` at line 5, Update the Callable import in command_queue.py to use collections.abc.Callable instead of typing.Callable, unless the repository configuration explicitly permits the typing alias; keep the existing annotations and Any import unchanged.Source: Linters/SAST tools
custom_components/ocpp/ocppv16.py (1)
605-605: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse one timeout exception spelling consistently.
Ruff UP041 flags
asyncio.TimeoutError; replace it with builtinTimeoutErrorwhere the declared Python target supports it, or verify the target and lint configuration before changing both sites.
custom_components/ocpp/ocppv16.py#L605-L605: update the production exception handler.tests/test_ocpp_timeout_and_queue.py#L149-L162: update the timeout fixture and assertion.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/ocppv16.py` at line 605, Use the builtin TimeoutError consistently for the production handler in custom_components/ocpp/ocppv16.py (605-605) and the timeout fixture and assertion in tests/test_ocpp_timeout_and_queue.py (149-162), after confirming the declared Python target and lint configuration support it.Source: Linters/SAST tools
tests/test_ocpp_timeout_and_queue.py (1)
137-173: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover replay, not only enqueue.
This test verifies that a timeout queues a command and triggers
stop(), but never exercisesQueuedCommand.execute()or reconnect draining. Add coverage for FIFO replay, successful execution after reconnect, requeue behavior on a second timeout, and suppression of duplicatestop()calls while reconnect is already in progress.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_ocpp_timeout_and_queue.py` around lines 137 - 173, Expand TestTimeoutHandling.test_timeout_triggers_reconnect or add focused tests that invoke QueuedCommand.execute() during reconnect draining, verifying queued commands replay in FIFO order and complete successfully after reconnect. Also cover a second timeout requeueing the command and ensure _call_with_timeout_handling does not call stop() again when _reconnect_in_progress is already true.custom_components/ocpp/chargepoint.py (1)
599-599: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueHoist
QueuedCommandimport to module level.
CommandQueueis already imported at the top of the file (Line 39); importQueuedCommandalongside it instead of re-importing inside the exception branch on every timeout.Proposed fix
-from .command_queue import CommandQueue +from .command_queue import CommandQueue, QueuedCommand- from .command_queue import QueuedCommand - cmd = QueuedCommand(🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/chargepoint.py` at line 599, Move the QueuedCommand import from the timeout exception branch in the relevant chargepoint command-handling method to the module-level imports, alongside CommandQueue. Remove the now-redundant local import while preserving the existing QueuedCommand usage and timeout behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@custom_components/ocpp/chargepoint.py`:
- Around line 601-621: Update CommandQueue.enqueue’s command coalescing list to
include the OCPP 2.0.1 call types RequestStartTransaction and
RequestStopTransaction alongside the existing transaction start/stop commands,
so repeated timeouts replace stale queued replies rather than accumulating them.
---
Duplicate comments:
In `@custom_components/ocpp/command_queue.py`:
- Around line 41-54: The command queue must coalesce charging-profile commands
by both connector and profile purpose, so update the filtering logic around the
queue coalescing block in custom_components/ocpp/command_queue.py. In
custom_components/ocpp/ocppv16.py, pass purpose metadata for custom profiles and
mark the relevant requests as TxProfile or TxDefaultProfile. Update
tests/test_ocpp_timeout_and_queue.py to verify same-purpose replacement,
different-purpose retention, and explicit purpose metadata in current-value
fixtures.
---
Nitpick comments:
In `@custom_components/ocpp/chargepoint.py`:
- Line 599: Move the QueuedCommand import from the timeout exception branch in
the relevant chargepoint command-handling method to the module-level imports,
alongside CommandQueue. Remove the now-redundant local import while preserving
the existing QueuedCommand usage and timeout behavior.
In `@custom_components/ocpp/command_queue.py`:
- Line 5: Update the Callable import in command_queue.py to use
collections.abc.Callable instead of typing.Callable, unless the repository
configuration explicitly permits the typing alias; keep the existing annotations
and Any import unchanged.
In `@custom_components/ocpp/ocppv16.py`:
- Line 605: Use the builtin TimeoutError consistently for the production handler
in custom_components/ocpp/ocppv16.py (605-605) and the timeout fixture and
assertion in tests/test_ocpp_timeout_and_queue.py (149-162), after confirming
the declared Python target and lint configuration support it.
In `@tests/test_ocpp_timeout_and_queue.py`:
- Around line 137-173: Expand
TestTimeoutHandling.test_timeout_triggers_reconnect or add focused tests that
invoke QueuedCommand.execute() during reconnect draining, verifying queued
commands replay in FIFO order and complete successfully after reconnect. Also
cover a second timeout requeueing the command and ensure
_call_with_timeout_handling does not call stop() again when
_reconnect_in_progress is already true.
🪄 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 Plus
Run ID: 1c7fed8e-c716-4c1e-be76-44b6843f70f9
📒 Files selected for processing (5)
custom_components/ocpp/chargepoint.pycustom_components/ocpp/command_queue.pycustom_components/ocpp/ocppv16.pycustom_components/ocpp/ocppv201.pytests/test_ocpp_timeout_and_queue.py
🚧 Files skipped from review as they are similar to previous changes (1)
- custom_components/ocpp/ocppv201.py
af1d5ed to
0f410c7
Compare
GitHub Actions
Failure
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (3)
custom_components/ocpp/chargepoint.py (3)
572-580: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winTrailing
clear()still wipes commands re-queued during replay.
dequeue_all()(line 574) already empties_command_queue. If a replayedcmd.execute()times out again,_call_with_timeout_handling()re-enqueues it — then line 580 immediately deletes it, along with anything enqueued concurrently while replay was running. This is unresolved from a prior review round.Proposed fix
async def _replay_queue(self) -> None: """Replay queued commands from last timeout/disconnect.""" commands = await self._command_queue.dequeue_all() for cmd in commands: try: await cmd.execute() except Exception as ex: _LOGGER.warning("Replay of %s failed: %s", cmd.call_type, ex) - await self._command_queue.clear()Given the ordering issue flagged below (replay runs before the receiver is listening), every replayed command will time out again on first reconnect attempt, get re-queued, and then be wiped by this
clear()— meaning the whole retry mechanism this PR adds is currently a no-op after the first timeout.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/chargepoint.py` around lines 572 - 580, The trailing _command_queue.clear() in _replay_queue must be removed so commands re-enqueued by cmd.execute() and commands added concurrently remain queued for later retry. Preserve dequeue_all() and the existing replay loop/error logging.
623-633: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReplay still runs before the OCPP receive loop is active.
_replay_queue()(line 631) executes queued calls viaself.call(), which awaits a response, butsuper().start()— the loop that reads incoming websocket frames and resolves those awaits — only starts afterward viaself.run(...)(line 633). Every replayed command will therefore time out again (sequentially, compounding the queue's configured timeout per item) before the connection even starts listening. Combined with the trailingclear()issue above, replayed commands are effectively guaranteed to be lost. This is unresolved from a prior review round.Proposed fix
self._metrics[(0, cstat.reconnects.value)].value += 1 - await self._replay_queue() # post connect now handled on receiving boot notification or with backstop in monitor connection - await self.run([super().start(), self.monitor_connection()]) + await self.run( + [super().start(), self.monitor_connection(), self._replay_queue()] + )Running replay as a gathered task alongside the receiver ensures responses can actually be matched to the queued calls.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/chargepoint.py` around lines 623 - 633, Update reconnect so the OCPP receive loop from super().start() runs concurrently with _replay_queue(), rather than awaiting the replay before starting the loop. Use the existing self.run orchestration to start both tasks together, while preserving monitor_connection and the current reconnect setup order.
619-621: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftQueued timeout still surfaces as a terminal failure to callers.
After enqueuing for replay, line 621 re-raises the
TimeoutErrorunchanged. OCPP 1.6/2.0.1 callers (e.g.set_availability,set_charge_rate) catch this and returnFalse/notify failure, even though the command is pending replay. This is unresolved from a prior review round; a distinct queued/pending outcome is still needed so callers don't report a permanent failure for a command that will be retried.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@custom_components/ocpp/chargepoint.py` around lines 619 - 621, The timeout path in the surrounding reconnect/replay logic must return a distinct queued or pending outcome after successfully enqueuing the command, rather than re-raising TimeoutError. Update the caller-facing handling used by OCPP 1.6/2.0.1 operations such as set_availability and set_charge_rate to recognize that outcome without reporting permanent failure, while preserving exception propagation for timeouts that are not queued.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@custom_components/ocpp/chargepoint.py`:
- Around line 613-619: Update the reconnect/timeout handling around self.stop()
and the surrounding call() flow so connection failures from sibling in-flight
OCPP operations are caught and queued for replay before teardown, or ensure
stopping the connection does not cancel unrelated self.tasks. Preserve replay
behavior for both TimeoutError and connection-closed/WebSocketException-style
failures.
---
Duplicate comments:
In `@custom_components/ocpp/chargepoint.py`:
- Around line 572-580: The trailing _command_queue.clear() in _replay_queue must
be removed so commands re-enqueued by cmd.execute() and commands added
concurrently remain queued for later retry. Preserve dequeue_all() and the
existing replay loop/error logging.
- Around line 623-633: Update reconnect so the OCPP receive loop from
super().start() runs concurrently with _replay_queue(), rather than awaiting the
replay before starting the loop. Use the existing self.run orchestration to
start both tasks together, while preserving monitor_connection and the current
reconnect setup order.
- Around line 619-621: The timeout path in the surrounding reconnect/replay
logic must return a distinct queued or pending outcome after successfully
enqueuing the command, rather than re-raising TimeoutError. Update the
caller-facing handling used by OCPP 1.6/2.0.1 operations such as
set_availability and set_charge_rate to recognize that outcome without reporting
permanent failure, while preserving exception propagation for timeouts that are
not queued.
🪄 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 Plus
Run ID: b4813901-f5e0-44f5-bae3-7d405286c864
📒 Files selected for processing (5)
custom_components/ocpp/chargepoint.pycustom_components/ocpp/command_queue.pycustom_components/ocpp/ocppv16.pycustom_components/ocpp/ocppv201.pytests/test_ocpp_timeout_and_queue.py
🚧 Files skipped from review as they are similar to previous changes (4)
- custom_components/ocpp/ocppv16.py
- custom_components/ocpp/command_queue.py
- custom_components/ocpp/ocppv201.py
- tests/test_ocpp_timeout_and_queue.py
|
|
||
| if not self._reconnect_in_progress: | ||
| self._reconnect_in_progress = True | ||
| try: | ||
| await self.stop() | ||
| finally: | ||
| self._reconnect_in_progress = False |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect the installed ocpp library's call() implementation to confirm what
# exception type is raised when the underlying connection closes mid-await.
python - <<'PY'
try:
import ocpp
print(ocpp.__file__)
except Exception as e:
print("ocpp not installed in sandbox:", e)
PY
fd -a 'charge_point\.py$' 2>/dev/null | xargs -r grep -n "async def call" -A 30Repository: lbbrhzn/ocpp
Length of output: 204
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== repository files matching chargepoint/ocpp =="
git ls-files | rg '(^|/)chargepoint\.py$|(^|/)charge_point\.py$|requirements.*|pyproject\.toml|setup\.cfg|setup\.py|README'
echo
echo "== chargepoint.py context around lines 560-660 =="
sed -n '560,660p' custom_components/ocpp/chargepoint.py | nl -ba -v560
echo
echo "== ocpp dependency versions == "
for f in requirements*.txt requirements/*.txt requirements/*.* requirements-custom.txt manifest.json pyproject.toml setup.py setup.cfg poetry.lock tasks.yaml; do
[ -f "$f" ] && { echo "--- $f"; rg -n "ocpp|websockets" "$f" || true; }
done
echo
echo "== imports/usages of call_with_timeout_handling/tasks/_tasks/call =="
rg -n "_call_with_timeout_handling|call_with_timeout_handling|self\.tasks|self\._tasks|QueueExecutor|reconnect_in_progress|async def call\(|def call\(" custom_components/ocpp -C 3Repository: lbbrhzn/ocpp
Length of output: 402
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== chargepoint.py context around timeout teardown =="
awk 'NR>=560 && NR<=680 {printf "%6d\t%s\n", NR, $0}' custom_components/ocpp/chargepoint.py
echo
echo "== queue/replay helpers and call definitions =="
awk 'NR>=1 && NR<=560 {printf "%6d\t%s\n", NR, $0}' custom_components/ocpp/chargepoint.py
echo
echo "== dependency files =="
sed -n '1,220p' requirements.txt
if [ -f custom_components/ocpp/manifest.json ]; then sed -n '1,220p' custom_components/ocpp/manifest.json; fi
if [ -f setup.cfg ]; then sed -n '1,220p' setup.cfg; fi
echo
echo "== exact dependency mentions =="
rg -n "ocpp|websockets" requirements.txt docs/requirements.txt custom_components/ocpp/manifest.json setup.cfg 2>/dev/null || trueRepository: lbbrhzn/ocpp
Length of output: 30941
🌐 Web query:
OCPP library v1.6 Python call function websocket exception raises when connection closed asyncio timeout
💡 Result:
When using the ocpp library, connection closures—whether due to network issues, keepalive timeouts, or graceful shutdowns—are commonly surfaced as websockets.exceptions.ConnectionClosed (or its subclasses, such as ConnectionClosedError) [1][2][3]. Because the ChargePoint.start method runs an internal loop that awaits incoming messages, these exceptions will propagate to your code if not caught, often leading to unhandled errors or application crashes [2][4]. To handle these exceptions, you must wrap the call to ChargePoint.start in a try/except block [1][2][4]. Handling Connection Closures: Wrap your start routine in the connection handler: async def on_connect(websocket): charge_point_id = websocket.path.strip("/") cp = ChargePoint(charge_point_id, websocket) try: await cp.start except websockets.exceptions.ConnectionClosed: # Handle the disconnection (e.g., log, cleanup, or trigger reconnection) print(f"Connection for {charge_point_id} closed.") finally: # Perform necessary cleanup pass If you are experiencing premature timeouts, it is often due to the websockets library's default keepalive mechanism, which sends pings every 20 seconds and expects a response within 20 seconds [3][5]. If the client (Charging Station) fails to respond, the server will raise a ConnectionClosedError [3][5]. You can adjust this by configuring the ping_interval and ping_timeout arguments when setting up the websocket server [3]. If you are encountering errors when attempting to close the connection manually, ensure you are properly awaiting the close coroutine: await websocket.close Additionally, note that recent versions of the websockets library have changed how the path is accessed. If you encounter a TypeError regarding missing arguments, access the path directly from the websocket object (e.g., websocket.path or websocket.request.path) rather than expecting it as a function argument [6].
Citations:
- 1: Can I catch websockets exception? mobilityhouse/ocpp#11
- 2: Disconnect will show error in the console mobilityhouse/ocpp#234
- 3: https://websockets.readthedocs.io/en/stable/faq/common.html
- 4: mobilityhouse/ocpp@d6b003d
- 5: Charge Point reconnects every 40 seconds mobilityhouse/ocpp#79
- 6: Getting TypeError: on_connect() missing 1 required positional argument: 'path'. V1.6 ocpp mobilityhouse/ocpp#708
🌐 Web query:
OCPP Python ocpp 2.1.0 Call call async close websocket exception ConnectionClosedError
💡 Result:
In the Python ocpp library (including version 2.1.0), websockets.exceptions.ConnectionClosedError occurs when the underlying WebSocket connection terminates unexpectedly, often due to network issues, timeouts, or the peer closing the connection without a proper handshake [1][2][3]. Since ocpp relies on the websockets library for transport, this exception bubbles up when you attempt to send or receive messages over a stale or broken connection [4][5][2]. To handle this exception, you must wrap the ChargePoint.start() call (or any high-level communication loop) in a try...except block [6][7]. ### Implementation Pattern When implementing a Charge Point (client) or Central System (server), you should structure your connection logic to catch the ConnectionClosedError to initiate graceful recovery or clean-up [6][4][7]. python import asyncio import websockets from ocpp.v201 import ChargePoint as cp # Example for OCPP 2.0.1 async def on_connect(websocket, path): charge_point_id = path.strip('/') charge_point = ChargePoint(charge_point_id, websocket) try: await charge_point.start() except websockets.exceptions.ConnectionClosedError as e: # Log the error and perform cleanup print(f"Connection closed abnormally: {e}") finally: # Ensure resources are released await websocket.close() ### Common Causes 1. Keepalive Timeout: The websockets library has a default ping_interval and ping_timeout (often 20 seconds each) [8][1]. If the peer (e.g., a physical Charge Point) stops responding to these pings, the library will raise a ConnectionClosedError with an internal error code [8][1]. 2. Network Instability: Unstable network connections often lead to TCP resets, which the websockets library perceives as an abnormal connection closure [1]. 3. Server/Client Resource Cleanup: If you manually call .close() on a websocket, ensure it is awaited correctly [6]. In custom integration logic, failing to set the connection object to None after a failure can lead to AttributeError or repeated attempts to use a dead connection [9]. ### Best Practices * Use a robust reconnection strategy: Wrap your start() method in a loop that handles retries, potentially with exponential backoff, rather than simply exiting [4]. * Logging: Configure the ocpp logger to DEBUG level to capture detailed transport-layer messages which can help distinguish between protocol-level errors and transport-level connection failures [8]. * Clean Shutdown: Always use a finally block to ensure that await websocket.close() is called and internal state is reset, preventing the application from attempting to route messages through a closed socket [9][7].
Citations:
- 1: https://websockets.readthedocs.io/en/stable/faq/common.html
- 2: https://websockets.readthedocs.io/en/16.0/reference/exceptions.html
- 3: https://dokk.org/documentation/python-websockets/9.1/api/utilities/
- 4: Offline behaviour. Reconnecting client properly troubles mobilityhouse/ocpp#164
- 5: https://stackoverflow.com/questions/70566624/what-causes-the-web-socket-connection-loss
- 6: Disconnect will show error in the console mobilityhouse/ocpp#234
- 7: Can I catch websockets exception? mobilityhouse/ocpp#11
- 8: Charge Point reconnects every 40 seconds mobilityhouse/ocpp#79
- 9: Only route messages if connection is not None mobilityhouse/ocpp#248
🌐 Web query:
OCPP python library call method implementation self._connection.close send wait_for unique id exception ConnectionClosedError
💡 Result:
The ConnectionClosedError (or ConnectionClosed) exception in the Python ocpp library typically occurs because the underlying WebSocket connection has been terminated unexpectedly—such as through a network failure, a timeout (e.g., keepalive ping timeout), or a deliberate closure by the peer [1][2][3]. When using the ocpp library, the ChargePoint.call() method (which uses _send and wait_for internally) relies on an active WebSocket connection. If the connection drops during this process, the websockets library raises a ConnectionClosedError [4][5]. ### Key Implementation and Debugging Considerations: 1. Handle Exceptions at the Call Site Because ChargePoint.start() or your individual call() operations may encounter this error, you should wrap your high-level communication logic in a try...except block to catch websockets.exceptions.ConnectionClosedError [6][7][8]. python import websockets # ... try: response = await cp.call(boot_notification_payload) except websockets.exceptions.ConnectionClosedError as e: # Handle the disconnection, log it, and trigger a reconnection logic logger.error(f"Connection lost: {e}") 2. Avoiding Deadlocks Early versions of the ocpp library had issues where self._call_lock would not release if a ConnectionClosedError occurred during a call(), preventing subsequent messages from being sent [4]. Ensure you are using an up-to-date version of the library. If you are overriding internal methods, ensure that any lock or resource cleanup is placed in a finally block [4][7]. 3. Graceful Closure If you intend to close the connection manually, always ensure you await self._connection.close() properly [6][7][8]. If your application needs to handle reconnections, it is common to: * Set a flag or event indicating the connection is down [6]. * Clean up the existing _connection (set it to None) [7]. * Re-initialize or re-instantiate the ChargePoint logic once a new WebSocket connection is established [6][7]. 4. Identifying the Cause If you see ConnectionClosedError frequently: * Check for keepalive/ping timeouts: The default websockets timeout is often 20 seconds. If your network or environment has high latency, you may need to adjust ping_interval and ping_timeout in your WebSocket configuration [1][3]. * Enable logging: Configure the ocpp logger to DEBUG level to see raw messages and detect if the server is actively sending close frames [1]. If you are encountering this during wait_for specifically, it indicates the response for the unique ID associated with the request was never received because the socket closed before the promise could be fulfilled [4]. You must treat the communication as failed and decide whether to retry the operation after re-establishing the connection [6][7].
Citations:
- 1: Charge Point reconnects every 40 seconds mobilityhouse/ocpp#79
- 2: https://stackoverflow.com/questions/70566624/what-causes-the-web-socket-connection-loss
- 3: https://websockets.readthedocs.io/en/stable/faq/common.html
- 4: self._call_lock never released when trying to send payload on a closed socket mobilityhouse/ocpp#46
- 5: https://websockets.readthedocs.io/en/16.0/reference/exceptions.html
- 6: Offline behaviour. Reconnecting client properly troubles mobilityhouse/ocpp#164
- 7: Only route messages if connection is not None mobilityhouse/ocpp#248
- 8: Disconnect will show error in the console mobilityhouse/ocpp#234
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== command queue implementation =="
rg -n "class CommandQueue|async def __init__|async def enqueue|async def dequeue_all|async def run|async def start|reconnect" custom_components/ocpp/command_queue.py -C 4
echo
echo "== command_queue.py =="
awk '{printf "%6d\t%s\n", NR, $0}' custom_components/ocpp/command_queue.pyRepository: lbbrhzn/ocpp
Length of output: 4511
Preserve in-flight OCPP messages during timeout teardown.
One call dropping into the TimeoutError branch calls self.stop(), which cancels self.tasks; sibling OCPP operations awaiting self.call() then fail with a connection-closed/WebSocketException-style error instead and are never queued for replay. Catch connection failures around call() as well and enqueue them before teardown, or otherwise avoid canceling unrelated in-flight messages while stopping the connection.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@custom_components/ocpp/chargepoint.py` around lines 613 - 619, Update the
reconnect/timeout handling around self.stop() and the surrounding call() flow so
connection failures from sibling in-flight OCPP operations are caught and queued
for replay before teardown, or ensure stopping the connection does not cancel
unrelated self.tasks. Preserve replay behavior for both TimeoutError and
connection-closed/WebSocketException-style failures.
0f410c7 to
9da7f1d
Compare
GitHub Actions
Failure
|
I've applied changes based on Coderabbit feedback, and I'm running them on my own Home Assistant set up to see if any issues crop up. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@custom_components/ocpp/ocppv16.py`:
- Around line 499-501: The SetChargingProfile handler must stop issuing sibling
profile calls after _call_with_timeout_handling() times out and disconnects.
Update the surrounding TxProfile and TxDefaultProfile flows, including the
corresponding calls at the referenced locations, to return immediately on the
timeout path or queue the remaining profiles without using the stopped
transport; preserve normal sequential calls when no timeout occurs.
- Around line 667-669: Update start_transaction() and stop_transaction() to
prevent timed-out RemoteStartTransaction and RemoteStopTransaction calls from
being queued for _replay_queue() execution. Exclude these non-idempotent
commands from generic replay, or reconcile transaction state after reconnect
before retrying, while preserving timeout handling for other calls.
🪄 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 Plus
Run ID: c69fdb1a-c7cf-48e9-b842-45fb6a74d028
📒 Files selected for processing (5)
custom_components/ocpp/chargepoint.pycustom_components/ocpp/command_queue.pycustom_components/ocpp/ocppv16.pycustom_components/ocpp/ocppv201.pytests/test_ocpp_timeout_and_queue.py
🚧 Files skipped from review as they are similar to previous changes (4)
- custom_components/ocpp/command_queue.py
- tests/test_ocpp_timeout_and_queue.py
- custom_components/ocpp/ocppv201.py
- custom_components/ocpp/chargepoint.py
9da7f1d to
492986c
Compare
GitHub Actions
Inactive
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #2001 +/- ##
==========================================
+ Coverage 94.96% 95.28% +0.31%
==========================================
Files 12 13 +1
Lines 2962 3032 +70
==========================================
+ Hits 2813 2889 +76
+ Misses 149 143 -6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@custom_components/ocpp/chargepoint.py`:
- Around line 617-629: Update the coalesced-command list in command_queue.py to
include RequestStartTransaction and RequestStopTransaction alongside the
existing transaction commands. Ensure the helper queuing via
_call_with_timeout_handling uses these entries so repeated OCPP 2.0.1
transaction requests replace stale queued replies.
🪄 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 Plus
Run ID: 5534668f-a4e4-42fe-84e7-0d978edc5e37
📒 Files selected for processing (5)
custom_components/ocpp/chargepoint.pycustom_components/ocpp/command_queue.pycustom_components/ocpp/ocppv16.pycustom_components/ocpp/ocppv201.pytests/test_ocpp_timeout_and_queue.py
🚧 Files skipped from review as they are similar to previous changes (4)
- custom_components/ocpp/command_queue.py
- custom_components/ocpp/ocppv16.py
- tests/test_ocpp_timeout_and_queue.py
- custom_components/ocpp/ocppv201.py
| cmd = QueuedCommand( | ||
| call_type=call_type, | ||
| call_fn=self._call_with_timeout_handling, | ||
| args=(req,), | ||
| kwargs={ | ||
| "call_type": call_type, | ||
| "connector_id": connector_id, | ||
| **call_kwargs, | ||
| }, | ||
| connector_id=connector_id, | ||
| profile_purpose=profile_purpose, | ||
| ) | ||
| await self._command_queue.enqueue(cmd) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n 'RequestStartTransaction|RequestStopTransaction|RemoteStartTransaction|RemoteStopTransaction' custom_components/ocpp/command_queue.pyRepository: lbbrhzn/ocpp
Length of output: 239
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== command_queue.py relevant outlines =="
ast-grep outline custom_components/ocpp/command_queue.py --view expanded || true
echo
echo "== command_queue.py around coalescing =="
sed -n '1,120p' custom_components/ocpp/command_queue.py
echo
echo "== chargepoint.py helper context =="
sed -n '580,645p' custom_components/ocpp/chargepoint.py
echo
echo "== ocppv201.py transaction helpers =="
rg -n 'def (start|stop)_transaction|RequestStartTransaction|RequestStopTransaction|_call_with_timeout_handling' custom_components/ocpp/ocppv201.py custom_components/ocpp/chargepoint.pyRepository: lbbrhzn/ocpp
Length of output: 8410
Add OCPP 2.0.1 transaction messages to the coalesce list
command_queue.py still coalesces only SetChargingProfile, RemoteStartTransaction, and RemoteStopTransaction, while ocppv201.py queues OCPP 2.0.1 RequestStartTransaction/RequestStopTransaction through this helper. Repeat transaction timeouts can therefore queue duplicate stale replies for replay; include both RequestStartTransaction and RequestStopTransaction in the coalesced list.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@custom_components/ocpp/chargepoint.py` around lines 617 - 629, Update the
coalesced-command list in command_queue.py to include RequestStartTransaction
and RequestStopTransaction alongside the existing transaction commands. Ensure
the helper queuing via _call_with_timeout_handling uses these entries so
repeated OCPP 2.0.1 transaction requests replace stale queued replies.
492986c to
315fdd0
Compare
GitHub Actions
Inactive
On asyncio.TimeoutError place failed commands in queue. Newer commands replace commands of the same type. On reconnect replay queued commands Changes: - New CommandQueue class with FIFO ordering and coalescing logic - New _call_with_timeout_handling() wrapper in ChargePoint to catch timeouts - Modified reconnect() to drain queue via _replay_queue() before re-establishing connection - Add _command_queue instance to ChargePoint.__init__ Integration: - ocppv16.py: wrap set_charge_rate() (4 calls) and set_availability() with timeout handler - ocppv201.py: wrap set_charge_rate() (2 calls), set_availability() (3 calls), start_transaction(), and stop_transaction() with timeout handler - Add asyncio import to ocppv16.py for exception handling Fixes: lbbrhzn#1999
315fdd0 to
6ca40a3
Compare
GitHub Actions
Inactive
1 participant
On asyncio.TimeoutError place failed commands in queue. Newer commands replace commands of the same type. On reconnect replay queued commands
Changes:
Integration:
Fixes: #1999
Summary by CodeRabbit
New Features
Bug Fixes
Tests