src/genro_asgi_multiworker_spa/orchestration/spa_commander.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 """SpaCommander: the vertex — who exists, where he is, and what was decided about him.16 17 The one object that knows the whole picture. It owns three indexes and nothing18 below it owns a copy of them: a group knows where ITS users live, a worker knows19 who is in ITS memory, and only here is there an answer to "who is this cid" or20 "is this user in the freezer".21 22 **The three indexes.** ``connection_user_map`` says whose a cid is — and it is23 ETERNAL, because the cookie is: a browser that comes back a week later is the24 same person, whatever happened to the process it used to talk to.25 ``page_connection_map`` says which connection a page belongs to, and that never26 changes for the life of the page. ``user_map`` is the anagraph, one row per27 identity:28 29 user_map[user] = {group, frozen, on_hold}30 31 Reading a row's meaning goes through the predicates (``user_is_frozen``), and32 ``on_hold`` is not read at all: it is RAISED, as ``UserOnHold``, by the one step33 that resolves an identity. So a caller cannot forget to look at it.34 35 **One identity, and it is the site's.** The cookie carries the hosted site's own36 connection id: the front mints nothing and keeps no state, and the vertex mints37 nobody. A request with no cookie travels ANONYMOUS to the default group's38 reception, the site names its own connection and its own guest while serving,39 and the fold of ``new_connection`` writes ``connection_user_map`` and the user40 row at the fact (``record_connection_user``). The answer carries that id back to41 the front, which writes it in the cookie, so the next request routes on it. This42 is what replaced the minted ``sticky_cid`` and the index that translated it43 (owner decision 2026-08-22): one identity space, no junction, and the maps say44 what their names promise. A cid whose USER ROW is gone — a cookie that outlived45 it — is healed with an empty row: the browser is still known, its state is not.46 47 **The master of the store lives here, and it is a Bag.** Every worker holds a48 replica of it and never writes it: what a worker wants written travels up, is49 written here, and comes back down as the whole content again. The Bag is where50 the store meets the application; the TYTX encoding is where it meets the channel,51 so it happens on the way out and nowhere else. The read-modify-write grant — one52 worker at a time holding the master while it computes a new value — is the lock,53 and it arrives with the request chain.54 55 **Two writers, both here.** The minting above is one; the other is the fold — the56 chain of the envelope, which turns what the processes announce into these57 indexes, one worker event at a time, synchronously. The mutators live on this58 class because the data does, and the chain calls them by name.59 60 **The groups are its own.** The grammar of the machine arrives as61 ``groups={name: kwargs}`` and one ``GroupHandler`` per entry is built right here,62 each handed ``memory_concession_bytes`` — the total it is a share of — so the one63 number of the cascade is never carried by hand from outside. Building a group by64 hand stays legitimate and is what the tests do; either way the group hangs itself65 in ``group_map``, in the order it was named. ``default_group`` is the group66 that receives whoever arrives with no past: the elected one, or the first67 declared.68 69 **A request walks the whole chain from here.** ``serve_request`` takes a70 cookie and gives back what the site answered: the cid becomes an identity, the71 identity names a group (his own, or the elected one when he has none yet), the72 group names the worker — placing him NOW if he has no home — and the request73 travels as the ``http`` form with the identity and the freezer verdict beside it.74 The front hands over a cid and a request and nothing else: it never names a75 group, a worker or a wire, and it keeps no state to name them with. What comes76 back is the child's whole REPLY, folded by the chain before this returns.77 78 The refusals travel as CLASSES, because the caller's next step is written in79 which one arrives: nobody could take him is ``AssignmentRefused`` carrying the80 seconds to come back in; a site that failed inside its process is81 ``SiteFailedRequest``; a wire that is gone is ``ConnectionError``. The82 waiting is the one that does not travel: a user between two homes is waited for83 here, on the budget the request gave, and the walk starts over at the top — the84 map is the authority at every step, so nothing is remembered across the wait.85 86 **The waiting room has a door.** ``on_hold`` on a row is what ``resolve_user``87 raises ``UserOnHold`` on; ``user_hold_event_map`` is what a request PARKS on while88 that lasts. One Event per user on hold, born with the hold and gone with its89 release — the same mutators, in the same breath: ``hold_user`` raises both,90 ``mark_user_frozen``, ``mark_user_adopted``, ``drop_user`` and91 ``release_user_hold`` — the ordered departure that did not happen — let both go.92 Nobody else writes either, so the row and the door cannot say different things.93 94 **Up and down.** ``start`` brings the base group's reception into being and95 only then starts the clock: a reception that has presented itself is what READY96 means, and the front serves from that instant. ``stop`` stops the clock and97 takes every group down dry — no mass freeze on the way out, because without the98 soft boot those files would be read by nobody.99 100 **The freezer is not on the ladder.** A worker parks a user's state on disk101 itself and announces it; the vertex only writes the mark. The one time the vertex102 touches the freezer is when nobody below can: pruning the traces of a wild death103 (what a dead process left behind is not to be trusted, so it is discarded and104 counted) and reaping what expired. Both go through the ``FreezeHandler``, which105 is the only thing in the project that talks to the filesystem.106 107 **Every order leaves a row; every decision leaves its reason.** ``log_order``108 writes the compact human account. ``log_decision`` writes JSONL beside it: the109 decision, its stable reason, the candidates the judge saw and the outcome. An110 order is mirrored there automatically; calculations that issue no order write111 directly. The two files rotate independently and never share stdout. A wild112 death gets an order row too, and it is nobody's decision.113 114 **The counters are aggregate, so they are here.** How many parcels were115 discarded, how much was waiting for somebody who is gone: numbers the level below116 cannot know because each of them sees only its own share.117 118 **The memory cascade starts here.** ``memory_max_percent`` is what this server119 may hold of the machine, and ``memory_concession_bytes`` is that share in bytes:120 the ONE total of the machine, from which a group takes its quota and a worker its121 ceiling, each as a percentage of the rung above. A machine that does not say how122 much memory it has leaves the whole cascade unmeasured, which is what an ungated123 pool honestly is.124 125 The machine is the CGROUP wherever there is one. A server in a container reads,126 through ``psutil.virtual_memory``, the memory of the host holding the127 container — 64 GiB where it may take 2 — so the limit written under128 ``/sys/fs/cgroup`` is read as well and stands in for both figures where it is129 smaller. ``memory_available_bytes`` is the second half of that reading: what is130 left free right now, everything charged to the cgroup counted, and the gate a131 group asks before forking a worker into it.132 133 **There is ONE orchestration clock in the machine, and it is here.**134 ``heartbeat_loop`` waits135 for its timer OR for any group's wake, whichever comes first: the timer gives a136 full round — every group a turn, and the vertex's own tasks each on its own count137 of beats — while a wake gives an anticipated round on THAT group alone, which is138 how the end of a wire is answered in milliseconds whatever the cadence. There is139 no caretaker object anywhere: the probe IS the beat, and the monitor gets a fresh140 photo by ringing the wake like everybody else. A group whose previous turn is141 still open is skipped rather than given a second one, so a mute process delays142 its own group and never the machine; every turn is awaited, an exception is a143 value and not a cancellation, and a round that fails is written down and144 followed by the next beat.145 146 Observation has a second, deliberately narrower cadence. ``cpu_meter_loop``147 reads each governed process's cumulative CPU clock through psutil at the configured148 cadence (100 ms by default). It sends no worker RPC and builds no photo; the same149 pass reconciles only CPU admission. Placement observes that gate, while offload150 reads the latest temperature on the ordinary heartbeat.151 152 **Three tasks are the vertex's own, because nobody below can do them.** The153 frozen whose age ran out have no process to notice them, so ``drop_expired_users``154 prunes the row and the disk itself — the declared exception to the rule that the155 levels below prune themselves. ``cleanup_frozen`` discards what the freezer holds156 for nobody the indexes know. ``check_resources`` reads the machine's memory157 against its alarm line and the freezer's storage against the reserve — the158 memory alone writes ``state``, a storage under reserve is said out loud — and159 calls ``need_resources``, which does nothing here and is where a commander that160 can grow its own machine says so. All three open the disk, so they read it OFF the161 loop: the vertex must never be the reason a healthy child reads as mute.162 """163 164 from __future__ import annotations165 166 import asyncio167 import hashlib168 import json169 import logging170 import os171 import time172 from collections import Counter173 from datetime import UTC, datetime174 from logging.handlers import RotatingFileHandler175 from pathlib import Path176 from typing import Any, overload177 178 from genro_routes import RoutingClass, route179 import psutil180 from genro_tytx import from_tytx, to_tytx181 from genro_asgi.channel.frame import Frame182 from genro_asgi.channel.control import ControlPayload183 184 from genro_asgi.orchestration_profile_store import (185 OrchestrationProfileNotFoundError,186 OrchestrationProfileStore,187 )188 from ..global_store import GlobalStoreLock189 from .beats import every190 from .envelope_handler import CommanderEnvelopeHandler191 from .exceptions import AssignmentRefused, UserOnHold, SiteFailedRequest192 from .freeze_handler import FreezeHandler193 from .group_handler import CHECK_OCCUPANCY_BEATS, GroupHandler194 from .group_policy import GroupPolicy, GroupPolicyError195 from .worker_handler import CENSUS_OP_PATH, EVAL_OP_PATH, OBSERVE_OP_PATH196 197 #: What a user with no name of his own is called: the prefix plus his cid. The198 #: name itself carries the rule — whoever reads it knows nobody logged in here.199 #: Redefined with its ratified value rather than imported: the machine it is200 #: shared with dies at the cutover.201 GUEST_PREFIX = "guest_"202 203 #: The logger the orchestration log is written on, whether or not a file is204 #: attached to it.205 ORDERS_LOGGER_NAME = "genro_asgi.orchestration.orders"206 207 #: The logger of the structured decision journal. Its records are JSON objects,208 #: one per line, separate from both stdout and the human order log.209 DECISIONS_LOGGER_NAME = "genro_asgi.orchestration.decisions"210 211 #: Seconds between two beats of the one clock — the twin of212 #: ``PROCESS_PING_INTERVAL``, which is the cadence a single process is beaten at.213 HEARTBEAT_SECONDS = 5.0214 215 #: Default cadence of the observation-only worker CPU thermometer.216 CPU_TEMPERATURE_SAMPLE_SECONDS = 0.1217 218 # Beats between two rounds of each task of the vertex — the cadences, each where219 # its own knowledge is: an expiry is hours away, so the frozen are read every few220 # minutes; the sweep of the freezer opens the disk over everything ever frozen,221 # which F18 measured in seconds at scale, so it goes hourly; the machine's gauges222 # are trends and not emergencies, and a minute is soon enough for a trend.223 DROP_EXPIRED_USERS_BEATS = 60224 CLEANUP_FROZEN_BEATS = 720225 CHECK_RESOURCES_BEATS = 12226 227 # The reserve line of the storage the freezer lives on: under this much free228 # room the sysop is told, and the machine asks the world outside for more. It is229 # a technical line and not a policy — a full disk is full for every installation.230 STORAGE_RESERVE_PERCENT = 10.0231 232 # The conversion the expiry hours of the grammar meet the clock through.233 SECONDS_PER_HOUR = 3600.0234 235 #: What a refused request is told to wait, in seconds. DERIVED and never a number236 #: of its own: it is exactly when the pool will have re-read its own shape and237 #: decided again, so what is promised to a browser stays true the day the beat238 #: changes.239 SHAPE_REVIEW_SECONDS = HEARTBEAT_SECONDS * CHECK_OCCUPANCY_BEATS240 241 #: The routing key every request of the hosted site travels under. Nothing routes242 #: on it — the child tells the http form by its payload — but it is what a human243 #: reads in a log, and it keeps a site page called ``/op/something`` from looking244 #: like one of the contract ops.245 SITE_PATH_PREFIX = "/site"246 247 #: The lane path of a channel command: the worker resolves what follows on its248 #: own dispatcher, so the front names the branch and nothing else.249 WSX_PATH_PREFIX = "/wsx/openchannel"250 251 #: Where the container's own memory limit is written, cgroup v2 first and v1252 #: after: the limit file and the usage file of each layout. Outside a container253 #: none of them is there, and the host figures stand.254 CGROUP_MEMORY_FILES = (255 ("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory.current"),256 (257 "/sys/fs/cgroup/memory/memory.limit_in_bytes",258 "/sys/fs/cgroup/memory/memory.usage_in_bytes",259 ),260 )261 262 #: Where a soft quit writes its photo while it is being taken. Beside the263 #: working deposit, never inside it, so the sweep never meets it.264 REBOOT_TEMP_NAME = "reboot_temp"265 266 #: The name that same directory takes once the photo is complete — the one a267 #: boot looks for, and the only thing that tells a governed quit from a crash.268 REBOOT_DATA_NAME = "reboot_data"269 270 __all__ = [271 "CGROUP_MEMORY_FILES",272 "DECISIONS_LOGGER_NAME",273 "GUEST_PREFIX",274 "HEARTBEAT_SECONDS",275 "ORDERS_LOGGER_NAME",276 "CommanderOperations",277 "GlobalStoreOperations",278 "SpaCommander",279 ]280 281 282 class GlobalStoreOperations(RoutingClass):283 """The ``store`` branch of the commander's dispatcher: the global store on the lane.284 285 The store is one dictionary living on the commander and nowhere else, so the286 simple read, write and removal and the two halves of a turn are CALLs a287 worker places on ``/commander/store/<op>``. Every one of them takes the same288 FIFO lock: while a turn is in force, the others wait. ``delete`` is served289 under the name ``del`` — the path the workers use, which is not a Python name.290 291 Args:292 spa_commander: the vertex whose ``global_register`` and ``global_lock``293 these operations act on.294 """295 296 def __init__(self, spa_commander: Any) -> None:297 self.spa_commander = spa_commander298 self._logger = logging.getLogger(__name__)299 300 def _check_key(self, key: Any) -> None:301 """Refuse anything but a string: the dictionary's keys are literal strings."""302 if not isinstance(key, str):303 raise TypeError(f"a global-store key is a string, not {type(key).__name__}")304 305 @route()306 async def get(self, key: str) -> dict[str, Any]:307 """Read one key under the lock.308 309 Args:310 key: the literal key to read.311 312 Returns:313 ``exists`` — whether the key is there — and ``value``, TYTX-encoded,314 ``None`` encoded when the key is absent: the client tells the two315 apart by ``exists``, never by the value.316 317 Acts on nothing; waits for a turn in force.318 """319 self._check_key(key)320 async with self.spa_commander.global_lock.lock:321 store = self.spa_commander.global_register322 return {"key": key, "exists": key in store, "value": to_tytx(store.get(key), "json")}323 324 @route()325 async def set(self, key: str, value: Any = None) -> dict[str, Any]:326 """Write one key under the lock, and answer once the master holds it.327 328 Args:329 key: the literal key to write.330 value: what to write, TYTX-encoded; decoded BEFORE the lock is taken,331 so an undecodable value never holds anybody up.332 333 Returns:334 The key written.335 336 Acts on ``global_register``.337 """338 self._check_key(key)339 decoded = from_tytx(value, "json")340 async with self.spa_commander.global_lock.lock:341 self.spa_commander.global_register[key] = decoded342 return {"key": key}343 344 @route(name="del")345 async def delete(self, key: str) -> dict[str, Any]:346 """Remove one key under the lock; an absent key is a no-op.347 348 Args:349 key: the literal key to remove.350 351 Returns:352 The key removed.353 354 Acts on ``global_register``.355 """356 self._check_key(key)357 async with self.spa_commander.global_lock.lock:358 self.spa_commander.global_register.pop(key, None)359 return {"key": key}360 361 @route()362 async def lock(363 self, worker: str, request_id: str, key: str | None = None364 ) -> dict[str, Any]:365 """Park on the FIFO lock, then hand the selected value to the winner.366 367 Args:368 worker: the process asking — whose death releases the turn.369 request_id: the turn's own id, which the release quotes back.370 key: the key selected, or None for the whole dictionary.371 372 Returns:373 ``exists`` and ``value`` (TYTX-encoded) as they stand at grant time:374 the key's value, or a snapshot of the whole dictionary. The answer IS375 the grant, so a worker whose call is still parked here has simply376 not been answered yet.377 378 Acts on ``global_lock``. A grant that cannot be encoded releases the379 turn and raises, so nothing stays held for an answer that never left.380 """381 if key is not None:382 self._check_key(key)383 global_lock = self.spa_commander.global_lock384 await global_lock.acquire(worker, request_id, key)385 try:386 store = self.spa_commander.global_register387 exists = True if key is None else key in store388 value = dict(store) if key is None else store.get(key)389 return {390 "request_id": request_id,391 "key": key,392 "exists": exists,393 "value": to_tytx(value, "json"),394 }395 except Exception:396 global_lock.release()397 raise398 399 @route()400 def unlock(401 self, request_id: str, apply: bool = True, value: Any = None402 ) -> dict[str, Any]:403 """Publish a holder's complete value and let the next waiter in — or just let it in.404 405 Args:406 request_id: the turn being given back.407 apply: True publishes ``value``; False aborts, the master untouched.408 value: the COMPLETE replacement, TYTX-encoded — the selected key's409 new value, or the whole dictionary for a turn with no key.410 411 Returns:412 ``applied``: whether the value was published. A release for a turn413 no longer in force publishes nothing and releases nothing — it must414 never free a newer turn.415 416 Acts on ``global_register`` and on ``global_lock``: the value is417 decoded and checked BEFORE anything is written, then published in one418 assignment, with no await between the publication and the release.419 """420 global_lock = self.spa_commander.global_lock421 if not global_lock.holds(request_id):422 self._logger.debug("store: the release of the turn %s is no longer in force", request_id)423 return {"applied": False}424 try:425 if apply:426 decoded = from_tytx(value, "json")427 store = self.spa_commander.global_register428 if global_lock.holder_key is None:429 if not isinstance(decoded, dict):430 raise TypeError("a whole-store turn must publish a dict")431 store.clear()432 store.update(decoded)433 else:434 store[global_lock.holder_key] = decoded435 finally:436 global_lock.release()437 return {"applied": apply}438 439 def release_worker_lock(self, worker: str) -> None:440 """Give the grant back for a worker that died holding it, applying nothing.441 442 Args:443 worker: the process whose wire has just ended.444 445 Acts on ``global_lock`` when that worker was the holder, and on nothing446 at all otherwise. The changes it had made live only on its own working447 copy, which died with it — the whole death rule, and the reason the448 protocol needs no rollback.449 """450 if not self.spa_commander.global_lock.held_by(worker):451 return452 self.spa_commander.global_lock.release()453 self._logger.info("Worker %s died holding the store: released, nothing applied", worker)454 455 456 class CommanderOperations(RoutingClass):457 """The commander's dispatcher: what a worker may call on the vertex, as a tree.458 459 A CALL that climbs the lane arrives at the worker's GROUP, whose dispatcher460 forwards every ``commander/…`` path here (#59, D59-15). The tree IS the461 table of the operations: ``observation`` is a leaf of this class, ``store``462 is :class:`GlobalStoreOperations`, and a consumer attaches its own class463 under a name of its own with ``add_branches`` — once, from its subclass of464 the commander. A path nobody serves raises ``NotFound`` when the node is465 called; the wire turns it into the error REPLY.466 467 Args:468 spa_commander: the vertex.469 """470 471 def __init__(self, spa_commander: Any) -> None:472 self.spa_commander = spa_commander473 self.global_store = GlobalStoreOperations(spa_commander)474 self.add_branches([{"name": "store", "instance": self.global_store}])475 476 @route()477 def observation(self, kind: str, source: str, data: dict[str, Any]) -> dict[str, Any]:478 """Take one observation off the lane and hand it to whoever watches.479 480 Args:481 kind: the mutation the child reports.482 source: the worker it happened in.483 data: the keys that name it.484 485 Returns:486 Nothing: the child does not read this answer, it only needs one.487 """488 self.spa_commander.publish_observation(kind, source, data)489 return {}490 491 492 class SingleGroupRequired(Exception):493 """A profile names setpoints and this machine has no single group to give them to."""494 495 496 class SpaCommander:497 """The vertex of the pool: the indexes, the minting, the master store, the log.498 499 Args:500 frozen_users_path: the freezer root — the same one the workers are given,501 since a parcel written on one side is read on the other.502 groups: the grammar of this machine's groups, ``{name: kwargs}`` — one503 ``GroupHandler`` per entry, each built with the concession this504 vertex owns. Building one by hand stays legitimate: it hangs itself505 here the same way.506 default_group: which group receives whoever arrives with no past;507 None elects the first declared.508 orchestration_log_path: where the log of the orders goes; None keeps them509 on the logger alone, which is what a test wants.510 orchestration_log_max_bytes: the size at which that file rotates.511 orchestration_log_backup_count: how many rotations are kept.512 user_expiry_hours: how long a frozen user is kept before the machine513 forgets him whole.514 guest_expiry_hours: the same for somebody who never logged in, and it is515 shorter — a guest is a browser, not a person the machine knows.516 machine_memory_alarm_percent: the health line of the WHOLE machine, not517 of what this server was conceded: past it nothing grows.518 memory_max_percent: what this server may hold OF THE MACHINE — the519 concession every percentage below it is a share of. All of it by520 default.521 profiles_path: the folder of the stored profiles, when this machine may522 be reconfigured by name; None leaves only the inline apply.523 recipe_settings: the setpoints the recipe declared, kept as their own524 immutable level — every apply recomposes from it.525 env_settings: the setpoints the environment overrode, the level ABOVE526 the profile, kept immutable the same way.527 active_profile: which stored profile boot put in force, if any.528 """529 530 def __init__(531 self,532 frozen_users_path: str | Path,533 *,534 groups: dict[str, dict[str, Any]] | None = None,535 default_group: str | None = None,536 orchestration_log_path: str | Path | None = None,537 orchestration_log_max_bytes: int = 10 * 1024 * 1024,538 orchestration_log_backup_count: int = 5,539 user_expiry_hours: float = 720.0,540 guest_expiry_hours: float = 24.0,541 machine_memory_alarm_percent: float = 90.0,542 memory_max_percent: float = 100.0,543 profiles_path: str | Path | None = None,544 recipe_settings: dict[str, Any] | None = None,545 env_settings: dict[str, Any] | None = None,546 active_profile: str | None = None,547 cpu_temperature_sample_seconds: float | None = CPU_TEMPERATURE_SAMPLE_SECONDS,548 ) -> None:549 self.freeze_handler = FreezeHandler(frozen_users_path)550 self.user_expiry_hours = user_expiry_hours551 self.guest_expiry_hours = guest_expiry_hours552 self.machine_memory_alarm_percent = machine_memory_alarm_percent553 self.memory_max_percent = memory_max_percent554 if (555 cpu_temperature_sample_seconds is not None556 and cpu_temperature_sample_seconds <= 0.0557 ):558 raise ValueError("cpu_temperature_sample_seconds must be greater than zero")559 self.cpu_temperature_sample_seconds = cpu_temperature_sample_seconds560 #: The global store itself: not a master over replicas any more, the561 #: ONLY copy there is. Every read and every write of the hosted sites562 #: reaches it as a CALL on the lane, answered once it has landed.563 self.global_register = self.new_global_store()564 #: The grant of that store for a read-modify-write hold: FIFO, one565 #: holder, and a holder whose process dies releases it applying nothing.566 self.global_lock = GlobalStoreLock()567 #: The tree of what a worker may call on the vertex: ``store/…``,568 #: ``observation`` and whatever a consumer attaches.569 self.commander_dispatcher = CommanderOperations(self)570 #: Where the whole machine stands: ``running`` or ``saturated`` (no room571 #: for a newcomer anywhere). Written by the check of the resources, which572 #: arrives with the heartbeat.573 self.state = "running"574 #: The aggregate counts, one key per thing worth counting.575 self.counters: Counter[str] = Counter()576 #: The queues watching the observation stream: empty means nobody is577 #: looking, which is what keeps the workers silent.578 self._observation_queues: set[asyncio.Queue[dict[str, Any]]] = set()579 #: The anagraph: one row per identity the machine knows. Read it through580 #: the predicates, and leave the writing to the mutators.581 self.user_map: dict[str, dict[str, Any]] = {}582 #: Whose each cid is. A cid stays here once written: the cookie outlives583 #: the process, the placement and the freezer.584 self.connection_user_map: dict[str, str] = {}585 #: Which connection each page belongs to; written once, only ever removed.586 self.page_connection_map: dict[str, str] = {}587 #: The groups of this machine, by name — a group hangs itself here when588 #: it is built, the way a worker hangs itself in its own group's map.589 self.group_map: dict[str, Any] = {}590 self._group_turns: dict[str, asyncio.Task[None]] = {}591 self._beat_timer: asyncio.Task[None] | None = None592 #: One row per periodic method of this vertex — turns seen, runs, errors593 #: and the last one's text: the dashboard of who is due and who is broken.594 self.beat_counts: dict[str, dict[str, Any]] = {}595 #: Whoever is waiting for a user to have a home again, one Event per user596 #: on hold. An entry is born with the hold and dies with its release, so597 #: outside that window this map is empty.598 self.user_hold_event_map: dict[str, asyncio.Event] = {}599 self._default_group = default_group600 self._heartbeat_task: asyncio.Task[None] | None = None601 self._cpu_meter_task: asyncio.Task[None] | None = None602 self._logger = logging.getLogger(__name__)603 self._decision_sequence = 0604 self._orders_logger = self._build_orders_logger(605 orchestration_log_path,606 orchestration_log_max_bytes,607 orchestration_log_backup_count,608 )609 self._decisions_logger = self._build_decisions_logger(610 orchestration_log_path,611 orchestration_log_max_bytes,612 orchestration_log_backup_count,613 )614 for name, group_settings in (groups or {}).items():615 GroupHandler(616 self,617 name,618 memory_concession_bytes=self.memory_concession_bytes,619 **group_settings,620 )621 #: Where the named profiles are read from; None means there are none.622 self.profile_store = (623 None if profiles_path is None else OrchestrationProfileStore(profiles_path)624 )625 #: The two immutable levels of the effective configuration. They are626 #: never merged into one another: every apply recomposes627 #: recipe ⊕ profile ⊕ env from these two and the profile of the moment.628 self.recipe_settings = dict(recipe_settings or {})629 self.env_settings = dict(env_settings or {})630 #: Which stored profile is in force; None after an inline apply.631 self.active_profile = active_profile632 #: How many effective configurations this machine has had — boot's is 1,633 #: and every successful apply adds one, an idempotent apply included.634 self.configuration_generation = 1635 #: The last apply ATTEMPT, applied or rejected: what an introspection636 #: reads to know what was tried, by whom and how it ended. Boot carries637 #: no digest — no apply has run.638 self.last_apply: dict[str, Any] = {639 "ts": datetime.now(UTC).isoformat(),640 "source": "boot",641 "active_profile": active_profile,642 "digest": None,643 "outcome": "applied",644 "generation": 1,645 }646 self._configuration_lock = asyncio.Lock()647 648 @property649 def memory_concession_bytes(self) -> int:650 """What this server may hold of the machine's memory, in bytes.651 652 Returns:653 The concession — the machine's whole memory times654 ``memory_max_percent``. It is the ONE total of the cascade: a655 group's quota and a worker's ceiling are shares of it.656 """657 total = self._machine_memory_gauges()["MemTotal"]658 return int(total * self.memory_max_percent / 100.0)659 660 @property661 def memory_available_bytes(self) -> float:662 """What the machine still has free, in bytes.663 664 Returns:665 What is left of the cgroup this server runs in, or of the whole666 machine when no cgroup limits it.667 668 The twin reading of ``memory_concession_bytes``, and the other half of669 every growth: the concession says how much of the machine this server670 MAY take, this says how much there IS. It counts everything charged to671 the cgroup — this process, the templates, whatever else shares the672 container — which the workers' own photos never see.673 """674 return self._machine_memory_gauges()["MemAvailable"]675 676 @property677 def default_group(self) -> str:678 """The group that receives whoever arrives with no past.679 680 Returns:681 The elected name, or the first group declared when none was elected —682 ``group_map`` keeps them in the order the recipe named them.683 684 Raises:685 KeyError: the elected name is nobody's, or there is no group at all;686 either way a newcomer has nowhere to go.687 """688 name = self._default_group or next(iter(self.group_map), None)689 if name not in self.group_map:690 raise KeyError(f"Vertex: no group to receive a newcomer ({name!r})")691 return name692 693 async def serve_request(694 self, cid: str | None, http: Frame, *, hold_timeout: float695 ) -> Frame:696 """Serve one request of the hosted site, from the cookie to the answer.697 698 Args:699 cid: the connection the request carries, None when it carries none —700 a browser the site has never named.701 http: the request in the form the child reads, without the cid.702 hold_timeout: the WHOLE time this request may spend waiting for a user703 who is between two homes, however many times it has to wait.704 705 Returns:706 The child's REPLY payload, untouched — reading it is the front's job.707 708 Raises:709 AssignmentRefused: nobody can take him now, and ``retry_after`` says710 when the machine will have decided again.711 SiteFailedRequest: his worker answered with a failure.712 ConnectionError: the wire of his worker is gone.713 714 Acts on the indexes only through the chain: a request with no connection,715 or with one the indexes never saw, travels ANONYMOUS to the default716 group's reception — the site baptises while serving, and the fold of its717 own announcements is what writes the indexes. A known user with no home718 is placed, as ever.719 """720 user, worker_handler = await self.resolve_worker(cid, hold_timeout=hold_timeout)721 return await self._call_worker(722 worker_handler,723 http.path,724 Frame(id=http.id, method="CALL", path=http.path,725 info={**http.info, "cid": cid}, payload=http.payload),726 user,727 )728 729 async def serve_wsx_request(730 self, cid: str | None, payload: dict[str, Any], *, hold_timeout: float731 ) -> dict[str, Any]:732 """Serve one channel command of a page, the way a request of the site is served.733 734 Args:735 cid: the connection the command came in on.736 payload: what the front composed — the ``wsx`` form, with no737 ``http`` dict in it.738 hold_timeout: the whole time this command may spend waiting for a739 user who is between two homes.740 741 Returns:742 The child's REPLY payload, untouched.743 744 Raises:745 AssignmentRefused: nobody can take him now.746 SiteFailedRequest: his worker answered with a failure.747 748 The barrier, the reception-first rule and the placement are the ones an749 http request meets — the same ``resolve_worker`` — because a channel750 command belongs to a user exactly as a request does: it writes on the751 row of one of his pages, and that row lives where he lives.752 """753 user, worker_handler = await self.resolve_worker(cid, hold_timeout=hold_timeout)754 return await self._call_worker(worker_handler, WSX_PATH_PREFIX, payload, user)755 756 async def resolve_worker(757 self, cid: str | None, *, hold_timeout: float758 ) -> tuple[str | None, Any]:759 """Who this connection is, and which worker will serve him.760 761 Args:762 cid: the connection the request carries, None when it carries none.763 hold_timeout: the whole time this request may spend waiting for a764 user who is between two homes, however many times it waits.765 766 Returns:767 The identity — ``None`` for a browser the site never named — and the768 handler of the worker that hosts him: his own, the one the placement769 just gave him, or his group's reception when he is a guest.770 771 Raises:772 AssignmentRefused: nobody can take him now, and ``retry_after`` says773 when the machine will have decided again.774 775 Every form of request comes through here, so the barrier, the776 reception-first rule and the placement are written once and every777 caller meets them the same way.778 """779 deadline = asyncio.get_running_loop().time() + hold_timeout780 while True:781 try:782 user = self.resolve_user(cid)783 break784 except UserOnHold as waiting:785 await self._wait_out_hold(waiting.user, deadline)786 if user is None or user.startswith(GUEST_PREFIX):787 # RECEPTION-FIRST, the ratified rule: as long as somebody is a788 # guest he never leaves the reception — anonymous first visit and789 # baptised guest alike. Only the login makes him placeable.790 group_handler = self.group_map[791 (self.user_map[user]["group"] if user is not None else None) or self.default_group792 ]793 reception = group_handler.reception794 # The saturation doctrine holds for the STRANGER: no room, polite795 # refusal. A guest already inside is served as ever.796 if reception is None or (user is None and group_handler.state == "saturated"):797 raise self._refused(798 AssignmentRefused(799 user or cid or "a newcomer",800 "no room for a newcomer: the pool is restricted",801 )802 ) from None803 return user, reception804 group_handler = self.group_map[self.user_map[user]["group"] or self.default_group]805 try:806 worker_name = group_handler.user_worker_map.get(807 user808 ) or await group_handler.assign_user(user)809 except AssignmentRefused as refusal:810 raise self._refused(refusal) from None811 return user, group_handler.worker_handler_map[worker_name]812 813 @overload814 async def _call_worker(815 self, worker_handler: Any, path: str, payload: Frame, user: str | None816 ) -> Frame: ...817 818 @overload819 async def _call_worker(820 self, worker_handler: Any, path: str, payload: dict[str, Any], user: str | None821 ) -> dict[str, Any]: ...822 823 async def _call_worker(824 self, worker_handler: Any, path: str, payload: dict[str, Any] | Frame, user: str | None825 ) -> dict[str, Any] | Frame:826 """Put one request on a worker's lane and hand its REPLY back.827 828 Args:829 worker_handler: the handler of the worker that will serve it.830 path: the path on the lane — the caller's own, because the forms do831 not share one.832 payload: what the caller composed, without the identity.833 user: whose request it is, added here with the freeze verdict,834 because both are the vertex's knowledge and not the caller's.835 836 Returns:837 The child's REPLY payload, untouched — reading it is the front's job.838 839 Raises:840 SiteFailedRequest: his worker answered with a failure.841 ConnectionError: the wire of his worker is gone.842 """843 trusted = {"identity": user,844 "user_frozen": self.user_is_frozen(user) if user is not None else False}845 if isinstance(payload, Frame):846 reply = await worker_handler.connector.call_frame(847 Frame(id=payload.id, method="CALL", path=path,848 info={**payload.info, **trusted}, payload=payload.payload)849 )850 if "error" in reply.info:851 raise SiteFailedRequest(user, str(reply.info["error"]), reply.info.get("status"))852 return reply853 reply_frame = await worker_handler.connector.call_frame(Frame(854 method="CALL", path=path, info={"format": "control-json", **trusted},855 payload=ControlPayload().encode(payload)))856 reply = {**(ControlPayload().decode(reply_frame.payload) or {}),857 **{key: value for key, value in reply_frame.info.items() if key != "format"}}858 if "error" in reply:859 raise SiteFailedRequest(user, str(reply["error"]), reply.get("status"))860 return reply861 862 async def _wait_out_hold(self, user: str, deadline: float) -> None:863 """Wait for a user to have a home again, inside what is left of the budget.864 865 A budget already spent is a wait of no seconds, which is the refusal866 itself: the request has waited as long as it said it would.867 """868 try:869 await self.await_user_release(user, deadline - asyncio.get_running_loop().time())870 except TimeoutError:871 raise self._refused(872 AssignmentRefused(user, "he is still between two homes")873 ) from None874 875 def _refused(self, refusal: AssignmentRefused) -> AssignmentRefused:876 """Count one request the pool could not take, and tell it when to come back."""877 self.counters["requests_refused"] += 1878 refusal.retry_after = SHAPE_REVIEW_SECONDS879 return refusal880 881 @property882 def console_targets(self) -> list[str]:883 """Every process the debug door can look into: this vertex, then the workers."""884 names = ["commander"]885 for group_handler in self.group_map.values():886 names.extend(group_handler.worker_handler_map)887 return names888 889 async def eval_in_target(self, target: str, expr: str) -> str:890 """Evaluate one debug expression in one process of the pool, repr back.891 892 Args:893 target: ``commander`` for this very process, or a worker's name.894 expr: a Python expression; the namespace holds ``commander`` here,895 ``worker`` inside a child.896 897 Returns:898 The ``repr`` of the value, whatever the expression reached — the899 point of an eval door is answering questions nobody predicted.900 901 Raises:902 KeyError: no such target; the ones there are travel in the error.903 RuntimeError: the child refused the expression — its error verbatim.904 905 Full eval power by construction: the door exists only where the906 console surface was mounted on purpose, never in production.907 """908 if target == "commander":909 return repr(eval(expr, {"commander": self}))910 for group_handler in self.group_map.values():911 worker_handler = group_handler.worker_handler_map.get(target)912 if worker_handler is not None:913 reply = await worker_handler.connector.call(EVAL_OP_PATH, {"expr": expr})914 if "error" in reply:915 raise RuntimeError(str(reply["error"]))916 return reply["result"]["repr"]917 raise KeyError(918 f"eval: no target {target!r} here — have: {', '.join(self.console_targets)}"919 )920 921 async def subscribe_observation(self, queue: asyncio.Queue[dict[str, Any]]) -> None:922 """Put one queue on the observation stream, switching the workers on if it is the first.923 924 Args:925 queue: where every observation is put from now on.926 927 Acts on the pool: the first subscriber turns the reporting on in every928 living process, so nothing is paid for while nobody watches.929 """930 first = not self._observation_queues931 self._observation_queues.add(queue)932 if first:933 await self.switch_observation(True)934 935 async def unsubscribe_observation(self, queue: asyncio.Queue[dict[str, Any]]) -> None:936 """Take one queue off the stream, switching the workers off with the last of them.937 938 Args:939 queue: the queue that stops watching; one that never subscribed is940 no error, the stream is a debug surface.941 942 Acts on the pool: the last leaving turns the reporting off everywhere.943 """944 self._observation_queues.discard(queue)945 if not self._observation_queues:946 await self.switch_observation(False)947 948 @property949 def observation_watched(self) -> bool:950 """Whether anybody is on the observation stream right now."""951 return bool(self._observation_queues)952 953 async def switch_observation(self, on: bool) -> None:954 """Tell every living worker whether to report its mutations.955 956 Args:957 on: True to report, False to fall silent.958 959 Acts on the processes. A worker that does not answer is logged and960 skipped: the stream is best-effort and never holds up the pool.961 """962 for group_handler in self.group_map.values():963 for worker_handler in group_handler.living_workers:964 try:965 await worker_handler.connector.call(OBSERVE_OP_PATH, {"on": on})966 except Exception as exc:967 self._logger.debug(968 "Observation switch %s refused by %s (%s)", on, worker_handler.name, exc969 )970 971 @property972 def envelope_handler(self) -> CommanderEnvelopeHandler:973 """The last layer of the envelope chain: what the fold does at this level.974 975 Read once by every ``GroupHandler`` at its birth, which hands the layer976 to its own. A consumer's commander returns its subclass of977 ``CommanderEnvelopeHandler`` here — an ``on_<op>`` of its own that calls978 the core's and then reads what the event carries for it, such as the979 tables a newborn page subscribes — and the fold stays one chain.980 """981 return CommanderEnvelopeHandler(self)982 983 def on_worker_presented(self, worker_handler: Any) -> None:984 """A process has just presented itself on its wire: the seam a consumer overrides.985 986 Args:987 worker_handler: the handler of the newborn process; its ``connector``988 is what a CALL to it is placed on.989 990 Called by ``WorkerHandler.read_envelope`` on the envelope that carries991 the presentation, once per process. The core has nothing to tell a992 newborn: a consumer that must (the source filter of a hosted site, say)993 overrides this and places its CALL on a task of its own, never holding994 up the envelope.995 """996 997 def new_global_store(self) -> dict[str, Any]:998 """The vertex's data at birth: one empty dictionary.999 1000 Returns:1001 ``dict[str, Any]``, the type the store protocol fixes: literal string1002 keys, opaque values. A consumer may fill it, never change its type.1003 """1004 return {}1005 1006 def publish_observation(self, kind: str, source: str, data: dict[str, Any]) -> None:1007 """Put one observation on every watching queue, and never raise.1008 1009 Args:1010 kind: the mutation being reported.1011 source: the worker it happened in, or ``commander`` for a fold here.1012 data: the keys that name it.1013 1014 A full queue drops the event: a slow observer loses what it could not1015 read, and the pool does not wait for it.1016 """1017 event = {"kind": kind, "source": source, "data": data}1018 for queue in list(self._observation_queues):1019 try:1020 queue.put_nowait(event)1021 except asyncio.QueueFull:1022 self._logger.debug("Observation %s dropped: a watcher is not reading", kind)1023 1024 async def get_pool_census(self) -> dict[str, Any]:1025 """The whole pool read out for a human: this vertex, then every worker.1026 1027 Returns:1028 The routing maps and counters of the vertex, one entry per group1029 with its placements and shape,1030 and one census per living worker under ``workers`` — a worker that1031 does not answer appears as ``{"error": ...}`` instead of raising.1032 1033 JSON-safe by construction: no live store and no object is in here.1034 """1035 census: dict[str, Any] = {1036 "user_map": {1037 user: {1038 "group": row["group"],1039 "frozen": row["frozen"],1040 "on_hold": row["on_hold"],1041 }1042 for user, row in self.user_map.items()1043 },1044 "connection_user_map": dict(self.connection_user_map),1045 "page_connection_map": dict(self.page_connection_map),1046 "counters": dict(self.counters),1047 "default_group": self.default_group,1048 "groups": {},1049 "workers": {},1050 }1051 for name, group_handler in self.group_map.items():1052 census["groups"][name] = {1053 "user_worker_map": dict(group_handler.user_worker_map),1054 "living_workers": [1055 worker_handler.name for worker_handler in group_handler.living_workers1056 ],1057 "memory_occupied_percent": group_handler.memory_occupied_percent,1058 "memory_accounting": group_handler.memory_accounting_kind,1059 "worker_max_number": group_handler.worker_max_number,1060 "workers": {1061 worker_handler.name: {1062 "state": worker_handler.state,1063 "memory_occupancy_percent": (1064 group_handler.get_memory_occupancy_percent(1065 worker_handler.worker_snapshot1066 )1067 ),1068 "rss_bytes": (worker_handler.worker_snapshot or {}).get("rss_bytes"),1069 "pss_bytes": (worker_handler.worker_snapshot or {}).get("pss_bytes"),1070 "accounted_memory_bytes": group_handler.get_memory_accounting(1071 worker_handler.worker_snapshot1072 )[0],1073 "memory_accounting": group_handler.get_memory_accounting(1074 worker_handler.worker_snapshot1075 )[1],1076 "cpu_temperature_percent": (1077 worker_handler.cpu_temperature_percent1078 ),1079 "cpu_temperature_sample_percent": (1080 worker_handler.cpu_temperature_sample_percent1081 ),1082 "cpu_temperature_interval_seconds": (1083 worker_handler.cpu_temperature_interval_seconds1084 ),1085 "cpu_temperature_age_seconds": (1086 None1087 if worker_handler.cpu_temperature_sampled_at is None1088 else max(1089 0.0,1090 time.monotonic()1091 - worker_handler.cpu_temperature_sampled_at,1092 )1093 ),1094 }1095 for worker_handler in group_handler.living_workers1096 },1097 }1098 for worker_handler in group_handler.living_workers:1099 worker_census = await self._get_worker_census(worker_handler)1100 worker_census["cpu_temperature_percent"] = (1101 worker_handler.cpu_temperature_percent1102 )1103 worker_census["cpu_temperature_sample_percent"] = (1104 worker_handler.cpu_temperature_sample_percent1105 )1106 worker_census["cpu_temperature_interval_seconds"] = (1107 worker_handler.cpu_temperature_interval_seconds1108 )1109 worker_census["cpu_temperature_age_seconds"] = (1110 None1111 if worker_handler.cpu_temperature_sampled_at is None1112 else max(1113 0.0,1114 time.monotonic() - worker_handler.cpu_temperature_sampled_at,1115 )1116 )1117 census["workers"][worker_handler.name] = worker_census1118 return census1119 1120 async def _get_worker_census(self, worker_handler: Any) -> dict[str, Any]:1121 """One worker's census off the lane, or the error entry when it does not answer."""1122 try:1123 reply = await worker_handler.connector.call(CENSUS_OP_PATH, {})1124 except Exception as exc:1125 return {"error": f"{type(exc).__name__}: {exc}"}1126 if "error" in reply:1127 return {"error": str(reply["error"])}1128 return dict(reply["result"])1129 1130 def resolve_user(self, cid: str | None) -> str | None:1131 """Whose cid this is — None for a browser the site has not named yet.1132 1133 Args:1134 cid: the connection the cookie carries, or None for no cookie.1135 1136 Returns:1137 The user this connection belongs to, or None: the vertex MINTS1138 NOBODY — the identity and the guest name are the hosted site's,1139 learned from its own ``new_connection`` through the fold. A None1140 routes to the reception, anonymous.1141 1142 Raises:1143 UserOnHold: this user is between two homes.1144 1145 Acts on the indexes only to heal a known cid whose user row is gone —1146 the cookie outlived the row, the browser is still known, its state is1147 not.1148 """1149 user = self.connection_user_map.get(cid)1150 if user is None:1151 return None1152 if user not in self.user_map:1153 self.user_map[user] = self._new_row()1154 row = self.user_map[user]1155 if row["on_hold"] is not None:1156 raise UserOnHold(user, row["on_hold"])1157 return user1158 1159 def record_connection_user(self, cid: str, user: str) -> None:1160 """The junction, written at the fact: the routing cookie learns whose it is.1161 1162 Args:1163 cid: the connection the site named while serving.1164 user: the identity the site baptised while serving it.1165 1166 Acts on ``connection_user_map`` and, for an identity never seen, on1167 ``user_map``. Called by the fold of ``new_connection`` — the one road1168 an identity enters the vertex by.1169 """1170 self.connection_user_map[cid] = user1171 if user not in self.user_map:1172 self.user_map[user] = self._new_row()1173 1174 def user_is_frozen(self, user: str) -> bool:1175 """Whether this user's state is in the freezer rather than in a process.1176 1177 Args:1178 user: the identity to judge.1179 1180 Returns:1181 True when the mark is on. An identity with no row at all is not1182 frozen — there is nothing of his anywhere.1183 """1184 row = self.user_map.get(user)1185 return bool(row and row["frozen"])1186 1187 def get_user_expiry_seconds(self, user: str) -> float:1188 """How long this identity is kept without a sign of life, in seconds.1189 1190 Args:1191 user: the identity; whether he is a guest is read off his name.1192 1193 Returns:1194 The horizon in seconds — the guest's is the shorter, because a guest1195 is a browser and not a person the machine knows.1196 1197 ONE horizon per identity, whatever is being measured against it: the age1198 of a parcel for whoever is in the deposit, the silence off the photo's1199 clocks for whoever is still on a worker.1200 """1201 guest = user.startswith(GUEST_PREFIX)1202 return (self.guest_expiry_hours if guest else self.user_expiry_hours) * SECONDS_PER_HOUR1203 1204 def hold_user(self, user: str, cause: str) -> None:1205 """Put a user in the waiting room: his next request waits instead of routing.1206 1207 Args:1208 user: the identity on his way out of the process he lives on.1209 cause: what put him there, kept for the log.1210 1211 Acts on his row AND on the barrier whoever asks for him will wait on;1212 a hold already there keeps its first cause and its own Event.1213 """1214 row = self.user_map[user]1215 if row["on_hold"] is None:1216 row["on_hold"] = cause1217 self.user_hold_event_map[user] = asyncio.Event()1218 1219 async def await_user_release(self, user: str, timeout: float) -> None:1220 """Wait until this user has a home again, or give up at the deadline.1221 1222 Args:1223 user: the identity somebody's request found on hold.1224 timeout: how long that request may wait — the caller's own patience.1225 1226 Raises:1227 TimeoutError: the hold outlived the deadline.1228 1229 Nothing is written. A user whose hold fell between the raise and this1230 call has no barrier left and is not waited for at all.1231 """1232 event = self.user_hold_event_map.get(user)1233 if event is not None:1234 await asyncio.wait_for(event.wait(), timeout)1235 1236 def release_user_hold(self, user: str) -> None:1237 """Let a user out of the waiting room, leaving him where he already was.1238 1239 Args:1240 user: the identity whose ordered departure did not happen.1241 1242 Acts on his row and on his barrier: the hold goes off and whoever waited1243 for him walks again. Nothing else is written, which is the whole point —1244 he is not frozen and not gone, he is still on the worker he was on.1245 """1246 self.user_map[user]["on_hold"] = None1247 self._release_hold(user)1248 1249 def _release_hold(self, user: str) -> None:1250 """Let go of whoever was waiting for this user, and forget his barrier."""1251 event = self.user_hold_event_map.pop(user, None)1252 if event is not None:1253 event.set()1254 1255 def drop_page(self, page_id: str) -> None:1256 """Forget a page.1257 1258 Args:1259 page_id: the page that is gone; one already forgotten is that same1260 outcome.1261 1262 Acts on ``page_connection_map``.1263 """1264 self.page_connection_map.pop(page_id, None)1265 1266 def drop_connection(self, cid: str) -> None:1267 """Forget a connection's pages, and keep the connection's identity.1268 1269 Args:1270 cid: the connection that is gone.1271 1272 Acts on ``page_connection_map``: the cid stays in ``connection_user_map``,1273 because the cookie is eternal.1274 """1275 for page_id in [page for page, owner in self.page_connection_map.items() if owner == cid]:1276 del self.page_connection_map[page_id]1277 1278 def drop_user(self, user: str) -> bool:1279 """Forget an identity whole: his row, his connections, his pages, his freezer state.1280 1281 Args:1282 user: the identity that is gone; one already forgotten is that same1283 outcome.1284 1285 Returns:1286 Whether the freezer was holding anything of his.1287 1288 Acts on all three indexes, on his barrier — whoever waited for him is1289 woken to find him gone and starts over — and on the freezer.1290 """1291 self.user_map.pop(user, None)1292 self._release_hold(user)1293 for cid in [cid for cid, owner in self.connection_user_map.items() if owner == user]:1294 self.drop_connection(cid)1295 del self.connection_user_map[cid]1296 had_state = self.freeze_handler.drop_user_folder(user)1297 if had_state:1298 self.counters["frozen_users_discarded"] += 11299 return had_state1300 1301 def change_connection_user(self, cid: str, user: str, previous_user: str) -> None:1302 """The login, as the surface sees it: a connection changes owner.1303 1304 Args:1305 cid: the connection that logged in.1306 user: the identity it belongs to from now on.1307 previous_user: who it belonged to a moment ago.1308 1309 Acts on two indexes and on nothing else: the cid points at its new owner,1310 whose row is brought into being when he is unknown here, and the guest1311 left behind goes — he had this one connection and nothing else, by1312 construction. A previous identity that is NOT a guest keeps his row: he1313 is a person with a life of his own, and losing a connection is not losing1314 him. Nothing is placed: where the user lives is his next request's1315 business, and the freezer is not touched — a guest never had a folder.1316 """1317 self.connection_user_map[cid] = user1318 if user not in self.user_map:1319 self.user_map[user] = self._new_row()1320 if previous_user.startswith(GUEST_PREFIX):1321 self.user_map.pop(previous_user, None)1322 1323 def record_user_group(self, user: str, group: str) -> None:1324 """Write down which group a user was placed on.1325 1326 Args:1327 user: the identity that has just been given a home.1328 group: the group that took him.1329 1330 Acts on his row. Called by the group in the same breath in which it1331 writes its own map, so the two can never say different things.1332 """1333 self.user_map[user]["group"] = group1334 1335 def mark_user_frozen(self, user: str) -> None:1336 """Write down that a user's state is on disk.1337 1338 Args:1339 user: the identity that left his process.1340 1341 Acts on his row and on his barrier: the mark goes on and the wait he may1342 have been in is over.1343 """1344 row = self.user_map[user]1345 row["frozen"] = True1346 row["on_hold"] = None1347 self._release_hold(user)1348 1349 def mark_user_adopted(self, user: str) -> None:1350 """Write down that a user came home from the freezer.1351 1352 Args:1353 user: the identity now living in a process again.1354 1355 Acts on his row and on his barrier: the mark goes off, the wait is over.1356 """1357 row = self.user_map[user]1358 row["frozen"] = False1359 row["on_hold"] = None1360 self._release_hold(user)1361 1362 def drop_users(self, users: list[str], *, cause: str) -> None:1363 """Take these users out of the machine and discard whatever they left on disk.1364 1365 Args:1366 users: the identities to forget.1367 cause: why, for the log.1368 1369 Acts on all three indexes and on the freezer, one user at a time, each1370 departure named in the log with whether it had state to lose.1371 """1372 for user in users:1373 had_state = self.drop_user(user)1374 self.log_order(1375 "vertex",1376 "drop_user",1377 user,1378 numbers={"had_state": had_state},1379 outcome=cause,1380 )1381 1382 def log_order(1383 self,1384 decided_by: str,1385 order: str,1386 subject: str | None = None,1387 *,1388 numbers: dict[str, Any] | None = None,1389 outcome: str | None = None,1390 reason: str = "order_issued",1391 ) -> None:1392 """Write one row of the orchestration log: an order, and what came of it.1393 1394 Args:1395 decided_by: who decided — a group, a handler, the vertex itself.1396 order: what was decided.1397 subject: on whom or on what.1398 numbers: what the decider had in front of it when it decided.1399 outcome: how it ended.1400 reason: the stable reason code carried by the structured journal.1401 """1402 self._orders_logger.info(1403 "decided_by=%s order=%s subject=%s numbers=%s outcome=%s",1404 decided_by,1405 order,1406 subject,1407 numbers,1408 outcome,1409 )1410 self.log_decision(1411 decided_by,1412 order,1413 outcome or "ordered",1414 reason=reason,1415 subject=subject,1416 numbers=numbers,1417 )1418 1419 def log_decision(1420 self,1421 decided_by: str,1422 decision: str,1423 outcome: str,1424 *,1425 reason: str,1426 subject: str | None = None,1427 numbers: dict[str, Any] | None = None,1428 candidates: list[dict[str, Any]] | None = None,1429 ) -> None:1430 """Write one structured judgment, including a stable reason code.1431 1432 A decision may issue no order: candidate selection, suppression and a1433 deliberate no-op belong here too. Records are JSONL so a monitor can1434 filter and correlate them without parsing prose.1435 """1436 self._decision_sequence += 11437 record = {1438 "schema": 1,1439 "decision_id": f"{os.getpid()}-{self._decision_sequence}",1440 "timestamp": datetime.now(UTC).isoformat(),1441 "decided_by": decided_by,1442 "decision": decision,1443 "subject": subject,1444 "outcome": outcome,1445 "reason": reason,1446 "numbers": numbers or {},1447 "candidates": candidates or [],1448 }1449 self._decisions_logger.info(1450 json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)1451 )1452 1453 @property1454 def configured_group(self) -> Any:1455 """The one group a profile governs.1456 1457 Raises:1458 SingleGroupRequired: this machine has zero or several groups, so a1459 profile names setpoints without saying whose.1460 """1461 if len(self.group_map) != 1:1462 raise SingleGroupRequired(1463 f"Vertex: a profile governs exactly one group, this machine has "1464 f"{len(self.group_map)} ({sorted(self.group_map)})"1465 )1466 return next(iter(self.group_map.values()))1467 1468 async def apply_group_settings(1469 self,1470 *,1471 profile: dict[str, Any] | None = None,1472 profile_name: str | None = None,1473 source: str = "inline",1474 ) -> dict[str, Any]:1475 """Put a new effective configuration in force on the one group, or refuse it whole.1476 1477 Args:1478 profile: the profile level given inline; the active profile becomes1479 None, since nothing stored is in force any more.1480 profile_name: the stored profile to read as that level instead, which1481 becomes the active one. Never both.1482 source: who asked — the word that reaches the audit and the answer.1483 1484 Returns:1485 The payload of the apply: ``outcome``, ``source``, ``active_profile``,1486 ``generation``, ``changed_settings`` and ``effective_settings``.1487 1488 Raises:1489 SingleGroupRequired: not exactly one group.1490 OrchestrationProfileNameError, OrchestrationProfileNotFoundError,1491 OrchestrationProfileContentError: the1492 stored profile could not be read.1493 GroupPolicyError: the composed settings are invalid, carrying every1494 violation found.1495 1496 Acts on the group's policy, on the CPU admission of its workers and on1497 this vertex's generation and record. Three stages: everything fallible1498 happens BEFORE anything moves, the swap itself is assignments only, and1499 the log and the wake come after and cannot undo it. The whole apply is1500 serialized on ``_configuration_lock``, the profile read included, so two1501 callers queue instead of colliding.1502 """1503 async with self._configuration_lock:1504 try:1505 group = self.configured_group1506 if profile_name is not None:1507 if self.profile_store is None:1508 raise OrchestrationProfileNotFoundError(1509 "this machine was given no profiles folder"1510 )1511 profile = await asyncio.to_thread(self.profile_store.read, profile_name)1512 new_policy = GroupPolicy.from_settings(1513 {**self.recipe_settings, **(profile or {}), **self.env_settings}1514 )1515 except Exception as error:1516 self._audit_settings_refusal(profile_name, source, error)1517 raise1518 in_force = group.policy.to_settings()1519 effective = new_policy.to_settings()1520 changed = {key: value for key, value in effective.items() if value != in_force[key]}1521 reconciliation = self._cpu_reconciliation(group, new_policy)1522 record = {1523 "ts": datetime.now(UTC).isoformat(),1524 "source": source,1525 "active_profile": profile_name,1526 "digest": self._settings_digest(effective),1527 "outcome": "applied",1528 "generation": self.configuration_generation + 1,1529 }1530 payload = {1531 "outcome": "applied",1532 "source": source,1533 "active_profile": profile_name,1534 "generation": record["generation"],1535 "changed_settings": changed,1536 "effective_settings": effective,1537 }1538 self._commit_group_settings(group, new_policy, reconciliation, record)1539 try:1540 self.log_order(1541 "vertex",1542 "apply_group_settings",1543 profile_name or source,1544 numbers={1545 "generation": record["generation"],1546 "digest": record["digest"],1547 "source": source,1548 "changed": changed,1549 },1550 outcome="applied",1551 )1552 self.log_order(1553 "vertex",1554 "cpu_policy_reconciled",1555 profile_name or source,1556 numbers=dict(reconciliation),1557 )1558 except Exception:1559 self._logger.exception("Vertex: the apply of the setpoints could not be audited")1560 try:1561 group.ping_now()1562 except Exception:1563 self._logger.exception("Vertex: the round after the apply could not be anticipated")1564 return payload1565 1566 def _commit_group_settings(1567 self,1568 group: Any,1569 new_policy: GroupPolicy,1570 reconciliation: list[tuple[str, bool]],1571 record: dict[str, Any],1572 ) -> None:1573 """Put the prepared configuration in force: guaranteed assignments only.1574 1575 Args:1576 group: the group the setpoints govern.1577 new_policy: the validated policy that replaces its current one.1578 reconciliation: the CPU admission each worker lands on, as stage one1579 judged it; a worker that left meanwhile is dropped here.1580 record: the audit row of this apply, generation included.1581 1582 No await and nothing that can raise: the loop is never yielded between1583 the swap, the generation and the record, so no task can read one of the1584 three without the other two.1585 """1586 group.apply_policy(1587 new_policy, [pair for pair in reconciliation if pair[0] in group.worker_handler_map]1588 )1589 self.active_profile = record["active_profile"]1590 self.configuration_generation = record["generation"]1591 self.last_apply = record1592 1593 def _cpu_reconciliation(1594 self, group: Any, policy: GroupPolicy1595 ) -> list[tuple[str, bool]]:1596 """Where each worker's CPU admission lands under the NEW thresholds.1597 1598 Args:1599 group: the group whose workers are judged.1600 policy: the policy about to govern them.1601 1602 Returns:1603 One ``(worker name, cpu_admission_open)`` pair per worker. The band1604 between the two new thresholds PRESERVES what the worker is now —1605 that state is the memory of the hysteresis; a policy that is off and1606 a worker with no photo are both open, and nothing is grown here.1607 """1608 reconciliation = []1609 for worker_handler in group.worker_handler_map.values():1610 cpu_temperature_percent = worker_handler.get_cpu_temperature_percent()1611 if policy.cpu_admission_close_percent is None or cpu_temperature_percent is None:1612 admission_open = True1613 elif cpu_temperature_percent > policy.cpu_admission_close_percent:1614 admission_open = False1615 elif cpu_temperature_percent < policy.cpu_admission_reopen_percent:1616 admission_open = True1617 else:1618 admission_open = worker_handler.cpu_admission_open1619 reconciliation.append((worker_handler.name, admission_open))1620 return reconciliation1621 1622 def _settings_digest(self, settings: dict[str, Any]) -> str:1623 """The fingerprint of one effective configuration: sha256 of its canonical JSON."""1624 canonical = json.dumps(settings, sort_keys=True, allow_nan=False)1625 return hashlib.sha256(canonical.encode()).hexdigest()1626 1627 def _audit_settings_refusal(1628 self, profile_name: str | None, source: str, error: Exception1629 ) -> None:1630 """Record an apply that never happened; the machine stays where it was.1631 1632 Args:1633 profile_name: the profile that was asked for, if any.1634 source: who asked.1635 error: what refused it — a ``GroupPolicyError`` carries every1636 violation, anything else speaks for itself.1637 1638 Acts on ``last_apply``, which is the last ATTEMPT: the generation and the1639 active profile stay the ones in force, and there is no digest because1640 there is no new configuration.1641 """1642 violations = error.violations if isinstance(error, GroupPolicyError) else [str(error)]1643 outcome = f"rejected: {violations[0]}"1644 if len(violations) > 1:1645 outcome += f"+{len(violations) - 1}"1646 self.last_apply = {1647 "ts": datetime.now(UTC).isoformat(),1648 "source": source,1649 "active_profile": self.active_profile,1650 "digest": None,1651 "outcome": outcome,1652 "generation": self.configuration_generation,1653 }1654 self.log_order(1655 "vertex",1656 "apply_group_settings",1657 profile_name or source,1658 numbers={"generation": self.configuration_generation, "violations": violations},1659 outcome=outcome,1660 )1661 1662 def adopt_frozen_registers(self) -> None:1663 """Become what the last soft quit froze, if it is there; boot clean if not.1664 1665 Acts on the disk — only ever through a ``FreezeHandler`` — and on the1666 indexes, in this order and no other. The working deposit is wiped FIRST1667 and ALWAYS (F4): nothing a previous run left there survives a start. A1668 leftover ``reboot_temp`` — a quit that died halfway — is dropped unread.1669 1670 Then ``reboot_data`` is asked for the frozen commander registers. They1671 are read BEFORE anything is moved: a read that fails must leave a clean1672 boot behind, not parcels no map knows about — those would be swept1673 within the hour as orphans. Read, their item is dropped, the directory1674 is renamed onto the working deposit — every lazy wake from here reads1675 the ordinary place, with the ordinary handler — and the three maps and1676 the global store become this vertex's, every user frozen. Anything1677 missing or unreadable means the current behaviour: boot clean, said1678 once in the log. Never a partial adoption.1679 """1680 self.freeze_handler.wipe_root()1681 FreezeHandler(self.reboot_temp_path).drop_root()1682 reboot = FreezeHandler(self.reboot_data_path)1683 try:1684 saved = reboot.read_commander_register_item()1685 except Exception:1686 self._logger.exception(1687 "Vertex: the frozen commander registers could not be read — booting clean"1688 )1689 saved = None1690 if saved is None:1691 reboot.drop_root()1692 return1693 reboot.drop_commander_register_item()1694 self.freeze_handler.drop_root()1695 reboot.rename_root(self.freeze_handler.root_path)1696 self.user_map = saved["user_map"]1697 self.connection_user_map = saved["connection_user_map"]1698 self.page_connection_map = saved["page_connection_map"]1699 self.global_register = saved["global_register"]1700 self.log_order(1701 "vertex", "adopt_frozen_registers", "-", numbers={"users": len(self.user_map)}1702 )1703 1704 async def start(self) -> None:1705 """Bring the machine up: the reception of the base group, then the beat.1706 1707 Acts on the base group — its reception is launched and awaited, so this1708 returns when the machine is READY to be served through — and on this1709 vertex, whose clock starts last. A reception that would not start leaves1710 its group ``broken``: the beat is running by then, and the group tries1711 again at its own round.1712 """1713 await asyncio.to_thread(self.adopt_frozen_registers)1714 await self.drop_expired_users(now=True)1715 await self.group_map[self.default_group].start_worker()1716 if self.cpu_temperature_sample_seconds is not None:1717 self._cpu_meter_task = asyncio.ensure_future(self.cpu_meter_loop())1718 self._heartbeat_task = asyncio.ensure_future(self.heartbeat_loop())1719 1720 @property1721 def reboot_temp_path(self) -> Path:1722 """Where a soft quit writes, beside the working deposit and never inside it."""1723 return self.freeze_handler.root_path.parent / REBOOT_TEMP_NAME1724 1725 @property1726 def reboot_data_path(self) -> Path:1727 """The same directory once the photo is complete — the name a boot looks for."""1728 return self.freeze_handler.root_path.parent / REBOOT_DATA_NAME1729 1730 async def quit(self) -> None:1731 """The soft quit: everybody parked in the photo, and the photo committed.1732 1733 Acts on this vertex — the clock stops first, so no round can write while1734 the photo is taken — on every group, each ordered to park its users in1735 ``reboot_temp``, and on the disk, where the vertex adds its own item and1736 then renames the directory to ``reboot_data``.1737 1738 The rename is the commit (F5): a directory under the final name is a1739 COMPLETE photo by construction, and a quit that dies halfway leaves the1740 provisional name, which no boot looks at. The vertex writes LAST because1741 it is the only one that knows the groups are done.1742 """1743 if self._heartbeat_task is not None:1744 self._heartbeat_task.cancel()1745 self._heartbeat_task = None1746 if self._cpu_meter_task is not None:1747 self._cpu_meter_task.cancel()1748 self._cpu_meter_task = None1749 photo = FreezeHandler(self.reboot_temp_path)1750 for group_handler in list(self.group_map.values()):1751 await group_handler.quit_all(str(photo.root_path))1752 await asyncio.to_thread(1753 photo.write_commander_register_item,1754 self.frozen_commander_registers,1755 writer="vertex",1756 cause="quit",1757 )1758 photo.rename_root(self.reboot_data_path)1759 self.log_order("vertex", "quit", "-", numbers={"users": len(self.user_map)})1760 1761 @property1762 def frozen_commander_registers(self) -> dict[str, Any]:1763 """What the vertex freezes of itself: its indexes and the global store.1764 1765 Returns:1766 The three maps and the store. The rows go in NORMALISED — everybody1767 frozen, nobody on hold, no pending event — because a boot adopts1768 nobody eagerly and the events that were waiting are stale by then.1769 1770 The indexes are saved and not rederived (D-h, owner 2026-08-25): the1771 cookie carries a cid, and only ``connection_user_map`` says whose it is.1772 The alternative was reading identities back off the filenames, which1773 ``user_to_userkey`` forbids and the sweep relies on it forbidding.1774 """1775 return {1776 "user_map": {1777 user: dict(row, frozen=True, on_hold=None)1778 for user, row in self.user_map.items()1779 },1780 "connection_user_map": dict(self.connection_user_map),1781 "page_connection_map": dict(self.page_connection_map),1782 "global_register": self.global_register,1783 "quit_ts": time.time(),1784 }1785 1786 async def stop(self) -> None:1787 """Take the machine down dry: the clock off, then every group.1788 1789 Acts on this vertex and, through each group, on every process it holds.1790 Nothing is frozen on the way out: without the soft boot those files1791 would be read by nobody, and the next boot wipes the working folder.1792 """1793 if self._heartbeat_task is not None:1794 self._heartbeat_task.cancel()1795 self._heartbeat_task = None1796 if self._cpu_meter_task is not None:1797 self._cpu_meter_task.cancel()1798 self._cpu_meter_task = None1799 for group_handler in list(self.group_map.values()):1800 await group_handler.stop()1801 1802 async def cpu_meter_loop(self) -> None:1803 """Continuously refresh every group's CPU telemetry without worker traffic.1804 1805 One task serves the whole vertex. Unavailable process rows are an absent1806 gauge rather than a failed observation. The only judge called here is1807 CPU admission; no placement, offload or shape round runs on this clock.1808 """1809 while True:1810 sampled_at = time.monotonic()1811 self.sample_cpu_temperatures(sampled_at=sampled_at)1812 elapsed = time.monotonic() - sampled_at1813 interval = self.cpu_temperature_sample_seconds1814 if interval is None:1815 return1816 await asyncio.sleep(max(0.0, interval - elapsed))1817 1818 def sample_cpu_temperatures(self, *, sampled_at: float | None = None) -> None:1819 """Read every living worker's local process clock for observation only."""1820 instant = time.monotonic() if sampled_at is None else sampled_at1821 for group_handler in list(self.group_map.values()):1822 for worker_handler in group_handler.living_workers:1823 try:1824 worker_handler.record_cpu_reading(1825 worker_handler.get_process_cpu_reading(), sampled_at=instant1826 )1827 except Exception:1828 self._logger.exception(1829 "Vertex: worker %s CPU temperature failed",1830 worker_handler.name,1831 )1832 if group_handler.cpu_admission_close_percent is not None:1833 group_handler._judge_cpu_admission(log_scan=False)1834 1835 async def heartbeat_loop(self) -> None:1836 """The one clock: a round at every beat, and never a death by a bad round.1837 1838 Never returns — whoever starts it cancels it. Acts through everything1839 the round acts on.1840 """1841 while True:1842 woken = await self._wait_beat()1843 try:1844 if woken:1845 await self.ping_groups(woken)1846 continue1847 await self.ping_groups()1848 except Exception:1849 self._logger.exception("Vertex: the round failed")1850 continue1851 await self.drop_expired_users()1852 await self.cleanup_frozen()1853 await self.check_resources()1854 1855 async def ping_groups(self, group_handlers: list[Any] | None = None) -> None:1856 """Give every group its turn, all at once, and wait for all of them.1857 1858 Args:1859 group_handlers: the groups to give a turn to; all of them when None,1860 which is what the timer asks for.1861 1862 Acts through the groups: one still in its turn is skipped, and a turn1863 that raises is a value here and cancels no sibling.1864 """1865 turns = []1866 for group_handler in group_handlers or list(self.group_map.values()):1867 running = self._group_turns.get(group_handler.name)1868 if running is not None and not running.done():1869 self._logger.warning("Vertex: group %s is still in its turn", group_handler.name)1870 continue1871 running = asyncio.get_running_loop().create_task(group_handler.ping())1872 self._group_turns[group_handler.name] = running1873 turns.append(running)1874 await asyncio.gather(*turns, return_exceptions=True)1875 1876 @every(DROP_EXPIRED_USERS_BEATS)1877 async def drop_expired_users(self) -> None:1878 """Forget the frozen whose age ran out — the row here, the folder on disk.1879 1880 Acts on the indexes and on the freezer; the disk is opened off the loop.1881 """1882 frozen_users = [user for user in self.user_map if self.user_is_frozen(user)]1883 expired = await asyncio.to_thread(self._expired_users, frozen_users)1884 if expired:1885 self.drop_users(expired, cause="expired")1886 1887 @every(CLEANUP_FROZEN_BEATS)1888 async def cleanup_frozen(self) -> None:1889 """Discard what the freezer holds for nobody the indexes know.1890 1891 Acts on the freezer, counting and naming each folder it discards; the1892 disk is opened off the loop.1893 """1894 claimed = {self.freeze_handler.user_to_userkey(user) for user in self.user_map}1895 sweep = self.freeze_handler.cleanup_frozen1896 for userkey in await asyncio.to_thread(sweep, claimed):1897 self.counters["orphan_folders_discarded"] += 11898 self.log_order("vertex", "cleanup_frozen", userkey, outcome="orphan")1899 1900 @every(CHECK_RESOURCES_BEATS)1901 async def check_resources(self) -> None:1902 """Read the machine's memory against its alarm line, the storage against the reserve.1903 1904 Acts on ``state`` — the MEMORY alone decides it — and calls1905 ``need_resources`` for as long as either alarm stands. A gauge the1906 platform does not offer alarms nobody. The gauges are read off the loop.1907 """1908 memory_percent, storage_free_percent = await asyncio.to_thread(self._read_resources)1909 over = memory_percent > self.machine_memory_alarm_percent1910 self.state = "saturated" if over else "running"1911 on_reserve = storage_free_percent < STORAGE_RESERVE_PERCENT1912 if over or on_reserve:1913 numbers = {"memory": memory_percent, "storage_free": storage_free_percent}1914 outcome = "saturated" if over else "on_reserve"1915 self.log_order("vertex", "check_resources", numbers=numbers, outcome=outcome)1916 self.need_resources()1917 1918 def need_resources(self) -> None:1919 """Ask the world outside this process for more room; here that is nothing.1920 1921 A commander that can grow its own machine says so by overriding this.1922 """1923 1924 async def _wait_beat(self) -> list[Any]:1925 """Wait for the timer or for any group's wake, whichever comes first.1926 1927 Returns:1928 The groups that rang, and an empty list when the timer came — which1929 is the full round. The timer SURVIVES the wakes it loses to: a group1930 ringing at every breath anticipates its own round as often as it1931 likes, but cannot postpone the full round — the beat every group and1932 every task of the vertex is owed — past its own due.1933 """1934 if self._beat_timer is not None and self._beat_timer.done():1935 # The beat expired while an anticipated round was running: it is1936 # owed as a full round, never discarded.1937 self._beat_timer = None1938 return []1939 if self._beat_timer is None:1940 self._beat_timer = asyncio.ensure_future(asyncio.sleep(HEARTBEAT_SECONDS))1941 wakes = {1942 asyncio.ensure_future(group_handler.ping_now_event.wait()): group_handler1943 for group_handler in self.group_map.values()1944 }1945 done, _pending = await asyncio.wait(1946 [self._beat_timer, *wakes], return_when=asyncio.FIRST_COMPLETED1947 )1948 for wake in wakes:1949 if wake not in done:1950 wake.cancel()1951 if self._beat_timer in done:1952 self._beat_timer = None1953 return []1954 return [group_handler for wake, group_handler in wakes.items() if wake in done]1955 1956 def _expired_users(self, users: list[str]) -> list[str]:1957 """Which of these frozen users are past their own expiry; runs off the loop.1958 1959 A frozen row with nothing on disk has no age to judge, and is left to1960 ``cleanup_frozen``.1961 """1962 now = time.time()1963 expired = []1964 for user in users:1965 header = self.freeze_handler.get_item_header(user)1966 if header and now - header["ts"] > self.get_user_expiry_seconds(user):1967 expired.append(user)1968 return expired1969 1970 def _read_resources(self) -> tuple[float, float]:1971 """The machine's memory used and the freezer's storage free, in percent; off the loop."""1972 return self._machine_memory_used_percent(), self.freeze_handler.storage_free_percent1973 1974 def _machine_memory_used_percent(self) -> float:1975 """How much of the WHOLE machine's memory is in use, in percent."""1976 gauges = self._machine_memory_gauges()1977 return 100.0 * (gauges["MemTotal"] - gauges["MemAvailable"]) / gauges["MemTotal"]1978 1979 def _machine_memory_gauges(self) -> dict[str, float]:1980 """The machine's whole and available memory in BYTES, both always there.1981 1982 Both are read through ``psutil.virtual_memory`` on every platform, so1983 the cascade of percentages is always anchored and how much of the1984 machine is in use is always judged.1985 1986 Both readings are the HOST's: psutil does not know the cgroup this1987 process runs in, so a server in1988 a container would read the memory of the machine hosting it and grow1989 until the kernel kills it. The limit of the cgroup is therefore read1990 too, and where it is smaller it takes the place of both: the whole1991 becomes the limit, and the available becomes what the limit still has1992 free — every process charged to the cgroup counted, this one included.1993 No cgroup, no limit, or a file that does not read as a number: the host1994 figures stand, exactly as they did.1995 1996 A limit that reads and a charge that does not is the one case answered1997 CONSERVATIVELY: the available is 0. The machine is measurable, so the1998 silence is a gauge that failed, not a platform that has none — and what1999 is not proven free is not free.2000 """2001 machine = psutil.virtual_memory()2002 gauges: dict[str, float] = {2003 "MemTotal": float(machine.total),2004 "MemAvailable": float(machine.available),2005 }2006 limit, current = self._cgroup_memory_gauges(gauges["MemTotal"])2007 if limit is None:2008 return gauges2009 gauges["MemTotal"] = limit2010 if current is None:2011 # The limit is known and what is charged to it is not. NOTHING is2012 # proven free, so nothing is claimed: a growth that assumes the whole2013 # limit is its own is the growth that meets the kernel's killer.2014 gauges["MemAvailable"] = 0.02015 return gauges2016 headroom = limit - current2017 available = min(gauges.get("MemAvailable", headroom), headroom)2018 gauges["MemAvailable"] = min(max(available, 0.0), limit)2019 return gauges2020 2021 def _cgroup_memory_gauges(self, host_total: float) -> tuple[float | None, float | None]:2022 """The container's memory limit and current charge in bytes; None where there is none.2023 2024 Args:2025 host_total: the whole memory of the machine. A limit that reaches it2026 limits nothing — that is how cgroup v1 writes "unlimited", with2027 an enormous sentinel instead of a word.2028 2029 Returns:2030 The limit and what is charged to it, the charge None when that file2031 alone does not answer with a count of bytes — missing, unreadable,2032 not a number or negative. Both None when no layout answers: outside2033 a container the files are not there, and an unlimited cgroup v22034 writes ``max`` in ``memory.max``, which is not a number.2035 """2036 for limit_path, current_path in CGROUP_MEMORY_FILES:2037 limit = self._read_gauge_file(limit_path)2038 if limit is not None and 0 < limit < host_total:2039 current = self._read_gauge_file(current_path)2040 return limit, None if current is None or current < 0 else current2041 return None, None2042 2043 def _read_gauge_file(self, path: str) -> float | None:2044 """The one count of bytes a cgroup file holds, or None when it holds no count.2045 2046 A cgroup gauge is a whole number of bytes, so a whole number is what is2047 read: ``max``, an empty file, a fraction and every spelling of infinity2048 and not-a-number are all refused the same way, and nothing that is not a2049 count of bytes ever reaches the arithmetic below.2050 """2051 try:2052 with open(path, encoding="ascii") as gauge:2053 return float(int(gauge.read().strip()))2054 except (OSError, ValueError):2055 return None2056 2057 def _new_row(self) -> dict[str, Any]:2058 """The row of an identity nobody knows anything about yet."""2059 return {"group": None, "frozen": False, "on_hold": None}2060 2061 def _build_orders_logger(2062 self, path: str | Path | None, max_bytes: int, backup_count: int2063 ) -> logging.Logger:2064 """The dedicated logger of the orders, with its own file in place of whatever was there."""2065 logger = logging.getLogger(ORDERS_LOGGER_NAME)2066 if path is None:2067 return logger2068 for attached in list(logger.handlers):2069 logger.removeHandler(attached)2070 attached.close()2071 handler = RotatingFileHandler(2072 Path(path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"2073 )2074 handler.setFormatter(logging.Formatter("%(asctime)s %(message)s"))2075 logger.addHandler(handler)2076 logger.setLevel(logging.INFO)2077 return logger2078 2079 def _build_decisions_logger(2080 self, path: str | Path | None, max_bytes: int, backup_count: int2081 ) -> logging.Logger:2082 """The JSONL journal beside the human orchestration log."""2083 logger = logging.getLogger(DECISIONS_LOGGER_NAME)2084 for attached in list(logger.handlers):2085 logger.removeHandler(attached)2086 attached.close()2087 if path is None:2088 logger.propagate = True2089 return logger2090 decision_path = Path(path).with_suffix(".decisions.jsonl")2091 handler = RotatingFileHandler(2092 decision_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"2093 )2094 handler.setFormatter(logging.Formatter("%(message)s"))2095 logger.addHandler(handler)2096 logger.setLevel(logging.INFO)2097 logger.propagate = False2098 return logger