src/genro_asgi_multiworker_spa/global_store.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 """The global store: one dictionary, living ONLY on the commander, behind one lock.16 17 The commander owns ``dict[str, Any]``. Keys are literal strings — a dot in a key18 is a character, never a path — and values are opaque: a scalar, a dict, a Bag of19 either kind, anything the TYTX codec knows. There are no replicas: every access20 a worker makes is a CALL on the lane, served under ONE FIFO lock (issue #74,21 owner decision 2026-09-07).22 23 - :class:`GlobalStoreLock` is the commander's lock: an ``asyncio.Lock`` (FIFO by24 construction) plus who holds the TURN — request id, worker and the key the25 turn selected. A simple ``get``/``set``/``delete`` takes the same lock for the26 length of its own operation and records no holder; a turn holds it from the27 grant to the release. No lease and no timer: the holder's channel EOF is the28 whole death protocol, and it applies nothing.29 - :class:`GlobalStoreClient` is what a worker holds as ``global_store``: the30 three simple operations, synchronous from a pool thread, and ``for_update``.31 - :class:`GlobalStoreLease` is one turn: ``with`` or ``async with``, because the32 vehicle follows the caller. It yields ITSELF, with ``value`` — the private33 working copy the grant decoded — and ``exists``, said at grant time. The exit34 sends the COMPLETE value back (``apply=True``) and the commander replaces the35 selected key, or the whole dictionary when no key was selected; a body that36 raises, or a lease that cannot decode its grant or encode its value, releases37 with ``apply=False`` and the master is exactly as the grant found it.38 39 **A turn recognises its own context.** The client keeps a ``ContextVar`` set40 while a lease is in force: a second ``for_update`` or a simple operation from41 the same task or pool thread raises at once instead of parking on the lock it42 already holds. Other threads of the same worker wait normally.43 44 **The wire is TYTX.** Every value travels ``to_tytx(..., "json")`` and is45 hydrated by its reader, so a Bag stays a Bag and a datetime a datetime; a46 ``get`` reply carries ``exists`` beside ``value``, so a stored ``None`` and an47 absent key are two answers.48 49 **A commit whose answer never came is uncertain, and says so.** The commit is a50 CALL like any other and waits as long as the wire lives; when the wire fails51 after the commit was sent, the commander may already have published the value.52 The lease then raises :class:`GlobalStoreCommitUnconfirmed` — never a retry, and53 no abort attempt on a wire that is gone. A commander that REFUSED the commit is54 not this case: that is ``CommanderCallFailed``, and nothing was published.55 """56 57 from __future__ import annotations58 59 import asyncio60 import contextvars61 import uuid62 from types import TracebackType63 from typing import Any64 65 from genro_tytx import from_tytx, to_tytx66 67 #: The routing keys of the global store on the commander's dispatcher.68 GLOBAL_STORE_SET_OP_PATH = "/commander/store/set"69 GLOBAL_STORE_DEL_OP_PATH = "/commander/store/del"70 GLOBAL_STORE_GET_OP_PATH = "/commander/store/get"71 GLOBAL_STORE_LOCK_OP_PATH = "/commander/store/lock"72 GLOBAL_STORE_UNLOCK_OP_PATH = "/commander/store/unlock"73 74 __all__ = [75 "GLOBAL_STORE_DEL_OP_PATH",76 "GLOBAL_STORE_GET_OP_PATH",77 "GLOBAL_STORE_LOCK_OP_PATH",78 "GLOBAL_STORE_SET_OP_PATH",79 "GLOBAL_STORE_UNLOCK_OP_PATH",80 "GlobalStoreClient",81 "GlobalStoreCommitUnconfirmed",82 "GlobalStoreLease",83 "GlobalStoreLock",84 ]85 86 87 class GlobalStoreCommitUnconfirmed(Exception):88 """The commit of a turn was sent and its answer never came: the value MAY be published.89 90 Args:91 request_id: the turn whose commit is unconfirmed.92 key: the key it selected, None for the whole dictionary.93 cause: what ended the wait — the transport failure, for the log.94 95 The caller must not repeat the write on its own: the commander may hold it96 already. Raised by ``GlobalStoreLease`` on the commit path alone.97 """98 99 def __init__(self, request_id: str, key: str | None, cause: BaseException) -> None:100 self.request_id = request_id101 self.key = key102 self.cause = cause103 target = "the whole store" if key is None else f"key {key!r}"104 super().__init__(105 f"the commit of turn {request_id} on {target} got no answer "106 f"({type(cause).__name__}: {cause}): the value may have been published"107 )108 109 110 class GlobalStoreLock:111 """The commander's lock on the dictionary: FIFO, one holder, no lease and no timer."""112 113 def __init__(self) -> None:114 self.lock = asyncio.Lock()115 # The turn in force: its request id, the worker whose channel death116 # releases it, and the key it selected (None = the whole dictionary).117 self.holder: str | None = None118 self.holder_worker: str | None = None119 self.holder_key: str | None = None120 121 async def acquire(self, worker: str, request_id: str, key: str | None = None) -> None:122 """Park until the lock is this request's, then record whose turn it is.123 124 ``asyncio.Lock`` wakes its waiters in arrival order, so the FIFO the125 protocol promises is the primitive's own and nothing here queues.126 """127 await self.lock.acquire()128 self.holder = request_id129 self.holder_worker = worker130 self.holder_key = key131 132 def holds(self, request_id: str) -> bool:133 """Whether this request is the turn in force.134 135 A release for a turn no longer in force is a real case, not a protocol136 violation: the holder's channel died while its release was on the wire,137 and the death released it first. Such a release must touch NOTHING —138 neither the master nor a newer turn.139 """140 return self.holder == request_id141 142 def held_by(self, worker: str) -> bool:143 """Whether this worker holds the turn — the death check."""144 return self.holder_worker == worker145 146 def release(self) -> None:147 """Let the next waiter in; the caller has established who holds it."""148 self.holder = None149 self.holder_worker = None150 self.holder_key = None151 self.lock.release()152 153 154 class GlobalStoreClient:155 """A worker's side of the global store: three simple operations and the turn.156 157 Args:158 worker: the ``SpaWorker`` whose lane the CALLs travel on.159 160 The simple operations are synchronous and block a pool thread on the161 worker's loop; ``for_update`` answers a lease usable with ``with`` from a162 pool thread or ``async with`` on the loop.163 """164 165 def __init__(self, worker: Any) -> None:166 self.worker = worker167 self.active_turn: contextvars.ContextVar[GlobalStoreLease | None] = (168 contextvars.ContextVar(f"global_store_turn:{worker.name}", default=None)169 )170 171 def refuse_inside_turn(self) -> None:172 """Raise when this context already holds a turn: a CALL would wait on itself."""173 turn = self.active_turn.get()174 if turn is not None:175 raise RuntimeError(176 f"the global store is already held by this context (turn {turn.request_id})"177 )178 179 def get(self, key: str, default: Any = None) -> Any:180 """Read one key: the stored value, ``None`` included, or ``default`` when absent."""181 self.refuse_inside_turn()182 reply = self.worker.run_on_loop(183 self.worker.call(GLOBAL_STORE_GET_OP_PATH, {"key": key})184 )185 return from_tytx(reply["value"], "json") if reply["exists"] else default186 187 def set(self, key: str, value: Any = None) -> None:188 """Write one key; the master holds the value when this returns."""189 self.refuse_inside_turn()190 self.worker.run_on_loop(191 self.worker.call(192 GLOBAL_STORE_SET_OP_PATH, {"key": key, "value": to_tytx(value, "json")}193 )194 )195 196 def delete(self, key: str) -> None:197 """Remove one key; an absent key is a no-op."""198 self.refuse_inside_turn()199 self.worker.run_on_loop(self.worker.call(GLOBAL_STORE_DEL_OP_PATH, {"key": key}))200 201 def for_update(self, key: str | None = None) -> GlobalStoreLease:202 """One read-modify-write turn on ``key``, or on the whole dictionary when None."""203 return GlobalStoreLease(self, key)204 205 206 class GlobalStoreLease:207 """One turn on the global store: ``with`` or ``async with``, yielding itself.208 209 Args:210 client: the worker's ``GlobalStoreClient``.211 key: the selected key, or None for the whole dictionary.212 213 ``value`` is the private working copy the grant decoded — assign it or214 mutate it, the master sees nothing until the exit; ``exists`` says whether215 the key was there at grant time (always True for the whole dictionary).216 A body that raises releases with ``apply=False``; so does a grant that217 cannot be decoded or a value that cannot be encoded, the original error218 re-raised; so does a turn on which ``abort`` was called, whatever the body219 did to ``value`` afterwards — the lock stays held until the exit either way.220 """221 222 def __init__(self, client: GlobalStoreClient, key: str | None) -> None:223 self.client = client224 self.key = key225 self.request_id = uuid.uuid4().hex226 self.value: Any = None227 self.exists = False228 self.aborted = False229 self._token: contextvars.Token[GlobalStoreLease | None] | None = None230 231 def abort(self) -> None:232 """Mark this turn as not to be published: the exit sends ``apply=False``.233 234 The lock stays held until the ``with`` block exits; once called, nothing235 the body does to ``value`` reaches the master.236 """237 self.aborted = True238 239 async def _acquire(self) -> None:240 worker = self.client.worker241 reply = await worker.call(242 GLOBAL_STORE_LOCK_OP_PATH,243 {"worker": worker.name, "request_id": self.request_id, "key": self.key},244 )245 try:246 self.value = from_tytx(reply["value"], "json")247 except Exception:248 await self._abort()249 raise250 self.exists = reply["exists"]251 252 async def _release(self, exc_type: type[BaseException] | None) -> None:253 if exc_type is not None or self.aborted:254 await self._abort()255 return256 try:257 text = to_tytx(self.value, "json")258 except Exception:259 await self._abort()260 raise261 try:262 await self.client.worker.call(263 GLOBAL_STORE_UNLOCK_OP_PATH,264 {"request_id": self.request_id, "apply": True, "value": text},265 )266 except ConnectionError as exc:267 # The wire ended after the commit left: a parked CALL is failed with268 # ConnectionError, a write on a dead socket raises one of its269 # subclasses. Anything else — the commander's own refusal included —270 # propagates as it is.271 raise GlobalStoreCommitUnconfirmed(self.request_id, self.key, exc) from exc272 273 async def _abort(self) -> None:274 await self.client.worker.call(275 GLOBAL_STORE_UNLOCK_OP_PATH, {"request_id": self.request_id, "apply": False}276 )277 278 def _mark_active(self) -> None:279 self.client.refuse_inside_turn()280 self._token = self.client.active_turn.set(self)281 282 def _mark_closed(self) -> None:283 if self._token is not None:284 self.client.active_turn.reset(self._token)285 self._token = None286 287 async def __aenter__(self) -> GlobalStoreLease:288 self._mark_active()289 try:290 await self._acquire()291 except BaseException:292 self._mark_closed()293 raise294 return self295 296 async def __aexit__(297 self,298 exc_type: type[BaseException] | None,299 exc: BaseException | None,300 traceback: TracebackType | None,301 ) -> None:302 self._mark_closed()303 await self._release(exc_type)304 305 def __enter__(self) -> GlobalStoreLease:306 self._mark_active()307 try:308 self.client.worker.run_on_loop(self._acquire())309 except BaseException:310 self._mark_closed()311 raise312 return self313 314 def __exit__(315 self,316 exc_type: type[BaseException] | None,317 exc: BaseException | None,318 traceback: TracebackType | None,319 ) -> None:320 self._mark_closed()321 self.client.worker.run_on_loop(self._release(exc_type))