src/genro_asgi_multiworker_spa/orchestration/group_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 """GroupHandler: the workers of one grammar, where a user lands, and the two crises.16 17 A group is the workers built from ONE grammar — the same child, the same policies18 — and it owns three things nobody else does: where each of its users lives19 (``user_worker_map``, ``None`` meaning "to be assigned"), the manoeuvres on its20 own workers, and the shape of the group itself.21 22 **Nobody says how many workers there are.** There is no target and no maximum. At23 boot the group brings ONE worker into being — the RECEPTION, which is a role and24 not a count — and after that it grows only on demand (nobody admits a newcomer,25 so one more is born if the memory quota affords it) and shrinks when the capacity26 of one of them is spare. The reception is simply the oldest living worker, and it27 is succeeded silently when it dies.28 29 **The placement is EAFP: the refusal IS the answer.** ``assign_user`` walks the30 workers from the FULLEST down — filling what is already warm rather than31 spreading everybody thin — and asks each one to take the user;32 ``WorkerHandler.assign_user`` judges itself on its own last photo and refuses by33 RAISING, so the reason is a class and never a flag somebody has to remember to34 read: over the setpoint is ``NoRoomError``, one on its way out35 ``WorkerQuittingError``. Candidates36 exhausted, the base rises — whoever asked answers 503 — and the wake rings on the37 way out, so the group grows before he tries again. Two placements in a row are38 judged on the same photo, so a group can overshoot by one newcomer: accepted, and39 cheaper than a lock.40 41 **The occupancy is the currency of all of it.** A worker's fullness is read off42 its last photo the way the pool has always read it: one clamped component per43 measurable gauge, the FULLEST of them wins, and the answer is a percentage — so44 the memory of a process, the cost of a user and the setpoint of a worker are all45 the same number and can be added. On Linux the memory component is PSS, which46 divides prefork-shared pages among the processes mapping them; RSS remains the47 conservative fallback wherever PSS is unavailable. A photo carrying neither48 reads 0, which is what a worker nobody has measured yet honestly is.49 50 **The memory is a CASCADE of percentages, and only the bottom of it is bytes.**51 One total is always handed in — ``memory_concession_bytes``, what the machine52 concedes — and everything below it is a share: ``memory_max_percent`` is this group's share53 of the concession, and ``worker_memory_max_percent`` is what ONE worker may hold54 of the group's own quota. The worker share is usually not written at all:55 ``worker_max_number`` names how many workers the quota is SIZED FOR — an56 intuitive count of slots instead of a percentage — and the share is derived as57 ``100 / worker_max_number``. It is a divisor of the size and nothing else: the58 number of processes stays a reading, never a setting. An explicit59 ``worker_memory_max_percent`` wins over the derivation. So the first gate on the60 growth compares ``memory_occupied_percent``, what the living workers hold read61 against the concession, PLUS the share one more worker may hold, with62 ``memory_max_percent``: percent against percent, never a byte count against a63 byte count.64 65 The cascade knows only the workers, and a container holds more than them. So a66 second gate answers in bytes, on ``SpaCommander.memory_available_bytes``: the67 ceiling of the newborn must be free where the process will actually live —68 inside the cgroup when there is one, on the machine when there is not. The69 commander, the group templates and every other tenant of the container are70 counted there and nowhere else. Both gates hold or nobody is born; a machine71 that measures nothing refuses nothing.72 73 **The clock is the vertex's, the counting is the group's.** ``ping`` is this74 group's turn of the one round there is: it settles every process whose end has75 not been read yet — the state is the only word on a death, and a round is where76 it is read, whether the process left as it was told to or died wild — it beats77 the workers nobody has heard from — a process fresh from traffic has just78 photographed itself — and it reads79 its own shape only when its own count of turns says so, or when its wake was80 rung, which is what a death or a placement nobody admitted does. The wake is81 consumed HERE, at the start of the turn, so the group that rings while its turn82 runs is given another one.83 84 **The shape is decided on ONE picture, and one step per round.**85 ``check_occupancy`` takes the occupancy of every living worker once and then does86 the FIRST thing that reading calls for: restart the worker whose MEMORY is past87 ``restart_occupancy_max_percent`` (it will not get better on its own), give a88 group with no living worker its reception back when the memory affords it, or89 close the COLDEST worker whose temperature, shared by the survivors, keeps every90 one of them under ``cpu_close_percent`` (unset, the reopen threshold itself), so a91 closure never creates the condition for the next birth (#36) — and whose memory,92 shared the same way, keeps every survivor under the veto. It births nothing for a93 user who is not there yet: the reception of an empty group is the only birth94 here, every other one happens inside ``assign_user`` for the user who needs it.95 A worker younger than ``worker_min_life_seconds`` is never the one closed, a96 worker with no temperature yet suspends the judgment, and a closure is refused97 when the survivors lack room BY HEADS for the spare's placed users. The next98 round re-reads: a decision is never carried over.99 100 **CPU admission and demand-driven birth (#43, experimental, off by default).**101 With ``cpu_admission_close_percent`` set, the commander's process thermometer samples each102 worker independently of traffic. A fresh ``cpu_temperature_percent`` above the103 threshold CLOSES it to new users (``cpu_admission_open``); it reopens only below104 ``cpu_admission_reopen_percent``; between the two thresholds it keeps its state — the105 band is hysteresis. The photo carries no CPU: the thermometer is the only106 source, and it never creates a process by itself. When a107 real user arrives, placement first tries the CPU-open workers; only when none108 admits that user does the same placement, under its lock, fork one worker and109 place that same user on it. A transient sample therefore cannot leave empty110 capacity behind, while the template fork keeps the demand path short. Sticky111 users are never moved. If memory or server state refuses the birth, a closed112 worker still under the hard ``worker_memory_admission_percent`` takes the user as a113 LOGGED fallback — the soft closure shapes the pool, never at the price of a114 premature 503. Admission is reconciled in the sampling pass, without waiting for115 the heartbeat. The RETIREMENT STANDS ASIDE while the CPU speaks: a living116 worker still CPU-closed, or any CPU event younger than117 ``cpu_retirement_quiet_seconds`` — a blocking or a reopening — suspends the118 closure judge, because119 closing the emptiest worker under standing demand hands its users back to the120 hot one, which regrows seconds later (the close→grow cycle the bench measured,121 churn 2026-08-28). The quiet is CONTINUOUS: every event restarts it whole, the122 reopen included. After it, the retirement is exactly what it always was.123 124 **A CPU-hot worker slims one user at a beat (#43, off by default).** With125 ``cpu_offload_percent`` set — above ``cpu_admission_close_percent``, since the offload126 stands on the admission closure — ``check_cpu_offload`` runs at EVERY beat using127 the latest fresh temperature: the hottest CPU-closed ``running`` worker past the threshold128 cedes ONE user through the same ``freeze_hosted_user`` road as every departure.129 WHO is judged against the window itself: a MATERIAL contributor holds at least130 half the fair share of the interval's service time (``s >= S/(2N)``) or has a131 call in flight; negligible activity is never a candidate. The cession takes132 the least busy material contributor WITHOUT calls in flight (least133 ``recent_service_seconds``, then least ``recent_call_count``, then name) — a134 user mid-call is never transferred, and material contributors all busy defer135 the cession to the next beat (``cpu_offload_deferred_pending_calls``). The136 source is closed, so his next request is placed elsewhere or births the137 capacity it needs; the light contributors leave one per beat and the heavy one138 stays — a single material contributor is never transferred, journaled once as139 ``single_user_overload``, the worker de facto dedicated to him. A cession140 stamps the CPU pressure clock; the standing conditions are journaled once per141 (condition, subject), never every beat.142 143 **The closure is the departure of a whole worker, in six steps.** The group144 orders the quit; the worker answers AT ONCE with the photo of everybody flagged145 for the freezer, so the vertex parks them; it then drains, freezing one user at a146 time; emptied, it ends itself; the end of its wire was awaited, so the state says147 ``quitted``; and at the round that reads it the group does ``drop_worker`` — the148 socket taken away, the worker out of the list — which is the same verb the149 bonifica of a wild death uses. A departure is settled on the LAST PHOTO, so a150 worker nobody ever photographed would take its users down with it: the order151 takes a photo first, which is what ``ping_process`` is for.152 153 **The soft quit blocks BEFORE it orders.** ``quit_all`` raises the hold on every154 user it places on a worker and only then sends that worker's order. It is the155 same barrier the photo's flags raise at the vertex — a flag is read there as a156 hold — moved ahead of the order, which closes the window the photo cannot: the157 photo rides the answer, and until it arrives a request of his would walk into a158 process already emptying. The closure of a single worker (``close_worker``,159 ``restart_worker``) blocks nobody in advance and meets its users on the photo, as160 before.161 162 **Both crises are a polite 503.** ``saturated`` says the memory quota is full and163 somebody has to leave before anybody else comes in; ``broken`` says a process164 could not be started at all. Residents are served as ever in either; newcomers165 and the woken get a 503 with a ``Retry-After``. A saturation is written by the placement that was166 refused and lifted by the next check, once the quota affords a birth again; a167 broken group is closed by the first process that starts.168 A user never changes group: there is no fallback and no policy key.169 170 **Putting one user to sleep is the group's own move, in one order.**171 ``freeze_hosted_user`` blocks him at the vertex FIRST — from that instant a172 request of his waits instead of walking into a process that is emptying — then173 orders his worker to park him and waits for the REPLY, which IS the174 confirmation. The worker judges nothing on that road; it only executes. A175 departure that did not happen gives the block back and leaves him where he was.176 177 **And WHO sleeps is decided here too, on the same photos.**178 ``check_user_activity`` is the group's second periodic: it reads the two179 real clocks of every active user off his worker's last photo, parks whoever has180 been silent past ``user_idle_freeze_minutes`` through the order above, and drops181 whoever is silent past his own expiry — the same horizon the vertex applies to a182 parcel, asked of the vertex. Nothing below this rung has a gauge of its own: the183 worker was where that judgment used to live, and it kept no policy after the184 freeze order was built.185 186 **The group never touches an index of the vertex and never opens the freezer.**187 It READS from a user's row what he is expected to cost, and it writes its own map188 only; the marks, the purges and the disk are the vertex's, and so is the189 orchestration log every order of this group leaves its row in.190 """191 192 from __future__ import annotations193 194 import asyncio195 import logging196 import math197 import time198 from collections import Counter199 from typing import Any200 201 from genro_routes import RoutingClass, route202 203 from .envelope_handler import GroupEnvelopeHandler204 from .group_policy import GroupPolicy205 from .exceptions import AssignmentRefused, NoRoomError206 from .beats import every207 from .template_connector import TemplateConnector208 from .worker_handler import DROP_USER_OP_PATH, FREEZE_USER_OP_PATH, WorkerHandler209 210 #: The states of a worker whose process has ended: it is in the list only until211 #: the round that reads it, and it is nobody's candidate.212 DEAD_STATES = ("quitted", "aborted")213 214 #: How many workers a group's quota is sized for when the recipe says nothing:215 #: the per-worker memory ceiling defaults to quota / this. A divisor of the216 #: size only — the number of living processes stays a reading, never a setting.217 WORKER_MAX_NUMBER = 6218 219 # How many turns of the group pass between two readings of its own shape. The220 # health of a process is every turn's business; the shape of the group is a221 # slower thing, and the number lives here because the knowledge does.222 CHECK_OCCUPANCY_BEATS = 6223 224 # And how many pass between two readings of who has gone quiet. Slower still,225 # because the silence it judges is declared in MINUTES: reading it oftener would226 # cost turns to answer a question whose answer cannot have changed.227 CHECK_USER_ACTIVITY_BEATS = 12228 229 #: The conversion the silence needs: it is a policy of the installation and comes230 #: in minutes, the clocks it is read against are seconds.231 SECONDS_PER_MINUTE = 60.0232 233 #: How long the group waits for the confirmation of ONE ordered departure — a234 #: freeze or a drop — before giving up, releasing the hold and leaving the user235 #: where he is. A beat: the round that sends the order is the group's own, and a236 #: round that waits longer than a beat stops the group from beating at all. ONE237 #: number for the two orders, because the reason for the ceiling is the same.238 DEPARTURE_ORDER_WAIT_LIMIT = 5.0239 240 __all__ = ["DEAD_STATES", "GroupHandler"]241 242 243 class GroupOperations(RoutingClass):244 """The ``group`` branch of the group's dispatcher: what a worker may call on its group.245 246 Today one operation, the announcement — the worker's own channel for what247 happened while no CALL was being served (#60): the envelope goes into the248 same fold a REPLY's does, on the handler of the worker that sent it.249 250 Args:251 group_handler: the group whose handlers serve the announcements.252 """253 254 def __init__(self, group_handler: Any) -> None:255 self.group_handler = group_handler256 257 @route()258 def announce(self, worker: str, **envelope: Any) -> dict[str, Any]:259 """Fold one announced envelope on the handler of the worker that sent it.260 261 Args:262 worker: the name of the announcing process, put in the payload by263 the worker itself — the tree does not know which wire a CALL264 came in on.265 envelope: the payload as the worker shaped it: the events, and the266 photo when due.267 268 Returns:269 Nothing: the REPLY is the acknowledgement.270 """271 self.group_handler.worker_handler_map[worker].read_envelope(envelope)272 return {}273 274 275 class ForwardToCommander(RoutingClass):276 """The ``commander`` branch of the group's dispatcher: every path under it is forwarded.277 278 The worker talks to its group; what is the vertex's goes on to the279 commander's own dispatcher from here, path unconsumed, so the group can move280 to a process of its own without a path changing (roadmap seed 2). A281 catch-all on ``default_entry`` (genro-routes' best-match resolution) hands282 the remaining segments over as they are; a path the commander does not283 serve raises ``NotFound`` from ITS tree.284 285 Args:286 spa_commander: the vertex whose ``commander_dispatcher`` serves the path.287 """288 289 def __init__(self, spa_commander: Any) -> None:290 self.spa_commander = spa_commander291 self.route.default_entry = "forward"292 293 @route()294 def forward(self, *segments: str, **data: Any) -> Any:295 """Resolve the remaining path on the commander's dispatcher and call it."""296 return self.spa_commander.commander_dispatcher.route.node("/".join(segments))(**data)297 298 299 class GroupDispatcher(RoutingClass):300 """The root of what a worker may call up the lane: ``group/…`` here, ``commander/…`` onward.301 302 One per group. The first segment of a path names the level that serves it,303 and the tree is the table: ``route.nodes()`` lists both branches without a304 call being placed.305 306 Args:307 group_handler: the group this dispatcher belongs to.308 """309 310 def __init__(self, group_handler: Any) -> None:311 self.group_handler = group_handler312 self.add_branches(313 [314 {"name": "group", "instance": GroupOperations(group_handler)},315 {"name": "commander", "instance": ForwardToCommander(group_handler.spa_commander)},316 ]317 )318 319 320 class GroupHandler:321 """One group: its workers, the placement of its users, its shape, its crises.322 323 Args:324 spa_commander: the vertex this group hangs under — the layer above in the325 chain, the rows it reads a user's estimate from, and the log every326 order goes to.327 name: the group's name; its workers are named ``<name>_<counter>``, short328 because the name is the socket's too.329 worker_memory_admission_percent: the memory veto — past this share of its330 ceiling a worker takes no new user, whatever its CPU says.331 worker_admission_interval_seconds: how long after admitting a user a worker332 is skipped by the placement, so its load shows in the temperature333 before the next one lands; 0 switches the rule off.334 restart_occupancy_max_percent: past this MEMORY occupancy a process is335 restarted rather than kept. CPU pressure grows and closes336 admission; it never replaces a healthy process.337 cpu_close_percent: a closure is ordered only when the spare's temperature,338 shared by the survivors, keeps every one of them under this. None,339 the default, means ``cpu_admission_reopen_percent`` itself; set, it340 must not exceed it, so the band between the two is the pool's341 normal state and never a condition to correct (#36).342 cpu_admission_close_percent: the soft-admission threshold (#43, experimental): a343 worker whose fresh commander-side CPU temperature crosses above it344 stops taking new users. CPU sampling never creates a worker; concrete placement345 does. None, the default, leaves the policy off.346 cpu_admission_reopen_percent: below this the worker's admission reopens. The347 band between the two is hysteresis: the previous state is retained.348 cpu_offload_percent: past this fresh CPU temperature a CPU-closed349 worker must slim: at every beat the group orders ONE of its active350 users — the least busy in the last interval — into the freezer, and351 the next request of his lands elsewhere, since this worker is352 closed. None, the default, offloads nobody. Set, it requires353 ``cpu_admission_close_percent`` and sits above it: reopen < close < offload.354 cpu_retirement_quiet_seconds: how long the CPU must stay SILENT — no355 worker blocked or reopened — before retirement judges again, with356 the policy on.357 Not the age of a worker (that is ``worker_min_life_seconds``): this358 is the quiet of the whole group, and every CPU event restarts it359 whole. Closing the emptiest worker while demand still speaks hands360 its users back to the hot one, which regrows seconds later361 (measured, churn 2026-08-28).362 cpu_heating_seconds: the time constant of the temperature filter while the363 worker heats up — how long a hotter sample takes to weigh in.364 cpu_cooling_seconds: the same while it cools down; longer, so a worker that365 just closed or just ceded a user stays closed while its load leaves.366 worker_min_life_seconds: a worker is no closure candidate before this367 age — younger, its occupancy measures its own birth, not its work.368 user_idle_freeze_minutes: the silence, IN MINUTES, past which this group369 parks a user in the freezer; with nothing said, silence never parks370 anybody. Minutes because it is a policy of the installation, and the371 comparison against the photo's clocks converts where it is made.372 memory_concession_bytes: what the machine concedes the whole pool, in373 bytes — the total every percentage below is read against.374 memory_max_percent: this group's share of that concession.375 worker_max_number: how many workers the group's quota is sized for —376 the per-worker ceiling divisor, never a cap on how many processes377 exist. It replaces the bridge-era RAM×0.8/workers derivation with378 one intuitive number of slots.379 worker_memory_max_percent: what ONE worker of this group may hold, as a380 share of the group's own quota; None derives it as381 ``100 / worker_max_number``, and an explicit value wins.382 worker_settings: what every ``WorkerHandler`` of this group is built383 with — the child's identity and the installation's paths — handed384 over verbatim.385 """386 387 def __init__(388 self,389 spa_commander: Any,390 name: str,391 *,392 worker_memory_admission_percent: float = 80.0,393 restart_occupancy_max_percent: float = 95.0,394 cpu_close_percent: float | None = None,395 cpu_admission_close_percent: float | None = None,396 cpu_admission_reopen_percent: float = 40.0,397 cpu_offload_percent: float | None = None,398 cpu_retirement_quiet_seconds: float = 60.0,399 cpu_heating_seconds: float = 1.0,400 cpu_cooling_seconds: float = 5.0,401 worker_admission_interval_seconds: float = 1.0,402 worker_min_life_seconds: float = 60.0,403 worker_max_users: float = math.inf,404 user_idle_freeze_minutes: float = math.inf,405 memory_concession_bytes: int,406 memory_max_percent: float = 100.0,407 worker_max_number: int = WORKER_MAX_NUMBER,408 worker_memory_max_percent: float | None = None,409 engine_factory: str | None = None,410 engine_kwargs: dict[str, Any] | None = None,411 **worker_settings: Any,412 ) -> None:413 if cpu_admission_close_percent is not None and not (414 0.0 <= cpu_admission_reopen_percent < cpu_admission_close_percent <= 100.0415 ):416 raise ValueError(417 f"Group {name}: cpu_admission_reopen_percent "418 f"({cpu_admission_reopen_percent}) must sit below "419 f"cpu_admission_close_percent ({cpu_admission_close_percent}), "420 "both inside 0-100 — the band "421 "between them is the hysteresis; without it a steady worker respawns forever"422 )423 self.spa_commander = spa_commander424 self.name = name425 #: Every setpoint of this group, validated together and read through the426 #: properties below. ``inf`` is how the constructor spells "unlimited"427 #: and ``null`` is how a profile spells it, so the two translate here.428 self.policy = GroupPolicy.from_settings(429 {430 "worker_memory_admission_percent": worker_memory_admission_percent,431 "restart_occupancy_max_percent": restart_occupancy_max_percent,432 "cpu_close_percent": cpu_close_percent,433 "cpu_admission_close_percent": cpu_admission_close_percent,434 "cpu_admission_reopen_percent": cpu_admission_reopen_percent,435 "cpu_offload_percent": cpu_offload_percent,436 "cpu_retirement_quiet_seconds": cpu_retirement_quiet_seconds,437 "cpu_heating_seconds": cpu_heating_seconds,438 "cpu_cooling_seconds": cpu_cooling_seconds,439 "worker_admission_interval_seconds": worker_admission_interval_seconds,440 "worker_min_life_seconds": worker_min_life_seconds,441 "worker_max_users": None if worker_max_users == math.inf else worker_max_users,442 "user_idle_freeze_minutes": (443 None if user_idle_freeze_minutes == math.inf else user_idle_freeze_minutes444 ),445 "memory_max_percent": memory_max_percent,446 "worker_max_number": worker_max_number,447 "worker_memory_max_percent": worker_memory_max_percent,448 }449 )450 self.memory_concession_bytes = memory_concession_bytes451 self.worker_settings = worker_settings452 #: This group's template, when a factory was declared for it: the process453 #: its workers are forked from. None means its workers are spawned instead.454 self.template = (455 None456 if engine_factory is None457 else TemplateConnector(458 self,459 engine_factory=engine_factory,460 engine_kwargs=engine_kwargs,461 executable=worker_settings.get("executable"),462 )463 )464 self.envelope_handler = GroupEnvelopeHandler(self, spa_commander.envelope_handler)465 #: What a worker of this group may call up the lane, as a tree (#59).466 self.group_dispatcher = GroupDispatcher(self)467 #: Where each user of this group lives, by worker name; None says his468 #: state is somewhere else and he is to be assigned on his next request.469 self.user_worker_map: dict[str, str | None] = {}470 #: The workers of this group, oldest first — the order the reception is471 #: read off.472 self.worker_handler_map: dict[str, WorkerHandler] = {}473 #: Where this group stands: ``running``, ``saturated`` or ``broken``.474 self.state = "running"475 #: The wake: idempotent, without content, and the only push in the476 #: system. What it says is which group rang it.477 self.ping_now_event = asyncio.Event()478 self._placement_lock = asyncio.Lock()479 #: When the CPU last spoke (#43): a worker blocked or reopened — an480 #: apply that actually moves a worker's admission included. None from birth — no481 #: artificial cooldown at boot: until a real CPU event this gate does482 #: not exist, and the retirement judges as it always did. The retirement483 #: resumes only after ``cpu_retirement_quiet_seconds`` of CONTINUOUS484 #: silence past this instant.485 self._cpu_pressure_monotonic: float | None = None486 self._logger = logging.getLogger(__name__)487 self._worker_counter = 0488 #: One row per periodic method of this group — turns seen, runs, errors489 #: and the last one's text.490 self.beat_counts: dict[str, dict[str, Any]] = {}491 self._closing_wires: set[asyncio.Task[None]] = set()492 spa_commander.group_map[name] = self493 494 # The setpoints are read through here and stored nowhere else: the names495 # are the ones every decision and every reader has always used, and after496 # a swap they answer the new policy at once.497 @property498 def worker_memory_admission_percent(self) -> float:499 return self.policy.worker_memory_admission_percent500 501 @property502 def restart_occupancy_max_percent(self) -> float:503 return self.policy.restart_occupancy_max_percent504 505 @property506 def cpu_close_percent(self) -> float | None:507 return self.policy.cpu_close_percent508 509 @property510 def cpu_admission_close_percent(self) -> float | None:511 return self.policy.cpu_admission_close_percent512 513 @property514 def cpu_admission_reopen_percent(self) -> float:515 return self.policy.cpu_admission_reopen_percent516 517 @property518 def cpu_offload_percent(self) -> float | None:519 return self.policy.cpu_offload_percent520 521 @property522 def cpu_retirement_quiet_seconds(self) -> float:523 return self.policy.cpu_retirement_quiet_seconds524 525 @property526 def cpu_heating_seconds(self) -> float:527 return self.policy.cpu_heating_seconds528 529 @property530 def cpu_cooling_seconds(self) -> float:531 return self.policy.cpu_cooling_seconds532 533 @property534 def worker_admission_interval_seconds(self) -> float:535 return self.policy.worker_admission_interval_seconds536 537 @property538 def worker_min_life_seconds(self) -> float:539 return self.policy.worker_min_life_seconds540 541 @property542 def worker_max_users(self) -> float:543 return self.policy.worker_max_users544 545 @property546 def user_idle_freeze_minutes(self) -> float:547 return self.policy.user_idle_freeze_minutes548 549 @property550 def memory_max_percent(self) -> float:551 return self.policy.memory_max_percent552 553 @property554 def worker_max_number(self) -> int:555 return self.policy.worker_max_number556 557 @property558 def worker_memory_max_percent(self) -> float:559 return self.policy.worker_memory_max_percent560 def apply_policy(561 self, new_policy: GroupPolicy, reconciliation: list[tuple[str, bool]]562 ) -> None:563 """Take the new setpoints and settle the CPU admission of the listed workers.564 565 Args:566 new_policy: the complete policy that governs this group from now on.567 reconciliation: one ``(worker name, cpu_admission_open)`` pair per568 worker the caller judged against the NEW thresholds.569 570 Guaranteed assignments only: no await, no order on a wire, no birth and571 no log line — everything fallible was done before this is called, so the572 swap cannot half-happen. Every listed worker receives its admission573 state atomically.574 575 An admission this apply actually MOVES is a CPU event like any other and576 restarts the retirement's quiet: new thresholds that close a worker, or577 reopen one, are the same fact the periodic judge would have recorded. An578 apply that moves nobody — a new quiet, another setpoint entirely — leaves579 the clock exactly where it was, so nothing invents a cooldown out of a580 reconfiguration. A policy switched OFF moves nobody either, whatever it581 reopens: those reopenings are the gate being dismantled, not the CPU582 speaking, and stamping them would leave a cooldown behind for whoever583 switches the policy back on.584 """585 cpu_policy_on = new_policy.cpu_admission_close_percent is not None586 self.policy = new_policy587 for name, admission_open in reconciliation:588 worker_handler = self.worker_handler_map[name]589 if cpu_policy_on and worker_handler.cpu_admission_open != admission_open:590 self.record_cpu_pressure()591 worker_handler.cpu_admission_open = admission_open592 593 def _policy_held(594 self, policy: GroupPolicy, order: str, subject: str | None = None595 ) -> bool:596 """Whether the policy a decision stood on is still this group's.597 598 Args:599 policy: the snapshot the decision bound at its top.600 order: the order about to be given, for the log.601 subject: on whom it was to be given, for the log.602 603 Returns:604 True when the effect may go ahead. False when the policy was swapped605 while the decision was waiting — the effect is suppressed and the606 suppression logged, and the swap's own round judges the group again.607 """608 if self.policy is policy:609 return True610 self.spa_commander.log_order(611 self.name, order, subject, outcome="suppressed: policy changed while deciding"612 )613 return False614 615 @property616 def living_workers(self) -> list[WorkerHandler]:617 """The workers whose process has not ended, oldest first."""618 return [619 worker_handler620 for worker_handler in self.worker_handler_map.values()621 if worker_handler.state not in DEAD_STATES622 ]623 624 @property625 def reception(self) -> WorkerHandler | None:626 """The worker that receives whoever arrives unplaced: the oldest living one."""627 living = self.living_workers628 return living[0] if living else None629 630 @property631 def memory_quota_bytes(self) -> float:632 """What this group may hold: its share of the concession, in bytes."""633 return self.memory_concession_bytes * self.memory_max_percent / 100.0634 635 @property636 def worker_memory_ceiling_bytes(self) -> float:637 """What ONE worker of this group may hold: its share of the quota, in bytes."""638 return self.memory_quota_bytes * self.worker_memory_max_percent / 100.0639 640 @property641 def memory_occupied_percent(self) -> float:642 """What this group's living workers hold, as a share of the concession.643 644 Returns:645 The summed accounted memory of their last photos over the646 concession, in percent. PSS is used where Linux reports it; RSS is647 the conservative fallback elsewhere. Read against648 ``memory_max_percent``, so the growth gate compares percent with649 percent without counting prefork-shared pages once per worker.650 """651 accounted_bytes = sum(652 self.get_memory_accounting(worker_handler.worker_snapshot)[0] or 0653 for worker_handler in self.living_workers654 )655 return 100.0 * accounted_bytes / self.memory_concession_bytes656 657 @staticmethod658 def get_memory_accounting(659 worker_snapshot: dict[str, Any] | None,660 ) -> tuple[float | None, str]:661 """Choose the memory gauge a worker decision may account.662 663 Args:664 worker_snapshot: the worker's latest photo, or ``None``.665 666 Returns:667 ``(bytes, kind)``. A finite non-negative PSS wins; otherwise a668 finite non-negative RSS is the conservative portable fallback.669 With neither, the worker is ``unmeasured`` and contributes no670 invented number, preserving the pre-existing no-photo semantics.671 """672 photo = worker_snapshot or {}673 for field, kind in (("pss_bytes", "pss"), ("rss_bytes", "rss_fallback")):674 value = photo.get(field)675 if not isinstance(value, (int, float)) or isinstance(value, bool):676 continue677 try:678 numeric = float(value)679 except OverflowError:680 continue681 if math.isfinite(numeric) and numeric >= 0:682 return numeric, kind683 return None, "unmeasured"684 685 @property686 def memory_accounting_kind(self) -> str:687 """How the living workers' memory is accounted in this group."""688 kinds = {689 self.get_memory_accounting(worker.worker_snapshot)[1]690 for worker in self.living_workers691 }692 if not kinds:693 return "unmeasured"694 if len(kinds) == 1:695 return kinds.pop()696 return "mixed"697 698 def ping_now(self) -> None:699 """Ring this group's wake: its round comes now instead of at its cadence."""700 self.ping_now_event.set()701 702 async def ping(self) -> None:703 """This group's turn of the round: bury the dead, beat the silent, read the shape.704 705 Acts on the group: it consumes the wake it was given — which brings the706 reading of the shape forward — settles every process whose end has not707 been read yet, and lets its two periodics take their step. The shape708 comes first: a worker restarted or closed by that step changes who there709 is to read the silence of.710 """711 woken = self.ping_now_event.is_set()712 self.ping_now_event.clear()713 for worker_handler in list(self.worker_handler_map.values()):714 if worker_handler.state in DEAD_STATES:715 worker_handler.envelope_handler.report_death()716 await self.ping_workers()717 await self.check_occupancy(now=woken)718 await self.check_cpu_offload()719 await self.check_user_activity()720 721 async def ping_workers(self) -> None:722 """Beat every silent worker of this group at once, and wait for all of them.723 724 Acts on the processes: a mute one is killed by its own handler, and a725 beat that raises cancels no sibling.726 """727 beats = [728 worker_handler.ping_process()729 for worker_handler in self.living_workers730 if worker_handler.requires_beat_ping731 ]732 await asyncio.gather(*beats, return_exceptions=True)733 734 def get_memory_occupancy_percent(735 self, worker_snapshot: dict[str, Any] | None736 ) -> float:737 """How full this worker is by memory alone, in percent.738 739 PSS is the Linux currency and RSS its conservative fallback, selected740 by ``get_memory_accounting``. This separate reading is what the restart741 judge uses: CPU pressure asks for capacity and soft admission, never the742 destruction of a process that still owns live sessions.743 """744 accounted_bytes, _kind = self.get_memory_accounting(worker_snapshot)745 if accounted_bytes is None:746 return 0.0747 return 100.0 * min(accounted_bytes / self.worker_memory_ceiling_bytes, 1.0)748 749 async def assign_user(self, user: str) -> str:750 """Place a user: the hottest open worker that admits him, or one born for him.751 752 Args:753 user: the identity to place.754 755 Returns:756 The name of the worker that took him.757 758 Raises:759 AssignmentRefused: the group gave up — nobody admits him, the group760 may not grow (its quota, or the memory the machine still has),761 or the launch failed. The surrender and nothing before it is what762 the front turns into a 503.763 764 The birth lives INSIDE the placement (owner, 2026-08-25): when no765 living worker admits him and the group may grow, this very call brings766 the worker into being — ``start_worker`` returns at its presentation,767 a moment under a template — and places him on it. The request is never768 parked and never sent away to come back: it waits right here.769 770 Three levels, each tried only when the one before surrendered. First771 the CPU-open workers, hottest-first (``_placement_candidate``). Then772 the birth. LAST, when the growth was refused or failed, a CPU-closed773 worker still under its hard cap takes him (``_fallback_candidate``) —774 the soft closure shapes the pool, it must never cost a 503 the hard775 limit would not have cost; the fallback is logged, never silent, and776 never reached while an open worker or a birth can serve.777 778 One placement at a time per group, under ``_placement_lock``: sixteen779 simultaneous arrivals must not father sixteen workers when one is780 enough. Whoever waited recomputes and finds the newborn.781 782 Acts on ``user_worker_map`` and, in the same breath, on the row at the783 vertex that says which group he is on.784 """785 policy = self.policy786 async with self._placement_lock:787 worker_handler = self._placement_candidate(user)788 placement_reason = "hottest_cpu_open_candidate"789 if (790 worker_handler is None791 and self._may_grow792 and self._policy_held(policy, "grow", user)793 ):794 worker_handler = await self.start_worker()795 if worker_handler is not None:796 try:797 worker_handler.assign_user(user)798 placement_reason = "new_worker_created_for_placement"799 except AssignmentRefused:800 worker_handler = None801 if worker_handler is None:802 worker_handler = self._fallback_candidate(user)803 placement_reason = "cpu_closed_hard_cap_fallback"804 if worker_handler is None:805 if not self._may_grow:806 # The refusal itself writes ``saturated`` where the front807 # can read it; the next check lifts it when the quota808 # affords a birth again.809 self._mark_saturated(policy)810 self.ping_now()811 self.spa_commander.log_decision(812 self.name,813 "placement",814 "refused",815 reason="no_worker_could_admit_user",816 subject=user,817 numbers={"workers": len(self.living_workers)},818 candidates=self._placement_decision_rows(),819 )820 raise AssignmentRefused(user, f"{self.name} cannot allocate him")821 self.user_worker_map[user] = worker_handler.name822 worker_handler.last_admission_monotonic = time.monotonic()823 self.spa_commander.record_user_group(user, self.name)824 if placement_reason != "hottest_cpu_open_candidate":825 self.spa_commander.log_decision(826 self.name,827 "placement",828 worker_handler.name,829 reason=placement_reason,830 subject=user,831 candidates=self._placement_decision_rows(),832 )833 return worker_handler.name834 835 def _placement_candidate(self, user: str) -> WorkerHandler | None:836 """The hottest CPU-open living worker that admits *user* right now, or None.837 838 The choice is a CALCULATION of the group (owner, 2026-08-25) — the839 numbers are all here: the temperatures, the photos, the counts. Each840 worker's own judgment (``WorkerHandler.assign_user``) stays the single841 gate a placement passes, so choosing and admitting cannot drift apart.842 A worker the CPU judge closed (``cpu_admission_open`` False) is not in843 the running: hottest-first is applied AMONG THE OPEN ONLY, and hottest844 means the filtered temperature — the group consolidates while a worker845 still has room under the close threshold.846 847 Two passes over the same order. The first skips a worker that admitted848 somebody less than ``worker_admission_interval_seconds`` ago: his load849 is not in the temperature yet, so the next one goes elsewhere. The850 second, only when the first pass skipped somebody for that reason,851 waives the interval — it orders the walk, it refuses nobody and never852 births a worker — and takes the hottest that admits, journaled as853 ``admission_interval_waived``. If no open worker admits the user, the854 placement itself creates capacity.855 """856 candidates = sorted(857 (858 worker_handler859 for worker_handler in self.living_workers860 if worker_handler.cpu_admission_open861 ),862 key=lambda worker_handler: -(worker_handler.get_cpu_temperature_percent() or 0.0),863 )864 decision_rows = [self._get_worker_decision_row(candidate) for candidate in candidates]865 rows_by_name = {row["name"]: row for row in decision_rows}866 for reason, skip_recent in (867 ("hottest_cpu_open_candidate", True),868 ("admission_interval_waived", False),869 ):870 for worker_handler in candidates:871 row = rows_by_name[worker_handler.name]872 if skip_recent and row["recently_admitted"]:873 row["skipped"] = "worker_recently_admitted"874 continue875 try:876 worker_handler.assign_user(user)877 except NoRoomError as refusal:878 row["refusal"] = str(refusal)879 row["skipped"] = (880 "worker_max_users_reached"881 if row["users"] >= self.policy.worker_max_users882 else "worker_memory_full"883 )884 continue885 except AssignmentRefused as refusal:886 self._logger.debug("Group %s: %s", self.name, refusal)887 row["refusal"] = str(refusal)888 continue889 row.pop("skipped", None)890 self.spa_commander.log_decision(891 self.name,892 "placement",893 worker_handler.name,894 reason=reason,895 subject=user,896 candidates=decision_rows,897 )898 return worker_handler899 if not any(row.get("skipped") == "worker_recently_admitted" for row in decision_rows):900 break901 self.spa_commander.log_decision(902 self.name,903 "placement_candidates",904 "none",905 reason="no_cpu_open_candidate_admitted_user",906 subject=user,907 candidates=decision_rows,908 )909 return None910 911 def _recently_admitted(self, worker_handler: WorkerHandler) -> bool:912 """Whether this worker admitted a user less than the admission interval ago."""913 last = worker_handler.last_admission_monotonic914 return (915 last is not None916 and time.monotonic() - last < self.policy.worker_admission_interval_seconds917 )918 919 def _get_worker_decision_row(self, worker_handler: WorkerHandler) -> dict[str, Any]:920 """The facts a placement or growth judge sees for one worker."""921 photo = worker_handler.worker_snapshot or {}922 return {923 "name": worker_handler.name,924 "state": worker_handler.state,925 "users": sum(926 1 for name in self.user_worker_map.values() if name == worker_handler.name927 ),928 "cpu_admission_open": worker_handler.cpu_admission_open,929 "cpu_temperature_percent": worker_handler.get_cpu_temperature_percent(),930 "recently_admitted": self._recently_admitted(worker_handler),931 "memory_occupancy_percent": self.get_memory_occupancy_percent(photo),932 }933 934 def _placement_decision_rows(self) -> list[dict[str, Any]]:935 """The whole living pool, in the order placement considers it."""936 return [937 self._get_worker_decision_row(worker_handler)938 for worker_handler in sorted(939 self.living_workers,940 key=lambda handler: -(handler.get_cpu_temperature_percent() or 0.0),941 )942 ]943 944 def _fallback_candidate(self, user: str) -> WorkerHandler | None:945 """The hottest CPU-closed worker that still admits *user* under the memory veto.946 947 Args:948 user: the newcomer nobody else could take.949 950 Returns:951 The worker that takes him, or None — the true surrender.952 953 The last level of the placement, reached only when no open worker954 admits him AND the growth was refused or failed: the soft closure is a955 shaping policy, and shaping must never turn into a 503 the hard956 ``worker_memory_admission_percent`` — which ``assign_user`` still enforces here —957 would not have given. Every fallback placement is logged as its own958 order: capacity served over the soft limit is a fact the bench must see.959 """960 candidates = sorted(961 (962 worker_handler963 for worker_handler in self.living_workers964 if not worker_handler.cpu_admission_open965 ),966 key=lambda worker_handler: -(worker_handler.get_cpu_temperature_percent() or 0.0),967 )968 for worker_handler in candidates:969 try:970 worker_handler.assign_user(user)971 except AssignmentRefused as refusal:972 self._logger.debug("Group %s: %s", self.name, refusal)973 continue974 self.spa_commander.log_order(975 self.name,976 "placement_fallback",977 worker_handler.name,978 numbers={979 "cpu_temperature_percent": (980 worker_handler.get_cpu_temperature_percent()981 ),982 "workers": len(self.living_workers),983 },984 outcome=f"{user} placed over the soft limit: no open worker, no growth",985 reason="cpu_closed_hard_cap_fallback",986 )987 return worker_handler988 return None989 990 async def freeze_hosted_user(self, user: str) -> bool:991 """Block one of this group's users, have his worker park him, let the block fall.992 993 Args:994 user: a user this group has placed; one it has not placed is a loud995 ``KeyError``, since nobody but the group orders this.996 997 Returns:998 True when he is in the deposit; False when the departure did not999 happen, its refusal named in the orchestration log.1000 1001 Acts on the vertex's barrier, and through the fold on everything else.1002 The hold goes up BEFORE the order and comes down at the other end. On1003 the confirmation there is nothing left to write: the ``user_frozen``1004 worker event travelled in that same REPLY and the fold reads an envelope1005 BEFORE the caller of the order is answered, so the mark, the barrier and1006 the placement already say what they must. On anything else the hold is1007 what this method gives back — a user must never stay blocked on a1008 departure that did not happen — and that includes the CANCELLATION of1009 this coroutine, which is why the release is in a ``finally``.1010 1011 The order has a deadline, ``DEPARTURE_ORDER_WAIT_LIMIT``: the round that1012 sends it is the group's own, and a user with a long call in flight would1013 otherwise hold that round — and with it the group's whole beat — for as1014 long as the call lasts. The expiry takes the road of a refusal and loses1015 nothing: the next round of ``check_user_activity`` judges him again.1016 1017 ACCEPTED, and stated as the choice it is: at the expiry only the future1018 is dropped, the order stays alive on the worker, and it may park the user1019 AFTER the hold has fallen. The window ``hold_user`` closes reopens in1020 that one case, and no code here covers it.1021 """1022 worker_handler = self.worker_handler_map[self.user_worker_map[user]]1023 self.spa_commander.hold_user(user, f"freeze on {worker_handler.name}")1024 refusal: str | None = "CancelledError"1025 try:1026 reply = await worker_handler.connector.call(1027 FREEZE_USER_OP_PATH, {"user": user}, timeout=DEPARTURE_ORDER_WAIT_LIMIT1028 )1029 refusal = reply.get("error")1030 except Exception as exc:1031 refusal = f"{type(exc).__name__}: {exc}"1032 finally:1033 if refusal is not None:1034 self.spa_commander.release_user_hold(user)1035 self.spa_commander.log_order(1036 self.name, "freeze_hosted_user", user, outcome=str(refusal)1037 )1038 return refusal is None1039 1040 def _mark_saturated(self, policy: GroupPolicy) -> None:1041 """Write ``saturated`` where the front reads it; journal the birth the memory refused."""1042 self.state = "saturated"1043 self.spa_commander.log_order(1044 self.name,1045 "grow",1046 numbers={1047 "memory_occupied_percent": self.memory_occupied_percent,1048 "memory_max_percent": policy.memory_max_percent,1049 "worker_memory_ceiling_bytes": self.worker_memory_ceiling_bytes,1050 "memory_available_bytes": self.spa_commander.memory_available_bytes,1051 "workers": len(self.living_workers),1052 },1053 outcome="saturated",1054 )1055 1056 @property1057 def _may_grow(self) -> bool:1058 """Whether one more worker is allowed right now — every birth obeys it.1059 1060 Two memory gates, and the growth passes BOTH. The first is the group's1061 quota, read PROSPECTIVELY: what its workers hold today plus the ceiling1062 of the one about to be born must still fit the quota — a worker born at1063 a quota already full is a worker born to be killed. The second is the1064 machine, read on ``memory_available_bytes``: the ceiling must be free1065 RIGHT NOW where the process will live. Only the second sees the1066 commander, the templates and everything else inside the container, so1067 the quota alone would let a fork walk into a cgroup that has no room1068 left for it.1069 """1070 if self.spa_commander.state != "running":1071 return False1072 ceiling_percent = self.memory_max_percent * self.worker_memory_max_percent / 100.01073 if self.memory_occupied_percent + ceiling_percent > self.memory_max_percent:1074 return False1075 return self.spa_commander.memory_available_bytes >= self.worker_memory_ceiling_bytes1076 1077 @every(CHECK_OCCUPANCY_BEATS)1078 async def check_occupancy(self) -> None:1079 """Read the group and take the ONE step that reading calls for.1080 1081 Acts on the group: restart when MEMORY is past the restart setpoint,1082 update soft CPU admission, give an empty group its reception back when1083 the memory affords it, close a worker the others can absorb — or, when a1084 living worker has no temperature yet, journal that and take no step —1085 and lift ``saturated`` once the memory quota affords a birth again. No1086 worker is born here for a user1087 who is not there yet: the only birth is the reception of a group with1088 no living worker, every other one happens inside ``assign_user`` for1089 the user who needs it.1090 """1091 policy = self.policy1092 snapshots = {1093 worker_handler.name: worker_handler.worker_snapshot1094 for worker_handler in self.living_workers1095 }1096 memory_picture = {1097 name: self.get_memory_occupancy_percent(photo) for name, photo in snapshots.items()1098 }1099 for name, memory_occupancy_percent in memory_picture.items():1100 if memory_occupancy_percent > policy.restart_occupancy_max_percent:1101 if self._policy_held(policy, "restart_worker", name):1102 await self.restart_worker(self.worker_handler_map[name])1103 return1104 self._judge_cpu_admission()1105 if not self.living_workers:1106 # A group must always have a reception: the ONE birth nobody asked1107 # for by arriving, under the same lock and the same memory veto as1108 # every other. Whoever waited on the lock rereads the group first.1109 async with self._placement_lock:1110 if self.living_workers or not self._policy_held(policy, "grow"):1111 return1112 if self._may_grow:1113 await self.start_worker()1114 else:1115 self._mark_saturated(policy)1116 return1117 if self.state == "saturated" and self._may_grow:1118 self.state = "running"1119 missing = sorted(1120 worker_handler.name1121 for worker_handler in self.living_workers1122 if worker_handler.get_cpu_temperature_percent() is None1123 )1124 if missing:1125 # The retirement is a judgment on temperature: a worker without one1126 # yet makes it unmakeable, and this is the ONLY retirement row of1127 # the round — neither the suspension nor the absent spare follows.1128 self.spa_commander.log_decision(1129 self.name,1130 "retirement",1131 "no_action",1132 reason="cpu_temperature_missing",1133 numbers={"workers": len(self.living_workers), "missing": missing},1134 )1135 return1136 if policy.cpu_admission_close_percent is not None:1137 # The retirement stands aside while the CPU policy is under1138 # pressure (#43): closing the emptiest worker while demand stands1139 # hands its users back to the hot one, which regrows seconds later1140 # — the close→grow cycle the bench measured. With the policy off1141 # this gate does not exist.1142 suspension = self.get_retirement_suspension(policy)1143 if suspension is not None:1144 self._logger.debug(1145 "Group %s: retirement suspended — %s", self.name, suspension1146 )1147 self.spa_commander.log_decision(1148 self.name,1149 "retirement",1150 "no_action",1151 reason="cpu_pressure_holds_retirement",1152 numbers={"detail": suspension, "workers": len(self.living_workers)},1153 candidates=self._placement_decision_rows(),1154 )1155 return1156 spare = self._spare_worker(policy)1157 if spare is not None and self._policy_held(policy, "close_worker", spare.name):1158 await self._order_quit(spare, "close_worker")1159 return1160 self.spa_commander.log_decision(1161 self.name,1162 "retirement",1163 "no_action",1164 reason="no_absorbable_spare_worker",1165 numbers={"workers": len(self.living_workers)},1166 candidates=self._placement_decision_rows(),1167 )1168 1169 @every(1)1170 async def check_cpu_offload(self) -> None:1171 """Slim ONE CPU-hot worker by one user: the least busy MATERIAL contributor.1172 1173 Acts on the group at EVERY beat, on the freshest photos: among the1174 living ``running`` workers already CPU-closed and past1175 ``cpu_offload_percent``, the hottest one cedes one user through1176 ``freeze_hosted_user`` — the ordered departure, timeout and hold1177 release included. One cession per beat and per group; the next beat1178 re-reads everything, so nothing is planned ahead.1179 1180 WHO counts is decided against the window itself, with no absolute1181 threshold: over the active users (recent work or a call in flight),1182 ``S`` their summed ``recent_service_seconds`` and ``N`` their count, a1183 MATERIAL contributor is one with ``s >= S/(2N)`` — half the fair share1184 of the window, the full share would disqualify anybody under the mean —1185 or with a call in flight, which is load present whatever its delta1186 reads. Negligible activity is never a candidate: it belongs to the1187 idle-freeze judgment. The cession takes the least busy material1188 contributor WITHOUT calls in flight (least ``recent_service_seconds``,1189 then least ``recent_call_count``, then name): a user mid-call is never1190 transferred — material contributors all busy means the cession is1191 DEFERRED to the next beat, journaled as1192 ``cpu_offload_deferred_pending_calls``, and no freeze is ordered.1193 1194 The worker being closed is what keeps the ceded user from coming back:1195 his next request goes through the ordinary placement, which skips1196 CPU-closed workers and, when no open one admits him, births the1197 capacity on the spot — the demand-driven road. Progressively the light1198 contributors leave and the one generating the load stays: a single1199 material contributor is never transferred, the condition is journaled1200 as ``single_user_overload`` and the worker is de facto dedicated to him.1201 1202 The three standing conditions — one material contributor, none at all,1203 all of them mid-call — would repeat every 5 seconds for as long as1204 they hold, so they are journaled ONCE per (condition, subject) through1205 the marker on the handler, cleared when the worker leaves the offload1206 picture. An actual cession is an action, journaled every time with the1207 numbers that rebuild the judgment (S, N, the threshold, the counts),1208 and it stamps ``record_cpu_pressure``: not to shield the destination —1209 the retirement is already suspended while a worker is CPU-closed — but1210 so the pressure history and the quiet after it stay coherent.1211 """1212 policy = self.policy1213 if policy.cpu_offload_percent is None:1214 return1215 over = [1216 worker_handler1217 for worker_handler in self.living_workers1218 if worker_handler.state == "running"1219 and not worker_handler.cpu_admission_open1220 and (worker_handler.get_cpu_temperature_percent() or 0.0)1221 > policy.cpu_offload_percent1222 ]1223 for worker_handler in self.living_workers:1224 if worker_handler not in over:1225 worker_handler.cpu_offload_condition = None1226 if not over:1227 return1228 target = max(1229 over,1230 key=lambda handler: handler.get_cpu_temperature_percent() or 0.0,1231 )1232 actives = self._active_user_rows(target)1233 window_service_seconds = sum(1234 item.get("recent_service_seconds", 0.0) for _, item in actives1235 )1236 material_threshold = (1237 window_service_seconds / (2 * len(actives)) if actives else 0.01238 )1239 # Material ⟺ s >= S/(2N) or a call in flight. A window that measured1240 # no work at all (S == 0) has no material service contributor: the1241 # degenerate threshold of 0 would call everybody material, so there1242 # the calls in flight are the only material fact.1243 material = [1244 (user, item)1245 for user, item in actives1246 if (1247 window_service_seconds > 0.01248 and item.get("recent_service_seconds", 0.0) >= material_threshold1249 )1250 or item.get("pending_call_count", 0)1251 ]1252 cedible = sorted(1253 (1254 (user, item)1255 for user, item in material1256 if not item.get("pending_call_count", 0)1257 ),1258 key=lambda pair: (1259 pair[1].get("recent_service_seconds", 0.0),1260 pair[1].get("recent_call_count", 0),1261 pair[0],1262 ),1263 )1264 numbers = {1265 "cpu_temperature_percent": target.get_cpu_temperature_percent(),1266 "cpu_offload_percent": policy.cpu_offload_percent,1267 "cpu_admission_close_percent": policy.cpu_admission_close_percent,1268 "cpu_admission_reopen_percent": policy.cpu_admission_reopen_percent,1269 "resident_users": sum(1270 1 for name in self.user_worker_map.values() if name == target.name1271 ),1272 "window_service_seconds": window_service_seconds,1273 "active_users": len(actives),1274 "material_threshold": material_threshold,1275 "material_contributors": len(material),1276 "cedible_contributors": len(cedible),1277 "workers": len(self.living_workers),1278 }1279 if not material:1280 self._note_offload_condition(1281 target, "cpu_offload_no_active_candidate", None, numbers1282 )1283 return1284 if len(material) == 1:1285 self._note_offload_condition(1286 target, "single_user_overload", material[0][0], numbers1287 )1288 return1289 if not cedible:1290 self._note_offload_condition(1291 target, "cpu_offload_deferred_pending_calls", None, numbers1292 )1293 return1294 target.cpu_offload_condition = None1295 user, item = cedible[0]1296 if not self._policy_held(policy, "cpu_offload", user):1297 return1298 self.record_cpu_pressure()1299 self.spa_commander.log_decision(1300 self.name,1301 "cpu_offload",1302 target.name,1303 reason="cpu_offload_threshold",1304 subject=user,1305 numbers=numbers,1306 candidates=[self._get_worker_decision_row(target)],1307 )1308 self.spa_commander.log_decision(1309 self.name,1310 "cpu_offload",1311 user,1312 reason="cpu_offload_user_selected",1313 subject=user,1314 numbers=numbers1315 | {1316 "recent_service_seconds": item.get("recent_service_seconds", 0.0),1317 "recent_call_count": item.get("recent_call_count", 0),1318 "pending_call_count": item.get("pending_call_count", 0),1319 },1320 )1321 frozen = await self.freeze_hosted_user(user)1322 self.spa_commander.log_order(1323 self.name,1324 "cpu_offload",1325 user,1326 numbers=numbers,1327 outcome="completed" if frozen else "refused: the departure did not happen",1328 reason="cpu_offload_completed" if frozen else "cpu_offload_refused",1329 )1330 1331 def _active_user_rows(1332 self, worker_handler: WorkerHandler1333 ) -> list[tuple[str, dict[str, Any]]]:1334 """The users of this worker with any activity in the last interval.1335 1336 Args:1337 worker_handler: the CPU-hot worker being slimmed.1338 1339 Returns:1340 ``(user, photo item)`` pairs: state ``active``, still placed here,1341 and showing recent work or a call in flight. Whoever shows neither1342 belongs to the idle-freeze judgment, not to this one; a photo that1343 has not caught up with a departure names nobody.1344 """1345 photo = worker_handler.worker_snapshot or {}1346 return [1347 (user, row["item"])1348 for user, row in (photo.get("users") or {}).items()1349 if row["item"]["state"] == "active"1350 and self.user_worker_map.get(user) == worker_handler.name1351 and (1352 row["item"].get("recent_service_seconds", 0.0)1353 or row["item"].get("recent_call_count", 0)1354 or row["item"].get("pending_call_count", 0)1355 )1356 ]1357 1358 def _note_offload_condition(1359 self,1360 worker_handler: WorkerHandler,1361 condition: str,1362 subject: str | None,1363 numbers: dict[str, Any],1364 ) -> None:1365 """Journal a standing offload condition once, until it changes.1366 1367 Args:1368 worker_handler: the worker the condition stands on.1369 condition: the stable reason code.1370 subject: on whom, when the condition names somebody.1371 numbers: what the judge had in front of it.1372 """1373 if worker_handler.cpu_offload_condition == (condition, subject):1374 return1375 worker_handler.cpu_offload_condition = (condition, subject)1376 self.spa_commander.log_decision(1377 self.name,1378 "cpu_offload",1379 "no_action",1380 reason=condition,1381 subject=subject or worker_handler.name,1382 numbers=numbers,1383 )1384 1385 @every(CHECK_USER_ACTIVITY_BEATS)1386 async def check_user_activity(self) -> None:1387 """Read the silence off the photos and send whoever is due where he belongs.1388 1389 Acts on the users of this group's living workers: whoever has been silent1390 past his own expiry is DROPPED — on the worker by the order, and at the1391 vertex through the fold that reads the announcement — and whoever is1392 silent past ``user_idle_freeze_minutes`` is parked through1393 ``freeze_hosted_user``. The silence is read on the REAL clocks, never on1394 ``last_refresh_ts``, which a beat alone keeps warm forever.1395 1396 BOTH departures block the user at the vertex before the order goes out,1397 so one coming back at that instant waits instead of being routed onto a1398 worker that is erasing him, and BOTH orders carry1399 ``DEPARTURE_ORDER_WAIT_LIMIT``: a child that does not answer would1400 otherwise hold this round, and with it the group's whole beat, for as1401 long as it stays mute. A drop that CONFIRMS releases nothing here —1402 ``SpaCommander.drop_user`` lets go of the hold when the fold reads the1403 announcement, and whoever was waiting wakes to find him gone — but a1404 drop that did not happen gives the block back on the spot, as the freeze1405 does, so no user is ever left blocked on a departure that failed. The1406 expiry of the deadline is one such refusal, and it loses nothing: the1407 next round judges him again.1408 1409 The judgment is this rung's alone: the worker keeps no gauge and takes no1410 departure decision of its own. The photo it is read off is as fresh as1411 the last envelope out — seconds, against a silence declared in minutes —1412 and only a user this group still places on that worker is judged, so a1413 photo that has not yet caught up with a departure names nobody.1414 """1415 policy = self.policy1416 now = time.time()1417 idle_limit = policy.user_idle_freeze_minutes * SECONDS_PER_MINUTE1418 for worker_handler in self.living_workers:1419 photo = worker_handler.worker_snapshot or {}1420 for user, row in photo.get("users", {}).items():1421 item = row["item"]1422 if item["state"] != "active":1423 continue1424 if self.user_worker_map.get(user) != worker_handler.name:1425 continue1426 idle = now - max(item["last_user_ts"], item["last_rpc_ts"])1427 if idle > self.spa_commander.get_user_expiry_seconds(user):1428 if not self._policy_held(policy, "drop_user", user):1429 continue1430 self.spa_commander.hold_user(user, f"expiry on {worker_handler.name}")1431 refusal: str | None = "CancelledError"1432 try:1433 await worker_handler.connector.call(1434 DROP_USER_OP_PATH,1435 {"user": user},1436 timeout=DEPARTURE_ORDER_WAIT_LIMIT,1437 )1438 refusal = None1439 except Exception as exc:1440 refusal = f"{type(exc).__name__}: {exc}"1441 finally:1442 if refusal is not None:1443 self.spa_commander.release_user_hold(user)1444 self.spa_commander.log_order(1445 self.name, "drop_user", user, outcome=refusal or "expired"1446 )1447 elif idle > idle_limit and self._policy_held(policy, "freeze_hosted_user", user):1448 await self.freeze_hosted_user(user)1449 1450 async def start_worker(self) -> WorkerHandler | None:1451 """Bring one more worker into this group and start its process.1452 1453 Returns:1454 The worker now serving, or None when its process could not be started.1455 1456 Acts on ``worker_handler_map`` and on ``state``: a launch that lands ends1457 both crises, a launch that fails is the ``broken`` one.1458 """1459 self._worker_counter += 11460 name = f"{self.name}_{self._worker_counter:04d}"1461 worker_handler = WorkerHandler(self, name, **self.worker_settings)1462 self.worker_handler_map[name] = worker_handler1463 try:1464 await worker_handler.launch_process()1465 except Exception as failure:1466 del self.worker_handler_map[name]1467 await worker_handler.connector.stop()1468 self.state = "broken"1469 self._logger.exception("Group %s: %s could not be started", self.name, name)1470 self.spa_commander.log_order(self.name, "start_worker", name, outcome=str(failure))1471 return None1472 self.state = "running"1473 self.spa_commander.log_order(1474 self.name, "start_worker", name, numbers={"workers": len(self.living_workers)}1475 )1476 return worker_handler1477 1478 async def stop(self) -> None:1479 """Take every process of this group down and close their wires.1480 1481 Acts on each of its workers: the death is DECLARED before it is dealt —1482 the wait an order parks is what tells an ordered death from a wild one,1483 and a shutdown is an order — then the process is killed and buried and1484 the socket is closed. One that has already died is left alone: there is1485 nothing to kill and its wire went with it. The template goes last, when1486 there is one: it is nobody's watcher, so nothing depends on the order, but1487 the workers it forked are collected by it and it should outlive them.1488 """1489 for worker_handler in list(self.worker_handler_map.values()):1490 if worker_handler.process is not None:1491 worker_handler.expect_death()1492 await worker_handler.terminate_process()1493 await worker_handler.connector.stop()1494 if self.template is not None:1495 await self.template.stop()1496 1497 async def restart_worker(self, worker_handler: WorkerHandler) -> WorkerHandler | None:1498 """Ask a worker to leave for good and put a fresh one in its place.1499 1500 Args:1501 worker_handler: the worker that is not coming back.1502 1503 Returns:1504 The worker born in its place, or None when that one could not start.1505 1506 Acts on the group: the departure is settled through the death of the old1507 process, so the placements it held are released before the new one exists.1508 """1509 await self._order_quit(worker_handler, "restart_worker")1510 worker_handler.envelope_handler.report_death()1511 return await self.start_worker()1512 1513 def drop_worker(self, name: str) -> None:1514 """Take a worker out of the group for good: its wire, its placements, itself.1515 1516 Args:1517 name: the worker that has ended.1518 1519 Raises:1520 KeyError: this group has no worker of that name.1521 1522 Acts on ``worker_handler_map`` and ``user_worker_map``; the socket is1523 taken away detached, since whoever calls this is the fold and cannot wait.1524 CPU admission lives ON the handler, so it dies with it here — no1525 per-name state survives a dropped worker.1526 """1527 worker_handler = self.worker_handler_map.pop(name)1528 closing = asyncio.get_running_loop().create_task(worker_handler.connector.stop())1529 self._closing_wires.add(closing)1530 closing.add_done_callback(self._closing_wires.discard)1531 for user in [user for user, worker in self.user_worker_map.items() if worker == name]:1532 del self.user_worker_map[user]1533 self.spa_commander.log_order(1534 self.name, "drop_worker", name, outcome=worker_handler.state1535 )1536 1537 def _spare_worker(self, policy: GroupPolicy) -> WorkerHandler | None:1538 """The coldest worker whose closure leaves the pool cool; None when there is none.1539 1540 Args:1541 policy: the setpoints this round decided on.1542 1543 Returns:1544 The worker to close, or None when there is none. Every living1545 worker has a temperature: the caller guarantees it.1546 1547 The CPU decides, the memory vetoes. Candidates: not the reception, not1548 on their way out, older than ``worker_min_life_seconds``. The spare is1549 the coldest; its temperature is read as if shared evenly by the1550 survivors, and every survivor must stay under ``cpu_close_percent`` —1551 unset, the reopen threshold itself; set, at or below it — so a closure1552 can never create the condition for the next birth (#36). Then the veto: every survivor, with1553 its even share of the spare's memory, must stay under1554 ``worker_memory_admission_percent``; and, LAST, room BY HEADS for the1555 spare's placed users within each survivor's ``worker_max_users``.1556 """1557 temperatures: dict[str, float] = {1558 worker_handler.name: worker_handler.get_cpu_temperature_percent()1559 for worker_handler in self.living_workers1560 }1561 candidates = [1562 worker_handler1563 for worker_handler in self.living_workers1564 if worker_handler is not self.reception1565 and worker_handler.state != "quitting"1566 and worker_handler.life_seconds >= policy.worker_min_life_seconds1567 ]1568 if not candidates:1569 return None1570 spare = min(candidates, key=lambda worker_handler: temperatures[worker_handler.name])1571 remaining = [1572 worker_handler1573 for worker_handler in self.living_workers1574 if worker_handler is not spare and worker_handler.state != "quitting"1575 ]1576 if not remaining:1577 return None1578 close_threshold = (1579 policy.cpu_admission_reopen_percent1580 if policy.cpu_close_percent is None1581 else policy.cpu_close_percent1582 )1583 shared_heat = temperatures[spare.name] / len(remaining)1584 if any(1585 temperatures[worker_handler.name] + shared_heat > close_threshold1586 for worker_handler in remaining1587 ):1588 return None1589 shared_memory = self.get_memory_occupancy_percent(spare.worker_snapshot) / len(remaining)1590 if any(1591 self.get_memory_occupancy_percent(worker_handler.worker_snapshot) + shared_memory1592 > policy.worker_memory_admission_percent1593 for worker_handler in remaining1594 ):1595 return None1596 placed = Counter(self.user_worker_map.values())1597 head_room = sum(1598 max(0, policy.worker_max_users - placed[worker_handler.name])1599 for worker_handler in remaining1600 )1601 if head_room < placed[spare.name]:1602 return None1603 return spare1604 1605 def _judge_cpu_admission(self, *, log_scan: bool = True) -> None:1606 """Open or close workers to newcomers from the CPU hysteresis.1607 1608 Off unless ``cpu_admission_close_percent`` is set. The filtered temperature controls1609 admission only: above the close threshold a worker stops taking NEW1610 users, below the reopen threshold it takes them again, and inside the1611 band it keeps its state. Sticky users never move.1612 1613 No process is born here. The template fork is cheap enough that demand1614 creates capacity just in time inside ``assign_user`` when no open worker1615 can admit the arriving user. Consequently every CPU-related birth has1616 a concrete first user and a quiet interval creates no empty process.1617 """1618 policy = self.policy1619 if policy.cpu_admission_close_percent is None:1620 self.spa_commander.log_decision(1621 self.name,1622 "cpu_admission_scan",1623 "no_action",1624 reason="cpu_policy_disabled",1625 candidates=self._placement_decision_rows(),1626 )1627 return1628 transitions = 01629 for worker_handler in self.living_workers:1630 cpu_temperature_percent = worker_handler.get_cpu_temperature_percent()1631 if cpu_temperature_percent is None:1632 continue1633 if cpu_temperature_percent > policy.cpu_admission_close_percent:1634 if worker_handler.cpu_admission_open:1635 worker_handler.cpu_admission_open = False1636 transitions += 11637 self.record_cpu_pressure()1638 self.spa_commander.log_order(1639 self.name,1640 "cpu_admission",1641 worker_handler.name,1642 numbers={"cpu_temperature_percent": cpu_temperature_percent},1643 outcome="blocked: over the close threshold",1644 reason="cpu_over_close_threshold",1645 )1646 elif cpu_temperature_percent < policy.cpu_admission_reopen_percent:1647 if not worker_handler.cpu_admission_open:1648 worker_handler.cpu_admission_open = True1649 transitions += 11650 self.record_cpu_pressure()1651 self.spa_commander.log_order(1652 self.name,1653 "cpu_admission",1654 worker_handler.name,1655 numbers={"cpu_temperature_percent": cpu_temperature_percent},1656 outcome="reopened: below the reopen threshold",1657 reason="cpu_below_reopen_threshold",1658 )1659 if not log_scan:1660 return1661 decision_rows = self._placement_decision_rows()1662 self.spa_commander.log_decision(1663 self.name,1664 "cpu_admission_scan",1665 "updated" if transitions else "no_action",1666 reason=(1667 "cpu_admission_transitions" if transitions else "no_cpu_admission_transition"1668 ),1669 numbers={1670 "transitions": transitions,1671 "cpu_admission_close_percent": policy.cpu_admission_close_percent,1672 "cpu_admission_reopen_percent": policy.cpu_admission_reopen_percent,1673 "worker_cpu_temperature": {1674 name: handler.get_cpu_temperature_percent()1675 for name, handler in self.worker_handler_map.items()1676 },1677 "workers": len(decision_rows),1678 "open_workers": sum(1679 1 for row in decision_rows if row["cpu_admission_open"]1680 ),1681 "empty_workers": sum(1 for row in decision_rows if row["users"] == 0),1682 },1683 candidates=decision_rows,1684 )1685 1686 def record_cpu_pressure(self) -> None:1687 """Stamp NOW as the instant the CPU last spoke — the retirement's quiet restarts.1688 1689 Called on every CPU event: a worker blocked over the close threshold,1690 a worker reopened below the reopen threshold — and an apply that actually1691 moves a worker's admission, which is the same fact said by a1692 reconfiguration instead of a photo.1693 1694 The reopen counts ON PURPOSE: a worker that was closed for minutes and1695 reopens must not meet the retirement at the very next beat — the bench1696 measured exactly that close, and the regrowth it caused 5 seconds later1697 (churn of 2026-08-28, reopen at 30).1698 """1699 self._cpu_pressure_monotonic = time.monotonic()1700 1701 def get_retirement_suspension(self, policy: GroupPolicy) -> str | None:1702 """Why the retirement stands aside right now; None when it may judge.1703 1704 Args:1705 policy: the setpoints THIS round decided on, so the answer belongs1706 to the same picture as everything else the round did.1707 1708 Returns:1709 The reason, ready for the log, or None when the closure judge may1710 run.1711 1712 Two reasons, in the order they are asked. The policy off is no reason at1713 all: with no ``cpu_admission_close_percent`` this is never consulted and the1714 retirement is exactly what it always was. A living worker still1715 CPU-closed is standing demand — its load has nowhere to consolidate1716 INTO. And a CPU event younger than ``cpu_retirement_quiet_seconds``1717 means the pressure only just ended: the quiet must be CONTINUOUS, so1718 every event — the reopen included — restarts the whole period. Born1719 None, the clock imposes no cooldown at boot: before any CPU event only1720 the first answer exists.1721 """1722 if any(1723 not worker_handler.cpu_admission_open for worker_handler in self.living_workers1724 ):1725 return "a worker is still CPU-closed"1726 last_pressure = self._cpu_pressure_monotonic1727 if last_pressure is None:1728 return None1729 elapsed = time.monotonic() - last_pressure1730 if elapsed < policy.cpu_retirement_quiet_seconds:1731 return (1732 f"the CPU spoke {elapsed:.1f}s ago, the quiet lasts "1733 f"{policy.cpu_retirement_quiet_seconds:.1f}s"1734 )1735 return None1736 1737 async def quit_all(self, freezer_path: str) -> None:1738 """Block the users of every process of this group, then tell each one to leave.1739 1740 Args:1741 freezer_path: the directory this group's parcels are written to —1742 the reboot directory, never the working deposit.1743 1744 Acts on the vertex's barrier and on each of its workers. Every user this1745 group places on a worker is BLOCKED before that worker is ordered away:1746 from that instant a request of his waits instead of walking into a1747 process that is emptying. The order stays ONE per worker — no per-user1748 order travels the wire here, the worker's own cycle parks everybody as it1749 has since wf/33 — and the holds fall as the freezes confirm, each through1750 the fold that reads ``user_frozen``, the death of the process saying it1751 for whoever's own announcement did not survive the closing wire. A worker1752 already dead is ordered nothing and blocks nobody: its death is written,1753 and the round that read it has already marked or purged whoever it held.1754 The template goes last: it is nobody's watcher, but it outlives the1755 workers it forked.1756 """1757 for worker_handler in list(self.worker_handler_map.values()):1758 if worker_handler.state in DEAD_STATES:1759 continue1760 for user, name in list(self.user_worker_map.items()):1761 if name == worker_handler.name:1762 self.spa_commander.hold_user(user, f"quit of {worker_handler.name}")1763 await self._order_quit(worker_handler, "quit_all", freezer_path=freezer_path)1764 if self.template is not None:1765 await self.template.stop()1766 1767 async def _order_quit(1768 self, worker_handler: WorkerHandler, order: str, freezer_path: str | None = None1769 ) -> None:1770 """Ask a worker's process to leave, having made sure a photo of it exists.1771 1772 The departure of everybody on board is settled on the LAST photo — who1773 was flagged for the freezer — so a worker that has never answered1774 anything is photographed first: without that, an ordered quit would purge1775 its users as if nobody had promised them the freezer. A worker whose1776 death got there first — before the order, or under that very beat — is1777 ordered nothing: the death is already written, and the round buries it.1778 """1779 if worker_handler.state in DEAD_STATES:1780 return1781 if worker_handler.worker_snapshot is None:1782 await worker_handler.ping_process()1783 if worker_handler.state in DEAD_STATES:1784 return1785 self.spa_commander.log_order(1786 self.name,1787 order,1788 worker_handler.name,1789 numbers={1790 "memory_occupancy_percent": self.get_memory_occupancy_percent(1791 worker_handler.worker_snapshot1792 ),1793 "cpu_temperature_percent": (1794 worker_handler.get_cpu_temperature_percent()1795 ),1796 "workers": len(self.living_workers),1797 },1798 )1799 await worker_handler.quit_process(freezer_path)