src/genro_asgi_multiworker_spa/orchestration/spa_worker.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 """SpaWorker: the three registers of one process, and the row a request lands on.16 17 A worker holds the users it was given, and for each of them the connections and18 the pages under them. That picture lives in three registers — ``user_register``,19 ``connection_register``, ``page_register`` — and one entry of a register is a20 **register item**: the same word the deposit files under, so the thing in memory21 and the thing on disk are never two words for one object.22 23 **The registers are the shared registry's.** They are built by24 ``build_registry`` — the seam a consumer replaces to pair its own row types with25 the tree — and the three names above are properties onto its ``user_items`` /26 ``connection_items`` / ``page_items``. So a row is born with the whole data27 plane already on it (the live store, and whatever a consumer's row class28 adds) and the worker's own fields — ``state``, the29 transfer flag, the three clocks — ride that same row: one object, whichever30 half of the machine is reading it. Reading goes through the register idioms31 (``get``, ``keys``, ``keys_by``, ``in``); writing stays where it already was, in32 the single-writer mutators, which speak the registry's own lifecycle vocabulary.33 34 **The tree lives in the items.** A page belongs to a CONNECTION and a connection35 to a USER. Downwards the edge is a set the parent carries (a user item's36 ``connections``, a connection item's ``pages``); upwards it is the parent key the37 child already holds (a connection item's ``user``, a page item's38 ``connection_id``). Both directions are written in the same gesture by ONE39 mutator, so they cannot disagree, and a page stores no user label at all: the40 owner is derived by walking up, and what is derived cannot go stale.41 42 **The unified row.** A request for a user finds its row ``active`` and is served;43 finds it ``frozen`` and the store is pulled back from the deposit first; finds no44 row at all and one is added ``frozen``, then pulled the same way. The arrival of45 somebody nobody has ever seen and the waking of a hibernated user are the SAME46 line of code — the pull simply finds nothing for the first and a parcel for the47 second. The three states are derived from what the worker holds, never oracle48 booleans somebody has to remember to set.49 50 **One trip to the freezer, and the sisters wait for it.** A page fires dozens of51 calls at once, so a burst on a frozen user arrives as many coroutines together.52 The FIRST marks the row ``unfreezing`` before it reads anything, and that mark is53 what the others find: they await the transition — never the service, which stays54 parallel per user — and then all go on together. So the disk is read once, by55 one call, however wide the burst. A pull that fails takes the row away instead56 of leaving it half-born: the parcel is still on disk and the mark still on at57 the vertex, so the next request of his carries the verdict again and the trip58 is retried by the unified row's own shape, with nothing to reconcile.59 60 **What may be adopted, and what may not.** The user store comes home ONLY when61 the envelope authorises it: the Commander, routing the request, attaches its own62 verdict under ``user_frozen``, and without that verdict the parcel on disk is63 residue and is never touched — the sweep's business, not the worker's. A64 CONNECTION needs no verdict: a worker that does not hold the connection a request65 names looks in the user's own folder by itself — found, it installs the66 connection and its pages and serves; not found, it starts that connection empty.67 One code path for both, which is why the stranger needs no special treatment.68 69 **Adoption reads, empties, then announces.** Read the parcel, delete the file —70 and the folder with it when that file was the last thing in it, so the deposit71 holds the frozen and nothing else — and only then announce. The user store72 announces ``user_adopted``, which is what turns the sleeping mark off at the73 fold. An adopted CONNECTION announces nothing of its own: it is born through the74 ordinary mutators and emits the ordinary ``new_connection``/``new_page`` — one75 birth path in the machine, not two.76 77 **Announcements ride the reply of the CALL that caused them — two channels,78 never one queue** (owner, 2026-09-04; #60). Every mutation queues its protocol79 name in the ``worker_events`` of the ``RequestSlot`` of the CALL being served:80 each CALL is served on a task of its own, the stitching runs on the pool under a81 copy of that task's context, and ``send_reply`` sends that slot's events and no82 other's, then closes the slot. The names are the inherited83 ``new_user``/``new_connection``/``new_page`` and84 ``drop_user``/``drop_connection``/``drop_connections``/``drop_page``/85 ``drop_pages``, plus ``user_adopted``, ``user_frozen``,86 ``connection_user_changed``, ``user_rows_released``. A cascade speaks the87 plural: dropping a connection announces its pages as one ``drop_pages``, dropping88 a user its connections as one ``drop_connections``. Outside any CALL there is no89 slot and a mutation raises. The ONE producer that answers no CALL — the transfer90 cycle, which is the quit's — gives each departure a slot of its own and sends91 what it announced with ONE CALL of this worker's, ``/group/announce``, folded at92 the vertex exactly as a reply's envelope is; the REPLY is the acknowledgement, and a93 vertex that does not answer loses that announcement, counted and logged, never94 retried.95 96 **A drop asks for absence.** Dropping something already gone is that same97 outcome — no error, and nothing announced, because nothing happened.98 99 **Three clocks, one climb.** ``last_refresh_ts`` is technical contact and every100 call stamps it, the beat included; ``last_user_ts`` is a real human event and is101 the prince; ``last_rpc_ts`` is a real call, the surrogate metre until the page102 protocol carries the human event of its own. Whoever judges idleness or expiry103 reads the real clocks and never ``last_refresh_ts``, which a beat alone can keep104 warm forever. A stamp climbs the chain — page, its connection, its user — with an105 instant the server takes itself: a client cannot buy immortality by claiming106 activity.107 108 **One lock.** Every mutation is serialized on ``dispatch_lock``; nothing awaits109 while holding it. Finer grain was measured and refused: at a couple of kilobytes110 per user, reading and unpickling a parcel costs microseconds.111 112 **Leaving is the mirror of arriving.** ``freeze_user`` writes what the adoption113 reads back — the store under the user, one parcel per connection carrying that114 connection and its pages — under the folder semaphore, which is the deposit's115 only coherence mechanism, and then says ``user_frozen``. WHERE he wakes is not116 this worker's business: the worker event keeps a ``placement`` slot for the117 vertex that will decide it and this worker never fills it, so every departure118 leaves with the placement to be assigned and the row leaves memory WHOLE. No119 drop is announced beside it: the freeze worker event already told the story, and120 the wake tells it back through the ordinary births. A write that fails aborts121 the departure whole — the semaphore goes back, the user stays alive exactly122 where he is, nothing is announced, and the failure is logged and counted. Nobody123 here kills what could not be saved.124 125 **The semaphore is waited for, never forever.** A folder somebody else holds is126 waited on with a coroutine, and the first miss says out loud whose it is; past127 ``DEPOSIT_LOCK_WAIT_LIMIT`` the wait gives up rather than hang silently — an128 adoption raises, and the caller's own REPLY carries the failure; a freeze takes129 the shape of any other refused departure. That bound is a floor against a130 semaphore nobody gives back, not a budget: how long a REQUEST may wait before131 the vertex answers it something else is the Commander's parking budget, and132 arrives with the fold.133 134 **Nothing is parked while a call of its user runs.** Every call opens under its135 user and closes there (``open_request`` / ``close_request``, WSGI stitching136 included); a freeze happens only at empty pendings, because a store photographed137 with live calls inside would take their work nowhere while the browser was told138 it was done. The question is asked ONCE, at the door: what holds that window139 shut is the BLOCK whoever orders the departure raises at the vertex before140 ordering it, so nothing of his can be born while his parcels are written and no141 check under the semaphore would have anything left to decide. The end of a call142 is therefore where a143 departure that had to wait for it happens — one mechanism, whether the worker is144 being emptied or a single user is being ceded — and a departure is CLAIMED145 before the first await of its path, so the cycle and the hook can never park the146 same user twice.147 148 **The departures are decided above.** At photo time ``plan_transfers`` pairs149 every user row with a ``transfer_flag``: ``None`` kept, ``'T'`` ceded. WHO is150 ceded is handed in — by the group, which reads the clocks the photo carries and151 judges the silence, or by the quit, which cedes everybody. No user is named here152 by a policy of this process: silence and expiry are the group's judgment, and153 this rung has no gauge of its own. Then THE GATE: the worker does not park154 anybody in the same turn it announced them. It waits ``TRANSFER_START_DELAY``,155 the time the fold needs to park the users just named, and only then lets them156 go, one at a time, the loop breathing between two. So there is ONE departure157 scheme and no special case: the158 window in which somebody could come back to a row that was already emptied159 cannot open, because whoever comes back either is already in the pendings and160 his freeze waits for him, or arrives after the fold parked him and starts again161 from the vertex with the verdict in hand.162 163 **The ordered freeze.** ``/group/freeze_user`` parks ONE user on the parent's164 order. The worker only executes: it waits for whatever holds him — a pull165 bringing him home, his calls in flight — parks him through the same departure166 every road uses, and only then answers, so the REPLY IS the confirmation and167 ``user_frozen`` rides it as always. A user this worker does not host is168 refused out loud in that same REPLY. The waits are serialization, not169 policy: the judgment of WHO sleeps is the caller's alone, and so is the block170 that keeps new work of his from arriving while he leaves.171 172 **The exit.** ``quit`` is that same departure applied to everybody — flag, gate,173 park as the last calls end, leave — and once it starts the plan is TERMINAL: a174 later shot may add a newborn to the departing, never take anybody off them. A175 single user's failure is contained where it happens, so one refused parcel176 cannot keep the process from ending. The worker has no verb of rebirth: whoever177 wants a successor launches one.178 179 **The wire is handed in, never opened here.** Whoever runs this worker in a180 process connects to the handler's socket and hands the stream over181 (``attach_stream``); the worker presents itself on it — its pid and the182 configuration it was built from — and the answer brings the whole global store183 down. Then it reads envelopes until the wire ends. A CALL coming down is served184 on its own task, so a long one cannot make this worker deaf to the next; a REPLY185 coming down is the answer to a call this worker placed UPWARD and is resolved186 inline, on the frame id the future was parked under. Any other kind of envelope187 is denounced.188 189 **The lane upward.** ``call`` places a CALL of this worker's own on the same190 wire and awaits the answer: the id makes the conversations independent, so calls191 placed without awaiting the first resolve each with its own REPLY in whatever192 order the parent gives them, and an answer carrying an ``error`` raises193 ``CommanderCallFailed`` rather than returning a result nobody made. What the194 site's verbs need is the sync door onto it: they run on a traffic-pool thread,195 and ``run_on_loop`` hops the coroutine onto the loop the wire lives on.196 197 **Two pools, and what runs where.** The TRAFFIC pool takes the WSGI stitching198 and the long calls, the SERVICE pool — much smaller — takes the deposit IO;199 their sizes come down in the spawn payload. Neither ever takes a wait: waiting200 for a busy folder is a coroutine on the loop, because whoever holds that201 semaphore is working, and a thread parked here would be a thread not doing that202 work.203 204 **The orders are a tree, and one form.** Every order the parent gives is205 resolved by name on ``worker_dispatcher`` (#59): the first segment names who206 issues it — ``group`` for the orchestration, ``commander`` for the vertex — and207 the last one is a ``@route`` method of ``GroupOrders`` or ``CommanderOrders``.208 ``/group/ping`` answers the health beat and nothing else — are you alive.209 ``/group/quit`` is answered AT ONCE with the photo that shows every user flagged210 for cession, the departure running on a task of its own after the answer because211 the process ends with it; ``/group/drop_user`` and ``/group/drop_connection`` are212 answered when the drop is done, so the worker events it made ride that same213 reply. A path nobody serves answers ``NotFound``. The http CALL form (an ``http`` dict beside the214 ``identity`` and the ``user_frozen`` verdict) is a request the front packed215 whole: it lands on the unified row FIRST — the store adopted when the verdict216 authorises it, the connection looked up in the deposit by itself, the clocks217 stamped — and only then goes to ``hosted_app_seam``, which is the one road to218 whatever this worker hosts. Both seams are ``None`` here: this class hosts no219 application, and the property says so out loud. A subclass assigns ONE of them220 — ``asgi_app`` for an ASGI application, or the ``wsgi_app`` shortcut, which the221 core wraps in a ``WsgiSeam`` of its own and runs on the traffic pool because222 WSGI is synchronous — and that is the whole contract with a consumer.223 224 **The photo rides out.** ``worker_snapshot`` is a slot ANY envelope leaving here225 may carry beside its own payload: the presentation carries it (a live process is226 never without a photo), every population change carries it (a user entering or227 leaving is when the thing the photo describes really changes) — on the reply or228 on the announcement, whichever carries the change — and any reply carries it229 once ``worker_snapshot_ttl`` has run out on the last one. So there is one road230 instead of three, and the beat keeps the only question its name asks.231 232 **The global store is NOT here at all.** There is no replica: the one copy — a233 dictionary — lives on the commander, and every access is a CALL on the lane234 through ``global_store``, a ``GlobalStoreClient``: ``get``/``set``/``delete``235 answered under the commander's lock, and ``for_update`` — the turn, whose236 ``GlobalStoreLease`` yields the private working value the grant decoded and237 sends the COMPLETE value back at the exit. A body that raises releases with238 nothing applied, and a process that dies holding the turn has the vertex give239 it back.240 241 **When the wire dies.** The handler watches the process and the process watches242 the wire: two guardians converging on the same safe state. A wire gone means243 nobody can be told anything, so the worker parks everybody in the deposit — the244 road to safety does not pass through the channel — and leaves. The245 worker events stay unsaid: whoever finds the parcels needs no telling.246 """247 248 from __future__ import annotations249 250 import asyncio251 import contextvars252 import inspect253 import copy254 import contextlib255 import functools256 import logging257 import os258 import threading259 import time260 from collections.abc import Iterable261 from concurrent.futures import ThreadPoolExecutor262 from typing import Any, Callable263 264 import psutil265 from genro_routes import RoutingClass, route266 from genro_tytx import to_tytx267 268 from genro_asgi.channel.control import ControlPayload269 from genro_asgi.http_record import HttpRecord270 from genro_asgi.transport_limits import FrameTooLarge271 from genro_asgi.channel.frame import REGISTER_METHOD, REGISTER_PATH, Frame, FrameStream272 from genro_asgi.exceptions import HTTPException273 from ..environ import AsgiSeam, WsgiSeam274 from ..global_store import (275     GLOBAL_STORE_DEL_OP_PATH,276     GLOBAL_STORE_GET_OP_PATH,277     GLOBAL_STORE_LOCK_OP_PATH,278     GLOBAL_STORE_SET_OP_PATH,279     GLOBAL_STORE_UNLOCK_OP_PATH,280     GlobalStoreClient,281 )282 from ..register import Register283 from ..register_registry import RegisterRegistry284 from .freeze_handler import FreezeHandler285 from .worker_connector import (286     CALL_METHOD,287     ENVELOPE_SLOT_PRESENTATION,288     ENVELOPE_SLOT_WORKER_EVENTS,289     ENVELOPE_SLOT_WORKER_SNAPSHOT,290     REPLY_METHOD,291     CommanderCallFailed,292 )293 from .worker_handler import ANNOUNCE_OP_PATH294 295 #: The reserved prefix that names an anonymous user — the daemon's own296 #: convention, so the name itself carries the guest rule. Redefined here with297 #: its ratified value rather than imported: the legacy machine dies at the298 #: cutover, this one must outlive it.299 GUEST_PREFIX = "guest_"300 301 #: How often a wait for a busy deposit folder looks at it again, in seconds.302 DEPOSIT_LOCK_RETRY_INTERVAL = 0.05303 304 #: How long a wait for a busy deposit folder goes on before it gives up, in305 #: seconds. A technical floor against a semaphore nobody gives back — never a306 #: budget for how long a request may wait, which is the vertex's to spend.307 DEPOSIT_LOCK_WAIT_LIMIT = 30.0308 309 #: How long the worker waits between announcing its departures and starting to310 #: park them, in seconds — the time the fold needs to park the users just named.311 #: A technical time, not a grammar of configuration.312 TRANSFER_START_DELAY = 2.0313 314 #: How long a photo already sent stays fresh enough, in seconds: past it, the315 #: next envelope out carries a new one.316 WORKER_SNAPSHOT_TTL = 0.5317 318 #: How long a quit waits for a user whose call is still in flight, in seconds.319 #: Past it the wait is dropped and he is parked without that call.320 PENDING_CALL_GRACE_SECONDS = 5.0321 322 #: Where a channel command is resolved on this worker's own dispatcher. The323 #: front names the branch on the lane; the leaf is this worker's business.324 WSX_COMMAND_PATH = "/wsx/openchannel"325 326 #: Where a message for a browser is placed on the lane: the front attached that327 #: branch under the commander's operations, because the delivery needs the328 #: registry of sockets and the vertex's own map of pages.329 WEBSOCKET_SEND_PATH = "/commander/websocket/send"330 331 #: How long the worker waits for the vertex to take an announcement, in332 #: seconds. Past it the announcement is counted lost: a wire that answers333 #: nothing is a vertex that is gone, and the cycle must not hang on it.334 ANNOUNCE_TIMEOUT_SECONDS = 10.0335 336 #: The three clocks every register item carries, in the order of their rank.337 CLOCK_NAMES = ("last_refresh_ts", "last_user_ts", "last_rpc_ts")338 339 #: The routing keys of the lane going UP are paths on the trees the group and340 #: the commander host (#59): the first segment names the level that serves, the341 #: next ones the operation class and the operation — ``/commander/store/get``.342 #: The routing keys of the global store are the store module's own, re-exported here.343 344 #: The routing key one observation climbs: a mutation of this process's345 #: registers, as it happens, for whoever is watching at the vertex.346 OBSERVATION_OP_PATH = "/commander/observation"347 348 # What the census puts in place of a field it cannot carry as JSON: the field is349 # left out of the reading entirely, and a sentinel says so without colliding350 # with a legitimate None.351 _NOT_JSON_SAFE = object()352 353 # The worker events that mean the population changed — a user entering or354 # leaving — and therefore that the next envelope out owes a fresh photo.355 POPULATION_WORKER_EVENTS = frozenset(356     {357         "new_user",358         "drop_user",359         "user_frozen",360         "user_adopted",361         "connection_user_changed",362         "user_rows_released",363     }364 )365 366 __all__ = [367     "CLOCK_NAMES",368     "DEPOSIT_LOCK_RETRY_INTERVAL",369     "DEPOSIT_LOCK_WAIT_LIMIT",370     "GLOBAL_STORE_DEL_OP_PATH",371     "GLOBAL_STORE_GET_OP_PATH",372     "GLOBAL_STORE_LOCK_OP_PATH",373     "GLOBAL_STORE_SET_OP_PATH",374     "GLOBAL_STORE_UNLOCK_OP_PATH",375     "GUEST_PREFIX",376     "TRANSFER_START_DELAY",377     "WORKER_SNAPSHOT_TTL",378     "RequestSlot",379     "SpaWorker",380 ]381 382 383 class RequestSlot:384     """What one request has produced so far, waiting for its own exchange.385 386     The events of a request belong to THAT request: they accumulate here and387     leave together at its end. ``worker_events`` are what happened on the388     registers while this CALL was being served, and they ride ITS reply — never389     another's (owner, 2026-09-04). A consumer's slot class adds what its own390     verbs produce per request. ``connection_id`` is the one field that travels back OUT: the front reads391     it off the reply to write its cookie with. ``connection_previous_user`` is set392     when THIS request logged the connection in, and it is what makes the tail of393     this request, and of no other, carry that connection to the deposit.394     """395 396     def __init__(self) -> None:397         self.worker_events: list[dict[str, Any]] = []398         #: The connection the site named while serving this request — born, or399         #: changed owner. None when it named none, which is every request that400         #: reused the connection its cookie already carried.401         self.connection_id: str | None = None402         #: Who owned the connection before this request logged it in; None when403         #: this request logged nobody in.404         self.connection_previous_user: str | None = None405 406 407 class GroupOrders(RoutingClass):408     """The ``group`` branch of the worker's dispatcher: the orders the GROUP gives.409 410     Orchestration, all of it: placement undone (``drop_user``,411     ``drop_connection``), the ordered freeze of one user, the departure, the412     beat. Each is a ``@route`` method the vertex reaches on413     ``/group/<order>``; what it answers is the ``result`` of the REPLY, what it414     raises is the ``error``.415 416     Args:417         spa_worker: the process these orders act on.418     """419 420     def __init__(self, spa_worker: Any) -> None:421         self.spa_worker = spa_worker422 423     @route()424     def ping(self) -> dict[str, Any]:425         """The beat: an answer is the whole point — the photo rides the envelope."""426         return {}427 428     @route()429     def drop_user(self, user: str) -> dict[str, Any]:430         """Take one user off this process."""431         self.spa_worker.drop_user(user)432         return {}433 434     @route()435     def drop_connection(self, cid: str) -> dict[str, Any]:436         """Take one connection off this process; one nobody holds is answered quietly."""437         connection = self.spa_worker.connection_register.get(cid)438         if connection is not None:439             self.spa_worker.drop_connection(connection["user"], cid)440         return {}441 442     @route()443     async def freeze_user(self, user: str) -> dict[str, Any]:444         """Park ONE user and answer only then: the REPLY is the confirmation."""445         return await self.spa_worker.freeze_designated_user(user)446 447     @route()448     def quit(self, freezer_path: str | None = None) -> dict[str, Any]:449         """Flag everybody for departure, answer, THEN leave.450 451         Args:452             freezer_path: where the parcels of this departure go, when not the453                 working deposit — the reboot directory of a soft quit.454 455         Acts on the flags before the answer, so the photo riding it shows every456         user ceded and the level above parks them all in one read; the457         departure itself is put on a task of its own, so the REPLY goes down458         the wire first.459         """460         self.spa_worker.begin_quit(freezer_path=freezer_path)461         return {}462 463 464 class CommanderOrders(RoutingClass):465     """The ``commander`` branch of the worker's dispatcher: the orders the VERTEX gives.466 467     Reading and switching, nothing of the placement: the census, the eval468     door, the observation switch. A consumer attaches its own order class under469     here with ``add_branches``.470 471     Args:472         spa_worker: the process these orders act on.473     """474 475     def __init__(self, spa_worker: Any) -> None:476         self.spa_worker = spa_worker477 478     @route()479     def observe(self, on: bool) -> dict[str, Any]:480         """Turn this process's observation on or off."""481         self.spa_worker.observation_on = bool(on)482         return {}483 484     @route()485     def census(self) -> dict[str, Any]:486         """The structured reading of the whole process, JSON-safe."""487         return self.spa_worker.census()488 489     @route()490     def eval(self, expr: str) -> dict[str, Any]:491         """Evaluate one expression inside the process, ``repr`` back."""492         return {"repr": self.spa_worker.eval_expression(expr)}493 494 495 class WsxCommands(RoutingClass):496     """The ``wsx`` branch of the worker's dispatcher: what a PAGE asks of its channel.497 498     The other two branches name who ISSUES an order — the group, the vertex.499     This one names a matter instead: nobody orders here, a page opens its500     channel and says how it wants to be served on it.501 502     Args:503         spa_worker: the process the command acts on.504     """505 506     def __init__(self, spa_worker: Any) -> None:507         self.spa_worker = spa_worker508 509     @route()510     def openchannel(self, page_id: str, parameters: Any = None) -> str:511         """Open the channel of one page, and say how its calls must be served.512 513         Args:514             page_id: the page opening its channel.515             parameters: how it wants to be served — nothing for the ordinary516                 page, a dict for one that asked for something (``sequential``517                 serialises its calls).518 519         Returns:520             The word the front turns into the answer of the command.521 522         Raises:523             KeyError: no page of that name lives here. A page is born by the524                 site while it serves, so a channel opened for a page nobody525                 created is a client talking about something that does not526                 exist, and it is told so.527 528         Acts on the page's row, under its own lock. Saying it twice changes529         nothing: a browser that reconnects opens its channel again.530         """531         worker = self.spa_worker532         with worker.dispatch_lock:533             row = worker.page_register.get(page_id)534         if row is None:535             raise KeyError(f"no page {page_id!r} on this worker: it was never born here")536         with row["item_lock"]:537             row["wsx"] = parameters if parameters else True538         return "channel open"539 540 541 class WorkerDispatcher(RoutingClass):542     """The root of the orders a worker takes: ``group/…`` and ``commander/…``.543 544     The first segment of a path names who issues the order; the tree is the545     table, and ``route.nodes()`` lists it without an order being placed. The two546     branches are kept as attributes so a consumer can attach its own class under547     the issuer it extends.548 549     Args:550         spa_worker: the process this dispatcher belongs to.551     """552 553     def __init__(self, spa_worker: Any) -> None:554         self.spa_worker = spa_worker555         self.group_orders = GroupOrders(spa_worker)556         self.commander_orders = CommanderOrders(spa_worker)557         self.wsx_commands = WsxCommands(spa_worker)558         self.add_branches(559             [560                 {"name": "group", "instance": self.group_orders},561                 {"name": "commander", "instance": self.commander_orders},562                 {"name": "wsx", "instance": self.wsx_commands},563             ]564         )565 566 567 class SpaWorker:568     """The users, connections and pages one worker process holds.569 570     Args:571         name: the worker's name, the one its handler minted; it stamps every572             worker event and holds the deposit semaphore.573         freeze_handler: the deposit surface — the only way to the parcels.574         group: the group this worker serves in; it goes in the diagnostic header575             of every parcel, which is read for counting and for the sysop.576         deposit_lock_retry_interval: how often a busy user folder is looked at577             again while waiting for its semaphore.578         deposit_lock_wait_limit: how long that wait goes on before it gives up579             loud.580         transfer_start_delay: how long the gate stays shut between announcing581             the departures and parking them.582         main_threadpool_size: the traffic pool's size — the WSGI stitching and583             the long calls; ``None`` leaves the interpreter's own default.584         aux_threadpool_size: the service pool's size — the deposit IO, and much585             smaller.586         worker_snapshot_ttl: how long a photo already sent stays fresh enough.587     """588 589     def __init__(590         self,591         name: str,592         *,593         freeze_handler: FreezeHandler,594         group: str = "",595         deposit_lock_retry_interval: float = DEPOSIT_LOCK_RETRY_INTERVAL,596         deposit_lock_wait_limit: float = DEPOSIT_LOCK_WAIT_LIMIT,597         transfer_start_delay: float = TRANSFER_START_DELAY,598         main_threadpool_size: int | None = None,599         aux_threadpool_size: int | None = None,600         worker_snapshot_ttl: float = WORKER_SNAPSHOT_TTL,601     ) -> None:602         self.name = name603         self.freeze_handler = freeze_handler604         self.group = group605         self.deposit_lock_retry_interval = deposit_lock_retry_interval606         self.deposit_lock_wait_limit = deposit_lock_wait_limit607         self.transfer_start_delay = transfer_start_delay608         self.worker_snapshot_ttl = worker_snapshot_ttl609         self.traffic_pool = ThreadPoolExecutor(610             max_workers=main_threadpool_size, thread_name_prefix=f"{name}-traffic"611         )612         self.service_pool = ThreadPoolExecutor(613             max_workers=aux_threadpool_size, thread_name_prefix=f"{name}-service"614         )615         #: The wire this worker speaks on, handed in by whoever runs it.616         self.stream: FrameStream | None = None617         #: The loop the wire lives on, taken with the wire: what a pool thread618         #: hops onto to place a call of its own.619         self.loop: asyncio.AbstractEventLoop | None = None620         #: One future per CALL this worker placed upward, by frame id: the read621         #: loop resolves them as the answers land, in whatever order they do.622         self._parent_calls: dict[str, asyncio.Future[Frame]] = {}623         self._parent_call_paths: dict[str, str] = {}624         self._abandoned_parent_calls: dict[str, str] = {}625         #: The consumer seam of the http CALL form: an ASGI application a626         #: subclass assigns. None here — this class hosts no application of its627         #: own. Whoever assigns it also gives it this worker, the way the628         #: genropy bridge gives its hosted site its ``spa_worker``: the core629         #: writes no live object into a scope (owner, 2026-09-06). A task the630         #: application spawns inherits the context at creation, but mutating631         #: the registers AFTER the reply has left raises: the work of a request632         #: ends inside the request, as it does for a WSGI site.633         self.asgi_app: Any = None634         #: The shortcut for whoever hosts WSGI only: a WSGI callable the core635         #: wraps in a ``WsgiSeam`` itself. Alternative to ``asgi_app``, never636         #: an addition — both assigned is a configuration error.637         self.wsgi_app: Callable[..., Any] | None = None638         self.dispatch_lock = threading.RLock()639         #: The rows of the three registers, and the lifecycle vocabulary that640         #: moves them: the shared registry, built through its own hook.641         self.registry = self.build_registry()642         #: Whether every register mutation of this process is reported up the643         #: lane as it happens. Off until somebody watches, and switched only by644         #: the vertex: a debug surface must not cost anything when unobserved.645         self.observation_on = False646         #: One slot per CALL being served. Each CALL is served on a task of its647         #: own, and the stitching runs on the pool under a copy of that task's648         #: context, so the loop and the thread of one request see ONE slot and649         #: two requests never see each other's. None outside any CALL.650         self._request_slot_var: contextvars.ContextVar[RequestSlot | None] = (651             contextvars.ContextVar(f"request_slot:{name}", default=None)652         )653         self._global_store = GlobalStoreClient(self)654         self._announce_failures = 0655         self._observation_tasks: set[asyncio.Task[None]] = set()656         self._unfreeze_waits: dict[str, asyncio.Event] = {}657         self._pendings: dict[str, int] = {}658         #: One event per user a freeze order is waiting on: set by the end of659         #: his last call, which is the instant the order may park him.660         self._freeze_order_waits: dict[str, asyncio.Event] = {}661         self._transfer_flags: dict[str, str] = {}662         #: One entry per connection that logged in during a call and is663         #: waiting for that call's tail to carry it away: the identity it664         #: belonged to BEFORE, which is the only fact the tail needs.665         self._departing_users: set[str] = set()666         self._transfers_start_ts = 0.0667         self._transfers_done = asyncio.Event()668         self._transfers_done.set()669         self._transfers_changed = asyncio.Event()670         self._quitting = False671         self._freeze_failures = 0672         self._exited = False673         self._service_tasks: set[asyncio.Task[None]] = set()674         self._quit_task: asyncio.Task[None] | None = None675         #: The orders this process takes, as a tree: ``group/…`` and ``commander/…``.676         self.worker_dispatcher = WorkerDispatcher(self)677         self._snapshot_sent_ts = 0.0678         self._population_changed = False679         self._logger = logging.getLogger(__name__)680 681     def build_registry(self) -> RegisterRegistry:682         """Build the registry this worker holds its rows in.683 684         Returns:685             A fresh ``RegisterRegistry``.686 687         The seam a consumer replaces to pair its own row types with the tree:688         whoever hosts a site subclasses this and returns its own registry, and689         nothing else in this class names the concrete class.690         """691         return RegisterRegistry()692 693     @property694     def hosted_app_seam(self) -> AsgiSeam:695         """The seam onto the hosted application, the one seam a consumer assigned.696 697         Returns:698             The ASGI seam: on ``asgi_app`` when a consumer assigned one, on the699             WSGI adapter around ``wsgi_app`` when it took the shortcut instead.700 701         Raises:702             RuntimeError: both seams are assigned, or neither is — and the two703                 are not the same kind of trouble. BOTH is a contradiction704                 somebody declared, and ``WorkerEntry`` reads this at boot for705                 exactly that case, so the process dies before the wire exists.706                 NEITHER is the base worker, which is legitimate: it serves its707                 orders and hosts nothing, and it learns so here, when an http708                 CALL finally asks it to serve a request.709         """710         if self.asgi_app is not None and self.wsgi_app is not None:711             raise RuntimeError(712                 f"Worker {self.name}: asgi_app and wsgi_app are both assigned; "713                 "the WSGI shortcut is an alternative to the ASGI seam, not an addition"714             )715         if self.asgi_app is not None:716             return AsgiSeam(self.asgi_app)717         if self.wsgi_app is not None:718             return AsgiSeam(WsgiSeam(self.wsgi_app, self))719         raise RuntimeError(720             f"Worker {self.name}: neither asgi_app nor wsgi_app is assigned; "721             "this worker hosts no application"722         )723 724     async def run_sync(self, work: Callable[[], Any]) -> Any:725         """Run one piece of synchronous work on the traffic pool, in this CALL's slot.726 727         Args:728             work: what to run there — a WSGI callable, a legacy database call.729 730         Returns:731             Whatever the work returned.732 733         The thread runs under a COPY of the calling task's context, so it finds734         the request slot of the CALL it serves and whatever it announces rides735         that CALL's reply. This is how a hosted ASGI application does its736         synchronous work: the legacy database its group engine built is737         synchronous, and neither the loop nor the service pool may be held738         behind it.739         """740         return await self._run_in_pool(self.traffic_pool, work)741 742     def build_request_slot(self) -> RequestSlot:743         """Build the slot of one request: the seam a consumer replaces with its own slot class.744 745         Returns:746             A fresh ``RequestSlot``; a subclass adds the per-request state its747             verbs carry.748         """749         return RequestSlot()750 751     def on_request_served(self) -> None:752         """What runs at the end of every served request, failed ones included.753 754         Called in the ``finally`` of the stitching, on the pool thread, with the755         request's slot still open. The seam a consumer overrides to deliver what756         its verbs left on the slot; the core leaves nothing there.757         """758 759     @property760     def user_register(self) -> Register:761         """The users this worker holds, by identity.762 763         Returns:764             The live register — read it, and leave the writing to the mutators.765         """766         return self.registry.user_items767 768     @property769     def connection_register(self) -> Register:770         """The connections this worker holds, by cid.771 772         Returns:773             The live register — read it, and leave the writing to the mutators.774         """775         return self.registry.connection_items776 777     @property778     def page_register(self) -> Register:779         """The pages this worker holds, by page id.780 781         Returns:782             The live register — read it, and leave the writing to the mutators.783         """784         return self.registry.page_items785 786     @property787     def worker_events(self) -> list[dict[str, Any]]:788         """The worker events of the CALL being served, waiting for its reply.789 790         Returns:791             The live list of the current request slot: whoever composes the792             envelope takes them from here.793 794         Raises:795             RuntimeError: no CALL is being served in this context.796         """797         return self.request_slot.worker_events798 799     @property800     def freeze_failures(self) -> int:801         """How many departures the deposit refused since this worker was born.802 803         Returns:804             The count. Every one of them left a user alive and a loud line in805             the log; a number that grows is a disk to look at.806         """807         return self._freeze_failures808 809     @property810     def exited(self) -> bool:811         """Whether this worker has already left.812 813         Returns:814             True once ``exit_process`` was reached.815         """816         return self._exited817 818     @property819     def rss_bytes(self) -> int | None:820         """The resident set size of this process, in bytes.821 822         Returns:823             What psutil reads for this process on every platform, or None when824             the kernel refuses the reading — the photo carries the counts825             either way.826         """827         try:828             return psutil.Process().memory_info().rss829         except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):830             return None831 832     @property833     def pss_bytes(self) -> int | None:834         """The proportional set size of this process, in bytes.835 836         Returns:837             The ``pss`` psutil reads from the Linux ``smaps_rollup``. Shared838             pages are divided among the processes mapping them, unlike RSS, so839             summing this gauge across prefork workers does not charge the840             template's pages once per child. ``None`` is the honest answer on841             platforms whose full memory info has no ``pss`` field (macOS,842             Windows) or when the kernel refuses the reading.843         """844         try:845             return getattr(psutil.Process().memory_full_info(), "pss", None)846         except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):847             return None848 849     @property850     def worker_snapshot(self) -> dict[str, Any]:851         """What this process honestly knows of itself, ready for the wire.852 853         Returns:854             The aggregates of the process, one row per connection with its three855             clocks, and one pair per user — his projected item and the856             ``transfer_flag`` the last photo decided, ``None`` when he is going857             nowhere. Scalars only: the stores and the working fields are the858             application's business, never the observer's. Each user row carries859             his cumulative service counters — ``served_call_count``,860             ``service_seconds``, ``pending_call_count`` — raw readings the861             envelope layer turns into per-interval deltas. The CPU is measured862             by the commander through psutil, never by this photo. The counters863             live in the register item and never reach a frozen parcel: the freeze864             persists the store and the connections, not the row itself.865         """866         with self.dispatch_lock:867             return {868                 "pid": os.getpid(),869                 "name": self.name,870                 "group": self.group,871                 "rss_bytes": self.rss_bytes,872                 "pss_bytes": self.pss_bytes,873                 "user_count": len(self.user_register),874                 "connection_count": len(self.connection_register),875                 "page_count": len(self.page_register),876                 "connections": {877                     cid: {878                         "user": item["user"],879                         **{clock: item[clock] for clock in CLOCK_NAMES},880                     }881                     for cid, item in self._get_register_rows(self.connection_register)882                 },883                 "users": {884                     user: {885                         "item": self._user_row(user, item),886                         "transfer_flag": self._transfer_flags.get(user),887                     }888                     for user, item in self._get_register_rows(self.user_register)889                 },890             }891 892     def add_worker_event(self, op: str, **payload: Any) -> dict[str, Any]:893         """Queue one worker event for the envelope out.894 895         Args:896             op: the protocol name of what happened.897             payload: the entity keys that name it.898 899         Returns:900             The worker event as it was queued.901 902         Appends to the ``worker_events`` of the CALL being served, and marks the903         photo due when what happened is a user entering or leaving. Outside any904         CALL there is no slot and this raises: an event nobody would ever send905         is a fault, not a queue.906         """907         event = {"op": op, "worker": self.name, **payload}908         self.request_slot.worker_events.append(event)909         if self.observation_on:910             self.report_observation(op, payload)911         if op in POPULATION_WORKER_EVENTS:912             self._population_changed = True913         return event914 915     def add_user(self, user: str, **fields: Any) -> dict[str, Any]:916         """Bring a user into being on this worker and announce it.917 918         Args:919             user: the user identity.920             fields: anything else the item should carry, stored verbatim.921 922         Returns:923             The user register item.924 925         Adds the item and announces ``new_user``.926         """927         with self.dispatch_lock:928             item = self._add_user_item(user, **fields)929             self.add_worker_event("new_user", user=user)930             return item931 932     def add_connection(self, cid: str, user: str | None = None, **fields: Any) -> dict[str, Any]:933         """Bring a connection into being, born guest unless it is given a user.934 935         Args:936             cid: the connection identity.937             user: the user it belongs to; ``None`` is the anonymous reception,938                 which names it ``GUEST_PREFIX`` + the cid.939             fields: anything else the item should carry, stored verbatim.940 941         Returns:942             The connection register item.943 944         Adds the item — with the user above it when that user is unseen — and945         announces the cascade in the order it happened.946         """947         with self.dispatch_lock:948             user = user or GUEST_PREFIX + cid949             if user not in self.user_register:950                 self.add_user(user)951             item = self._add_connection_item(cid, user, **fields)952             self.request_slot.connection_id = cid953             self.add_worker_event("new_connection", user=user, connection_id=cid)954             return item955 956     def add_page(957         self, page_id: str, cid: str, user: str | None = None, **fields: Any958     ) -> dict[str, Any]:959         """Bring a page into being under its connection and announce it.960 961         Args:962             page_id: the page identity.963             cid: the connection the page belongs to.964             user: the user to hang an unseen connection from; ignored when the965                 connection is already here.966             fields: anything else the item should carry, stored verbatim.967 968         Returns:969             The page register item.970 971         Adds the item — with the connection and the user above it when they are972         unseen — and announces the cascade in the order it happened.973         """974         with self.dispatch_lock:975             if cid not in self.connection_register:976                 self.add_connection(cid, user)977             item = self._add_page_item(page_id, cid, **fields)978             self.add_worker_event(979                 "new_page",980                 user=self.registry.page_user(page_id),981                 page_id=page_id,982                 connection_id=cid,983                 **item.announcement_fields(),984             )985             return item986 987     def new_connection(self, identity: str, **fields: Any) -> dict[str, Any]:988         """Open a connection in the form the site calls it, identity first.989 990         Args:991             identity: the connection identity — the session id.992             fields: anything else the row should carry; ``user`` names the owner993                 and its absence is the anonymous reception.994 995         Returns:996             The connection register item.997 998         Acts on the registers through ``add_connection``, so the announcements999         rise exactly as they do for the reception.1000         """1001         return self.add_connection(identity, fields.pop("user", None), **fields)1002 1003     def new_page(1004         self, identity: str, page_id: str, **fields: Any1005     ) -> dict[str, Any]:1006         """Open a page under its user in the form the site calls it.1007 1008         Args:1009             identity: the user the page belongs to.1010             page_id: the page identity.1011             fields: anything else the row should carry; ``connection_id``1012                 names the connection it hangs from.1013 1014         Returns:1015             The page register item.1016 1017         Acts on the registers through ``add_page``, which brings the connection1018         and the user above it into being when they are unseen and announces the1019         cascade in the order it happened.1020         """1021         cid = fields.pop("connection_id", None)1022         return self.add_page(page_id, cid, identity, **fields)1023 1024     def drop_page(self, identity: str, page_id: str) -> None:1025         """Take one page off this worker, and whatever it was the last of.1026 1027         Args:1028             identity: the user the site names as the page's owner; the owner1029                 this worker acts on is derived through the chain.1030             page_id: the page to be gone.1031 1032         Removes the item and announces ``drop_page``, then the1033         ``drop_connection`` and ``drop_user`` its departure empties. A page1034         already gone is the same outcome: nothing happens and nothing is said.1035         """1036         with self.dispatch_lock:1037             page = self.page_register.get(page_id)1038             if page is None:1039                 return1040             cid = page["connection_id"]1041             user = self.registry.page_user(page_id)1042             self._remove_page_item(page_id)1043             self.add_worker_event("drop_page", user=user, page_id=page_id)1044             if not self.connection_register.get(cid)["pages"]:1045                 self._remove_connection_item(cid)1046                 self.add_worker_event("drop_connection", user=user, connection_id=cid)1047                 self._drop_emptied_user(user)1048 1049     def drop_connection(self, identity: str, connection_id: str) -> None:1050         """Take a whole connection off this worker, its pages first.1051 1052         Args:1053             identity: the user the site names as the connection's owner; the1054                 owner this worker acts on is read off the row.1055             connection_id: the connection to be gone.1056 1057         Raises:1058             KeyError: this worker holds no such connection.1059 1060         Removes the pages and the connection, announcing ``drop_pages`` (when it1061         had any), ``drop_connection``, and ``drop_user`` if it was the user's1062         last.1063         """1064         cid = connection_id1065         with self.dispatch_lock:1066             item = self.connection_register.get(cid)1067             if item is None:1068                 raise KeyError(f"drop_connection: no connection {cid!r} here")1069             user = item["user"]1070             page_ids = sorted(item["pages"])1071             for page_id in page_ids:1072                 self._remove_page_item(page_id)1073             if page_ids:1074                 self.add_worker_event("drop_pages", user=user, page_ids=page_ids)1075             self._remove_connection_item(cid)1076             self.add_worker_event("drop_connection", user=user, connection_id=cid)1077             self._drop_emptied_user(user)1078 1079     def drop_user(self, user: str) -> None:1080         """Take a user off this worker with everything under him.1081 1082         Args:1083             user: the user to be gone.1084 1085         Removes the pages, the connections and the user, announcing1086         ``drop_pages`` and ``drop_connections`` for what he had and ``drop_user``1087         last. A user already gone is the same outcome.1088         """1089         with self.dispatch_lock:1090             item = self.user_register.get(user)1091             if item is None:1092                 return1093             connection_ids = sorted(item["connections"])1094             page_ids = sorted(1095                 page_id1096                 for cid in connection_ids1097                 for page_id in self.connection_register.get(cid)["pages"]1098             )1099             for page_id in page_ids:1100                 self._remove_page_item(page_id)1101             if page_ids:1102                 self.add_worker_event("drop_pages", user=user, page_ids=page_ids)1103             for cid in connection_ids:1104                 self._remove_connection_item(cid)1105             if connection_ids:1106                 self.add_worker_event(1107                     "drop_connections", user=user, connection_ids=connection_ids1108                 )1109             self._remove_user_item(user)1110             self._unfreeze_waits.pop(user, None)1111             self.add_worker_event("drop_user", user=user)1112 1113     def refresh_chain(self, page_id: str, *clocks: str) -> float:1114         """Stamp a page and the chain above it with the server's own instant.1115 1116         Args:1117             page_id: the page the contact came in on.1118             clocks: the clocks the contact deserves besides ``last_refresh_ts``,1119                 which every contact stamps — ``last_user_ts`` for a human event,1120                 ``last_rpc_ts`` for a real call.1121 1122         Returns:1123             The instant written, the same on all three levels.1124 1125         Raises:1126             KeyError: no such page here.1127 1128         Stamps the page item, its connection item and its user item.1129         """1130         with self.dispatch_lock:1131             owner = self.registry.page_user(page_id)1132             page = self.page_register.get(page_id)1133             connection = self.connection_register.get(page["connection_id"])1134             user = self.user_register.get(owner)1135             return self._stamp_items((page, connection, user), clocks)1136 1137     # ------------------------------------------------------------------1138     # The local data plane: what a page hears about, what it collects, and1139     # the drain that hands both species over. Every write addressed here is1140     # LOCAL by stickiness — the target page lives on this worker of fact —1141     # so nothing ascends, while the addressing the second pass will route1142     # already rides in the signatures.1143     # ------------------------------------------------------------------1144 1145     @property1146     def request_slot(self) -> RequestSlot:1147         """The slot of the CALL being served in this context.1148 1149         Raises:1150             RuntimeError: no CALL is being served here — nothing opened a slot.1151         """1152         slot = self._request_slot_var.get()1153         if slot is None:1154             raise RuntimeError(f"Worker {self.name}: no request slot open in this context")1155         return slot1156 1157     def open_request_slot(self) -> RequestSlot:1158         """Put a fresh slot in this context, so no CALL inherits another's events.1159 1160         Returns:1161             The slot just opened.1162 1163         Called on the task that serves a CALL before anything is done for it;1164         the pool thread the stitching runs on inherits it through the copied1165         context. Whatever a previous slot of this context held goes with it.1166         """1167         slot = self.build_request_slot()1168         self._request_slot_var.set(slot)1169         return slot1170 1171     @property1172     def global_store(self) -> GlobalStoreClient:1173         """This worker's side of the global store: simple operations and ``for_update``.1174 1175         NEVER open a turn while holding ``dispatch_lock``: both halves place a1176         lane call, and a holder would park on its own lock.1177         """1178         return self._global_store1179 1180     def attach_stream(self, stream: FrameStream) -> None:1181         """Take the wire this worker speaks on.1182 1183         Args:1184             stream: the frame codec over the connection to the handler.1185 1186         Sets ``stream`` and ``loop``. The worker never opens the wire: whoever1187         runs this worker in a process connects, and hands the connection over —1188         from that process's own loop, which is the one the lane then lives on.1189         """1190         self.stream = stream1191         self._abandoned_parent_calls.clear()1192         self.loop = asyncio.get_running_loop()1193 1194     async def send_presentation(self, config: dict[str, Any]) -> None:1195         """Present this process on the wire and install the store that answers.1196 1197         Args:1198             config: the spawn payload this process was built from, echoed back1199                 so the handler sees what its child understood of it.1200 1201         Returns once the handler has answered, which is what tells this process1202         it is on the wire.1203         """1204         await self.stream.write(1205             Frame(1206                 method=REGISTER_METHOD,1207                 path=REGISTER_PATH,1208                 info=self._outbound({ENVELOPE_SLOT_PRESENTATION: os.getpid(), "config": config}),1209             )1210         )1211         await self.stream.read()1212 1213     async def receive_frames(self) -> None:1214         """Read the handler's envelopes until the wire ends.1215 1216         Returns when the wire is gone — EOF, the death signal on a same-host1217         socket — or a protocol violation closed it. What a worker without a wire1218         does is not decided here: the caller asks for ``on_wire_lost``. Every1219         CALL still parked for an answer is failed first, with ``ConnectionError``:1220         an answer that can no longer arrive must not be waited for.1221         """1222         try:1223             while True:1224                 try:1225                     frame = await self.stream.read()1226                 except ValueError:1227                     self._logger.exception(1228                         "Worker %s: protocol violation from its handler; leaving the wire",1229                         self.name,1230                     )1231                     return1232                 if frame is None:1233                     return1234                 try:1235                     self.handle_frame(frame)1236                 except ValueError:1237                     self._logger.exception(1238                         "Worker %s: protocol violation routing a handler reply; leaving the wire",1239                         self.name,1240                     )1241                     await self.stream.close()1242                     return1243         finally:1244             self._fail_parent_calls(ConnectionError("the wire to the handler ended"))1245 1246     def _fail_parent_calls(self, cause: BaseException) -> None:1247         """Wake every parked CALL with ``cause``; the answers will never come."""1248         for future in list(self._parent_calls.values()):1249             if not future.done():1250                 future.set_exception(cause)1251 1252     def handle_frame(self, frame: Frame) -> None:1253         """Route one envelope from the handler: a CALL to serve, a REPLY to resolve.1254 1255         Args:1256             frame: the envelope as it came off the wire.1257 1258         A CALL is served on its own task, so a long op cannot make this worker1259         deaf to the next one; a REPLY is the answer to a call this worker placed1260         upward and is resolved inline, on the frame id it was parked under.1261         """1262         if frame.method == CALL_METHOD:1263             task = asyncio.create_task(self._guarded_call(frame))1264             self._service_tasks.add(task)1265             task.add_done_callback(self._service_tasks.discard)1266         elif frame.method == REPLY_METHOD:1267             self._resolve_parent_reply(frame)1268         else:1269             self._logger.warning(1270                 "Worker %s: unexpected envelope %s from its handler", self.name, frame.method1271             )1272 1273     def eval_expression(self, expr: str) -> str:1274         """Evaluate one debug expression against this process, repr back.1275 1276         Args:1277             expr: a Python expression; ``worker`` names this instance.1278 1279         Returns:1280             The ``repr`` of the value — the debug door's whole answer, so1281             anything unforeseen is readable without a tool having predicted it.1282 1283         Evaluates under ``dispatch_lock``, so the rows it reads are coherent.1284         Full eval power by construction: the door exists only where the1285         console surface was mounted on purpose, never in production.1286         """1287         with self.dispatch_lock:1288             return repr(eval(expr, {"worker": self}))1289 1290     def report_observation(self, kind: str, data: dict[str, Any]) -> None:1291         """Send one observation up the lane, and forget about it.1292 1293         Args:1294             kind: the mutation it reports, named as the worker event is.1295             data: the keys that name what moved, JSON-safe as they travel.1296 1297         Best-effort by design: the answer is never read and a failure is logged1298         and dropped. Nothing here may raise into the traffic path — an observer1299         that changes what it observes is worse than no observer.1300         """1301         payload = {"kind": kind, "source": self.name, "data": data}1302         try:1303             self.loop.call_soon_threadsafe(self._start_observation_call, payload)1304         except (AttributeError, RuntimeError) as exc:1305             self._logger.debug("Worker %s: observation %s dropped (%s)", self.name, kind, exc)1306 1307     def _start_observation_call(self, payload: dict[str, Any]) -> None:1308         """Put one observation on the lane as a detached task, on the loop."""1309         task = asyncio.create_task(self._deliver_observation(payload))1310         self._observation_tasks.add(task)1311         task.add_done_callback(self._observation_tasks.discard)1312 1313     async def _deliver_observation(self, payload: dict[str, Any]) -> None:1314         """Await the observation's own answer, so a failure is logged and no more."""1315         try:1316             await self.call(OBSERVATION_OP_PATH, payload)1317         except Exception as exc:1318             self._logger.debug(1319                 "Worker %s: observation %s lost (%s)", self.name, payload["kind"], exc1320             )1321 1322     def census(self) -> dict[str, Any]:1323         """The whole process read out for a human: every register, JSON-safe.1324 1325         Returns:1326             The three registers key by key with their scalar fields. Live1327             objects are left out by construction — only what survives1328             ``json.dumps`` is in here.1329         """1330         with self.dispatch_lock:1331             return {1332                 "name": self.name,1333                 "group": self.group,1334                 "pid": os.getpid(),1335                 "user_register": self._census_register(self.user_register),1336                 "connection_register": self._census_register(self.connection_register),1337                 "page_register": self._census_register(self.page_register),1338             }1339 1340     def _census_register(self, register: Register) -> dict[str, Any]:1341         """One register key by key, each item reduced to its JSON-safe fields."""1342         return {1343             key: {1344                 field: self._census_field(value)1345                 for field, value in register.get(key).items()1346                 if self._census_field(value) is not _NOT_JSON_SAFE1347             }1348             for key in register.keys()1349         }1350 1351     def _census_field(self, value: Any) -> Any:1352         """One field as the census carries it, or ``_NOT_JSON_SAFE`` to leave it out.1353 1354         Args:1355             value: whatever the item holds under that name.1356 1357         Returns:1358             The value itself when it is a scalar, the sorted elements of a1359             container of scalars (the keys, for a dict), the count when those1360             elements are objects — never the objects themselves.1361         """1362         if value is None or isinstance(value, (str, int, float, bool)):1363             return value1364         if isinstance(value, dict):1365             value = list(value)1366         if isinstance(value, (set, frozenset, list, tuple)):1367             if all(1368                 element is None or isinstance(element, (str, int, float, bool))1369                 for element in value1370             ):1371                 return sorted(str(element) for element in value)1372             return len(value)1373         return _NOT_JSON_SAFE1374 1375     async def answer_call(self, frame: Frame) -> None:1376         """Answer one CALL: the http form, or an order resolved on the dispatcher.1377 1378         Args:1379             frame: the CALL as it came off the wire.1380 1381         Sends exactly one REPLY, whatever the outcome: the order's result, or1382         its error — ``NotFound`` for a path nobody serves, ``TypeError`` for a1383         payload that does not fit the order's signature, both raised before any1384         body runs. The HTTP form is selected by format metadata; only this1385         destination opens the application's HTTP record.1386         """1387         wire_format = frame.info.get("format")1388         if wire_format == "http":1389             info = frame.info1390             if set(info) - {"format", "cid", "page_id", "reply_path", "identity", "user_frozen"}:1391                 raise ValueError("unexpected HTTP routing metadata")1392             if "cid" not in info:1393                 raise ValueError("HTTP routing metadata is missing cid")1394             for key in ("cid", "page_id", "reply_path", "identity"):1395                 if info.get(key) is not None and not isinstance(info[key], str):1396                     raise ValueError(f"invalid HTTP routing {key}")1397             if not isinstance(info.get("user_frozen", False), bool):1398                 raise ValueError("user_frozen must be a boolean")1399             scope, body = HttpRecord().decode_request(frame.payload)1400             http = {**scope, "body": body,1401                     "headers": [(n.decode("latin-1"), v.decode("latin-1"))1402                                 for n, v in scope.get("headers", [])],1403                     "query_string": scope.get("query_string", b"").decode("latin-1"),1404                     "cid": frame.info.get("cid")}1405             for key in ("page_id", "reply_path"):1406                 if key in frame.info:1407                     http[key] = frame.info[key]1408             await self.serve_http(frame, {**frame.info, "http": http})1409             return1410         if wire_format not in (None, "control-json"):1411             raise ValueError("unsupported parent call payload format")1412         payload = ControlPayload().decode(frame.payload) or {}1413         if "http" in payload:1414             raise ValueError("HTTP requests require an opaque HTTP frame")1415         if "wsx" in payload:1416             if (frame.info.get("identity") is not None1417                     and not isinstance(frame.info["identity"], str)):1418                 raise ValueError("invalid WSX routing identity")1419             if not isinstance(frame.info.get("user_frozen", False), bool):1420                 raise ValueError("WSX user_frozen must be a boolean")1421             payload = {**payload, "identity": frame.info.get("identity"),1422                        "user_frozen": frame.info.get("user_frozen", False)}1423             await self.serve_wsx(frame, payload)1424             return1425         try:1426             result = self.worker_dispatcher.route.node(frame.path)(**payload)1427             if inspect.isawaitable(result):1428                 result = await result1429         except Exception as exc:1430             await self.send_reply(frame, error=f"{type(exc).__name__}: {exc}")1431         else:1432             await self.send_reply(frame, result=result)1433 1434     async def serve_http(self, frame: Frame, payload: dict[str, Any]) -> None:1435         """Serve the http CALL form through the WSGI seam, or refuse it.1436 1437         Args:1438             frame: the CALL being answered.1439             payload: its whole payload — the ``http`` dict the front packed, the1440                 ``identity`` to route on and the ``user_frozen`` verdict, which1441                 belong together because the row is resolved from all three.1442 1443         Whether this worker hosts anything at all is ``hosted_app_seam``'s1444         judgment, and it was already made at boot: what can still go wrong here1445         comes back as this same REPLY — a caller is answered once, always, and1446         a deposit that refuses a parcel must not leave a browser waiting for a1447         timeout.1448         """1449         try:1450             result: Any = await self._serve_request(payload)1451             error: Any = None1452             status: int | None = None1453         except HTTPException as refused:1454             # A refusal of the CLIENT: it knows what to answer, and the words1455             # are the ones the browser must read.1456             result, error, status = None, refused.detail, refused.status1457         except Exception as exc:1458             self._logger.exception("Worker %s: http CALL %s failed", self.name, frame.path)1459             result, error, status = None, f"{type(exc).__name__}: {exc}", None1460         await self.send_reply(frame, result=result, error=error, status=status)1461 1462     async def serve_wsx(self, frame: Frame, payload: dict[str, Any]) -> None:1463         """Serve the ``wsx`` CALL form: a channel command of one page.1464 1465         Args:1466             frame: the CALL being answered.1467             payload: its whole payload — the ``wsx`` dict the front composed,1468                 the ``identity`` to route on and the ``user_frozen`` verdict.1469 1470         The command runs through the SAME prologue an http request does: the1471         call is written in the user's pendings, and his row is put in order —1472         adopted from the deposit when the verdict says he was frozen, because a1473         page's row is not in memory until its user is, and there would be1474         nowhere to write the channel. What differs is the middle: instead of1475         the hosted application, the command is resolved on this worker's own1476         dispatcher.1477         """1478         try:1479             result: Any = await self._serve_command(payload)1480             error: Any = None1481             status: int | None = None1482         except HTTPException as refused:1483             result, error, status = None, refused.detail, refused.status1484         except Exception as exc:1485             self._logger.exception("Worker %s: wsx CALL %s failed", self.name, frame.path)1486             result, error, status = None, f"{type(exc).__name__}: {exc}", None1487         await self.send_reply(frame, result=result, error=error, status=status)1488 1489     async def send_reply(1490         self, frame: Frame, *, result: Any = None, error: Any = None, status: int | None = None1491     ) -> None:1492         """Answer a CALL, carrying what happened here while it was being served.1493 1494         Args:1495             frame: the CALL being answered; its id is what makes this a REPLY.1496             result: the answer, when there is one.1497             error: what went wrong instead.1498             status: what the parent should answer, when this worker knows —1499                 only a refusal knows. Absent from every other failure, and the1500                 front then answers as it always did.1501 1502         Empties the ``worker_events`` of this CALL's slot onto the envelope — the1503         worker events are delivered once, and the send IS the delivery — and1504         attaches the photo when it is due. Then the slot is closed: an event1505         offered after the reply of the CALL that could have carried it is a1506         fault, and ``request_slot`` says so.1507         """1508         with self.dispatch_lock:1509             slot = self.request_slot1510             events = slot.worker_events1511         info: dict[str, Any] = {ENVELOPE_SLOT_WORKER_EVENTS: events}1512         if error is not None:1513             info["error"] = error1514             if status is not None:1515                 info["status"] = status1516             payload = ControlPayload().encode({"error": error, **({"status": status} if status else {})})1517             info["format"] = "control-json"1518         elif frame.info.get("format") == "http":1519             info["format"] = "http"1520             info["connection_id"] = result.get("connection_id")1521             payload = HttpRecord().encode_response(1522                 {key: result[key] for key in ("status", "headers", "body")}1523             )1524         else:1525             info["format"] = "control-json"1526             payload = ControlPayload().encode({"result": result})1527         reply = Frame(id=frame.id, method=REPLY_METHOD, path=frame.path,1528                       info=self._outbound(info), payload=payload)1529         # Serialization failures above leave this call's slot available for an1530         # explicit error reply. Once sending begins, failure is uncertain: close1531         # the wire so its caller fails instead of waiting forever or replaying.1532         try:1533             await self.stream.write(reply)1534         except FrameTooLarge:1535             # No bytes were written. Keep the events for the error reply and1536             # ensure the snapshot rejected with this response is sent again.1537             if ENVELOPE_SLOT_WORKER_SNAPSHOT in reply.info:1538                 self._population_changed = True1539             raise1540         except Exception:1541             self._request_slot_var.set(None)1542             await self.stream.close()1543             raise1544         with self.dispatch_lock:1545             slot.worker_events = []1546         self._request_slot_var.set(None)1547 1548     async def call(1549         self, path: str, data: Any = None, timeout: float | None = None1550     ) -> Any:1551         """CALL the commander on the lane and await its answer.1552 1553         Args:1554             path: the routing key of the call, which is what the parent serves.1555             data: the payload, JSON-serializable.1556             timeout: the caller's own deadline; None waits until the answer lands.1557 1558         Returns:1559             The ``result`` the parent put in its REPLY.1560 1561         Raises:1562             CommanderCallFailed: the answer carried an ``error`` instead.1563 1564         The frame carries its own id and the future is parked under it, so calls1565         placed without awaiting the first resolve each with its own answer, in1566         whatever order the parent gives them. Reachable from a pool thread1567         through ``run_on_loop``.1568         """1569         frame = Frame(method=CALL_METHOD, path=path, info={"format": "control-json"},1570                       payload=ControlPayload().encode(data))1571         reply_frame = await self.call_frame(frame, timeout=timeout)1572         reply = ControlPayload().decode(reply_frame.payload) or {}1573         if "error" in reply:1574             raise CommanderCallFailed(path, str(reply["error"]))1575         return reply.get("result")1576 1577     async def call_frame(self, frame: Frame, timeout: float | None = None) -> Frame:1578         """Call the parent with opaque bytes; parked calls belong to this wire."""1579         if frame.method != CALL_METHOD:1580             raise ValueError("parent request/reply calls require method CALL")1581         if frame.id in self._parent_calls or frame.id in self._abandoned_parent_calls:1582             raise ValueError("duplicate in-flight correlation id")1583         if len(self._parent_calls) >= 256:1584             raise ConnectionError("parent call capacity exhausted; request not sent")1585         future: asyncio.Future[Frame] = asyncio.get_running_loop().create_future()1586         self._parent_calls[frame.id] = future1587         self._parent_call_paths[frame.id] = frame.path1588         sent = False1589         try:1590             sent = True1591             await self.stream.write(frame)1592             if timeout is None:1593                 return await asyncio.shield(future)1594             return await asyncio.wait_for(asyncio.shield(future), timeout)1595         except (asyncio.CancelledError, TimeoutError):1596             if sent and not future.done():1597                 if len(self._abandoned_parent_calls) >= 256:1598                     await self.stream.close()1599                 else:1600                     self._abandoned_parent_calls[frame.id] = frame.path1601             raise1602         finally:1603             self._parent_calls.pop(frame.id, None)1604             self._parent_call_paths.pop(frame.id, None)1605             if not future.done():1606                 future.cancel()1607 1608     def send_message(self, page_id: str, path: str, data: Any = None) -> bool:1609         """Write one message of the site onto the page's websocket.1610 1611         Args:1612             page_id: the page to address — one of this worker's own.1613             path: what the client routes the message on.1614             data: the payload, as a Python value.1615 1616         Returns:1617             ``True`` when the message reached a socket, ``False`` when that1618             page speaks on none.1619 1620         Callable from a traffic-pool thread, which is where the site's own code1621         runs: the CALL goes up on the loop and this thread waits for its REPLY.1622         Fire and forget in meaning, not in mechanics — delivered says written1623         to the socket, never executed by the page (W-12).1624 1625         The payload is serialised HERE, before it reaches the lane: the lane is1626         opaque bytes. The browser's codec reads the JSON TYTX value; the1627         frontend only wraps that serialized text in the public WSX envelope.1628         """1629         with self.dispatch_lock:1630             row = self.page_register.get(page_id)1631             cid = row.get("connection_id") if row is not None else None1632         encoded = to_tytx(data, "json") if data is not None else None1633         reply_frame = self.run_on_loop(self.call_frame(Frame(1634             method=CALL_METHOD, path=WEBSOCKET_SEND_PATH,1635             info={"format": "wsx-json", "page_id": page_id, "reply_path": path,1636                   "cid": cid, "data_present": encoded is not None},1637             payload=encoded.encode("utf-8") if encoded is not None else b"")))1638         reply = ControlPayload().decode(reply_frame.payload) or {}1639         if "error" in reply:1640             raise CommanderCallFailed(WEBSOCKET_SEND_PATH, str(reply["error"]))1641         return bool((reply.get("result") or {}).get("delivered"))1642 1643     def run_on_loop(self, coro: Any) -> Any:1644         """Run a coroutine on this worker's loop from a pool thread, and wait.1645 1646         Args:1647             coro: what to run there — a ``call`` of this worker's, in practice.1648 1649         Returns:1650             Whatever the coroutine returned, on the calling thread.1651 1652         The bridge the site's own verbs need: they are served on a traffic-pool1653         thread, where blocking costs nothing, and the lane lives on the loop.1654         """1655         return asyncio.run_coroutine_threadsafe(coro, self.loop).result()1656 1657     def _resolve_parent_reply(self, frame: Frame) -> None:1658         """Hand the answer to the parked call; a caller already gone drops it."""1659         abandoned_path = self._abandoned_parent_calls.get(frame.id)1660         if abandoned_path is not None:1661             if frame.path != abandoned_path:1662                 raise ValueError("abandoned parent REPLY does not belong to its route")1663             self._abandoned_parent_calls.pop(frame.id, None)1664             return1665         future = self._parent_calls.get(frame.id)1666         if future is None or future.done():1667             self._logger.debug("Worker %s: the REPLY %s has no parked call", self.name, frame.id)1668             return1669         if frame.path != self._parent_call_paths[frame.id]:1670             raise ValueError("parent REPLY does not belong to its parked route")1671         future.set_result(frame)1672 1673     async def on_wire_lost(self) -> None:1674         """The wire is gone: leave, and save nothing.1675 1676         A process that lost its wire is unhealthy, so what it holds is not1677         vouched for: it writes no parcel to the deposit. Its users are lost at1678         the vertex, which drops what such a worker leaves and counts them in1679         ``frozen_users_discarded``.1680         """1681         self._logger.warning("Worker %s: its wire is gone — leaving, saving nothing", self.name)1682         self.exit_process()1683 1684     async def adopt_user(self, user: str) -> dict[str, Any]:1685         """Bring a user's store home from the deposit — the pull of the unified row.1686 1687         Args:1688             user: the user the envelope authorised, under its ``user_frozen``1689                 verdict.1690 1691         Returns:1692             The user register item, ``active``.1693 1694         Raises:1695             Whatever the deposit raised, to the call that made the trip; the1696             sisters of that burst are woken with a failure of their own.1697 1698         Adds the item as ``frozen`` when the user is unknown, marks it1699         ``unfreezing`` for the one call that makes the trip — the sisters of a1700         burst await that transition and read nothing — installs the parcel,1701         deletes it from the deposit and announces ``user_adopted``. A pull that1702         fails leaves NO row behind: his parcel is still in the deposit and the1703         mark is still on at the vertex, so the next request of his carries the1704         verdict again and the adoption is retried by construction. A resident1705         row would have to be reconciled with that verdict; an absent one is1706         already the truth.1707         """1708         with self.dispatch_lock:1709             item = self.user_register.get(user)1710             if item is None:1711                 item = self._add_user_item(user, state="frozen")1712             if item["state"] == "active":1713                 return item1714             waiting = self._unfreeze_waits.get(user)1715             if waiting is None:1716                 waiting = self._unfreeze_waits[user] = asyncio.Event()1717                 item["state"] = "unfreezing"1718                 mine = True1719             else:1720                 mine = False1721         if not mine:1722             await waiting.wait()1723             item = self.user_register.get(user)1724             if item is None:1725                 raise RuntimeError(1726                     f"the adoption of {user} failed in the call that made the trip; "1727                     "his parcel is still in the deposit"1728                 )1729             return item1730         try:1731             store = await self._take_from_deposit(user, self._read_user_parcel)1732             if store is None:1733                 self._logger.warning(1734                     "Worker %s: %s was announced frozen but has no store in the deposit",1735                     self.name,1736                     user,1737                 )1738             with self.dispatch_lock:1739                 if store is not None:1740                     item["store"] = store1741                 item["state"] = "active"1742                 self.add_worker_event("user_adopted", user=user)1743         finally:1744             with self.dispatch_lock:1745                 del self._unfreeze_waits[user]1746                 if item["state"] == "unfreezing":1747                     self._release_rows(user)1748                     self._logger.error(1749                         "Worker %s: the deposit would not give %s back; his row goes "1750                         "and the verdict on his next request retries the adoption",1751                         self.name,1752                         user,1753                     )1754             waiting.set()1755         return item1756 1757     async def adopt_connection(self, user: str, cid: str) -> dict[str, Any] | None:1758         """Look for this connection of ``user`` in the deposit, install it if it is there.1759 1760         Args:1761             user: the user the connection belongs to.1762             cid: the connection the request carries.1763 1764         Returns:1765             The connection register item — or None when nothing is held and1766             nothing is parked: the rows are the site's to bear, and it baptises1767             again while being served.1768 1769         Reads the parcel by itself (no verdict authorises a connection), deletes1770         it from the deposit and brings the connection and its pages into being1771         through the ordinary mutators: the worker events are the natural1772         ``new_connection``/``new_page``, never one of its own. ONE key is looked1773         up, because there is one identity: the deposit files the parcel under1774         the very id the cookie carries. A connection already held is already1775         home and spares the trip — a living row and a parked parcel of the same1776         connection cannot both exist.1777         """1778         with self.dispatch_lock:1779             item = self.connection_register.get(cid)1780         if item is not None:1781             return item1782         parcel = await self._take_from_deposit(user, self._read_connection_parcel, cid)1783         if parcel is None:1784             return None1785         with self.dispatch_lock:1786             resident = user in self.user_register1787             self.add_connection(cid, user, **parcel.get("connection", {}))1788             for page_id, fields in parcel.get("pages", {}).items():1789                 replayed = {1790                     key: fields.pop(key)1791                     for key in self.registry.page_row_class.fields_replayed1792                     if key in fields1793                 }1794                 page = self._add_page_item(page_id, cid, **fields)1795                 page.replay_fields(self.registry, replayed)1796                 # Announced AFTER the replay, so the event carries the1797                 # subscriptions the vertex rebuilds its index from.1798                 self.add_worker_event(1799                     "new_page",1800                     user=self.registry.page_user(page_id),1801                     page_id=page_id,1802                     connection_id=cid,1803                     **page.announcement_fields(),1804                 )1805             self._install_carried_store(user, parcel.get("store"), resident)1806             return self.connection_register.get(cid)1807 1808     def _install_carried_store(self, user: str, store: Any, resident: bool) -> None:1809         """Give a login's store to the row it belongs to, or let it die out loud.1810 1811         A connection that logged in carries what its guest had accumulated. It1812         becomes the user's own store when the row was born a moment ago with1813         this very connection; when a row of his was already here — his own state1814         came home first, or he is living on this worker already — the RESIDENT1815         wins and what the guest did before logging in dies, said out loud rather1816         than silently dropped. The caller holds the lock.1817         """1818         if store is None:1819             return1820         if resident:1821             self._logger.info(1822                 "Worker %s: %s was already here, so what his guest accumulated is dropped",1823                 self.name,1824                 user,1825             )1826             return1827         self.user_register.get(user)["store"] = store1828 1829     def change_connection_user(self, cid: str, user: str, **fields: Any) -> None:1830         """The login: this connection stops being anonymous and becomes his.1831 1832         Args:1833             cid: the connection that logged in.1834             user: the identity it belongs to from now on.1835             fields: whatever else the site puts on the connection row, stored1836                 verbatim.1837 1838         Raises:1839             ValueError: the target carries ``GUEST_PREFIX`` — nobody logs in as a1840                 guest, and the value crosses a border of trust to get here.1841             KeyError: this worker holds no such connection.1842 1843         Acts on the registers AT ONCE — the row changes owner, the user is born1844         here if he was unknown (the worker's way, with ``state`` and the three1845         clocks the photo reads), and the pages follow their connection without1846         being touched, their owner being derived through it — on the flag the1847         tail of this call reads, and on the departure a GUEST may have been1848         promised: one that is ceasing to exist is not carried to the deposit, so1849         his flag is dropped in this same breath. A previous identity that is no1850         guest KEEPS his: he stays here with whatever else he holds, the round1851         that promised his departure still means it, and cancelling it would leave1852         the wait on him at the vertex with nothing to release it. Announces the1853         login, which is what the vertex folds. The connection is written in this1854         request's slot as well: the login is the second way a request settles on1855         a connection the browser does not carry yet, and the cookie the front1856         writes on the way out is whatever the slot ends up holding.1857         """1858         if user.startswith(GUEST_PREFIX):1859             raise ValueError(f"{user!r} is reserved: nobody logs in as a guest")1860         with self.dispatch_lock:1861             connection = self.connection_register.get(cid)1862             if connection is None:1863                 raise KeyError(f"change_connection_user: no connection {cid!r} here")1864             previous_user = connection["user"]1865             if user not in self.user_register and not previous_user.startswith(GUEST_PREFIX):1866                 # An avatar switch onto an identity unknown here: the registry1867                 # would bring his row into being bare, and the photo reads1868                 # ``state`` and the three clocks off every row. Born here, the1869                 # worker's way, so the registry finds him and joins.1870                 self._add_user_item(user)1871             self.registry.change_connection_user(cid, user, **fields)1872             if previous_user.startswith(GUEST_PREFIX):1873                 self._transfer_flags.pop(previous_user, None)1874             slot = self.request_slot1875             slot.connection_id = cid1876             slot.connection_previous_user = previous_user1877             self.add_worker_event(1878                 "connection_user_changed",1879                 user=user,1880                 previous_user=previous_user,1881                 connection_id=cid,1882             )1883 1884     def open_request(self, user: str) -> None:1885         """Write one live call under the user it is for.1886 1887         Args:1888             user: the user the call belongs to.1889 1890         Adds to his pendings: nothing of his is parked while it is open.1891         """1892         with self.dispatch_lock:1893             self._pendings[user] = self._pendings.get(user, 0) + 11894 1895     async def close_request(self, user: str) -> None:1896         """Close one live call, and execute the departure that was waiting for it.1897 1898         Args:1899             user: the user the call belonged to.1900 1901         Takes the call out of his pendings and, when it was his last and a1902         departure of his is past the gate, lets him go now — the closure of a1903         whole worker and the cession of a single user hang on this same hook.1904         A user with nothing open was CUT by a quit that would not wait for this1905         call any longer: he is already parked, and there is nothing to close.1906         """1907         with self.dispatch_lock:1908             if user not in self._pendings:1909                 return1910             self._pendings[user] -= 11911             if self._pendings[user]:1912                 return1913             del self._pendings[user]1914             flag = self._transfer_flags.get(user)1915             waiting_order = self._freeze_order_waits.pop(user, None)1916         if waiting_order is not None:1917             waiting_order.set()1918         if flag is not None and self._transfers_open:1919             await self._execute_transfer(user, flag)1920 1921     async def freeze_designated_user(self, user: str) -> dict[str, Any]:1922         """Park one user on the parent's order, waiting for whatever holds him.1923 1924         Args:1925             user: the user the order names.1926 1927         Returns:1928             ``{"frozen": user}`` — sent only once he is in the deposit, so the1929             REPLY that carries it IS the confirmation.1930 1931         Raises:1932             KeyError: this worker does not host him.1933             RuntimeError: he stayed here for good — a departure already under1934                 way (a state the caller's own serialization makes impossible),1935                 or a deposit that refused his parcels.1936 1937         The waits are serialization, never policy: a pull bringing him home is1938         awaited on its own event, a call of his in flight on the event the end1939         of that call sets — and a call born while the parcels were on the disk1940         sends the order back to that same wait. The judgment of WHO sleeps is1941         the caller's alone; this verb only executes.1942         """1943         while True:1944             with self.dispatch_lock:1945                 if user not in self.user_register:1946                     raise KeyError(f"freeze order refused: no user {user!r} here")1947                 adopting = self._unfreeze_waits.get(user)1948                 drained = None1949                 if adopting is None and user in self._pendings:1950                     drained = self._freeze_order_waits.get(user)1951                     if drained is None:1952                         drained = self._freeze_order_waits[user] = asyncio.Event()1953             if adopting is not None:1954                 await adopting.wait()1955                 continue1956             if drained is not None:1957                 await drained.wait()1958                 continue1959             if not self._claim_departure(user):1960                 raise RuntimeError(1961                     f"freeze order refused: a departure of {user!r} is already under way"1962                 )1963             try:1964                 parked = await self.freeze_user(user)1965             finally:1966                 self._release_departure(user)1967             if parked:1968                 return {"frozen": user}1969             if parked is False:1970                 raise RuntimeError(f"freeze order failed: {user!r} stayed here")1971 1972     async def freeze_user(self, user: str) -> bool | None:1973         """Park a user in the deposit and announce that he left.1974 1975         Args:1976             user: the user leaving memory.1977 1978         Returns:1979             True when he went to the deposit; None when a call of his is what1980             holds him at the door — DEFERRED: that call's own end is where his1981             departure happens, and the flag that sent him here must stay1982             untouched; False when he STAYED for good as far as this attempt1983             goes — a row that is not ``active``, a semaphore that never came1984             free, or a deposit that refused the parcels (both failures counted,1985             B1).1986 1987         Writes his store and one parcel per connection under the folder1988         semaphore, announces ``user_frozen`` — placement always ``None``, the1989         vertex's to decide — and takes his rows out of memory whole. The row is1990         judged ONCE, at the door: whoever orders a departure BLOCKS him at the1991         vertex first (``GroupHandler.freeze_hosted_user``), so no work of his1992         can be born while the parcels are being written and nothing is ever1993         taken back off the deposit. A failed write aborts the whole departure:1994         the semaphore goes back, he stays alive where he is, nothing is1995         announced, and the failure is logged and counted.1996         """1997         with self.dispatch_lock:1998             item = self.user_register.get(user)1999             if item is None or item["state"] != "active":2000                 return False2001             if user in self._pendings:2002                 return None2003         try:2004             await self._take_folder_lock(user)2005         except TimeoutError:2006             self._freeze_failures += 12007             self._logger.error(2008                 "Worker %s: the folder of %s never came free; he stays here", self.name, user2009             )2010             return False2011         try:2012             store, connection_parcels = self._get_user_parcels(item)2013             await self._run_in_pool(2014                 self.service_pool,2015                 functools.partial(self._write_parcels, user, store, connection_parcels),2016             )2017             with self.dispatch_lock:2018                 self.add_worker_event("user_frozen", user=user, placement=None)2019                 self._release_rows(user)2020         except Exception:2021             self._freeze_failures += 12022             self._logger.exception(2023                 "Worker %s: the deposit refused the parcels of %s; he stays here",2024                 self.name,2025                 user,2026             )2027             return False2028         finally:2029             self.freeze_handler.release_lock(user, self.name)2030         return True2031 2032     async def freeze_connection(self, cid: str, previous_user: str) -> bool:2033         """Carry one logged-in connection to the deposit, under its new identity.2034 2035         Args:2036             cid: the connection the call that has just ended logged in.2037             previous_user: who owned it before that login — read off the slot of2038                 that very call, so no other call's tail can do this.2039 2040         Returns:2041             True when it went to the deposit; False when it stayed (somebody else2042             is already taking the previous identity away, or the deposit refused2043             the parcel, which is counted).2044 2045         Writes ONE parcel under the identity the connection now belongs to — the2046         connection, its pages, and the store the previous identity accumulated2047         when that identity was a guest, which is the only thing that makes an2048         anonymous visit survive its own login — then takes the rows out of memory:2049         the connection, its pages, the guest left behind, and the new identity2050         too when this was all it had here, so that his own next request finds a2051         row just born and installs the carried store instead of discarding it.2052         A departure that does not happen leaves EVERYTHING alive and announces2053         nothing: the identity stays resident on this worker with its connection2054         attached, which is a legitimate shape of the machine, and the failure is2055         counted. BOTH ways of not happening end there — a folder that never comes2056         free and a deposit that refuses the parcel — so the claim taken on the2057         previous identity is given back on every road out. A claim kept by a2058         departure that gave up would be held forever, and the whole worker could2059         never finish leaving.2060         """2061         with self.dispatch_lock:2062             user = self.connection_register.get(cid)["user"]2063         if not self._claim_departure(previous_user):2064             return False2065         try:2066             try:2067                 await self._take_folder_lock(user)2068             except TimeoutError:2069                 self._freeze_failures += 12070                 self._logger.error(2071                     "Worker %s: the folder of %s never came free; the connection of his "2072                     "login stays here",2073                     self.name,2074                     user,2075                 )2076                 return False2077             try:2078                 with self.dispatch_lock:2079                     parcel = self._connection_parcel(cid)2080                     if previous_user.startswith(GUEST_PREFIX):2081                         parcel["store"] = self.user_register.get(user)["store"]2082                     parcel = copy.deepcopy(parcel)2083                     self._detach_parcel_capture(parcel)2084                 await self._run_in_pool(2085                     self.service_pool,2086                     functools.partial(2087                         self.freeze_handler.write_connection_register_item,2088                         user,2089                         cid,2090                         parcel,2091                         writer=self.name,2092                         cause="login",2093                         group=self.group,2094                     ),2095                 )2096             except Exception:2097                 self._freeze_failures += 12098                 self._logger.exception(2099                     "Worker %s: the deposit refused the connection of %s; he stays here",2100                     self.name,2101                     user,2102                 )2103                 return False2104             finally:2105                 self.freeze_handler.release_lock(user, self.name)2106         finally:2107             self._release_departure(previous_user)2108         with self.dispatch_lock:2109             self._release_login_rows(cid, user, previous_user)2110         return True2111 2112     def plan_transfers(2113         self, *, transfer_users: Iterable[str] = ()2114     ) -> dict[str, tuple[dict[str, Any], str | None]]:2115         """Pair every user with the flag the next photo carries, and shut the gate.2116 2117         Args:2118             transfer_users: the users this round cedes, named by whoever judged2119                 them — the group reading the silence off the photo's clocks, or2120                 the quit ceding everybody. This rung names nobody.2121 2122         Returns:2123             Every user, mapped to his register item and his flag: ``None`` kept,2124             ``'T'`` ceded.2125 2126         Remembers the flags that are not ``None`` and starts the clock of the2127         gate: nothing departs before ``transfer_start_delay`` has passed. A row2128         that is not active is left to the vertex. Once ``quit`` has begun the2129         plan is terminal: everybody is ceded, a row still mid-adoption included2130         (he is a straggler, ceded as soon as his pull lands), no flag already2131         given is taken back, and the gate already open is not shut again. Every2132         plan that leaves flags behind wakes the cycle, so a man named here while2133         the quit waits for a straggler leaves with the others.2134         """2135         now = time.time()2136         ceded = set(transfer_users)2137         transfers: dict[str, tuple[dict[str, Any], str | None]] = {}2138         with self.dispatch_lock:2139             if not self._quitting:2140                 self._transfer_flags = {}2141             for user, item in self._get_register_rows(self.user_register):2142                 flag = self._transfer_flags.get(user)2143                 if flag is None and (self._quitting or user in ceded):2144                     if item["state"] == "active" or self._quitting:2145                         flag = "T"2146                 if flag is not None:2147                     self._transfer_flags[user] = flag2148                 transfers[user] = (item, flag)2149             if not self._quitting:2150                 self._transfers_start_ts = now + self.transfer_start_delay2151             if self._transfer_flags:2152                 # A flag is a promise the vertex must read: the photo that2153                 # carries it is due whatever the throttle says.2154                 self._population_changed = True2155                 self._transfers_done.clear()2156                 self._transfers_changed.set()2157             elif not self._quitting:2158                 self._transfers_done.set()2159         return transfers2160 2161     async def execute_transfers(self) -> None:2162         """Wait out the gate, then let the flagged users go, one at a time.2163 2164         The ceded go to the deposit as soon as no call of theirs is in flight;2165         whoever still has one is taken by the end of that call. The loop2166         breathes between two users.2167 2168         An ordinary cycle passes over the flags it found and comes back. The2169         cycle of a ``quit`` does not: it re-reads the flag map at every pass —2170         so a man a later shot names, or one whose adoption has only just landed,2171         leaves with the others — and returns only when the departures are over,2172         which is no flag left and nobody on his way out. Between two passes it2173         sleeps on the changes, never on a clock: a flag added, a departure2174         ended, and it looks again.2175         """2176         await asyncio.sleep(self._transfers_start_ts - time.time())2177         while True:2178             self._transfers_changed.clear()2179             for user, flag in list(self._transfer_flags.items()):2180                 await self._execute_transfer(user, flag)2181                 await asyncio.sleep(0)2182             if not self._quitting:2183                 return2184             self._settle_transfers()2185             if self._transfers_done.is_set():2186                 return2187             await self._transfers_changed.wait()2188 2189     async def quit(self, *, freezer_path: str | None = None) -> None:2190         """Leave: everybody departs, the last call is waited for, the process ends.2191 2192         Args:2193             freezer_path: where the parcels of THIS departure go, when they must2194                 not go to the working deposit — the reboot directory of a soft2195                 quit. The handler is replaced for good: nothing else will use2196                 this process's deposit.2197 2198         Flags every user for cession, waits the gate, parks them as their calls2199         end, and only then leaves the process. From here the plan is this2200         routine's: a shot taken while it waits for a straggler may name a2201         newborn among the departing, but can take nobody off them, and the2202         cycle picks that newborn up itself. A straggler is whoever is not gone2203         yet — a call of his in flight, or a pull of his still on the way — and2204         the exit is behind the last of them, because a process that left with2205         an adoption in flight would shut the pool it is running on. A departure2206         that fails is counted where it fails, so the exit is reached whatever2207         the disk says. Rebirth is not the worker's: whoever wants a successor2208         launches one.2209         """2210         if freezer_path is not None:2211             self.freeze_handler = FreezeHandler(freezer_path)2212         self._flag_everybody_for_departure()2213         departures = asyncio.ensure_future(self.execute_transfers())2214         done, _ = await asyncio.wait({departures}, timeout=PENDING_CALL_GRACE_SECONDS)2215         if not done:2216             self._cut_stragglers()2217             await departures2218         self.exit_process()2219 2220     def _cut_stragglers(self) -> None:2221         """Give up on the calls still in flight so their users can be parked.2222 2223         A call is not interrupted — the site runs it on a thread of the traffic2224         pool and a thread cannot be killed. What is dropped is the WAIT: the2225         users are taken out of the pendings, which is the one thing keeping2226         ``freeze_user`` from parking them, and the cycle is woken to take them.2227         An ordered freeze parked on the end of one of those calls is woken with2228         them: the wait it is on would otherwise never be set, since the call it2229         was waiting for is the one being given up on.2230         The call finishes into a process that is leaving and its answer is lost;2231         the front turns that into the same 503 a refusal gets, because the wire2232         died on a server that is quitting.2233 2234         Accepted, and weighed: a call stuck this long is waiting on something,2235         not writing, so the parcel it photographs is almost always quiet. The2236         rare loser is a write that lands after the photo — lost — or a pickle2237         that meets a store mid-change, which fails loudly on that user alone.2238         """2239         with self.dispatch_lock:2240             cut = list(self._pendings)2241             self._pendings.clear()2242             for waiting_order in self._freeze_order_waits.values():2243                 waiting_order.set()2244             self._freeze_order_waits.clear()2245         if cut:2246             self._logger.warning(2247                 "Worker %s: %s user(s) still had a call in flight %.1fs into the quit — "2248                 "parking them without it: %s",2249                 self.name,2250                 len(cut),2251                 PENDING_CALL_GRACE_SECONDS,2252                 ", ".join(sorted(cut)),2253             )2254         self._transfers_changed.set()2255 2256     def exit_process(self) -> None:2257         """Leave the process — the last act of ``quit`` and of ``on_wire_lost``.2258 2259         Closes the wire, which is what ends the read the shell is parked on, and2260         stops the two pools. Nothing here kills a process: the shell returns2261         from its run and the process ends with it.2262         """2263         self._exited = True2264         if self.stream is not None:2265             self.stream.writer.close()2266         self.traffic_pool.shutdown(wait=False)2267         self.service_pool.shutdown(wait=False)2268 2269     @property2270     def _snapshot_due(self) -> bool:2271         """Whether the next envelope out owes a photo: a change, or a stale one."""2272         return (2273             self._population_changed2274             or time.time() - self._snapshot_sent_ts >= self.worker_snapshot_ttl2275         )2276 2277     @property2278     def _transfers_open(self) -> bool:2279         """Whether the gate opened on the departures last announced."""2280         return time.time() >= self._transfers_start_ts2281 2282     def _outbound(self, data: dict[str, Any]) -> dict[str, Any]:2283         """Attach the photo to an envelope going out, when one is due."""2284         if not self._snapshot_due:2285             return data2286         data[ENVELOPE_SLOT_WORKER_SNAPSHOT] = self.worker_snapshot2287         self._snapshot_sent_ts = time.time()2288         self._population_changed = False2289         return data2290 2291     def begin_quit(self, *, freezer_path: str | None = None) -> None:2292         """Flag everybody for departure now, and leave on a task of its own.2293 2294         Args:2295             freezer_path: where the parcels of THIS departure go; see ``quit``.2296 2297         Acts on the flags before it returns, so the REPLY the caller sends next2298         carries the photo with every user ceded; ``quit`` runs on2299         ``_quit_task`` and reaches the wire only after that REPLY is on it.2300         """2301         self._flag_everybody_for_departure()2302         self._quit_task = asyncio.create_task(self.quit(freezer_path=freezer_path))2303 2304     def _flag_everybody_for_departure(self) -> None:2305         """Cede every user and make the plan terminal: sets ``_quitting`` and the flags."""2306         self._quitting = True2307         self.plan_transfers(transfer_users=self.user_register.keys())2308 2309     async def _guarded_call(self, frame: Frame) -> None:2310         """Serve one CALL on a slot of its own, with the guard inside the task.2311 2312         The slot is opened HERE, on the task that is this CALL's and nobody2313         else's: whatever the service announces — on the loop or on the pool2314         thread the stitching runs on — lands in it and leaves with THIS reply.2315         The guard keeps the task from dying unretrieved.2316         """2317         self.open_request_slot()2318         try:2319             await self.answer_call(frame)2320         except Exception as failure:2321             self._logger.exception("Worker %s: service of CALL %s failed", self.name, frame.path)2322             if self._request_slot_var.get() is not None:2323                 try:2324                     await self.send_reply(frame, error=f"{type(failure).__name__}: {failure}")2325                 except Exception:2326                     # A malformed/oversized control envelope cannot carry even2327                     # the error. Link loss must release every parked caller.2328                     await self.stream.close()2329                     self._logger.exception("Worker %s: error reply failed", self.name)2330 2331     async def _serve_request(self, payload: dict[str, Any]) -> dict[str, Any]:2332         """The pendings, the row and the stitching — everything that can fail as one.2333 2334         The call is written in the user's pendings FIRST, before the row is put2335         in order: the pendings cover the adoption too, so no departure can wake2336         in the gap between the loading and the serving of the same call. Then2337         the row (the store adopted when the verdict authorises it, the2338         connection found by itself, the clocks stamped), and then the hosted2339         application, awaited on the task of this CALL — the slot is that task's,2340         so nothing needs copying; a WSGI site behind the shortcut goes to the2341         traffic pool from inside the adapter, because WSGI is synchronous. The2342         end of the call is where a2343         departure that had to wait for it happens — on the connection the2344         SERVICE settled on, which is the one the site named while serving when2345         it named one, and the one the request came in on otherwise.2346         """2347         user = payload.get("identity")2348         served: dict[str, Any] = {}2349         async with self._serving(payload, payload["http"]["cid"]):2350             service_started = time.monotonic()2351             try:2352                 try:2353                     # The message is admitted FIRST: whether this page may speak2354                     # at all is the client's own business, and it is answered2355                     # before anybody asks what would have served it.2356                     async with self._page_queue(payload["http"].get("page_id")):2357                         seam = self.hosted_app_seam2358                         served = await seam.serve(payload["http"], payload.get("identity"))2359                 finally:2360                     # On a pool thread, with this request's slot still open and2361                     # BEFORE the counters, the pendings and the login's tail:2362                     # a consumer delivers here what its verbs left on the slot,2363                     # and does it from a thread, because it may call up the lane.2364                     await self.run_sync(self.on_request_served)2365                 served["connection_id"] = self.request_slot.connection_id2366             finally:2367                 # Counted whatever the stitching did: a call that failed or ran2368                 # long is exactly the one the measure must not lose.2369                 if user is not None:2370                     self._record_service(user, time.monotonic() - service_started)2371             return served2372 2373     @contextlib.asynccontextmanager2374     async def _page_queue(self, page_id: str | None) -> Any:2375         """Refuse a page with no open channel, or hold its queue for the whole call.2376 2377         Args:2378             page_id: the page this call belongs to, ``None`` for a request that2379                 names none — an ordinary http request of the site.2380 2381         Raises:2382             HTTPException: 409, when the page named here never opened its2383                 channel or was never born on this worker. A message of a page2384                 is refused before anything is served: ``openchannel`` is what2385                 makes a page addressable, and a message that skips it is a2386                 client out of step with its own row. It is the CLIENT's2387                 mistake, not the site's, so it carries a status of its own and2388                 reaches the browser with these words rather than as the 502 of2389                 an upstream that broke (#70).2390 2391         A page that opened its channel with ``sequential`` is served one call at2392         a time: the lock lives on ITS row, so pages never wait for each other,2393         and it is taken around the WHOLE call — the slot is open and the call is2394         already in the user's pendings, so a freeze waits for the queue too,2395         which is what it must do. Every other call passes straight through.2396         """2397         if page_id is None:2398             yield2399             return2400         with self.dispatch_lock:2401             row = self.page_register.get(page_id)2402         if row is None or row.get("wsx") is None:2403             raise HTTPException(2404                 409,2405                 f"page {page_id!r} has no open channel on this worker: "2406                 "send openchannel before any message of its own",2407             )2408         # The site owns the connection name; it may differ from the cookie2409         # carried by the request. This gate checks channel readiness and order.2410         channel = row["wsx"]2411         if not isinstance(channel, dict) or not channel.get("sequential"):2412             yield2413             return2414         async with row["call_lock"]:2415             yield2416 2417     @contextlib.asynccontextmanager2418     async def _serving(self, payload: dict[str, Any], cid: str | None) -> Any:2419         """The opening and the closing every form of call shares.2420 2421         Args:2422             payload: the CALL's whole payload — the identity is read from it.2423             cid: the connection this call came in on.2424 2425         The call is written in the user's pendings FIRST, before the row is put2426         in order: the pendings cover the adoption too, so no departure can wake2427         in the gap between the loading and the serving of the same call. At the2428         end the call leaves the pendings — which is where a departure that was2429         waiting for it happens — and the login's tail runs, on the connection2430         the SERVICE settled on.2431 2432         What differs between the forms is only what happens in between: the2433         hosted application for a request, the dispatcher for a channel command.2434         """2435         user = payload.get("identity")2436         if user is not None:2437             self.open_request(user)2438         try:2439             await self._resolve_row(user, cid, payload)2440             yield user2441         finally:2442             if user is not None:2443                 await self.close_request(user)2444             slot = self.request_slot2445             if slot.connection_previous_user is not None:2446                 await self.freeze_connection(slot.connection_id, slot.connection_previous_user)2447 2448     async def _serve_command(self, payload: dict[str, Any]) -> Any:2449         """The shared prologue, and the command resolved on this worker's tree."""2450         command = payload["wsx"]2451         async with self._serving(payload, command.get("cid")):2452             return self.worker_dispatcher.route.node(WSX_COMMAND_PATH)(2453                 **{key: value for key, value in command.items() if key != "cid"}2454             )2455 2456     async def _resolve_row(self, user: str | None, cid: str, payload: dict[str, Any]) -> None:2457         """Put the row of an incoming request in order.2458 2459         Who the user IS was decided by the caller: the identity the front2460         routed on, or None for a cookie the indexes do not carry yet — the2461         ANONYMOUS first visit, which touches no register: the site baptises2462         while serving, and the rows are born from its own verbs. For a known2463         user the store comes home only if the envelope authorises it, and the2464         connection is looked for in the deposit with no authorisation at all.2465         """2466         if user is None:2467             return2468         if payload.get("user_frozen"):2469             await self.adopt_user(user)2470         await self.adopt_connection(user, cid)2471         self._stamp_request(user, cid)2472 2473     def _record_service(self, user: str, seconds: float) -> None:2474         """Add one served call to the user's cumulative counters.2475 2476         Args:2477             user: whom the call belonged to.2478             seconds: how long the stitching held a traffic thread for it.2479 2480         Acts on his register item: ``served_call_count`` and ``service_seconds``2481         grow monotonically, raw readings for the envelope layer's deltas. A user2482         whose row is gone — dropped mid-call — is counted nowhere, silently.2483         """2484         with self.dispatch_lock:2485             item = self.user_register.get(user)2486             if item is None:2487                 return2488             item["served_call_count"] = item.get("served_call_count", 0) + 12489             item["service_seconds"] = item.get("service_seconds", 0.0) + seconds2490 2491     def _stamp_request(self, user: str, cid: str) -> None:2492         """Stamp the user a request came in for, and his connection under it.2493 2494         The http form names no page: what it proves is a real call of that2495         user, which is the clock the group's judgment reads off the photo.2496         """2497         with self.dispatch_lock:2498             connection = self.connection_register.get(cid)2499             row = self.user_register.get(user)2500             items = [item for item in (connection, row) if item is not None]2501             self._stamp_items(items, ("last_rpc_ts",))2502 2503     def _stamp_items(self, items: Iterable[dict[str, Any]], clocks: Iterable[str]) -> float:2504         """Write the server's own instant on the items given.2505 2506         ``last_refresh_ts`` always, the clocks named besides it: a client cannot2507         buy immortality by claiming activity, so the instant is taken here.2508         """2509         now = time.time()2510         for item in items:2511             item["last_refresh_ts"] = now2512             for clock in clocks:2513                 item[clock] = now2514         return now2515 2516     def _user_row(self, user: str, item: dict[str, Any]) -> dict[str, Any]:2517         """One user item projected for the photo: state, clocks, service counters.2518 2519         Args:2520             user: whose row it is — the pendings are keyed by him, not by the item.2521             item: his user register item.2522 2523         Returns:2524             The scalar projection: his state, his connection count, his three2525             clocks, and the three service counters — the two cumulatives the2526             calls of his wrote (0 before his first), plus how many are open now.2527         """2528         row: dict[str, Any] = {2529             "state": item["state"],2530             "connection_count": len(item["connections"]),2531             "served_call_count": item.get("served_call_count", 0),2532             "service_seconds": item.get("service_seconds", 0.0),2533             "pending_call_count": self._pendings.get(user, 0),2534         }2535         row.update({clock: item[clock] for clock in CLOCK_NAMES})2536         return row2537 2538     async def _run_in_pool(self, pool: ThreadPoolExecutor, work: Callable[[], Any]) -> Any:2539         """Run one piece of synchronous work on the pool it belongs to.2540 2541         The work runs under a COPY of the calling task's context, so the thread2542         finds the request slot of the CALL it serves: what the site announces2543         there lands in that CALL's events.2544         """2545         return await asyncio.get_running_loop().run_in_executor(2546             pool, contextvars.copy_context().run, work2547         )2548 2549     async def _execute_transfer(self, user: str, flag: str) -> None:2550         """Let one flagged user go to the deposit.2551 2552         The departure is CLAIMED before the first await — the transfer cycle,2553         the end-of-call hook and the mass cycle all come through the same claim,2554         and whoever arrives second finds it taken and nothing to do. A row still2555         mid-adoption is WAITED for and never parked under its own pull: that is2556         how the quit keeps a straggler whose store is still travelling. What2557         goes wrong for one user is counted here and goes no further: a whole2558         worker leaving must not be stopped by one refused parcel. The flag is2559         the promise (owner, 2026-08-16): only a departure that HAPPENED, a2560         counted failure or the man's own absence consumes it — a freeze2561         deferred to a call's tail keeps it, and the wakeup set below lets the2562         quit's cycle find it again, so no instant between a closing call and a2563         releasing claim can drop a man between two hands.2564 2565         No CALL is being answered here, so the departure gets a slot of its own2566         and what it announces goes up the second channel — ONE CALL of this2567         worker's per departure, placed once the departure is over.2568         """2569         slot_token = self._request_slot_var.set(self.build_request_slot())2570         try:2571             await self._execute_transfer_in_slot(user, flag)2572         finally:2573             announced = self.request_slot.worker_events2574             self._request_slot_var.reset(slot_token)2575         if announced:2576             await self.announce_worker_events(announced)2577 2578     async def _execute_transfer_in_slot(self, user: str, flag: str) -> None:2579         """The departure itself: claim, wait out an adoption, freeze, settle the flag."""2580         with self.dispatch_lock:2581             if self._transfer_flags.get(user) != flag:2582                 return2583             if user in self._pendings:2584                 return2585             if not self._claim_departure(user):2586                 return2587             adopting = self._unfreeze_waits.get(user)2588         settled = True2589         try:2590             if adopting is not None:2591                 await adopting.wait()2592             settled = await self.freeze_user(user) is not None2593         except Exception:2594             settled = True2595             self._freeze_failures += 12596             self._logger.exception(2597                 "Worker %s: the departure of %s fell over; the others go on", self.name, user2598             )2599         finally:2600             with self.dispatch_lock:2601                 if settled or user not in self.user_register:2602                     self._transfer_flags.pop(user, None)2603                 self._release_departure(user)2604                 self._transfers_changed.set()2605 2606     async def announce_worker_events(self, worker_events: list[dict[str, Any]]) -> None:2607         """Send up what happened while no CALL was being served — the second channel.2608 2609         Args:2610             worker_events: the events of a slot no reply will ever carry.2611 2612         ONE CALL to the vertex, its envelope shaped as a reply's — the photo2613         rides it when due — and folded there the same way. The REPLY is the2614         acknowledgement. A vertex that answers an error or nothing within2615         ``ANNOUNCE_TIMEOUT_SECONDS`` has lost this announcement: counted and2616         logged, never retried — a wire that does not answer is a vertex that is2617         gone, and its death settles what this process held.2618         """2619         data = self._outbound({ENVELOPE_SLOT_WORKER_EVENTS: worker_events})2620         data["worker"] = self.name2621         try:2622             await self.call(ANNOUNCE_OP_PATH, data, timeout=ANNOUNCE_TIMEOUT_SECONDS)2623         except (CommanderCallFailed, TimeoutError):2624             self._announce_failures += 12625             self._logger.exception(2626                 "Worker %s: %d worker event(s) lost, the vertex did not take the announcement",2627                 self.name,2628                 len(worker_events),2629             )2630 2631     def _claim_departure(self, user: str) -> bool:2632         """Take the one departure a user is allowed at a time.2633 2634         Args:2635             user: the user about to leave.2636 2637         Returns:2638             True when the claim is the caller's; False when somebody is already2639             taking him away.2640 2641         Marks him departing. Three roads reach a freeze — the transfer cycle,2642         the end-of-call hook, the mass cycle of a lost wire — and the second to2643         arrive must find the door shut, or it would queue on a folder semaphore2644         this same worker is holding.2645         """2646         with self.dispatch_lock:2647             if user in self._departing_users:2648                 return False2649             self._departing_users.add(user)2650             return True2651 2652     def _release_departure(self, user: str) -> None:2653         """Give the claim back, say so if that was the last departure, wake the cycle.2654 2655         The wakeup is owed to whoever found this claim TAKEN and went away with2656         nothing done: the ordered freeze and the transfer cycle reach the same2657         user by two roads, and the loser leaves the flag where it is. Without2658         this the cycle of a quit would sleep on a change that nobody else is2659         going to announce.2660         """2661         with self.dispatch_lock:2662             self._departing_users.discard(user)2663             self._settle_transfers()2664             self._transfers_changed.set()2665 2666     def _settle_transfers(self) -> None:2667         """Declare the departures over: no flag left, and nobody on his way out.2668 2669         Both halves are asked, because a flag popped by the man who is at that2670         instant writing his parcels would otherwise let a ``quit`` leave from2671         under him.2672         """2673         with self.dispatch_lock:2674             if not self._transfer_flags and not self._departing_users:2675                 self._transfers_done.set()2676 2677     def _get_user_parcels(self, item: dict[str, Any]) -> tuple[Any, dict[str, dict[str, Any]]]:2678         """The store payload and the connection parcels, photographed off the registers.2679 2680         Args:2681             item: the user register item leaving memory.2682 2683         Returns:2684             Two things: the payload of his store, and one parcel per connection2685             in the shape the adoption reads back, in the order they are written.2686 2687         The photograph is DEEP and is taken under the dispatch lock: at the2688         scale of these parcels the copy is memory work of microseconds, and2689         nothing live then crosses onto the service pool, where the deposit2690         pickles what it is handed with no lock of ours at all. The write itself2691         is disk and runs with the lock let go, because a loop-side mutation must2692         never wait on a spinning disk.2693         """2694         with self.dispatch_lock:2695             parcels = (2696                 item["store"],2697                 {cid: self._connection_parcel(cid) for cid in sorted(item["connections"])},2698             )2699             store, connection_parcels = copy.deepcopy(parcels)2700             for parcel in connection_parcels.values():2701                 self._detach_parcel_capture(parcel)2702             return store, connection_parcels2703 2704     def _detach_parcel_capture(self, parcel: dict[str, Any]) -> None:2705         """Take the capture off the copied page stores, before they are pickled.2706 2707         A copied store arrives with the subscriber of the LIVE row still on it —2708         the copy would feed a queue that is not its own, and a subscriber does2709         not pickle at all. The birth on the other side attaches its own.2710         """2711         for page_id, fields in parcel["pages"].items():2712             self.registry.detach_page({**fields, "register_item_id": page_id})2713 2714     def _write_parcels(2715         self, user: str, store: Any, connection_parcels: dict[str, dict[str, Any]]2716     ) -> None:2717         """Write the store and the connection parcels already copied out of the registers.2718 2719         Runs on the service pool — this is real disk work — and holds NO lock of2720         the dispatch: what it writes was photographed before it was handed over,2721         so nothing here reads a register a mutation could be changing.2722         """2723         self.freeze_handler.write_user_register_item(2724             user, store, writer=self.name, cause="freeze", group=self.group2725         )2726         for cid, parcel in connection_parcels.items():2727             self.freeze_handler.write_connection_register_item(2728                 user, cid, parcel, writer=self.name, cause="freeze", group=self.group2729             )2730 2731     def _connection_parcel(self, cid: str) -> dict[str, Any]:2732         """One connection with its pages, in the shape the adoption reads back.2733 2734         What each row leaves behind is the row's own knowledge2735         (``fields_left_behind`` on its class): the edges of the tree, which the2736         folder already says; the lock; the live objects bound to the Bags of2737         THIS process, which the birth on the other side makes anew. What2738         travels but must be put back after the birth is the row's too2739         (``fields_replayed``), and the wake asks it.2740         """2741         item = self.connection_register.get(cid)2742         pages = {page_id: self.page_register.get(page_id) for page_id in sorted(item["pages"])}2743         return {2744             # The parcel names its own connection: the deposit filename hashes2745             # the id one-way, and the wake reads it back from here.2746             "connection_id": cid,2747             "connection": {2748                 key: value for key, value in item.items() if key not in item.fields_left_behind2749             },2750             "pages": {2751                 page_id: {2752                     key: value for key, value in page.items() if key not in page.fields_left_behind2753                 }2754                 for page_id, page in pages.items()2755             },2756         }2757 2758     def _release_login_rows(self, cid: str, user: str, previous_user: str) -> None:2759         """Take out of memory what the login left here: the caller holds the lock.2760 2761         The connection and its pages are gone to the deposit; the guest that used2762         to own it has nothing left anywhere and goes with them; and the identity2763         that received it goes too when this connection was all he had here — a2764         row left empty would make his own next request look like a resident and2765         throw away the store his connection is carrying. A previous identity that2766         is NOT a guest STAYS, empty if this was his last connection: he is a2767         person the machine knows, and the idleness sweep is what parks him.2768 2769         Losing that row is ANNOUNCED, and with a word of its own: he has not gone2770         to the deposit and he has not left the machine — he lives wherever he2771         lived before this login, and the only rung that has to hear it is the2772         handler of this process, whose list of who is on board is what a wild2773         death is settled on. A death reading a name whose rows are gone would2774         report the loss of somebody who is perfectly well somewhere else.2775         The pages leaving are announced as ``drop_pages`` in every case — the2776         vertex's projection follows the page rows, resident or not.2777         """2778         page_ids = sorted(self.connection_register.get(cid)["pages"])2779         if page_ids:2780             self.add_worker_event("drop_pages", user=user, page_ids=page_ids)2781         for page_id in page_ids:2782             self._remove_page_item(page_id)2783         self._remove_connection_item(cid)2784         if previous_user.startswith(GUEST_PREFIX) and previous_user in self.user_register:2785             self._remove_user_item(previous_user)2786         resident = self.user_register.get(user)2787         if resident is not None and not resident["connections"]:2788             self._remove_user_item(user)2789             self.add_worker_event("user_rows_released", user=user)2790 2791     def _release_rows(self, user: str) -> None:2792         """Take a user's rows out of memory; his departure names him, this names his pages.2793 2794         Everything of his goes — pages, connections, the user row itself. No2795         emptied row is left resident: he is the vertex's to place now, and2796         whatever comes back for him starts from the parcel in the deposit. Two2797         departures end here — the freeze that parked him, and the pull that2798         failed to bring him home. The pages leaving are announced as2799         ``drop_pages``: what the vertex keeps per page is a projection of the2800         page rows, and a page taken out of memory is taken out of the2801         projection — the wake's own announcements rebuild it.2802         """2803         item = self.user_register.get(user)2804         for cid in sorted(item["connections"]):2805             page_ids = sorted(self.connection_register.get(cid)["pages"])2806             if page_ids:2807                 self.add_worker_event("drop_pages", user=user, page_ids=page_ids)2808             for page_id in page_ids:2809                 self._remove_page_item(page_id)2810             self._remove_connection_item(cid)2811         self._remove_user_item(user)2812         self._unfreeze_waits.pop(user, None)2813 2814     def _add_user_item(self, user: str, **fields: Any) -> dict[str, Any]:2815         """Put a user item in the register, born stamped and with a live store."""2816         fields.setdefault("state", "active")2817         return self.registry.new_user(user, **self._stamped(**fields))2818 2819     def _add_connection_item(self, cid: str, user: str, **fields: Any) -> dict[str, Any]:2820         """Put a connection item in the register and join it to its user."""2821         return self.registry.new_connection(cid, user, **self._stamped(**fields))2822 2823     def _add_page_item(self, page_id: str, cid: str, **fields: Any) -> dict[str, Any]:2824         """Put a page item in the register and join it to its connection."""2825         return self.registry.new_page(2826             page_id,2827             user=self.connection_register.get(cid)["user"],2828             connection_id=cid,2829             **self._stamped(**fields),2830         )2831 2832     def _remove_page_item(self, page_id: str) -> None:2833         """Take a page item out of the register, capture and edge with it."""2834         self.registry.drop_page(page_id, cascade=False)2835 2836     def _remove_connection_item(self, cid: str) -> None:2837         """Take a connection item out of the register and off its user."""2838         self.registry.drop_connection(cid, cascade=False)2839 2840     def _remove_user_item(self, user: str) -> None:2841         """Take a user item out of the register."""2842         self.registry.user_items.drop(user)2843 2844     def _get_register_rows(self, register: Register) -> list[tuple[str, dict[str, Any]]]:2845         """Every key of a register paired with its live item, in one snapshot."""2846         return [(key, register.get(key)) for key in register.keys()]2847 2848     def _drop_emptied_user(self, user: str) -> None:2849         """Take the user away when the connection just removed was his last."""2850         if not self.user_register.get(user)["connections"]:2851             self._remove_user_item(user)2852             self._unfreeze_waits.pop(user, None)2853             self.add_worker_event("drop_user", user=user)2854 2855     def _stamped(self, **fields: Any) -> dict[str, Any]:2856         """An item born with the three clocks on the server's own instant."""2857         now = time.time()2858         for clock in CLOCK_NAMES:2859             fields.setdefault(clock, now)2860         return fields2861 2862     async def _take_folder_lock(self, user: str) -> None:2863         """Wait on the loop until the semaphore of the user's folder is this worker's.2864 2865         Args:2866             user: the user whose folder is being entered.2867 2868         Raises:2869             TimeoutError: ``deposit_lock_wait_limit`` passed and it never came2870                 free — a folder nobody gave back, which is a disk to look at and2871                 never something to go on waiting for in silence.2872 2873         The wait is a coroutine and never a thread: whoever holds the semaphore2874         is working, and a thread parked here would be a thread not doing that2875         work. The FIRST miss says out loud who is holding it, once for this2876         wait and not once per look. The limit is a technical floor, not a2877         budget: how long a REQUEST may wait before the vertex answers it2878         something else is the Commander's parking budget, and arrives with the2879         fold.2880         """2881         if self.freeze_handler.take_lock(user, self.name):2882             return2883         self._logger.warning(2884             "Worker %s: the deposit folder of %s is held by %s; waiting for it",2885             self.name,2886             user,2887             self.freeze_handler.lock_holder(user),2888         )2889         deadline = time.time() + self.deposit_lock_wait_limit2890         while not self.freeze_handler.take_lock(user, self.name):2891             if time.time() >= deadline:2892                 raise TimeoutError(2893                     f"the deposit folder of {user} was held by "2894                     f"{self.freeze_handler.lock_holder(user)!r} for "2895                     f"{self.deposit_lock_wait_limit}s"2896                 )2897             await asyncio.sleep(self.deposit_lock_retry_interval)2898 2899     async def _take_from_deposit(self, user: str, read: Any, *args: Any) -> Any:2900         """Hold the user's folder, read one parcel and delete it, then let go.2901 2902         The reading is real disk work and runs on the service pool; the wait for2903         the semaphore is not, and stays a coroutine on the loop. Releasing the2904         semaphore takes the folder away when the parcel read was the last thing2905         in it. A semaphore that never comes free raises out of here and travels2906         to the caller, whose REPLY says so: an adoption nobody can make is an2907         answered failure, never a request left hanging.2908         """2909         await self._take_folder_lock(user)2910         try:2911             return await self._run_in_pool(self.service_pool, functools.partial(read, user, *args))2912         finally:2913             self.freeze_handler.release_lock(user, self.name)2914 2915     def _read_user_parcel(self, user: str) -> Any:2916         """Read the user's store off the deposit and take the parcel away."""2917         payload = self.freeze_handler.read_user_register_item(user)2918         self.freeze_handler.drop_user_register_item(user)2919         return payload2920 2921     def _read_connection_parcel(self, user: str, cid: str) -> Any:2922         """Read one connection with its pages off the deposit and take it away."""2923         payload = self.freeze_handler.read_connection_register_item(user, cid)2924         self.freeze_handler.drop_connection_register_item(user, cid)2925         return payload