tests/spa/orchestration/test_orchestration_spa_commander.py

Source from this local checkout, regenerated when the reader rebuilds. Line links use #L<number>; a GitHub line range opens its first line.

1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 #     https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The vertex on its own: who it writes, what it answers, what it discards.16 17 The chain's tests look at the fold from above — an envelope arrives and the18 indexes move. These look at the same object from the front: the birth of a row19 for whoever shows up, the predicates the rest of the machine reads it20 through, the waiting room that is a raised exception and not a field, and the two21 things the vertex does that nobody below it can — discarding what a dead process22 left on disk, and writing the account of every order.23 """24 25 from __future__ import annotations26 27 import asyncio28 import json29 import logging30 import pickle31 32 import pytest33 34 35 from genro_asgi_multiworker_spa.orchestration import SpaCommander, UserOnHold36 from genro_asgi_multiworker_spa.orchestration import FreezeHandler37 from genro_asgi_multiworker_spa.orchestration.freeze_handler import USER_REGISTER_ITEM_NAME38 from genro_asgi_multiworker_spa.orchestration.spa_commander import GUEST_PREFIX39 40 WORKER_NAME = "standard_0001"41 42 43 44 def minted(commander, cid: str) -> str:45     """The identity the site would baptise for this cookie, learned by the vertex.46 47     The old mint died with the doctrine (the cookie routes, the site names):48     tests stage the junction the fold of ``new_connection`` would have written.49     """50     user = f"guest_{cid}"51     commander.record_connection_user(cid, user)52     return user53 54 55 @pytest.fixture56 def commander(short_root):57     return SpaCommander(short_root / "frozen_users")58 59 60 def parked_state(commander: SpaCommander, user: str) -> None:61     """What a freeze leaves on disk, written the way a worker writes it."""62     commander.freeze_handler.take_lock(user, WORKER_NAME)63     commander.freeze_handler.write_user_register_item(64         user, {"store": "whatever"}, writer=WORKER_NAME, cause="freeze", group="standard"65     )66     commander.freeze_handler.release_lock(user, WORKER_NAME)67 68 69 def test_whoever_shows_up_is_minted_before_anything_descends(commander):70     user = minted(commander, "cid-a")71 72     assert user == f"{GUEST_PREFIX}cid-a"73     assert commander.connection_user_map == {"cid-a": user}74     assert commander.user_map[user] == {"group": None, "frozen": False, "on_hold": None}75 76 77 def test_a_cid_already_known_is_answered_and_nothing_is_written_twice(commander):78     first = minted(commander, "cid-a")79     commander.user_map[first]["group"] = "kept"80 81     assert minted(commander, "cid-a") == first82     assert commander.user_map[first]["group"] == "kept"83 84 85 def test_a_cookie_that_outlived_its_row_is_still_that_person(commander):86     commander.connection_user_map["cid-a"] = "mario"87 88     assert commander.resolve_user("cid-a") == "mario"89     assert commander.user_map["mario"]["frozen"] is False90 91 92 def test_a_user_on_his_way_out_is_not_routed_but_raised(commander):93     user = minted(commander, "cid-a")94     commander.hold_user(user, "transfer_flag T")95 96     with pytest.raises(UserOnHold) as refusal:97         commander.resolve_user("cid-a")98 99     assert refusal.value.user == user100     assert refusal.value.cause == "transfer_flag T"101     assert str(refusal.value) == f"{user} is on hold: transfer_flag T"102 103 104 def test_the_cause_of_a_hold_is_the_one_that_explains_the_wait(commander):105     user = minted(commander, "cid-a")106     commander.hold_user(user, "transfer_flag T")107     commander.hold_user(user, "transfer_flag X")108 109     assert commander.user_map[user]["on_hold"] == "transfer_flag T"110 111 112 def test_nobody_is_frozen_until_it_is_written_down(commander):113     user = minted(commander, "cid-a")114 115     assert commander.user_is_frozen("somebody nobody knows") is False116     assert commander.user_is_frozen(user) is False117 118     commander.mark_user_frozen(user)119 120     assert commander.user_is_frozen(user) is True121 122 123 def test_a_freeze_ends_the_wait_it_was_the_reason_for(commander):124     user = minted(commander, "cid-a")125     commander.hold_user(user, "transfer_flag T")126 127     commander.mark_user_frozen(user)128 129     assert minted(commander, "cid-a") == user130 131 132 def test_an_adoption_takes_the_mark_off(commander):133     user = minted(commander, "cid-a")134     commander.mark_user_frozen(user)135 136     commander.mark_user_adopted(user)137 138     assert commander.user_is_frozen(user) is False139 140 141 def test_dropping_what_is_already_gone_is_that_same_outcome(commander):142     commander.drop_page("never-existed")143     commander.drop_connection("never-existed")144     commander.drop_user("never-existed")145 146     assert commander.page_connection_map == {}147     assert commander.user_map == {}148 149 150 def test_a_user_who_is_gone_takes_his_connections_pages_and_freezer_state_with_him(commander):151     user = minted(commander, "cid-a")152     commander.connection_user_map["cid-b"] = user153     commander.page_connection_map["p1"] = "cid-a"154     commander.page_connection_map["p2"] = "cid-b"155     parked_state(commander, user)156 157     assert commander.drop_user(user) is True158 159     assert commander.user_map == {}160     assert commander.connection_user_map == {}161     assert commander.page_connection_map == {}162     # Nothing of an identity nobody answers for is left behind: what the sweep of163     # the freezer finds later is only what a row lost WITHOUT a drop.164     assert commander.freeze_handler.user_folders == set()165     assert commander.counters["frozen_users_discarded"] == 1166 167 168 def test_what_a_dead_process_left_on_disk_is_discarded_and_counted(commander, caplog):169     caplog.set_level(logging.INFO)170     user = minted(commander, "cid-a")171     parked_state(commander, user)172     without_state = minted(commander, "cid-b")173 174     commander.drop_users([user, without_state], cause="process_aborted")175 176     assert commander.freeze_handler.user_folders == set()177     assert commander.user_map == {}178     assert commander.counters["frozen_users_discarded"] == 1179     assert caplog.text.count("order=drop_user") == 2180 181 182 def test_every_order_leaves_its_row_on_the_file_of_the_orders(short_root):183     log_path = short_root / "orchestration.log"184     commander = SpaCommander(short_root / "frozen_users", orchestration_log_path=log_path)185 186     commander.log_order(187         "standard",188         "quit_process",189         WORKER_NAME,190         numbers={"occupancy_percent": 12.0},191         outcome="quitted",192     )193 194     row = log_path.read_text().strip()195     assert "decided_by=standard" in row196     assert "order=quit_process" in row197     assert f"subject={WORKER_NAME}" in row198     assert "numbers={'occupancy_percent': 12.0}" in row199     assert "outcome=quitted" in row200 201 202 def test_every_order_also_leaves_a_structured_decision(short_root):203     log_path = short_root / "orchestration.log"204     commander = SpaCommander(short_root / "frozen_users", orchestration_log_path=log_path)205 206     commander.log_order(207         "standard",208         "quit_process",209         WORKER_NAME,210         numbers={"occupancy_percent": 12.0},211         outcome="quitted",212         reason="worker_was_ordered_away",213     )214 215     decision_path = log_path.with_suffix(".decisions.jsonl")216     row = json.loads(decision_path.read_text())217     assert row["schema"] == 1218     assert row["decision_id"].endswith("-1")219     assert row["decided_by"] == "standard"220     assert row["decision"] == "quit_process"221     assert row["subject"] == WORKER_NAME222     assert row["outcome"] == "quitted"223     assert row["reason"] == "worker_was_ordered_away"224     assert row["numbers"] == {"occupancy_percent": 12.0}225     assert row["candidates"] == []226     assert row["timestamp"].endswith("+00:00")227 228 229 def test_a_calculation_can_be_recorded_without_inventing_an_order(short_root):230     log_path = short_root / "orchestration.log"231     commander = SpaCommander(short_root / "frozen_users", orchestration_log_path=log_path)232 233     commander.log_decision(234         "standard",235         "placement_candidates",236         "standard_0002",237         reason="hottest_cpu_open_candidate",238         subject="mario",239         candidates=[240             {"name": "standard_0001", "cpu_admission_open": False},241             {"name": "standard_0002", "cpu_admission_open": True},242         ],243     )244 245     assert log_path.read_text() == ""246     row = json.loads(log_path.with_suffix(".decisions.jsonl").read_text())247     assert row["decision"] == "placement_candidates"248     assert row["outcome"] == "standard_0002"249     assert row["reason"] == "hottest_cpu_open_candidate"250     assert [candidate["name"] for candidate in row["candidates"]] == [251         "standard_0001",252         "standard_0002",253     ]254 255 256 def test_one_process_has_one_vertex_and_the_log_is_its_own(short_root):257     first_path = short_root / "first.log"258     second_path = short_root / "second.log"259     SpaCommander(short_root / "frozen_users", orchestration_log_path=first_path)260     second = SpaCommander(short_root / "frozen_users", orchestration_log_path=second_path)261 262     second.log_order("standard", "quit_process", WORKER_NAME, outcome="quitted")263 264     assert len(logging.getLogger("genro_asgi.orchestration.orders").handlers) == 1265     assert "order=quit_process" in second_path.read_text()266     assert first_path.read_text() == ""267     assert "quit_process" in second_path.with_suffix(".decisions.jsonl").read_text()268     assert first_path.with_suffix(".decisions.jsonl").read_text() == ""269 270 271 def test_the_machine_starts_running_and_holds_the_master_of_the_store(commander):272     assert commander.state == "running"273     assert commander.global_register == {}274 275 276 def test_the_concession_is_this_servers_share_of_the_whole_machine(commander, monkeypatch):277     monkeypatch.setattr(278         commander,279         "_machine_memory_gauges",280         lambda: {"MemTotal": 8_000_000_000.0, "MemAvailable": 6_000_000_000.0},281     )282     commander.memory_max_percent = 25.0283 284     assert commander.memory_concession_bytes == 2_000_000_000285     # And the alarm line is read against the machine, not against the concession.286     assert commander._machine_memory_used_percent() == 25.0287 288 289 def test_the_machine_total_is_read_off_the_platform_itself(commander):290     # No monkeypatch: os.sysconf answers on every platform this suite runs on,291     # /proc/meminfo or not.292     assert commander.memory_concession_bytes > 0293 294 295 def test_the_elected_group_receives_the_newcomer(commander):296     commander.group_map["stable"] = object()297     commander.group_map["canary"] = object()298 299     # Nobody elected: the first declared is the one the recipe named first.300     assert commander.default_group == "stable"301 302     commander._default_group = "canary"303     assert commander.default_group == "canary"304 305 306 def test_a_vertex_with_nowhere_to_put_a_newcomer_says_so(commander):307     with pytest.raises(KeyError):308         commander.default_group309 310     commander._default_group = "nobody"311     commander.group_map["stable"] = object()312     with pytest.raises(KeyError):313         commander.default_group314 315 316 def test_the_group_of_a_user_is_recorded_where_the_placement_decided_it(commander):317     user = minted(commander, "cid-a")318 319     commander.record_user_group(user, "stable")320 321     assert commander.user_map[user]["group"] == "stable"322 323 324 async def test_a_request_for_a_user_on_hold_waits_for_his_release(commander):325     user = minted(commander, "cid-a")326     commander.hold_user(user, "transfer_flag T")327 328     waiting = asyncio.ensure_future(commander.await_user_release(user, timeout=5.0))329     await asyncio.sleep(0)330     assert not waiting.done()331 332     commander.mark_user_adopted(user)333 334     await waiting335     assert commander.user_hold_event_map == {}336     assert minted(commander, "cid-a") == user337 338 339 async def test_the_freezer_mark_releases_the_wait_too(commander):340     user = minted(commander, "cid-a")341     commander.hold_user(user, "transfer_flag F")342     waiting = asyncio.ensure_future(commander.await_user_release(user, timeout=5.0))343     await asyncio.sleep(0)344 345     commander.mark_user_frozen(user)346 347     await waiting348     assert commander.user_hold_event_map == {}349 350 351 async def test_a_user_dropped_while_held_wakes_whoever_waited_for_him(commander):352     user = minted(commander, "cid-a")353     commander.hold_user(user, "transfer_flag T")354     waiting = asyncio.ensure_future(commander.await_user_release(user, timeout=5.0))355     await asyncio.sleep(0)356 357     commander.drop_users([user], cause="expired")358 359     await waiting360     assert commander.user_hold_event_map == {}361 362 363 async def test_a_wait_that_outlives_its_own_deadline_gives_up(commander):364     user = minted(commander, "cid-a")365     commander.hold_user(user, "transfer_flag T")366 367     with pytest.raises(TimeoutError):368         await commander.await_user_release(user, timeout=0.01)369 370     # The hold is still up: giving up on a wait decides nothing about the user.371     assert user in commander.user_hold_event_map372 373 374 async def test_nobody_waits_on_a_user_who_is_not_held(commander):375     user = minted(commander, "cid-a")376 377     await commander.await_user_release(user, timeout=0.01)378 379 380 def test_a_hold_already_up_keeps_its_first_cause_and_its_own_door(commander):381     user = minted(commander, "cid-a")382     commander.hold_user(user, "transfer_flag T")383     door = commander.user_hold_event_map[user]384 385     commander.hold_user(user, "transfer_flag X")386 387     assert commander.user_map[user]["on_hold"] == "transfer_flag T"388     assert commander.user_hold_event_map[user] is door389 390 391 class QuietGroup:392     """A group that records the order it was given and parks nobody."""393 394     def __init__(self) -> None:395         self.ordered_into: list[str] = []396 397     async def quit_all(self, freezer_path: str) -> None:398         self.ordered_into.append(freezer_path)399 400 401 async def test_the_quit_orders_every_group_and_commits_the_photo_by_renaming(commander):402     group = QuietGroup()403     commander.group_map["standard"] = group404     commander.record_connection_user("cid-a", "mario")405 406     await commander.quit()407 408     assert group.ordered_into == [str(commander.reboot_temp_path)]409     assert not commander.reboot_temp_path.exists()410     assert commander.reboot_data_path.exists()411     saved = FreezeHandler(commander.reboot_data_path).read_commander_register_item()412     assert saved["connection_user_map"] == {"cid-a": "mario"}413 414 415 async def test_the_saved_rows_are_normalised_for_a_boot_that_adopts_nobody(commander):416     commander.group_map["standard"] = QuietGroup()417     commander.record_connection_user("cid-a", "mario")418     commander.hold_user("mario", "moving")419 420     await commander.quit()421 422     saved = FreezeHandler(commander.reboot_data_path).read_commander_register_item()423     row = saved["user_map"]["mario"]424     assert row["frozen"] is True425     assert row["on_hold"] is None426 427 428 async def test_a_quit_that_dies_before_the_rename_leaves_no_photo(commander, monkeypatch):429     commander.group_map["standard"] = QuietGroup()430 431     def refuse(*args, **kwargs):432         raise OSError("the disk said no")433 434     monkeypatch.setattr(FreezeHandler, "write_commander_register_item", refuse)435 436     with pytest.raises(OSError):437         await commander.quit()438 439     assert commander.reboot_temp_path.exists()440     assert not commander.reboot_data_path.exists()441 442 443 def photographed(commander, user="mario", cid="cid-a", ts=None):444     """A photo on disk as a soft quit leaves it: one parcel, and the vertex's item."""445     photo = FreezeHandler(commander.reboot_data_path)446     photo.take_lock(user, WORKER_NAME)447     photo.write_user_register_item(448         user, {"store": "whatever"}, writer=WORKER_NAME, cause="quit", group="standard"449     )450     photo.release_lock(user, WORKER_NAME)451     if ts is not None:452         path = photo.root_path / photo.user_to_userkey(user) / USER_REGISTER_ITEM_NAME453         envelope = pickle.loads(path.read_bytes())454         envelope["header"]["ts"] = ts455         path.write_bytes(pickle.dumps(envelope))456     photo.write_commander_register_item(457         {458             "user_map": {user: dict(commander._new_row(), frozen=True, group="standard")},459             "connection_user_map": {cid: user},460             "page_connection_map": {},461             "global_register": {},462             "quit_ts": 0.0,463         },464         writer="vertex",465         cause="quit",466     )467     return photo468 469 470 def test_a_boot_with_frozen_registers_becomes_them_and_leaves_the_parcels_where_wakes_look(471     commander,472 ):473     photographed(commander)474 475     commander.adopt_frozen_registers()476 477     assert commander.connection_user_map == {"cid-a": "mario"}478     assert commander.user_is_frozen("mario") is True479     assert not commander.reboot_data_path.exists()480     assert commander.freeze_handler.read_user_register_item("mario") is not None481 482 483 def test_a_boot_with_no_frozen_registers_wipes_the_working_deposit_and_starts_clean(commander):484     parked_state(commander, "mario")485 486     commander.adopt_frozen_registers()487 488     assert commander.user_map == {}489     assert commander.freeze_handler.user_folders == set()490 491 492 def test_frozen_registers_that_cannot_be_read_boot_clean(commander):493     photo = photographed(commander)494     (photo.root_path / "commander_register_item.pickle").write_bytes(b"not a pickle")495 496     commander.adopt_frozen_registers()497 498     assert commander.user_map == {}499     assert not commander.reboot_data_path.exists()500     assert commander.freeze_handler.user_folders == set()501 502 503 def test_a_reboot_temp_left_by_a_dead_quit_is_never_read(commander):504     FreezeHandler(commander.reboot_temp_path)505 506     commander.adopt_frozen_registers()507 508     assert not commander.reboot_temp_path.exists()509     assert commander.user_map == {}510 511 512 async def test_a_user_past_his_expiry_is_dropped_at_the_boot_not_woken(commander):513     photographed(commander, ts=0.0)514     commander.adopt_frozen_registers()515 516     await commander.drop_expired_users(now=True)517 518     assert "mario" not in commander.user_map519     assert commander.freeze_handler.user_folders == set()520 521 522 async def test_the_cookie_survives_a_quit_and_the_boot_that_follows(commander):523     """The whole point: the same cid still names the same person on the other side."""524     commander.group_map["standard"] = QuietGroup()525     commander.record_connection_user("cid-a", "mario")526     parked_state(commander, "mario")527     commander.mark_user_frozen("mario")528     await commander.quit()529 530     reborn = SpaCommander(commander.freeze_handler.root_path)531     reborn.adopt_frozen_registers()532 533     assert reborn.resolve_user("cid-a") == "mario"534     assert reborn.user_is_frozen("mario") is True