tests/spa/orchestration/test_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 """The group's own life: it is born, it grows, it restarts, it closes, it breaks.16 17 Every worker here is a REAL child process: a scripted one written to disk at test18 time, started by the group itself through its own ``WorkerHandler``, configured by19 the very payload the handler puts in ``GENRO_ASGI_WORKER``. The child answers20 every order with a photo it was told to carry — how much memory it holds, and who21 is flagged for the freezer — leaves when it is asked to, or never shows up at all,22 which is the group's ``broken`` crisis.23 24 The vertex above the group is real too, so the marks a departure leaves are read25 where they are really written. What no test doubles is the group: it is the26 subject.27 28 The sockets live under a short ``mkdtemp`` root: the system caps a UDS path at29 about a hundred characters and pytest's own directory is already past it, which is30 the very reason worker names are short.31 """32 33 from __future__ import annotations34 35 import asyncio36 import os37 import shutil38 import tempfile39 import time as real_time40 from pathlib import Path41 from typing import Any42 43 import pytest44 45 from genro_asgi_multiworker_spa.orchestration import AssignmentRefused, GroupHandler, SpaCommander46 from genro_asgi_multiworker_spa.orchestration import group_handler as group_handler_module47 from genro_asgi_multiworker_spa.orchestration.worker_connector import (48 ENVELOPE_SLOT_WORKER_EVENTS,49 ENVELOPE_SLOT_WORKER_SNAPSHOT,50 )51 from genro_asgi_multiworker_spa.orchestration.worker_handler import (52 DROP_USER_OP_PATH,53 FREEZE_USER_OP_PATH,54 QUIT_OP_PATH,55 WORKER_ENV_VAR,56 WorkerHandler,57 )58 from tests.spa.orchestration.frame_helpers import control_frame59 60 from .conftest import kill_process, wait_for61 62 63 def warm(worker_handler, cpu_temperature_percent: float) -> None:64 """Declare a fresh filtered temperature, as the meter would have measured it."""65 worker_handler.cpu_temperature_percent = cpu_temperature_percent66 worker_handler.cpu_temperature_sampled_at = real_time.monotonic()67 worker_handler.cpu_temperature_interval_seconds = 0.168 69 70 CHILD_SCRIPT = '''71 """A scripted worker of a group: one photo, one answer, one departure."""72 73 import asyncio74 import json75 import os76 import time77 78 from tests.spa.orchestration.frame_helpers import call_endpoint, control_frame, read_control79 80 from genro_asgi.channel.frame import REGISTER_METHOD, REGISTER_PATH, Frame, FrameStream81 82 83 async def live() -> None:84 payload = json.loads(os.environ["{env_var}"])85 kwargs = payload["kwargs"]86 if kwargs["behaviour"] == "absent":87 await asyncio.sleep(60)88 return89 # The row of a user as a real photo carries it: his flag, his state, and his90 # three clocks. The two REAL ones are pushed into the past by the silence the91 # story declares for him; ``last_refresh_ts`` is always NOW, which is a beat92 # keeping the row warm and proving nobody. A story whose users are RESIDENTS93 # and not departing declares no flag: a flag is read at the vertex as a hold.94 now = time.time()95 photo = {{96 "pid": os.getpid(),97 "rss_bytes": kwargs["rss_bytes"],98 "users": {{99 user: {{100 "transfer_flag": kwargs["transfer_flag"],101 "item": {{102 "state": "active",103 "connection_count": 1,104 "last_refresh_ts": now,105 "last_user_ts": now - kwargs["user_silence"].get(user, 0),106 "last_rpc_ts": now - kwargs["user_silence"].get(user, 0),107 }},108 }}109 for user in kwargs["users"]110 }},111 }}112 reader, writer = await asyncio.open_unix_connection(payload["uds_url"].removeprefix("uds:"))113 stream = FrameStream(reader, writer)114 await stream.write(115 control_frame(116 method=REGISTER_METHOD,117 path=REGISTER_PATH,118 data={{"pid": os.getpid(), "{snapshot_key}": photo}},119 )120 )121 await stream.read()122 while True:123 frame = await stream.read()124 if frame is None:125 return126 if frame.method == "CALL":127 data = {{"result": {{}}}}128 # The ordered freeze: the REPLY IS the confirmation, and the worker129 # event of the departure rides it. Refused on demand, which is the130 # departure that did not happen.131 if frame.path == "{freeze_path}":132 user = (read_control(frame) or {{}})["user"]133 # Taken and never answered: the order that outlives the caller's134 # own deadline, which is the wait the group puts a ceiling on.135 if kwargs["freeze_unanswered"]:136 continue137 if kwargs["freeze_refused"]:138 data = {{"error": "the deposit refused the parcels of " + user}}139 else:140 photo["users"].pop(user, None)141 data = {{142 "result": {{"frozen": user}},143 "{events_key}": [144 {{145 "op": "user_frozen",146 "worker": payload["name"],147 "user": user,148 "placement": None,149 }}150 ],151 }}152 await stream.write(153 control_frame(id=frame.id, method="REPLY", path=frame.path, data=data)154 )155 continue156 # The order to forget somebody: the row goes and the departure is157 # ANNOUNCED, which is what prunes the indexes above.158 if frame.path == "{drop_path}":159 user = (read_control(frame) or {{}})["user"]160 # Taken and never answered, like the freeze above: the drop order161 # has a ceiling of its own and the group must not sit on it.162 if kwargs["drop_unanswered"]:163 continue164 photo["users"].pop(user, None)165 await stream.write(166 control_frame(167 id=frame.id,168 method="REPLY",169 path=frame.path,170 data={{171 "result": {{}},172 "{events_key}": [173 {{"op": "drop_user", "worker": payload["name"], "user": user}}174 ],175 }},176 )177 )178 continue179 # A real worker attaches its photo only when one is DUE, so an180 # answer — the one to the order to leave included — may carry none.181 if frame.path != "{quit_path}" or kwargs["photo_on_quit"]:182 data["{snapshot_key}"] = photo183 await stream.write(184 control_frame(id=frame.id, method="REPLY", path=frame.path, data=data)185 )186 # Asked to leave it leaves, after answering: the answer is what187 # carries the photo of everybody it is about to park.188 if frame.path == "{quit_path}":189 await stream.close()190 return191 192 193 asyncio.run(live())194 '''.format(195 env_var=WORKER_ENV_VAR,196 snapshot_key=ENVELOPE_SLOT_WORKER_SNAPSHOT,197 events_key=ENVELOPE_SLOT_WORKER_EVENTS,198 quit_path=QUIT_OP_PATH,199 freeze_path=FREEZE_USER_OP_PATH,200 drop_path=DROP_USER_OP_PATH,201 )202 203 CHILD_MODULE = "scripted_group_child"204 205 #: What the machine concedes these groups. Their quota is the whole of it by206 #: default; what ONE worker of theirs may hold is half, so that a group of this207 #: file can hold two of them and still have room for a third to be born.208 MEMORY_CEILING = 1_000_000209 210 #: The ceiling every photo below is read against: an rss written as a fraction211 #: of THIS is that fraction of one worker's occupancy.212 WORKER_CEILING = MEMORY_CEILING // 2213 214 215 @pytest.fixture216 def instance_root(monkeypatch):217 """A short root holding the sockets, the deposit and the scripted child."""218 root = Path(tempfile.mkdtemp(prefix="gnrgh_"))219 (root / f"{CHILD_MODULE}.py").write_text(CHILD_SCRIPT)220 inherited = os.environ.get("PYTHONPATH", "")221 monkeypatch.setenv("PYTHONPATH", os.pathsep.join([str(root), inherited]).rstrip(os.pathsep))222 yield root223 shutil.rmtree(root, ignore_errors=True)224 225 226 @pytest.fixture227 def commander(instance_root):228 return SpaCommander(instance_root / "frozen_users")229 230 231 @pytest.fixture232 def group_settings(instance_root):233 """What a group of this file is built with: the child's identity and the paths."""234 return {235 "instance_dir": instance_root / "i",236 "frozen_users_path": instance_root / "frozen_users",237 "entry_module": CHILD_MODULE,238 # The scenarios of this file size ONE worker over half the quota, so two239 # of them fit it and a third may still be born; the core default sizes a240 # group for worker_max_number workers instead.241 "worker_memory_max_percent": 50.0,242 # Every worker here is newborn: the minimum life would exempt them all243 # from closure. The one test about the minimum life sets its own.244 "worker_min_life_seconds": 0.0,245 "worker_kwargs": {246 "behaviour": "answer",247 "users": [],248 "user_silence": {},249 "transfer_flag": "T",250 "rss_bytes": 0,251 "photo_on_quit": True,252 "freeze_refused": False,253 "freeze_unanswered": False,254 "drop_unanswered": False,255 },256 # Wide: the same value bounds the wait for the presentation, and a257 # fresh interpreter on a loaded machine can take seconds to get there.258 # The one scenario about a process that never shows up sets its own.259 "process_ping_timeout": 10.0,260 }261 262 263 @pytest.fixture264 async def make_group(commander, group_settings):265 """Build groups, and let no process or socket of theirs outlive the test."""266 groups: list[GroupHandler] = []267 268 def build(269 *,270 behaviour: str = "answer",271 users: list[str] | None = None,272 user_silence: dict[str, float] | None = None,273 transfer_flag: str | None = "T",274 rss_bytes: int = 0,275 photo_on_quit: bool = True,276 freeze_refused: bool = False,277 freeze_unanswered: bool = False,278 drop_unanswered: bool = False,279 **policies: Any,280 ) -> GroupHandler:281 settings = dict(group_settings)282 settings["worker_kwargs"] = {283 "behaviour": behaviour,284 "users": users or [],285 "user_silence": user_silence or {},286 "transfer_flag": transfer_flag,287 "rss_bytes": rss_bytes,288 "photo_on_quit": photo_on_quit,289 "freeze_refused": freeze_refused,290 "freeze_unanswered": freeze_unanswered,291 "drop_unanswered": drop_unanswered,292 }293 settings.update(policies)294 group = GroupHandler(295 commander,296 "standard",297 memory_concession_bytes=MEMORY_CEILING,298 **settings,299 )300 groups.append(group)301 return group302 303 yield build304 for group in groups:305 for worker_handler in list(group.worker_handler_map.values()):306 if worker_handler.process is not None:307 kill_process(worker_handler.process)308 await wait_for(lambda: not worker_handler.process.alive)309 await worker_handler.connector.stop()310 311 312 def known_at_the_vertex(commander, cid: str, user: str) -> None:313 """What the login will do in Macro 4: this cid is that person's, and he has a row."""314 commander.connection_user_map[cid] = user315 commander.resolve_user(cid)316 317 318 async def test_the_first_worker_of_a_group_is_its_reception(make_group):319 group = make_group()320 assert group.reception is None321 322 worker_handler = await group.start_worker()323 324 assert group.worker_handler_map == {"standard_0001": worker_handler}325 assert group.reception is worker_handler326 assert worker_handler.state == "running"327 assert worker_handler.worker_snapshot["pid"] == worker_handler.process.pid328 assert group.state == "running"329 330 331 async def test_a_group_of_one_closes_nobody(make_group):332 group = make_group()333 reception = await group.start_worker()334 335 await group.check_occupancy(now=True)336 337 # Empty as it is, the reception is still the one that receives whoever338 # arrives unplaced: it is nobody's spare capacity.339 assert list(group.worker_handler_map) == ["standard_0001"]340 assert reception.state == "running"341 342 343 async def test_a_growth_the_quota_refuses_saturates_the_group_until_there_is_room(344 make_group, commander345 ):346 # Half the concession is this group's, and one worker of it may hold half of347 # that: two of them at 85% of what they may hold — past the memory veto —348 # stand together at 42.5% of the concession, and a third one's ceiling would349 # not fit the group's share.350 quota = MEMORY_CEILING // 2351 ceiling = quota // 2352 group = make_group(rss_bytes=int(0.85 * ceiling), memory_max_percent=50.0)353 reception = await group.start_worker()354 spare = await group.start_worker()355 356 # The refusal writes the saturation: nobody admits him, and the quota357 # refuses the birth that would.358 known_at_the_vertex(commander, "cid-a", "mario")359 with pytest.raises(AssignmentRefused):360 await group.assign_user("mario")361 362 assert sorted(group.worker_handler_map) == ["standard_0001", "standard_0002"]363 assert group.memory_occupied_percent == 42.5364 assert group.state == "saturated"365 366 # Somebody left: the next check finds the quota affords a birth again, and367 # the crisis is over without anybody having to say so.368 reception.worker_snapshot = {"rss_bytes": ceiling // 5}369 spare.worker_snapshot = {"rss_bytes": 3 * ceiling // 5}370 await group.check_occupancy(now=True)371 372 assert group.memory_occupied_percent == 20.0373 assert group.state == "running"374 assert sorted(group.worker_handler_map) == ["standard_0001", "standard_0002"]375 376 377 async def test_a_process_that_never_starts_breaks_the_group_until_one_does(make_group, caplog):378 # The short window here bounds the wait for the absent one, and nothing379 # else: the test would otherwise sit the whole file default out.380 group = make_group(behaviour="absent", process_ping_timeout=2.0)381 382 with caplog.at_level("ERROR"):383 assert await group.start_worker() is None384 385 assert group.state == "broken"386 assert group.worker_handler_map == {}387 assert "could not be started" in caplog.text388 389 # The first process that starts closes the crisis — nothing else does.390 # It presents itself, so it gets the wide window back.391 group.worker_settings["worker_kwargs"]["behaviour"] = "answer"392 group.worker_settings["process_ping_timeout"] = 10.0393 assert await group.start_worker() is not None394 assert group.state == "running"395 396 397 async def test_a_worker_past_the_restart_setpoint_is_replaced_by_a_fresh_one(make_group, commander):398 group = make_group(rss_bytes=int(0.99 * WORKER_CEILING), users=["mario"])399 known_at_the_vertex(commander, "cid-a", "mario")400 doomed = await group.start_worker()401 doomed.worker_snapshot["pss_bytes"] = int(0.99 * WORKER_CEILING)402 doomed.hosted_users.add("mario")403 group.user_worker_map["mario"] = doomed.name404 405 await group.check_occupancy(now=True)406 407 assert list(group.worker_handler_map) == ["standard_0002"]408 await wait_for(lambda: not doomed.process.alive)409 assert doomed.state == "quitted"410 # His state went to the freezer with the process that held it, and his411 # placement is to be assigned: his next request decides where he wakes.412 assert commander.user_is_frozen("mario") is True413 assert group.user_worker_map == {"mario": None}414 assert group.state == "running"415 416 417 async def test_shared_rss_does_not_restart_a_worker_whose_pss_is_small(make_group):418 group = make_group(rss_bytes=int(0.99 * WORKER_CEILING))419 worker = await group.start_worker()420 worker.worker_snapshot["pss_bytes"] = int(0.20 * WORKER_CEILING)421 422 await group.check_occupancy(now=True)423 424 assert list(group.worker_handler_map) == ["standard_0001"]425 assert worker.state == "running"426 427 428 async def test_a_death_that_outruns_the_close_order_leaves_no_zombie(make_group, commander):429 group = make_group()430 known_at_the_vertex(commander, "cid-a", "mario")431 doomed = await group.start_worker()432 doomed.hosted_users.add("mario")433 group.user_worker_map["mario"] = doomed.name434 await wait_for(lambda: doomed.worker_snapshot is not None)435 kill_process(doomed.process)436 await wait_for(lambda: doomed.state == "aborted")437 438 fresh = await group.restart_worker(doomed)439 440 # The order found the death already written and ordered nothing: the wild441 # death was not overwritten into a departure nobody would ever settle, the442 # dead one is buried with the users it held, and the fresh one serves.443 assert doomed.state == "aborted"444 assert list(group.worker_handler_map) == [fresh.name]445 assert group.reception is fresh446 assert group.user_worker_map == {}447 448 449 async def test_a_dead_worker_never_photographed_is_ordered_nothing(make_group):450 group = make_group()451 doomed = await group.start_worker()452 kill_process(doomed.process)453 await wait_for(lambda: doomed.state == "aborted")454 doomed.worker_snapshot = None455 456 # The state is read BEFORE the photo beat: no beat is thrown at a dead wire.457 fresh = await group.restart_worker(doomed)458 459 assert doomed.state == "aborted"460 assert list(group.worker_handler_map) == [fresh.name]461 462 463 async def test_a_worker_on_its_way_out_is_nobodys_spare(make_group):464 group = make_group()465 await group.start_worker()466 leaving = await group.start_worker()467 leaving.state = "quitting"468 469 await group.check_occupancy(now=True)470 471 # Its closure is already somebody's order: it is not a candidate for a472 # second one, and its room counts for nobody.473 assert leaving.state == "quitting"474 475 476 async def test_the_closure_of_a_spare_worker_goes_through_its_six_steps(make_group, commander):477 group = make_group(users=["mario"])478 known_at_the_vertex(commander, "cid-a", "mario")479 reception = await group.start_worker()480 spare = await group.start_worker()481 spare.hosted_users.add("mario")482 group.user_worker_map["mario"] = spare.name483 warm(reception, 1.0)484 warm(spare, 1.0)485 486 # 1. the group decides on one reading: what the spare one holds, the others487 # can hold and still admit.488 await group.check_occupancy(now=True)489 490 # 2-4. it answered the order at once, with the photo of everybody flagged for491 # the freezer; then it drained and ENDED BY ITSELF, and that end was awaited.492 assert spare.worker_snapshot["users"]["mario"]["transfer_flag"] == "T"493 assert spare.state == "quitted"494 await wait_for(lambda: not spare.process.alive)495 assert reception.state == "running"496 497 # 5. at the round that reads the ended state, the group takes it out: out of498 # the list, its wire away, its placements released — and the vertex marks the499 # user whose own worker event died with the wire.500 spare.envelope_handler.report_death()501 502 assert list(group.worker_handler_map) == ["standard_0001"]503 assert group.reception is reception504 assert commander.user_is_frozen("mario") is True505 assert group.user_worker_map == {"mario": None}506 await wait_for(lambda: not spare.connector.socket_path.exists())507 508 509 async def test_a_closure_the_memory_veto_refuses_is_not_made(make_group):510 group = make_group()511 reception = await group.start_worker()512 spare = await group.start_worker()513 reception.worker_snapshot = {"rss_bytes": int(0.78 * WORKER_CEILING)}514 spare.worker_snapshot = {"rss_bytes": int(0.05 * WORKER_CEILING)}515 warm(reception, 1.0)516 warm(spare, 1.0)517 518 await group.check_occupancy(now=True)519 520 # Cool as both are, the reception would stand at 83 of memory with the521 # spare's share: the veto refuses a closure the CPU would have allowed.522 assert sorted(group.worker_handler_map) == ["standard_0001", "standard_0002"]523 assert spare.state == "running"524 525 526 async def test_a_worker_in_its_first_seconds_is_no_closure_candidate(make_group):527 group = make_group(worker_min_life_seconds=60.0)528 reception = await group.start_worker()529 young = await group.start_worker()530 warm(reception, 1.0)531 warm(young, 1.0)532 533 await group.check_occupancy(now=True)534 535 # Empty as it reads, its occupancy measures its own birth, not its work:536 # a newborn is otherwise both the proof that growth was needed and the537 # proof that closing is safe (#36).538 assert young.state == "running"539 540 541 async def test_a_closure_leaving_a_survivor_warm_is_not_ordered(make_group):542 # 35 shared onto 35 puts the survivor at 70 of CPU: under the admission close543 # threshold —544 # one threshold for both decisions would close here and grow the round545 # after (#36) — but over cpu_close_percent, so the pool holds:546 # the band between the two thresholds is its normal state.547 group = make_group()548 reception = await group.start_worker()549 spare = await group.start_worker()550 warm(reception, 35.0)551 warm(spare, 35.0)552 553 await group.check_occupancy(now=True)554 555 assert spare.state == "running"556 557 558 async def test_a_closure_the_survivors_cannot_take_by_heads_is_refused(make_group):559 # The occupancy says yes — everybody near zero — but the survivors' own560 # worker_max_users cannot seat the spare's placed users. The ceiling is561 # asked LAST, after the occupancy has spoken, the same order the placement562 # asks it in (#36).563 group = make_group(worker_max_users=2)564 reception = await group.start_worker()565 spare = await group.start_worker()566 group.user_worker_map.update(567 {"anna": reception.name, "bruno": reception.name, "carla": spare.name, "dario": spare.name}568 )569 warm(reception, 1.0)570 warm(spare, 1.0)571 572 await group.check_occupancy(now=True)573 574 assert spare.state == "running"575 576 577 async def test_a_placement_pointing_at_a_worker_that_died_goes_with_it(make_group, commander):578 group = make_group()579 worker_handler = await group.start_worker()580 user = "guest_legacy1"581 commander.record_connection_user("cid-a", user)582 assert await group.assign_user(user) == worker_handler.name583 584 # The process dies before it ever said the user had arrived in it: nobody585 # names him at the death, so his placement goes with the worker holding it.586 kill_process(worker_handler.process)587 await wait_for(lambda: worker_handler.state == "aborted")588 worker_handler.envelope_handler.report_death()589 590 assert group.worker_handler_map == {}591 assert group.user_worker_map == {}592 593 594 async def test_a_death_reported_for_a_worker_this_group_does_not_have_is_loud(make_group):595 group = make_group()596 await group.start_worker()597 598 with pytest.raises(KeyError):599 group.drop_worker("standard_9999")600 601 assert list(group.worker_handler_map) == ["standard_0001"]602 603 604 async def test_an_ordered_quit_photographs_the_worker_first_so_nobody_is_lost(605 make_group, commander606 ):607 # A worker whose answer to the order carries no photo — which is what the608 # throttle of a real one does when none is due.609 group = make_group(users=["mario"], photo_on_quit=False)610 known_at_the_vertex(commander, "cid-a", "mario")611 worker_handler = await group.start_worker()612 worker_handler.hosted_users.add("mario")613 # And of which there is no photo at all: a departure is settled on the last614 # one, so without the beat the order takes first, this user would be purged615 # as lost instead of parked in the freezer.616 worker_handler.worker_snapshot = None617 618 await group.restart_worker(worker_handler)619 620 assert commander.user_is_frozen("mario") is True621 assert "mario" in commander.user_map622 assert list(group.worker_handler_map) == ["standard_0002"]623 624 625 async def test_a_photo_past_the_restart_setpoint_brings_the_round_forward(make_group):626 group = make_group()627 worker_handler = await group.start_worker()628 group.ping_now_event.clear()629 630 worker_handler.read_envelope(631 {ENVELOPE_SLOT_WORKER_SNAPSHOT: {"rss_bytes": WORKER_CEILING // 2}}632 )633 assert group.ping_now_event.is_set() is False634 635 worker_handler.read_envelope({ENVELOPE_SLOT_WORKER_SNAPSHOT: {"rss_bytes": WORKER_CEILING}})636 637 assert group.ping_now_event.is_set() is True638 639 640 async def test_every_order_of_the_group_leaves_its_row_in_the_orchestration_log(641 make_group, commander, caplog642 ):643 group = make_group()644 with caplog.at_level("INFO", logger="genro_asgi.orchestration.orders"):645 worker_handler = await group.start_worker()646 await group.restart_worker(worker_handler)647 648 rows = [record.getMessage() for record in caplog.records]649 assert any("order=start_worker subject=standard_0001" in row for row in rows)650 assert any("order=restart_worker subject=standard_0001" in row for row in rows)651 assert any(652 "order=drop_worker subject=standard_0001 numbers=None outcome=quitted" in row653 for row in rows654 )655 assert any("order=start_worker subject=standard_0002" in row for row in rows)656 657 658 async def test_the_vertex_builds_its_groups_with_the_concession_already_inside(659 instance_root, group_settings660 ):661 vertex = SpaCommander(instance_root / "frozen_users", groups={"standard": group_settings})662 663 group = vertex.group_map["standard"]664 assert isinstance(group, GroupHandler)665 assert group.spa_commander is vertex666 assert group.memory_concession_bytes == vertex.memory_concession_bytes667 assert vertex.default_group == "standard"668 669 670 async def test_a_group_built_without_the_concession_says_so_at_once(commander, group_settings):671 with pytest.raises(TypeError):672 GroupHandler(commander, "standard", **group_settings)673 674 675 async def test_start_brings_the_reception_up_and_stop_leaves_no_child_alive(676 instance_root, group_settings677 ):678 vertex = SpaCommander(instance_root / "frozen_users", groups={"standard": group_settings})679 group = vertex.group_map["standard"]680 try:681 await vertex.start()682 683 # READY is the reception having presented itself: start returns there.684 reception = group.reception685 assert reception is not None686 assert reception.state == "running"687 process = reception.process688 assert process.alive689 finally:690 await vertex.stop()691 692 assert not process.alive693 assert not reception.connector.connected694 # The death of a shutdown is ORDERED: no alarm is owed for it, and the log695 # of a clean stop must not read like N processes died on their own.696 assert reception.state == "quitted"697 698 699 async def test_the_group_of_a_user_is_written_where_he_is_placed(make_group, commander):700 group = make_group()701 await group.start_worker()702 known_at_the_vertex(commander, "cid-a", "mario")703 704 await group.assign_user("mario")705 706 assert group.user_worker_map["mario"] == "standard_0001"707 assert commander.user_map["mario"]["group"] == "standard"708 709 710 async def test_a_placement_nobody_took_writes_no_group(make_group, commander):711 # The vertex is saturated, so the group may not grow: the surrender path.712 group = make_group()713 commander.state = "saturated"714 known_at_the_vertex(commander, "cid-a", "mario")715 716 with pytest.raises(AssignmentRefused):717 await group.assign_user("mario")718 719 assert "mario" not in group.user_worker_map720 assert commander.user_map["mario"]["group"] is None721 722 723 async def test_the_group_orders_every_worker_into_the_reboot_directory(make_group, monkeypatch):724 """One order per worker, each carrying where its parcels go."""725 group = make_group()726 await group.start_worker()727 await group.start_worker()728 ordered = []729 730 async def record(self, freezer_path=None):731 ordered.append((self.name, freezer_path))732 733 monkeypatch.setattr(WorkerHandler, "quit_process", record)734 735 await group.quit_all("/tmp/reboot_temp")736 737 assert ordered == [738 ("standard_0001", "/tmp/reboot_temp"),739 ("standard_0002", "/tmp/reboot_temp"),740 ]741 742 743 async def test_the_quit_blocks_a_worker_s_users_before_its_order_leaves(744 make_group, commander, monkeypatch745 ):746 """The block is up when the order goes out, and only for that worker's users."""747 group = make_group()748 first = await group.start_worker()749 second = await group.start_worker()750 known_at_the_vertex(commander, "cid-a", "mario")751 known_at_the_vertex(commander, "cid-b", "lucia")752 group.user_worker_map["mario"] = first.name753 group.user_worker_map["lucia"] = second.name754 held_when_ordered = []755 756 async def record(self, freezer_path=None):757 held_when_ordered.append(758 (self.name, sorted(u for u, r in commander.user_map.items() if r["on_hold"]))759 )760 761 monkeypatch.setattr(WorkerHandler, "quit_process", record)762 763 await group.quit_all("/tmp/reboot_temp")764 765 # The users of the worker being ordered are already blocked; the one placed766 # on the worker whose turn has not come is not — his own order raises his.767 assert held_when_ordered == [768 (first.name, ["mario"]),769 (second.name, ["lucia", "mario"]),770 ]771 assert commander.user_map["mario"]["on_hold"] == f"quit of {first.name}"772 773 774 async def test_a_worker_already_dead_is_ordered_nothing_and_blocks_nobody(775 make_group, commander, monkeypatch776 ):777 """Its death is written: the round that read it already said what became of him."""778 group = make_group()779 worker_handler = await group.start_worker()780 known_at_the_vertex(commander, "cid-a", "mario")781 group.user_worker_map["mario"] = worker_handler.name782 worker_handler.state = "aborted"783 ordered = []784 monkeypatch.setattr(785 WorkerHandler, "quit_process", lambda self, freezer_path=None: ordered.append(self.name)786 )787 788 await group.quit_all("/tmp/reboot_temp")789 790 assert ordered == []791 assert commander.user_map["mario"]["on_hold"] is None792 793 794 async def test_the_quit_gives_every_hold_back_as_the_freezes_confirm(make_group, commander):795 """The whole soft quit: blocked before the order, frozen and free after it."""796 # The child's photo carries mario flagged for the freezer, which is what a797 # real worker's answer to the order says of everybody on board.798 group = make_group(users=["mario"])799 # His row exists before the process does: the REGISTER photo already carries800 # his flag, and a flag is read at the vertex as a hold.801 known_at_the_vertex(commander, "cid-a", "mario")802 worker_handler = await group.start_worker()803 group.user_worker_map["mario"] = worker_handler.name804 worker_handler.hosted_users.add("mario")805 806 await group.quit_all("/tmp/reboot_temp")807 808 # Blocked he is; the cause reads `transfer_flag T` and not the quit's own,809 # because this scripted child carries the flag from birth and a hold keeps810 # its first cause. What the cause says is test 1's business.811 assert commander.user_map["mario"]["on_hold"] is not None812 assert worker_handler.state == "quitted"813 814 # The death is read at the group's round, and it says of every flagged user815 # what his own announcement would have said had the wire outlived it.816 worker_handler.envelope_handler.report_death()817 818 assert commander.user_is_frozen("mario") is True819 assert commander.user_map["mario"]["on_hold"] is None820 assert group.user_worker_map["mario"] is None821 822 823 async def test_a_worker_at_its_user_ceiling_makes_the_placement_father_a_new_one(824 make_group, commander825 ):826 """worker_max_users: the policy the bench sets to 1 — and the birth lives827 INSIDE the placement (owner, 2026-08-25): the second user is never sent828 away with a 503, his own placement brings his worker into being."""829 group = make_group(worker_max_users=1)830 await group.start_worker()831 commander.record_connection_user("cid-a", "guest_first1")832 commander.record_connection_user("cid-b", "guest_second1")833 assert await group.assign_user("guest_first1") == "standard_0001"834 835 assert await group.assign_user("guest_second1") == "standard_0002"836 837 assert sorted(group.worker_handler_map) == ["standard_0001", "standard_0002"]838 assert sorted(group.user_worker_map.values()) == ["standard_0001", "standard_0002"]839 840 841 async def test_at_the_ceiling_with_no_way_to_grow_the_placement_surrenders(make_group, commander):842 group = make_group(worker_max_users=1)843 await group.start_worker()844 commander.record_connection_user("cid-a", "guest_first1")845 commander.record_connection_user("cid-b", "guest_second1")846 assert await group.assign_user("guest_first1") == "standard_0001"847 commander.state = "saturated"848 group.ping_now_event.clear()849 850 with pytest.raises(AssignmentRefused):851 await group.assign_user("guest_second1")852 853 assert group.ping_now_event.is_set()854 assert group.user_worker_map == {"guest_first1": "standard_0001"}855 856 857 async def test_without_the_ceiling_nothing_changes(make_group, commander):858 group = make_group()859 await group.start_worker()860 commander.record_connection_user("cid-a", "guest_first1")861 commander.record_connection_user("cid-b", "guest_second1")862 assert await group.assign_user("guest_first1") == "standard_0001"863 assert await group.assign_user("guest_second1") == "standard_0001"864 865 866 async def test_the_group_orders_the_freeze_and_the_confirmation_settles_everything(867 make_group, commander868 ):869 """The whole sequence for one user: block, order, confirmation, block gone."""870 group = make_group()871 worker_handler = await group.start_worker()872 known_at_the_vertex(commander, "cid-a", "mario")873 assert await group.assign_user("mario") == worker_handler.name874 worker_handler.hosted_users.add("mario")875 876 assert await group.freeze_hosted_user("mario") is True877 878 # Nothing of this was written by the method: the worker event travelled in879 # the REPLY that confirms the order, and the fold read it before the caller880 # was answered.881 assert commander.user_is_frozen("mario") is True882 assert commander.user_map["mario"]["on_hold"] is None883 assert group.user_worker_map["mario"] is None884 assert worker_handler.hosted_users == set()885 886 887 async def test_a_freeze_the_worker_refuses_leaves_the_user_unblocked_and_where_he_was(888 make_group, commander889 ):890 """A departure that did not happen gives the block back."""891 group = make_group(freeze_refused=True)892 worker_handler = await group.start_worker()893 known_at_the_vertex(commander, "cid-a", "mario")894 await group.assign_user("mario")895 worker_handler.hosted_users.add("mario")896 897 assert await group.freeze_hosted_user("mario") is False898 899 assert commander.user_is_frozen("mario") is False900 assert commander.user_map["mario"]["on_hold"] is None901 assert commander.resolve_user("cid-a") == "mario"902 assert group.user_worker_map["mario"] == worker_handler.name903 assert worker_handler.hosted_users == {"mario"}904 905 906 async def test_a_request_arriving_under_the_order_waits_and_is_served_after_it(907 make_group, commander908 ):909 """#40: no request of his can reach the emptying process, and none is refused.910 911 The order is held ON THE WIRE while a request of his arrives, which is the912 window the block exists for: it parks on his barrier instead of walking into913 a worker that is writing his parcels, and it is served once the confirmation914 has dropped the block — at the destination the vertex assigns him again.915 """916 group = make_group()917 worker_handler = await group.start_worker()918 known_at_the_vertex(commander, "cid-a", "mario")919 await group.assign_user("mario")920 worker_handler.hosted_users.add("mario")921 ordered = asyncio.Event()922 confirm = asyncio.Event()923 on_the_wire = []924 placed_call = worker_handler.connector.call925 926 async def held_order(path, data=None, timeout=None):927 on_the_wire.append(path)928 if path == FREEZE_USER_OP_PATH:929 ordered.set()930 await confirm.wait()931 return await placed_call(path, data, timeout)932 933 worker_handler.connector.call = held_order934 935 order = asyncio.ensure_future(group.freeze_hosted_user("mario"))936 await ordered.wait()937 request = asyncio.ensure_future(938 commander.serve_request(939 "cid-a",940 control_frame(method="CALL", path="/invoices", data={"http": {"path": "/invoices"}}),941 hold_timeout=5.0,942 )943 )944 await asyncio.sleep(0.05)945 946 # He waits, and NOTHING of his went down that wire: no page of his can be947 # born on the worker that is writing his parcels, which is #40 itself.948 assert not request.done()949 assert on_the_wire == [FREEZE_USER_OP_PATH]950 951 confirm.set()952 953 assert await order is True954 assert "error" not in (await request).info955 assert group.user_worker_map["mario"] == worker_handler.name956 957 958 async def test_the_group_parks_whoever_has_gone_quiet_and_spares_the_active(make_group, commander):959 """The valve is the GROUP's: it reads the silence off the photo and orders.960 961 Mario has been silent a minute past a valve set to half of one, and his962 ``last_refresh_ts`` is NOW: a beat keeps a row warm and proves nobody. Anna963 has just spoken, and nobody touches her.964 """965 group = make_group(966 users=["mario", "anna"],967 user_silence={"mario": 60},968 transfer_flag=None,969 user_idle_freeze_minutes=0.5,970 )971 worker_handler = await group.start_worker()972 for cid, user in (("cid-a", "mario"), ("cid-b", "anna")):973 known_at_the_vertex(commander, cid, user)974 await group.assign_user(user)975 worker_handler.hosted_users.add(user)976 977 await group.check_user_activity(now=True)978 979 assert commander.user_is_frozen("mario") is True980 assert group.user_worker_map["mario"] is None981 assert commander.user_map["mario"]["on_hold"] is None982 assert commander.user_is_frozen("anna") is False983 assert group.user_worker_map["anna"] == worker_handler.name984 assert worker_handler.hosted_users == {"anna"}985 986 987 async def test_whoever_is_past_his_own_expiry_is_dropped_and_not_parked(make_group, commander):988 """The other verdict: he is forgotten whole, and nothing of his is written.989 990 The horizon is the vertex's own — the one it applies to a parcel in the991 deposit — asked of it per identity, so a guest's shorter life needs no second992 setting anywhere. Expiry wins over the valve on the same user.993 """994 commander.user_expiry_hours = 0.5995 group = make_group(996 users=["ugo"],997 user_silence={"ugo": 3600},998 transfer_flag=None,999 user_idle_freeze_minutes=0.1,1000 )1001 worker_handler = await group.start_worker()1002 known_at_the_vertex(commander, "cid-c", "ugo")1003 await group.assign_user("ugo")1004 worker_handler.hosted_users.add("ugo")1005 1006 await group.check_user_activity(now=True)1007 1008 assert "ugo" not in commander.user_map1009 assert "ugo" not in group.user_worker_map1010 assert worker_handler.hosted_users == set()1011 assert commander.freeze_handler.user_folders == set()1012 1013 1014 async def test_a_drop_order_nobody_answers_expires_and_gives_the_block_back(1015 make_group, commander, monkeypatch1016 ):1017 """The same ceiling on the other order: the round gives up rather than hang.1018 1019 The worker takes the drop and never answers it. Without a deadline this1020 round would wait as long as the child stays mute, and the vertex gathers the1021 group turns, so its whole clock would stop with it. The expiry takes the road1022 of a refusal: the block falls, the user stays where he was, and the next1023 round judges him again.1024 """1025 monkeypatch.setattr(group_handler_module, "DEPARTURE_ORDER_WAIT_LIMIT", 0.2)1026 commander.user_expiry_hours = 0.51027 group = make_group(1028 users=["ugo"],1029 user_silence={"ugo": 3600},1030 transfer_flag=None,1031 drop_unanswered=True,1032 )1033 worker_handler = await group.start_worker()1034 known_at_the_vertex(commander, "cid-c", "ugo")1035 await group.assign_user("ugo")1036 worker_handler.hosted_users.add("ugo")1037 1038 await group.check_user_activity(now=True)1039 1040 assert "ugo" in commander.user_map1041 assert commander.user_map["ugo"]["on_hold"] is None1042 assert "ugo" not in commander.user_hold_event_map1043 assert group.user_worker_map["ugo"] == worker_handler.name1044 1045 1046 async def test_a_drop_order_that_cannot_be_sent_frees_the_user_and_the_round(make_group, commander):1047 """A wire gone under the order: the hold falls and the other users are judged.1048 1049 The expiry road raises the block before it orders, so an order that cannot1050 even be sent must give it back — otherwise every request of his would park1051 at the vertex and answer 503 for as long as the process lives. And the round1052 must go on: anna, silent past the valve, is parked in the same turn.1053 """1054 commander.user_expiry_hours = 0.51055 group = make_group(1056 users=["ugo", "anna"],1057 user_silence={"ugo": 3600, "anna": 60},1058 transfer_flag=None,1059 user_idle_freeze_minutes=0.5,1060 )1061 worker_handler = await group.start_worker()1062 for cid, user in (("cid-c", "ugo"), ("cid-b", "anna")):1063 known_at_the_vertex(commander, cid, user)1064 await group.assign_user(user)1065 worker_handler.hosted_users.add(user)1066 worker_handler.connector._stream = None1067 1068 await group.check_user_activity(now=True)1069 1070 assert "ugo" in commander.user_map1071 assert commander.user_map["ugo"]["on_hold"] is None1072 assert "ugo" not in commander.user_hold_event_map1073 assert commander.user_map["anna"]["on_hold"] is None1074 1075 1076 async def test_an_order_nobody_answers_expires_and_leaves_the_user_where_he_was(1077 make_group, commander, monkeypatch1078 ):1079 """The deadline on the order: the round gives up rather than stop beating.1080 1081 The worker takes the order and never answers it. Without a deadline this1082 round — and with it every later beat of the group — would wait as long as1083 the call it is stuck behind lasts. The expiry takes the road of a refusal:1084 the block falls, the user stays on his worker, and the next round judges him1085 again.1086 """1087 monkeypatch.setattr(group_handler_module, "DEPARTURE_ORDER_WAIT_LIMIT", 0.2)1088 group = make_group(freeze_unanswered=True)1089 worker_handler = await group.start_worker()1090 known_at_the_vertex(commander, "cid-a", "mario")1091 await group.assign_user("mario")1092 worker_handler.hosted_users.add("mario")1093 1094 assert await group.freeze_hosted_user("mario") is False1095 1096 assert commander.user_is_frozen("mario") is False1097 assert commander.user_map["mario"]["on_hold"] is None1098 assert group.user_worker_map["mario"] == worker_handler.name1099 1100 1101 async def test_an_order_cancelled_under_the_await_gives_the_block_back(make_group, commander):1102 """The quit cancels the beat: a user caught mid-order must not stay blocked."""1103 group = make_group(freeze_unanswered=True)1104 worker_handler = await group.start_worker()1105 known_at_the_vertex(commander, "cid-a", "mario")1106 await group.assign_user("mario")1107 worker_handler.hosted_users.add("mario")1108 1109 order = asyncio.ensure_future(group.freeze_hosted_user("mario"))1110 await asyncio.sleep(0.05)1111 assert commander.user_map["mario"]["on_hold"] is not None1112 1113 order.cancel()1114 with pytest.raises(asyncio.CancelledError):1115 await order1116 1117 assert commander.user_map["mario"]["on_hold"] is None1118 assert "mario" not in commander.user_hold_event_map1119 assert group.user_worker_map["mario"] == worker_handler.name1120 1121 1122 async def test_a_request_arriving_under_the_expiry_order_waits_instead_of_routing(1123 make_group, commander1124 ):1125 """#40 on the OTHER road: the drop blocks him at the vertex the freeze does.1126 1127 The order to forget him is held on the wire while a request of his arrives.1128 Without the block it would be routed onto the very worker that is erasing his1129 rows; with it, it parks on his barrier and is served after — as the newcomer1130 he is once his identity is gone, since the fold that prunes the indexes is1131 also what lets the barrier go.1132 """1133 commander.user_expiry_hours = 0.51134 group = make_group(1135 users=["ugo"],1136 user_silence={"ugo": 3600},1137 transfer_flag=None,1138 user_idle_freeze_minutes=0.1,1139 )1140 worker_handler = await group.start_worker()1141 known_at_the_vertex(commander, "cid-c", "ugo")1142 await group.assign_user("ugo")1143 worker_handler.hosted_users.add("ugo")1144 ordered = asyncio.Event()1145 confirm = asyncio.Event()1146 on_the_wire = []1147 placed_call = worker_handler.connector.call1148 1149 async def held_order(path, data=None, timeout=None):1150 on_the_wire.append(path)1151 if path == DROP_USER_OP_PATH:1152 ordered.set()1153 await confirm.wait()1154 return await placed_call(path, data, timeout)1155 1156 worker_handler.connector.call = held_order1157 1158 round_of_the_group = asyncio.ensure_future(group.check_user_activity(now=True))1159 await ordered.wait()1160 request = asyncio.ensure_future(1161 commander.serve_request(1162 "cid-c",1163 control_frame(method="CALL", path="/invoices", data={"http": {"path": "/invoices"}}),1164 hold_timeout=5.0,1165 )1166 )1167 await asyncio.sleep(0.05)1168 1169 assert commander.user_map["ugo"]["on_hold"] == f"expiry on {worker_handler.name}"1170 assert not request.done()1171 assert on_the_wire == [DROP_USER_OP_PATH]1172 1173 confirm.set()1174 await round_of_the_group1175 1176 assert "ugo" not in commander.user_map1177 assert "ugo" not in commander.user_hold_event_map1178 assert "error" not in (await request).info