src/genro_asgi/channel/hub.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Channel hub — the parent side of the channel, with typed envelopes.16 17 The hub binds the socket the children connect to (``uds:`` in a private18 directory, or ``tcp:`` as the multi-host door), keeps the rubric of the19 registered members and routes envelopes. It is **transport-only**: it never20 interprets an op name and never looks inside ``data`` beyond the three keys21 the REPLY contract owns (``result``, ``error``, ``events``). The rubric key22 is the full channel name the member declares in its REGISTER frame23 (``W:<name>`` for workers) — the typing happens at the member, not here.24 25 Three envelope kinds ride the frame protocol of ``frame.py`` (which is not26 modified: ``method`` carries the kind, ``path`` the routing key):27 28 - ``CALL`` — a request. ``call()`` parks an ``asyncio.Future`` on the frame29 id and awaits the matching ``REPLY``. A CALL arriving FROM a member has no30 ratified consumer: it is logged as an unexpected envelope.31 - ``REPLY`` — the answer to a CALL, reusing its id. ``data`` is32 ``{result | error, events: [...]}`` and ``call()`` returns it **verbatim**:33 the barrier lives outside the transport, so the hub neither folds the34 events nor interprets ``result``/``error``. The consumer does both, in its35 own coroutine, after the future resolves. A CALL has **no default36 deadline**: the internal leg waits, because a member that applied the37 lifecycle MUST report it. A caller with an outer surface to protect passes38 its own ``timeout``. The terminator is member death: on EOF — and on a39 deliberate ``stop()`` — every pending CALL addressed to that member fails40 with ``ConnectionError``.41 - ``EVENT`` — fire-and-forget: ``post()`` outbound, ``on_event(member,42 frame)`` inbound. An inbound EVENT is SERVED on its own task: resolving a43 REPLY is O(1) and stays inline, but running a consumer is work, and a slow44 one must not hold the member's receive loop away from the REPLY behind it.45 One task per EVENT also means per-member EVENT ordering is NOT preserved:46 a consumer that needs ordering must provide it itself.47 48 An in-process member joins through ``attach_local(local_channel)`` — the same49 REGISTER frame and the same receive loop over a queue-backed codec twin, so50 the rubric holds one kind of member however it got here (``local.py``).51 52 Liveness is the frame protocol's: EOF reports connection loss, so a member53 whose stream ends is dropped from the rubric and ``on_channel_lost(member)``54 fires — sweep and relaunch belong to the commander, not here. A deliberate55 ``stop()`` is not a death and fires nothing. A ``ValueError`` from the codec56 is a protocol violation of ONE member: that connection is closed, the hub57 and its other members are untouched. Callbacks may be sync or async and58 their exceptions are logged, never fatal — a consumer bug must not fake a59 member death.60 """61 62 from __future__ import annotations63 64 import asyncio65 import contextlib66 import inspect67 import logging68 import os69 import shutil70 import tempfile71 from typing import Any, Callable72 73 from .control import ControlPayload74 from .frame import REGISTER_METHOD, Frame, FrameStream75 from .local import LocalChannel, LocalFrameStream76 77 __all__ = [78 "CALL_METHOD",79 "EVENT_METHOD",80 "REPLY_METHOD",81 "ChannelCallError",82 "ChannelHub",83 "ChannelMember",84 ]85 86 CALL_METHOD = "CALL"87 REPLY_METHOD = "REPLY"88 EVENT_METHOD = "EVENT"89 MAX_PENDING_CALLS = 102490 MAX_EVENT_TASKS = 102491 92 93 class ChannelCallError(Exception):94 """A CALL answered with an error REPLY; ``error`` is the member's payload.95 96 ``payload`` is the whole REPLY the error arrived in: an errored REPLY can97 carry more than its error (the spa delivery keys ride it), and whoever98 turns this exception into a response needs those keys untouched.99 """100 101 def __init__(102 self, member_name: str, path: str, error: Any, payload: dict[str, Any] | None = None103 ) -> None:104 super().__init__(f"call {path} on {member_name} failed: {error}")105 self.member_name = member_name106 self.path = path107 self.error = error108 self.payload = payload or {}109 110 111 class ChannelMember:112 """A child connection in the rubric, keyed by the name it declared."""113 114 __slots__ = ("hub", "name", "pid", "stream")115 116 def __init__(117 self,118 hub: ChannelHub,119 name: str,120 pid: int,121 stream: FrameStream | LocalFrameStream,122 ) -> None:123 self.hub = hub124 self.name = name125 self.pid = pid126 self.stream = stream127 128 async def write(self, frame: Frame) -> None:129 """Send one frame, surfacing a link failure to its caller."""130 await self.stream.write(frame)131 132 def __repr__(self) -> str:133 return f"<ChannelMember {self.name} pid={self.pid}>"134 135 136 class ChannelHub:137 """Parent-side endpoint: binds the socket, tracks members, routes envelopes.138 139 Give ``path`` for UDS, ``host`` (with ``port=0`` to let the OS choose) for140 TCP, or neither to get a socket in a private 0700 directory the hub owns141 and removes at ``stop()``.142 """143 144 REGISTER_TIMEOUT = 10.0145 146 def __init__(147 self,148 *,149 path: str | None = None,150 host: str | None = None,151 port: int = 0,152 on_member_joined: Callable[..., Any] | None = None,153 on_channel_lost: Callable[..., Any] | None = None,154 on_event: Callable[..., Any] | None = None,155 max_size: int | None = None,156 max_pending_calls: int = MAX_PENDING_CALLS,157 max_event_tasks: int = MAX_EVENT_TASKS,158 ) -> None:159 if path is not None and host is not None:160 raise ValueError("give path (uds) or host (tcp), not both")161 if max_pending_calls < 1 or max_event_tasks < 1:162 raise ValueError("max_pending_calls and max_event_tasks must be positive")163 self.on_member_joined = on_member_joined164 self.on_channel_lost = on_channel_lost165 self.on_event = on_event166 self.max_size = max_size167 self.control_payload = ControlPayload()168 self.max_pending_calls = max_pending_calls169 self.max_event_tasks = max_event_tasks170 self.logger = logging.getLogger(__name__)171 self._owned_dir: str | None = None172 if path is None and host is None:173 self._owned_dir = tempfile.mkdtemp(prefix="gnrhub_")174 os.chmod(self._owned_dir, 0o700)175 path = os.path.join(self._owned_dir, "hub.sock")176 self.path = path177 self.host = host178 self.port = port179 self._server: asyncio.Server | None = None180 self._members: dict[str, ChannelMember] = {}181 self._pending: dict[str, tuple[ChannelMember, str, asyncio.Future[Frame]]] = {}182 self._abandoned: dict[str, tuple[ChannelMember, str]] = {}183 self._local_loops: set[asyncio.Task[None]] = set()184 self._event_tasks: set[asyncio.Task[None]] = set()185 self._closing = False186 187 @property188 def started(self) -> bool:189 """Whether the socket is bound."""190 return self._server is not None191 192 @property193 def address(self) -> str:194 """The connectable address (``uds:<path>`` or ``tcp:<host>:<port>``)."""195 if self._server is None:196 raise RuntimeError("hub not started")197 if self.path is not None:198 return f"uds:{self.path}"199 return f"tcp:{self.host}:{self.port}"200 201 @property202 def members(self) -> dict[str, ChannelMember]:203 """Snapshot of the rubric, by channel name."""204 return dict(self._members)205 206 async def start(self) -> None:207 """Bind the socket and start accepting children."""208 if self.path is not None:209 self._server = await asyncio.start_unix_server(self._handle_connection, path=self.path)210 else:211 self._server = await asyncio.start_server(self._handle_connection, self.host, self.port)212 self.port = self._server.sockets[0].getsockname()[1]213 self.logger.info("Channel hub listening on %s", self.address)214 215 async def stop(self) -> None:216 """Deliberate shutdown: close every member without firing channel_lost.217 218 The members go first: ``Server.wait_closed()`` waits for the219 connection handlers, and a handler parked on a live member's read220 would never return.221 """222 if self._server is None:223 return224 self._closing = True225 for member in list(self._members.values()):226 await member.stream.close()227 self._server.close()228 await self._server.wait_closed()229 self._server = None230 self._members.clear()231 self._fail_pending(None, "hub stopped")232 if self.path is not None and os.path.exists(self.path):233 os.unlink(self.path)234 if self._owned_dir is not None:235 shutil.rmtree(self._owned_dir, ignore_errors=True)236 self._owned_dir = None237 self.logger.info("Channel hub stopped")238 239 async def attach_local(self, local: LocalChannel) -> ChannelMember | None:240 """Register an in-process member arriving over a ``LocalChannel``.241 242 The single attachment point for the local wire: the REGISTER frame and243 the receive loop are the socket ones, so the rubric holds one kind of244 member however it got here.245 """246 member = await self._register_connection(local.hub_stream)247 if member is not None:248 loop_task = asyncio.create_task(self._receive_loop(member))249 self._local_loops.add(loop_task)250 loop_task.add_done_callback(self._local_loops.discard)251 return member252 253 def resolve(self, name: str) -> ChannelMember | None:254 """The member registered under this channel name, or ``None``."""255 return self._members.get(name)256 257 async def post(self, name: str, path: str, data: Any = None) -> str:258 """Send one EVENT to one member (fire-and-forget); returns the frame id."""259 member = self._members.get(name)260 if member is None:261 raise LookupError(f"no member named {name!r}")262 frame = Frame(method=EVENT_METHOD, path=path, payload=self.control_payload.encode(data))263 await member.write(frame)264 return frame.id265 266 async def call(267 self, name: str, path: str, data: Any = None, timeout: float | None = None268 ) -> Any:269 """CALL one member and await its REPLY; returns the REPLY ``data`` verbatim.270 271 The payload is the member's ``{result | error, events}`` dict, handed272 over untouched — reading it is the consumer's job. ``timeout`` is the273 caller's own deadline and expires with ``TimeoutError``; ``None`` waits274 indefinitely, until the REPLY lands or the member dies275 (``ConnectionError``).276 """277 member = self._members.get(name)278 if member is None:279 raise LookupError(f"no member named {name!r}")280 frame = Frame(method=CALL_METHOD, path=path, payload=self.control_payload.encode(data))281 reply = await self.call_frame(name, frame, timeout=timeout)282 return self.control_payload.decode(reply.payload)283 284 async def call_frame(self, name: str, frame: Frame, timeout: float | None = None) -> Frame:285 """Send an opaque CALL frame and return its matching opaque REPLY."""286 if frame.method != CALL_METHOD:287 raise ValueError("call_frame requires a CALL frame")288 member = self._members.get(name)289 if member is None:290 raise LookupError(f"no member named {name!r}")291 if len(self._pending) >= self.max_pending_calls:292 raise RuntimeError(f"channel has {self.max_pending_calls} outstanding calls")293 if frame.id in self._pending:294 raise RuntimeError(f"a call with id {frame.id!r} is already pending")295 if frame.id in self._abandoned:296 raise RuntimeError(f"a call with id {frame.id!r} is awaiting a late reply")297 future: asyncio.Future[Frame] = asyncio.get_running_loop().create_future()298 self._pending[frame.id] = (member, frame.path, future)299 sent = False300 try:301 # A cancellation during drain cannot prove that no bytes reached302 # the peer, so crossing the write boundary is conservatively sent.303 sent = True304 await member.write(frame)305 if timeout is None:306 return await future307 return await asyncio.wait_for(future, timeout)308 except (TimeoutError, asyncio.CancelledError):309 if (310 sent311 and (not future.done() or future.cancelled())312 and self._members.get(member.name) is member313 ):314 if len(self._abandoned) >= self.max_pending_calls:315 # Cleanup must not replace this caller's timeout/cancellation.316 with contextlib.suppress(Exception, asyncio.CancelledError):317 await member.stream.close()318 else:319 self._abandoned[frame.id] = (member, frame.path)320 raise321 finally:322 self._pending.pop(frame.id, None)323 324 async def _handle_connection(325 self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter326 ) -> None:327 """Per-connection task: require REGISTER as the first frame, then relay."""328 stream = FrameStream(329 reader, writer, max_size=self.max_size330 )331 member = await self._register_connection(stream)332 if member is not None:333 await self._receive_loop(member)334 335 async def _register_connection(336 self, stream: FrameStream | LocalFrameStream337 ) -> ChannelMember | None:338 """Read and validate the presentation frame; reject anything else."""339 try:340 frame = await asyncio.wait_for(stream.read(), timeout=self.REGISTER_TIMEOUT)341 except (TimeoutError, ValueError):342 self.logger.warning("Connection rejected: no valid REGISTER frame")343 await stream.close()344 return None345 if frame is None or frame.method != REGISTER_METHOD:346 self.logger.warning("Connection rejected: first frame is not %s", REGISTER_METHOD)347 await stream.close()348 return None349 try:350 info = self.control_payload.decode(frame.payload)351 except ValueError:352 self.logger.warning("Connection rejected: invalid REGISTER control payload")353 await stream.close()354 return None355 if not isinstance(info, dict):356 self.logger.warning("Connection rejected: REGISTER payload is not an object")357 await stream.close()358 return None359 name = info.get("name")360 if not isinstance(name, str) or not name:361 self.logger.warning("Connection rejected: REGISTER without a valid name")362 await stream.close()363 return None364 if name in self._members:365 # Names are minted by the commander and never reused, so a name366 # already in the rubric is a protocol violation of the newcomer:367 # the registered member is the real one and stays.368 self.logger.warning("Connection rejected: name %s is already registered", name)369 await stream.close()370 return None371 try:372 pid = int(info.get("pid", 0))373 except (TypeError, ValueError, OverflowError):374 self.logger.warning("Connection rejected: REGISTER with invalid pid")375 await stream.close()376 return None377 member = ChannelMember(self, name, pid, stream)378 self._members[name] = member379 await self._fire(self.on_member_joined, member)380 self.logger.info("Member joined: %s", member)381 return member382 383 async def _receive_loop(self, member: ChannelMember) -> None:384 """Read this member's frames until the channel ends; EOF → channel lost."""385 try:386 while True:387 try:388 frame = await member.stream.read()389 except ValueError:390 self.logger.exception(391 "Protocol violation from %s; closing that member", member.name392 )393 break394 if frame is None:395 break396 await self._dispatch(member, frame)397 except asyncio.CancelledError:398 return399 finally:400 await member.stream.close()401 if self._members.get(member.name) is member:402 del self._members[member.name]403 self._fail_pending(member, f"channel to {member.name} lost")404 self._abandoned = {405 frame_id: expected406 for frame_id, expected in self._abandoned.items()407 if expected[0] is not member408 }409 if not self._closing:410 self.logger.info("Channel lost: %s", member.name)411 await self._fire(self.on_channel_lost, member)412 413 async def _dispatch(self, member: ChannelMember, frame: Frame) -> None:414 """Route one inbound frame by envelope kind: resolve inline, serve on a task.415 416 A REPLY only hands a payload to a parked future — O(1), so it stays in417 the receive loop. An EVENT runs a consumer, so it goes on its own task418 and the loop returns to the wire; the ref is held here because the419 loop keeps only a weak one. One task per EVENT means per-member EVENT420 ordering is not preserved: an ordering-sensitive consumer provides its own.421 """422 if frame.method == REPLY_METHOD:423 await self._resolve_reply(member, frame)424 elif frame.method == EVENT_METHOD:425 if len(self._event_tasks) >= self.max_event_tasks:426 self.logger.warning(427 "Dropping EVENT %s from %s: event task limit %s reached",428 frame.path,429 member.name,430 self.max_event_tasks,431 )432 return433 task = asyncio.create_task(self._fire(self.on_event, member, frame))434 self._event_tasks.add(task)435 task.add_done_callback(self._event_tasks.discard)436 else:437 self.logger.warning("Unknown envelope %s from %s", frame.method, member.name)438 439 async def _resolve_reply(self, member: ChannelMember, frame: Frame) -> None:440 """Hand the REPLY payload to the parked future, verbatim.441 442 A REPLY whose caller already went away — its deadline expired, or it443 was cancelled — is dropped: nobody is left to read the envelope.444 """445 abandoned = self._abandoned.get(frame.id)446 if abandoned is not None:447 if abandoned == (member, frame.path):448 del self._abandoned[frame.id]449 else:450 self.logger.warning(451 "REPLY %s from %s does not match its abandoned CALL", frame.id, member.name452 )453 await member.stream.close()454 return455 parked = self._pending.get(frame.id)456 if parked is None or parked[2].done():457 self.logger.debug("REPLY %s from %s has no parked caller", frame.id, member.name)458 return459 if parked[0] is not member or parked[1] != frame.path:460 self.logger.warning(461 "REPLY %s from %s does not match its pending CALL", frame.id, member.name462 )463 await member.stream.close()464 return465 parked[2].set_result(frame)466 467 def _fail_pending(self, member: ChannelMember | None, reason: str) -> None:468 """Fail one member's pending CALLs (``None`` = all) with ``ConnectionError``.469 470 The entries stay in ``_pending``: each caller pops its own in the471 ``finally`` of ``call()``.472 """473 for expected_member, _path, future in self._pending.values():474 if (member is None or expected_member is member) and not future.done():475 future.set_exception(ConnectionError(reason))476 477 async def _fire(self, callback: Callable[..., Any] | None, *args: Any) -> None:478 """Run a sync-or-async callback; a consumer bug must not sever the channel."""479 if callback is None:480 return481 try:482 result = callback(*args)483 if inspect.isawaitable(result):484 await result485 except Exception:486 self.logger.exception("Channel callback %r failed", callback)