src/genro_asgi_multiworker_spa/orchestration/freeze_handler.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 """FreezeHandler: the freezer on disk, and the only direct filesystem access.16 17 A user who leaves memory leaves it here. One DIRECTORY per user — named by18 ``user_to_userkey``, which goes ONE WAY: every reader starts from the identity19 and computes the name forward, nothing ever derives an identity back from a20 directory name. Inside: ``user_register_item.pickle`` for the user's own store, one21 ``connection_register_item_<cid>.pickle`` per connection, carrying that connection AND22 its pages. Beside them the semaphore, ``.lock``.23 24 **The freezer only through this surface.** The house rule says the filesystem25 is reached through storage nodes; this class is the declared exception, and the26 exception is what buys it: the semaphore needs real exclusive creation27 (``O_CREAT|O_EXCL``), which no logical-volume surface offers. The deal is that28 NOBODY else computes a path under the root — freezing, adoption, the sweep and29 the photo all speak to this surface.30 31 **The semaphore is the only coherence mechanism.** There is no temporary file32 and no rename: whoever holds the lock writes DIRECTLY over the destination.33 Nobody can read half a file, because a reader waits for the lock before34 looking; the half file a crash leaves behind is covered elsewhere — the dead35 worker's folder is discarded by the cleanup that follows its death, and every36 server start wipes the working freezer anyway. ONE declared exception, accepted37 by weighed probability (2026-08-18): the vertex reads headers and drops folders38 WITHOUT the lock. A parcel is complete before the vertex ever marks its user39 frozen, so the only collision left is a drop (expiry, forget) against a lazy40 wake in flight — and it ends in a loud error on a user the machine was41 forgetting anyway, never in silent corruption.42 43 **Waiting is the caller's, on its own loop.** ``take_lock`` tries once and says44 yes or no. Whoever finds it taken retries as a coroutine — never holding a45 thread of the service pool, which exists for real disk work and would starve46 itself waiting for the operation meant to release the lock.47 48 **The lock owns the empty folder.** Dropping items never touches the lock: a49 folder is alive while an operation runs, whatever it has left inside. It is50 ``release_lock`` — the end of that operation — that removes the folder when51 nothing but the lock remains, so the root holds the frozen and nothing else.52 53 **A drop asks for absence.** Every ``drop_*`` says what must no longer be54 there, and a thing already gone is that same outcome: no error. The cleanup55 after a dead worker is the ordinary caller, and it walks over parcels the dead56 one may or may not have written — it must not have to ask first.57 58 **The header is diagnostic and only that.** Every payload goes to disk wrapped59 with who wrote it, when, for which cause and from which group. It is read for60 counting and for the sysop; no decision is ever taken on it — what is true61 about a user is the mark in the indexes, never the file.62 """63 64 from __future__ import annotations65 66 import os67 import pickle68 import shutil69 import time70 import urllib.parse71 from pathlib import Path72 from typing import Any73 74 LOCK_NAME = ".lock"75 USER_REGISTER_ITEM_NAME = "user_register_item.pickle"76 CONNECTION_REGISTER_ITEM_PREFIX = "connection_register_item_"77 COMMANDER_REGISTER_ITEM_NAME = "commander_register_item.pickle"78 79 __all__ = [80 "COMMANDER_REGISTER_ITEM_NAME",81 "CONNECTION_REGISTER_ITEM_PREFIX",82 "LOCK_NAME",83 "USER_REGISTER_ITEM_NAME",84 "FreezeHandler",85 ]86 87 88 class FreezeHandler:89 """The freezer: one directory per frozen user, under one root.90 91 Args:92 root_path: the freezer root, created private (0700) if missing.93 """94 95 def __init__(self, root_path: str | Path):96 self.root_path = Path(root_path)97 self.root_path.mkdir(mode=0o700, parents=True, exist_ok=True)98 99 def user_to_userkey(self, user: str) -> str:100 """The directory name ``user`` is filed under: its identity, percent-encoded.101 102 Args:103 user: the user identity.104 105 Returns:106 The key, with every separator quoted away — no identity can name a107 directory outside the root. ONE WAY: no reverse exists.108 """109 return urllib.parse.quote(user, safe="")110 111 @property112 def storage_free_percent(self) -> float:113 """How much of the storage the freezer lives on is still free, in percent.114 115 Returns:116 The free share of the whole filesystem, not the room the freezer117 takes: what runs out is the storage, and the freezer is only one of118 the things filling it.119 """120 usage = shutil.disk_usage(self.root_path)121 return 100.0 * usage.free / usage.total122 123 @property124 def user_folders(self) -> set[str]:125 """The keys of every folder in the freezer, as one set.126 127 Returns:128 The folder names, unopened. The sweep subtracts the keys of the129 users it manages from this set and discards the difference whole.130 """131 return {entry.name for entry in os.scandir(self.root_path) if entry.is_dir()}132 133 def take_lock(self, user: str, holder: str) -> bool:134 """Try ONCE to take the semaphore of ``user``, creating the folder if needed.135 136 Args:137 user: the user whose folder is being entered.138 holder: the name answering for the operation, written inside the lock.139 140 Returns:141 True if the semaphore is now this holder's, False if somebody holds it.142 143 Creates the user folder and the lock file.144 """145 folder = self._user_folder(user)146 folder.mkdir(mode=0o700, parents=True, exist_ok=True)147 try:148 fd = os.open(folder / LOCK_NAME, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)149 except FileExistsError:150 return False151 with os.fdopen(fd, "w") as lock_file:152 lock_file.write(holder)153 return True154 155 def release_lock(self, user: str, holder: str) -> None:156 """Give back the semaphore of ``user``, and the folder if nothing else is left.157 158 Args:159 user: the user whose folder is being left.160 holder: the name that took it — a mismatch is a protocol break.161 162 Raises:163 RuntimeError: the semaphore is not this holder's.164 165 Removes the lock file, and the folder when the lock was all it had.166 """167 current = self.lock_holder(user)168 if current != holder:169 raise RuntimeError(170 f"release of {user}: the semaphore is {current!r}, not {holder!r}"171 )172 folder = self._user_folder(user)173 os.remove(folder / LOCK_NAME)174 if not os.listdir(folder):175 os.rmdir(folder)176 177 def lock_holder(self, user: str) -> str | None:178 """Who holds the semaphore of ``user`` right now.179 180 Args:181 user: the user whose folder is asked about.182 183 Returns:184 The holder name, or None if the semaphore is free.185 """186 try:187 return (self._user_folder(user) / LOCK_NAME).read_text()188 except FileNotFoundError:189 return None190 191 def write_user_register_item(192 self, user: str, payload: Any, *, writer: str, cause: str, group: str193 ) -> None:194 """Write the user's own store, directly over whatever was there.195 196 Args:197 user: the user the store belongs to.198 payload: the store, pickled as it comes.199 writer: the name answering for the write.200 cause: why it is being written (freeze, login, ...).201 group: the group the writer belongs to.202 203 Writes the file. The caller holds the semaphore.204 """205 self._write_item(self._user_folder(user) / USER_REGISTER_ITEM_NAME, payload, writer, cause, group)206 207 def write_connection_register_item(208 self, user: str, cid: str, payload: Any, *, writer: str, cause: str, group: str209 ) -> None:210 """Write one connection of ``user`` — the connection and its pages.211 212 Args:213 user: the user the connection belongs to.214 cid: the connection identity.215 payload: the connection with its pages, pickled as it comes.216 writer: the name answering for the write.217 cause: why it is being written (freeze, login, ...).218 group: the group the writer belongs to.219 220 Writes the file. The caller holds the semaphore.221 """222 self._write_item(223 self._connection_path(user, cid), payload, writer, cause, group224 )225 226 def write_commander_register_item(self, payload: Any, *, writer: str, cause: str) -> None:227 """Write the vertex's own item at the ROOT — a file, never a folder.228 229 Args:230 payload: what the vertex saves of itself, pickled as it comes.231 writer: the name answering for the write.232 cause: why it is being written.233 234 Writes the file. It sits beside the user folders and outside them, so235 the sweep — which only ever looks at directories — cannot see it.236 """237 self._write_item(self.root_path / COMMANDER_REGISTER_ITEM_NAME, payload, writer, cause, "")238 239 def read_commander_register_item(self) -> Any:240 """The vertex's own item, or None when there is no such file."""241 envelope = self._read_envelope(self.root_path / COMMANDER_REGISTER_ITEM_NAME)242 return None if envelope is None else envelope["payload"]243 244 def drop_commander_register_item(self) -> None:245 """Remove the vertex's own item; a thing already gone is that same outcome."""246 (self.root_path / COMMANDER_REGISTER_ITEM_NAME).unlink(missing_ok=True)247 248 def wipe_root(self) -> None:249 """Empty the root and stand it up again — what every start does first (F4).250 251 Removes everything the root held and recreates it bare, with the same252 permissions ``__init__`` gives it.253 """254 shutil.rmtree(self.root_path, ignore_errors=True)255 self.root_path.mkdir(mode=0o700, parents=True, exist_ok=True)256 257 def drop_root(self) -> None:258 """Remove the root itself, whole; a root already gone is that same outcome."""259 shutil.rmtree(self.root_path, ignore_errors=True)260 261 def rename_root(self, destination: str | Path) -> None:262 """Move this whole root to *destination*, and follow it.263 264 Args:265 destination: where the root goes — the same filesystem, so the move266 is one atomic rename and never a copy.267 268 Acts on the disk and on this handler, which points at the new name269 afterwards. The rename is the only commit this class has: a directory270 that appears under its final name is complete by construction.271 """272 destination = Path(destination)273 os.rename(self.root_path, destination)274 self.root_path = destination275 276 def read_user_register_item(self, user: str) -> Any:277 """Read back the user's own store.278 279 Args:280 user: the user the store belongs to.281 282 Returns:283 The payload as it was written, or None if there is no such file.284 """285 envelope = self._read_envelope(self._user_folder(user) / USER_REGISTER_ITEM_NAME)286 return None if envelope is None else envelope["payload"]287 288 def read_connection_register_item(self, user: str, cid: str) -> Any:289 """Read back one connection of ``user`` with its pages.290 291 Args:292 user: the user the connection belongs to.293 cid: the connection identity.294 295 Returns:296 The payload as it was written, or None if there is no such file.297 """298 envelope = self._read_envelope(self._connection_path(user, cid))299 return None if envelope is None else envelope["payload"]300 301 def get_item_header(self, user: str, cid: str | None = None) -> dict[str, Any] | None:302 """The diagnostic header of an item — for counting and for the sysop.303 304 Args:305 user: the user the item belongs to.306 cid: a connection identity, or None for the user's own store.307 308 Returns:309 The header (writer, ts, cause, group), or None if there is no such310 file. Never a ground for a decision.311 """312 path = (313 self._user_folder(user) / USER_REGISTER_ITEM_NAME if cid is None314 else self._connection_path(user, cid)315 )316 envelope = self._read_envelope(path)317 return None if envelope is None else envelope["header"]318 319 def drop_user_register_item(self, user: str) -> None:320 """Discard the user's own store, adopted or spent.321 322 Args:323 user: the user the store belongs to.324 325 Removes the file if it is there — an absence is the same outcome, not an326 error. The folder stays until the semaphore is released.327 """328 (self._user_folder(user) / USER_REGISTER_ITEM_NAME).unlink(missing_ok=True)329 330 def drop_connection_register_item(self, user: str, cid: str) -> None:331 """Discard one connection of ``user``, adopted or spent.332 333 Args:334 user: the user the connection belongs to.335 cid: the connection identity.336 337 Removes the file if it is there — an absence is the same outcome, not an338 error. The folder stays until the semaphore is released.339 """340 self._connection_path(user, cid).unlink(missing_ok=True)341 342 def drop_user_folder(self, user: str) -> bool:343 """Discard everything ``user`` has in the freezer, semaphore included.344 345 Args:346 user: the user leaving the freezer for good.347 348 Returns:349 Whether the freezer was holding anything of his.350 351 Raises:352 RuntimeError: the folder survived its own removal.353 354 Removes the folder and verifies it is gone.355 """356 return self._drop_folder(self.user_to_userkey(user))357 358 def cleanup_frozen(self, claimed: set[str]) -> list[str]:359 """Sweep the freezer of everything that belongs to nobody.360 361 Args:362 claimed: the keys somebody still answers for — the caller computes363 them forward from its own identities, since no identity ever364 comes back from a folder name.365 366 Returns:367 The keys swept away, so the caller can count and name them.368 369 Removes those folders, each verified gone.370 """371 orphans = sorted(self.user_folders - claimed)372 for userkey in orphans:373 self._drop_folder(userkey)374 return orphans375 376 def _drop_folder(self, userkey: str) -> bool:377 """Remove one folder of the freezer by its key, verify it is gone, say if it was."""378 folder = self.root_path / userkey379 was_there = folder.exists()380 shutil.rmtree(folder, ignore_errors=True)381 if folder.exists():382 raise RuntimeError(f"freezer folder {userkey}: it survived its removal")383 return was_there384 385 def _user_folder(self, user: str) -> Path:386 return self.root_path / self.user_to_userkey(user)387 388 def _connection_path(self, user: str, cid: str) -> Path:389 name = f"{CONNECTION_REGISTER_ITEM_PREFIX}{self.user_to_userkey(cid)}.pickle"390 return self._user_folder(user) / name391 392 def _write_item(self, path: Path, payload: Any, writer: str, cause: str, group: str) -> None:393 header = {"writer": writer, "ts": time.time(), "cause": cause, "group": group}394 path.write_bytes(pickle.dumps({"header": header, "payload": payload}))395 396 def _read_envelope(self, path: Path) -> dict[str, Any] | None:397 try:398 return pickle.loads(path.read_bytes())399 except FileNotFoundError:400 return None