Add connection pool with failover and keepalive - #4
Conversation
|
Hey Balake, thanks for the guidance posted on the HA PR thread. Opened this PR as you suggested. To answer your questions: |
Moved connection pool, failover, keepalive, and command throttling logic into pydeako (DeakoLights/pydeako#4). The HA integration now delegates to DeakoConnectionPool and only handles HA-specific concerns: zeroconf discovery, config flow, device registry, and failover scanning. - Removed translations/en.json (auto-generated in core) - Removed version field from manifest.json - Removed unused same_host error from strings.json Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
| Maintained for backward compatibility. New code should | ||
| prefer peek_address() or get_all_addresses(). |
There was a problem hiding this comment.
Can you explain this? Why? Deako devices are safe to use as singular connections, meaning if you use an address, it should not be available to use again as multiple TCP connections to the same device are not well supported. Has that changed?
There was a problem hiding this comment.
You're right, the helpers had no consumers in pydeako or the HA integration. Reverted pydeako/discover/_address_pool.py and its tests to the pre-PR state. The pool handles failover entirely inside DeakoConnectionPool and does not rely on non-destructive peeks at the discovery address pool.
There was a problem hiding this comment.
My overall piece of feedback for this is that you are re-defining much of what is already defined. We have a socket connection util here. Why not use that?
Furthermore, the manager already does keepalive pinging through the integration itself. Why not use that for connection pooling? I'm sure some modifications/hooks are needed, but much of what you're implementing here already exists.
There was a problem hiding this comment.
Thanks for catching this. Two changes:
_KeepAliveSocket now composes _SocketConnection rather than carrying its own socket path. All real socket I/O stays in utils/_socket.py.
The in-pool health monitor is gone. Liveness is detected by _Manager's existing ping loop through a new on_connection_lost callback (optional kwarg on Deako.init, forwarded to _Manager). When a pong is missed, _Manager fires the callback once; the pool schedules a single failover switch. No parallel keepalive loop, single source of truth for "primary is dead."
Also added NoSocketException(OSError) in utils/_socket.py so send failures from the socket wrapper surface as a distinct, catchable type through the manager and pool.
There was a problem hiding this comment.
I still don't understand why you need the _KeepAliveSocket. That doesn't do any of the pinging to validate that the integration is working. You could just use another _Manager and that'll also verify that the integration is working.
| self._keepalive = None | ||
|
|
||
| # Give the bridge time to release the keepalive session | ||
| await asyncio.sleep(BRIDGE_RECYCLE_DELAY_S) |
There was a problem hiding this comment.
Is there no way to determine this besides waiting some arbitrary amount of time?
There was a problem hiding this comment.
Replaced the fixed sleep with a bounded poll loop. A private TCP probe opens a connection to the failover host and closes immediately; a wait helper polls it at a fixed interval up to a module-level timeout before giving up. The pool only proceeds once the bridge is actually accepting connections. No blind sleeps remain in the failover path.
Adds NoSocketException(OSError) in utils/_socket.py and raises it from send_bytes / read_bytes when self.sock is None. Subclassing OSError means callers can keep a single narrow catch covering both real socket send failures (sock_sendall / sock_recv raising OSError) and the "no socket at all" case, without widening to bare Exception. This is a narrow externally observable behavior change: callers that previously relied on send_bytes / read_bytes raising a different exception type will now see NoSocketException. Existing catches on OSError continue to work. Also threads the exception through utils/_connection.py so upstream send-failure paths can distinguish "socket gone" from other errors and react by reconnecting rather than retrying in place. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Adds an optional on_connection_lost callback to _Manager, invoked once from maintain_connection_worker when a pong is missed, before close() and the reconnect task start. The callback is wrapped in a broad try/except logged at WARNING so a buggy callback cannot stall reconnection. This is the single source of truth for primary-connection liveness. External supervisors (for example DeakoConnectionPool) hook here instead of running parallel health loops, addressing review feedback about duplicating existing keepalive logic. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Adds on_connection_lost kwarg to Deako.__init__ and forwards it to the underlying _Manager. Adds Deako.is_connected() as a thin wrapper over the manager's connection state for callers (the pool) that need to check liveness without poking manager internals. Existing Deako(get_address, client_name=...) callers are unaffected. Both additions are optional and backward compatible. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Adds an opt-in DeakoConnectionPool that keeps one bridge as the
active primary and the other warm via a held TCP connection on
port 23 (_KeepAliveSocket, which composes _SocketConnection so
all socket I/O stays in utils/_socket.py).
Failover triggers:
* _Manager's ping loop calls a sync hook wired as Deako's
on_connection_lost, which schedules one failover switch. No
parallel in-pool health monitor.
* A failed send inside control_device (OSError or
NoSocketException from Deako._control_device_strict) triggers
exactly one failover switch and exactly one retry on the new
active before raising NoSocketException.
Switch coordination uses asyncio.Lock + asyncio.Event: the lock
serializes the switch, the event signals completion to in-flight
callers so they either land on the new primary or fail cleanly
on timeout.
Bridge readiness on the new primary is detected with a bounded
TCP-probe poll, capped by a module-level timeout, rather than a
blind sleep.
Public surface:
* DeakoConnectionPool (opt-in; existing Deako usage unchanged).
* ConnectionPoolState (frozen dataclass, five fields:
primary_host, failover_host, primary_connected,
failover_keepalive_active, started).
* NoSocketException (re-exported at pydeako top level;
OSError subclass).
Other behaviors:
* Idempotent start() / stop().
* Sync state callbacks only; async callbacks are out of scope
for this PR.
* Module logger uses __package__ to match the rest of the
library.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
|
Thanks for the excellent review. Heads up, about to force-push this branch to condense from 14 to 4 commits with refactoring to address your review comments. Didn't bump the version but happy to if you'd like. Replies to each comment coming right after the push. |
f9b3baf to
2dc1382
Compare
pjens45
changed the title
Close three stop-race windows in DeakoConnectionPool where a terminal stop() that arrived while a long connect was in flight would still see the freshly connected Deako installed on self.active after the await returned. start(), _switch_to_failover(), and _attempt_recovery() now re-check self._stopped after the connect path returns and, if stopped, disconnect the new Deako under wait_for(STEP_TIMEOUT_S) and bail without mutating the host map. Also enforce the docstring promise that single-bridge pools are unsupported: the constructor now raises ValueError when primary_host == failover_host rather than silently returning a pool whose failover target equals its primary. Adds four pytest cases covering the three stop-race paths and the constructor guard. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
| except OSError as exc: | ||
| _LOGGER.error("Error sending data: %s", exc) | ||
| self.state = ConnectionState.ERROR | ||
| raise |
There was a problem hiding this comment.
Can you explain why you are adding this here?
There was a problem hiding this comment.
The state flip ahead of the re-raise keeps is_connected() accurate for the next pool-level check. The pool's failover logic polls is_connected(), not the exception path, so without the flip the pool could see the socket as still healthy after a failed send. Rewrote the docstring to say that directly instead of just describing the code.
| Narrow catch: only OSError (which includes NoSocketException | ||
| via its OSError inheritance) is caught to flip state to ERROR | ||
| before re-raising. Other exceptions propagate unchanged, so | ||
| programming errors and unrelated failures surface normally. | ||
| Callers get OSError propagation on real send failure. | ||
| """ | ||
| _LOGGER.debug("[%s] Sending data: %s", self.address, data_to_send) | ||
| try: | ||
| await self.socket.send_bytes(str.encode(data_to_send)) | ||
| except Exception as exc: # pylint: disable=broad-exception-caught | ||
| except OSError as exc: |
There was a problem hiding this comment.
Can you better explain this behavioral change? The comment just explains what the code does, not why.
There was a problem hiding this comment.
Fair, the comment was all "what" and no "why". Rewrote it: narrowed from except Exception so programming errors (TypeError on malformed payload, AttributeError on stale refs, etc.) propagate as bugs rather than getting masked as connection-state transitions. Only socket-level failures (OSError, which NoSocketException subclasses) mean the socket is actually dead and warrant the state flip.
| _LOGGER.info("Connecting to %s", self.address) | ||
| address, port = self.address.split(":") | ||
| await self.loop.sock_connect(self.sock, (address, port)) | ||
| await self.loop.sock_connect(self.sock, (address, int(port))) |
There was a problem hiding this comment.
Why does is this type cast necessary?
There was a problem hiding this comment.
self.address.split(":") returns strings, while loop.sock_connect() expects (host, int_port). The cast makes that contract explicit instead of relying on any implementation-specific coercion. Added a one-line inline comment next to the cast so a future reader does not wonder the same thing.
| ) -> None: | ||
| """Add control request to queue. | ||
|
|
||
| Preserves 0.x external behavior per decision 29: any send |
There was a problem hiding this comment.
Sorry, that was a leftover ref to internal design notes. The substance: control_device()'s existing 0.x behavior was to swallow send failures silently. The pool needs a raising version to detect failures and drive failover, so I added _control_device_strict() (raises on failure) as an internal API while keeping the public control_device() swallow-behavior unchanged. Removed the ref and updated the comment to say that directly in a follow-up docs cleanup commit.
| def is_connected(self) -> bool: | ||
| """Return True iff the active primary is currently connected. | ||
|
|
||
| Uses `Deako.is_connected()` (decision 21) so the pool does |
There was a problem hiding this comment.
Another internal doc ref, sorry. The underlying point: the pool checks connection health through Deako.is_connected() rather than reaching through Deako.connection_manager to the raw socket, so the pool stays coupled to Deako's public surface only. Comment rewritten without the ref.
| Returns True iff a TCP connection to ``host:port`` could be | ||
| established within ``timeout`` seconds. This is a reachability | ||
| probe, not a readiness proof; a bridge that answers here may | ||
| still reject the application-level handshake. Callers follow |
There was a problem hiding this comment.
Can you explain what the application level handshake is?
There was a problem hiding this comment.
Clarified the docstring. By "application-level handshake" I meant the Deako protocol exchange after TCP connect succeeds. Not just "port 23 accepted a socket," but that the bridge can actually complete Deako.connect() / find_devices() and enter normal request/response flow. _tcp_probe() only proves the port is open, so callers still use _connect_primary() as the real readiness gate.
| # own keepalive lives on failover_host, and we are | ||
| # about to connect there; without this it fights us | ||
| # for the single TCP slot on that bridge. None-guard | ||
| # is defensive per section 7.2 note. |
There was a problem hiding this comment.
What is the section 7.2 note?
There was a problem hiding this comment.
Internal doc ref, sorry. The actual note: the bridge only accepts one TCP session on port 23, so the standby keepalive has to be dropped before opening a real Deako connection to the failover host, otherwise the pool blocks its own switch. Comment rewritten inline without the ref.
|
|
||
| # ----- Switch and recovery --------------------------------- | ||
|
|
||
| # pylint: disable-next=too-many-return-statements,too-many-branches |
There was a problem hiding this comment.
I don't think this should be disabled. This function is very difficult to follow
There was a problem hiding this comment.
Agreed. Refactored _switch_to_failover into a short orchestration method that calls smaller private helpers (_wait_for_in_flight_switch, _teardown_active, _teardown_keepalive, _discard_if_stopped, _install_new_active, _execute_failover_switch). Removed the disable; pylint is clean at 10/10 without it.
| "on_failover_switch callback error: %s", exc, | ||
| ) | ||
|
|
||
| # pylint: disable-next=too-many-return-statements,too-many-branches,too-many-statements |
There was a problem hiding this comment.
I don't think this should be disabled either as it's leading to a difficult to follow function
There was a problem hiding this comment.
Same treatment: pulled out _wait_for_in_flight_recovery for the concurrent-caller branch and _try_recovery_candidate(target) for the per-host probe-and-connect, reused the shared _teardown_active / _teardown_keepalive / _discard_if_stopped helpers, and removed the disable. The body now reads as the candidate loop with a concurrent-caller branch in front of it.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2 participants
Summary
Adds
DeakoConnectionPool, a thin supervisor over twoDeakoinstances that keeps one bridge as the active primary and the
other warm as a failover standby. Failover is triggered by two
paths into a single coordinator inside the pool:
_Manager'sping loop (via the new
on_connection_lostcallback) and afailed send inside
control_device. There is no parallelin-pool health monitor.
What is new, public-facing
pydeako.DeakoConnectionPool: high-level pool withstart(),stop(),control_device(),set_state_callback(),state().pydeako.ConnectionPoolState: frozen dataclass snapshot.pydeako.NoSocketException:OSErrorsubclass raised by_SocketConnectionwhen there is no live socket to send on.Deako.__init__:on_connection_lost.Deako.is_connected().Externally observable behavior change
NoSocketExceptionis a new exception class that subclassesOSError._SocketConnection.send_bytesandread_bytesnowraise it when there is no socket to operate on. Callers that
previously caught a different exception type for that condition
will see
NoSocketExceptioninstead; catches onOSErrorcontinue to work.
This is the only non-additive change in the PR.
Key design points (round 2 changes)
_AddressPoolchanges reverted. The pool did not neednon-destructive reads.
_KeepAliveSocketcomposes_SocketConnection, so socket I/Ostays in
utils/_socket.py._Managerpings. When a pong ismissed, the pool's on-lost hook schedules one failover switch.
TCP-probe poll, capped by a module-level timeout, rather than a
blind sleep.
asyncio.Lock+asyncio.Eventcoordinate the switch: thelock serializes, the event signals completion to in-flight
callers so they either land on the new primary or fail cleanly
on timeout.
ConnectionPoolStateisfrozen=True.start()/stop().__package__to match the rest of thelibrary.
Version
Mostly additive public surface, with one narrow externally
observable change noted above (
NoSocketExceptionnow subclassesOSError). Leaving the version bump to maintainer discretion;semver minor seems appropriate for the new surface, happy to bump
if you'd like me to.
Companion HA integration PR
The
deakointegration inhome-assistant/coreis the firstconsumer of the new public surface. The companion PR wires the
integration through
DeakoConnectionPooland the newDeako.on_connection_lostkwarg. The 250ms inter-command throttlelives in the integration rather than the library, so pydeako stays
free of consumer-specific pacing. The HA PR is draft and will stay
draft until a pydeako release carrying this change is on PyPI.
Link: home-assistant/core#167234
Tests
pydeako/deako/test_manager.py: coverage for the newon_connection_lostparameter (invocation on ping timeout,exception swallow, no-callback path).
pydeako/deako/test_deako.py: coverage for the new parameterforwarding and for
is_connected().pydeako/deako/test_connection_pool.py: TCP probe behavior,keepalive socket lifecycle,
ConnectionPoolStateimmutability,pool init validation, idempotent
start/stop, switchcoordination, readiness poll, full failover switch path (happy,
stopped-early, host-not-ready, retry exhaustion, callback
replay), and the
on_connection_lostscheduling path.pydeako/deako/utils/test_socket.py,pydeako/deako/utils/test_connection.py: send-failurefoundation and
NoSocketExceptionraising paths.Local gates run