fix(delivery): fit bounded health probes into startup budgets

backend/tests/test_delivery.py

@@ -2233,46 +2233,159 @@ def test_proxy_render_is_fixed_to_a_selected_snapshot_without_business_credentia
     assert "EnvironmentFile=" not in text and "LoadCredential=" not in text
 
 
-@pytest.mark.parametrize("name", ["queues", "vhost_limits", "users", "permissions"])
 @pytest.mark.parametrize("failure", ["deadline", "exit", "output"])
-def test_broker_query_failure_identifies_fixed_operation_without_leaking_process_data(
-    monkeypatch, name, failure
-):
-    import subprocess
+def test_broker_snapshot_failure_keeps_process_details_private(monkeypatch, failure):
     from types import SimpleNamespace
 
-    from scripts.release import broker_health
+    from scripts.release import broker_snapshot
 
     secret = "synthetic-private-process-detail"
 
     def failed(command, *, stdout, stderr, timeout, env):
-        assert stderr == subprocess.DEVNULL
-        assert timeout == 8
-        assert command[-1] == "--formatter=json"
+        assert stderr == subprocess.DEVNULL and timeout == 30
+        assert command == [
+            str(broker_snapshot.CTL),
+            "-q",
+            "-n",
+            "kf@localhost",
+            "--timeout",
+            "5",
+            "eval",
+            broker_snapshot.SNAPSHOT_EXPRESSION,
+            "--formatter=json",
+        ]
+        assert env["ERL_EPMD_ADDRESS"] == "127.0.0.1"
+        assert "rpc:call(node(), erlang, apply," in command[7]
+        assert command[7].endswith("end, []], 5000).")
         if failure == "deadline":
             raise subprocess.TimeoutExpired([secret], timeout, stderr=secret.encode())
         if failure == "output":
             stdout.write(b"x" * 32769)
         return SimpleNamespace(returncode=1 if failure == "exit" else 0)
 
-    monkeypatch.setattr(broker_health.subprocess, "run", failed)
+    monkeypatch.setattr(broker_snapshot.subprocess, "run", failed)
     suffix = {"deadline": "exceeded deadline", "exit": "failed", "output": "output exceeds bound"}
     with pytest.raises(policy.DeliveryError) as raised:
-        broker_health.command(name)
-    assert str(raised.value) == "broker " + name + " query " + suffix[failure]
+        broker_snapshot.command()
+    assert str(raised.value) == "broker snapshot query " + suffix[failure]
     assert secret not in str(raised.value)
 
 
+@pytest.mark.parametrize("change", [None, "node", "result", "rpc_failure", "extra", "duplicate"])
+def test_broker_snapshot_accepts_only_the_pinned_node_envelope(monkeypatch, change):
+    from types import SimpleNamespace
+
+    from scripts.release import broker_snapshot
+
+    snapshot = {"queues": [], "vhost_limits": [], "users": [], "permissions": []}
+    envelope = {"result": "ok", "node": "kf@localhost", "value": json.dumps(snapshot)}
+    if change == "node":
+        envelope["node"] = "other@localhost"
+    elif change == "result":
+        envelope["result"] = "error"
+    elif change == "rpc_failure":
+        envelope["value"] = ["badrpc", "timeout"]
+    elif change == "extra":
+        envelope["unknown"] = True
+    raw = json.dumps(envelope).encode()
+    if change == "duplicate":
+        raw = b'{"node":"kf@localhost",' + raw[1:]
+
+    def query(command, *, stdout, **kwargs):
+        stdout.write(raw)
+        return SimpleNamespace(returncode=0)
+
+    monkeypatch.setattr(broker_snapshot.subprocess, "run", query)
+    if change is None:
+        assert broker_snapshot.command() == snapshot
+    else:
+        with pytest.raises(policy.DeliveryError):
+            broker_snapshot.command()
+
+
 @pytest.mark.parametrize("name", ["delete_queue", "users --delete", None])
-def test_broker_query_selector_never_accepts_extra_commands_or_flags(monkeypatch, name):
-    from scripts.release import broker_health
+def test_broker_snapshot_has_no_command_or_expression_selector(monkeypatch, name):
+    from scripts.release import broker_snapshot
 
     def unexpected(*args, **kwargs):
-        pytest.fail("unknown selector launched a process")
+        pytest.fail("caller text launched a process")
 
-    monkeypatch.setattr(broker_health.subprocess, "run", unexpected)
-    with pytest.raises(policy.DeliveryError, match="unknown broker metadata query"):
-        broker_health.command(name)
+    monkeypatch.setattr(broker_snapshot.subprocess, "run", unexpected)
+    with pytest.raises(TypeError):
+        broker_snapshot.command(name)
+
+
+@pytest.mark.parametrize(
+    "kind,value", [("longstr", "classic"), ("longstr", "quorum"), ("long", "classic")]
+)
+def test_broker_arguments_allow_only_exact_automatic_classic_type(kind, value):
+    args = [
+        [k, "longstr" if k == "x-overflow" else "long", v]
+        for k, v in dependencies.QUEUE_ARGUMENTS.items()
+    ]
+    args.append(["x-queue-type", kind, value])
+    if (kind, value) == ("longstr", "classic"):
+        assert broker_health.queue_arguments(args) == dependencies.QUEUE_ARGUMENTS
+        supplied = {**dependencies.QUEUE_ARGUMENTS, "x-queue-type": "classic"}
+        assert broker_health.queue_arguments(supplied) == dependencies.QUEUE_ARGUMENTS
+        assert supplied["x-queue-type"] == "classic"
+        args[-1] = args[0].copy()
+    with pytest.raises(policy.DeliveryError):
+        broker_health.queue_arguments(args)
+
+
+@pytest.mark.parametrize(
+    "name", ["broker-health", "factory-health", "import-health", "outbox-health"]
+)
+@pytest.mark.parametrize("cpu", ["100000 100000", "max 100000", "200000 100000"])
+def test_fixed_health_guard_still_requires_exact_helper_and_node_cpu(monkeypatch, name, cpu):
+    import resource
+
+    from scripts.release import operator_units
+
+    _, uid, memory, pids, expected_cpu, _ = operator_units.PROFILES[name]
+    assert expected_cpu == 100
+    monkeypatch.setattr(operator_units.os, "getuid", lambda: uid)
+    monkeypatch.setattr(operator_units.os, "geteuid", lambda: uid)
+    monkeypatch.setattr(operator_units.os, "getgid", lambda: uid)
+    monkeypatch.setattr(operator_units.os, "getgroups", lambda: [uid])
+    group = Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service")
+    monkeypatch.setattr(operator_units.preflight, "current_cgroup", lambda: group)
+    observed = {
+        "memory.max": str(memory * 1024**2),
+        "memory.swap.max": "0",
+        "pids.max": str(pids),
+        "cpu.max": cpu,
+    }
+    node = {
+        "memory.max": str(3840 * 1024**2),
+        "memory.swap.max": "0",
+        "pids.max": "384",
+        "cpu.max": "200000 100000",
+    }
+
+    def read(path, *args, **kwargs):
+        if str(path) == "/proc/self/status":
+            return "NoNewPrivs: 1\n" + "\n".join(
+                k + ": 0" for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb")
+            )
+        return (observed if path.parent == group else node)[path.name]
+
+    monkeypatch.setattr(Path, "read_text", read)
+    limits = {
+        resource.RLIMIT_CORE: 0,
+        resource.RLIMIT_FSIZE: operator_units.FILE_LIMIT_MIB.get(name, 16) * 1024**2,
+        resource.RLIMIT_NOFILE: 256,
+        resource.RLIMIT_STACK: 16 * 1024**2,
+    }
+    monkeypatch.setattr(
+        operator_units.resource, "getrlimit", lambda key: (limits[key], limits[key])
+    )
+    if cpu == "100000 100000":
+        operator_units.guarded(name)
+    else:
+        with pytest.raises(policy.DeliveryError, match="budgets differ"):
+            operator_units.guarded(name)
 
 
 @pytest.mark.parametrize(

scripts/release/README.md

@@ -415,9 +415,16 @@ end-to-end target still requires measured acceptance.
 Fixed health helpers include interpreter startup in their wall-clock deadline.
 Outbox and import queries have 60 seconds each; factory status has 30 seconds and
 broker metadata has 45 seconds. Node readiness has 300 seconds for the sequential
-checks. Each helper keeps its fixed CPU, memory, process and privilege limits.
-These operator deadlines do not change student execution budgets. An expired or
-failed query keeps admission closed.
+checks. Each ordinary helper has one CPU within the node's two-CPU ceiling;
+its memory, process and privilege limits remain fixed. These operator budgets
+do not change student execution budgets. Startup opens admission only after all
+required readiness queries succeed.
+
+Broker metadata uses one fixed read-only RPC with a five-second remote deadline
+and a 30-second CLI process deadline, within the helper's 45-second limit. It reads
+queue metadata, vhost limits, usernames/tags and permissions; it does not read
+messages or password hashes. Readiness requires exactly the configured classic
+queue and budgets, including RabbitMQ's optional `x-queue-type=classic` argument.
 
 The broker helper has a 64 MiB per-file ceiling for Erlang JIT mappings.
 [OTP 27.3.4](https://github.com/erlang/otp/blob/OTP-27.3.4/erts/emulator/beam/jit/beam_jit_main.cpp)

scripts/release/broker_health.py

@@ -4,56 +4,33 @@ import argparse
 import json
 import stat
 import subprocess
-import tempfile
 from pathlib import Path
 
+from . import broker_snapshot, operator_units, preflight
 from . import dependency_policy as dependencies
-from . import operator_units, preflight
 from .policy import DeliveryError, canonical, fields, integer
 
-CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl")
-QUERIES = {
-    "queues": [
-        "list_queues",
-        "-p",
-        "kf",
-        "name",
-        "durable",
-        "auto_delete",
-        "exclusive",
-        "type",
-        "state",
-        "arguments",
-        "messages_ready",
-        "messages_unacknowledged",
-        "message_bytes",
-        "consumers",
-        "policy",
-        "operator_policy",
-        "effective_policy_definition",
-    ],
-    "vhost_limits": ["list_vhost_limits", "--global"],
-    "users": ["list_users"],
-    "permissions": ["list_permissions", "-p", "kf"],
-}
+CTL = broker_snapshot.CTL
 
 
 def queue_arguments(value) -> dict:
     if isinstance(value, dict):
-        result = value
-    elif isinstance(value, list) and len(value) == 4:
+        result = dict(value)
+    elif isinstance(value, list) and len(value) in (4, 5):
         result = {}
         for row in value:
             if not isinstance(row, list) or len(row) != 3 or row[0] in result:
                 raise DeliveryError("duplicate/unknown broker argument")
             name, kind, argument = row
-            if name not in dependencies.QUEUE_ARGUMENTS or kind != (
-                "longstr" if name == "x-overflow" else "long"
+            if name not in {*dependencies.QUEUE_ARGUMENTS, "x-queue-type"} or kind != (
+                "longstr" if name in {"x-overflow", "x-queue-type"} else "long"
             ):
                 raise DeliveryError("broker argument type differs")
             result[name] = argument
     else:
         raise DeliveryError("broker argument shape differs")
+    if "x-queue-type" in result and result.pop("x-queue-type") != "classic":
+        raise DeliveryError("actual broker queue type differs")
     if canonical(result) != canonical(dependencies.QUEUE_ARGUMENTS):
         raise DeliveryError("actual broker queue budgets differ")
     return result
@@ -148,49 +125,6 @@ def validate_snapshot(value: dict) -> dict:
     }
 
 
-def command(name: str):
-    if name not in QUERIES:
-        raise DeliveryError("unknown broker metadata query")
-    # A file-backed capped stream prevents malicious metadata/log output from
-    # allocating an unbounded capture buffer. The unit adds a hard file limit.
-    env = {
-        "PATH": "/usr/bin:/bin",
-        "LC_ALL": "C.UTF-8",
-        "HOME": dependencies.STORAGE["rabbitmq"]["path"],
-        "ERL_EPMD_ADDRESS": "127.0.0.1",
-        "RABBITMQ_NODENAME": "kf@localhost",
-        "RABBITMQ_CTL_ERL_ARGS": (
-            "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}"
-        ),
-    }
-    with tempfile.TemporaryFile() as output:
-        try:
-            result = subprocess.run(
-                [
-                    str(CTL),
-                    "-q",
-                    "-n",
-                    "kf@localhost",
-                    "--timeout",
-                    "5",
-                    *QUERIES[name],
-                    "--formatter=json",
-                ],
-                stdout=output,
-                stderr=subprocess.DEVNULL,
-                timeout=8,
-                env=env,
-            )
-        except subprocess.TimeoutExpired as error:
-            raise DeliveryError("broker " + name + " query exceeded deadline") from error
-        if result.returncode:
-            raise DeliveryError("broker " + name + " query failed")
-        if output.tell() > 32768:
-            raise DeliveryError("broker " + name + " query output exceeds bound")
-        output.seek(0)
-        return operator_units.decode(output.read(32769))
-
-
 def observe() -> dict:
     operator_units.guarded("broker-health")
     preflight.protected(CTL)
@@ -202,8 +136,7 @@ def observe() -> dict:
         or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1204, 1204, 0o400)
     ):
         raise DeliveryError("ordinary private broker cookie required")
-    data = {name: command(name) for name in QUERIES}
-    return validate_snapshot(data)
+    return validate_snapshot(broker_snapshot.command())
 
 
 def main() -> None:

scripts/release/broker_snapshot.py

@@ -0,0 +1,96 @@
+"""One fixed read-only RabbitMQ 4.3.6 metadata RPC; no caller expressions."""
+
+import subprocess
+import tempfile
+from pathlib import Path
+
+from . import dependency_policy, operator_units
+from .policy import DeliveryError, fields
+
+CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl")
+# Fixed getters only: no caller text, mutation, message reads or password hashes.
+SNAPSHOT_EXPRESSION = r"""
+rpc:call(node(), erlang, apply, [fun() ->
+    Count = rabbit_amqqueue:count(<<"kf">>),
+    true = (Count =:= 0 orelse Count =:= 1),
+    Normalize = fun
+        (name, {resource, <<"kf">>, queue, Value}) -> Value;
+        (arguments, Value) -> [[K, atom_to_binary(T, utf8), V] || {K,T,V} <- Value];
+        (type, rabbit_classic_queue) -> <<"classic">>;
+        (K, Value) when K =:= policy; K =:= operator_policy ->
+            case Value of undefined -> <<>>; none -> <<>>; '' -> <<>>; _ -> Value end;
+        (effective_policy_definition, Value) when is_list(Value) -> maps:from_list(Value);
+        (tags, Value) -> [case T of
+            A when is_atom(A) -> atom_to_binary(A,utf8); _ -> T end || T <- Value];
+        (_, Value) when is_atom(Value), Value =/= true, Value =/= false ->
+            atom_to_binary(Value,utf8);
+        (_, Value) -> Value
+    end,
+    ToMap = fun(Items) -> maps:from_list(
+        [{atom_to_binary(K,utf8),Normalize(K,V)} || {K,V} <- Items]) end,
+    Queues = [ToMap(Row) || Row <- rabbit_amqqueue:info_all(<<"kf">>,
+        [name,durable,auto_delete,exclusive,type,state,arguments,messages_ready,
+         messages_unacknowledged,message_bytes,consumers,policy,operator_policy,
+         effective_policy_definition])],
+    Limits = [#{<<"vhost">> => V,
+                <<"limits">> => iolist_to_binary(rabbit_json:encode(maps:from_list(L)))}
+              || {V,L} <- rabbit_vhost_limit:list()],
+    Users = [ToMap(Row) || Row <- rabbit_auth_backend_internal:list_users()],
+    Permissions = [ToMap(Row) || Row <-
+        rabbit_auth_backend_internal:list_vhost_permissions(<<"kf">>)],
+    iolist_to_binary(rabbit_json:encode(#{<<"queues">> => Queues, <<"vhost_limits">> => Limits,
+        <<"users">> => Users, <<"permissions">> => Permissions}))
+end, []], 5000).
+""".strip()
+
+
+def command() -> dict:
+    env = {
+        "PATH": "/usr/bin:/bin",
+        "LC_ALL": "C.UTF-8",
+        "HOME": dependency_policy.STORAGE["rabbitmq"]["path"],
+        "ERL_EPMD_ADDRESS": "127.0.0.1",
+        "RABBITMQ_NODENAME": "kf@localhost",
+        "RABBITMQ_CTL_ERL_ARGS": (
+            "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}"
+        ),
+    }
+    with tempfile.TemporaryFile() as output:
+        try:
+            result = subprocess.run(
+                [
+                    str(CTL),
+                    "-q",
+                    "-n",
+                    "kf@localhost",
+                    "--timeout",
+                    "5",
+                    "eval",
+                    SNAPSHOT_EXPRESSION,
+                    "--formatter=json",
+                ],
+                stdout=output,
+                stderr=subprocess.DEVNULL,
+                timeout=30,
+                env=env,
+            )
+        except subprocess.TimeoutExpired as error:
+            raise DeliveryError("broker snapshot query exceeded deadline") from error
+        if result.returncode:
+            raise DeliveryError("broker snapshot query failed")
+        if output.tell() > 32768:
+            raise DeliveryError("broker snapshot query output exceeds bound")
+        output.seek(0)
+        value = operator_units.decode(output.read(32769))
+    fields(value, {"result", "node", "value"}, "broker snapshot envelope")
+    if (
+        value["result"] != "ok"
+        or value["node"] != "kf@localhost"
+        or not isinstance(value["value"], str)
+    ):
+        raise DeliveryError("broker snapshot result or node differs")
+    return fields(
+        operator_units.decode(value["value"].encode()),
+        {"queues", "vhost_limits", "users", "permissions"},
+        "broker snapshot",
+    )

scripts/release/node_health.py

@@ -66,6 +66,7 @@ def binding() -> tuple[dict, dict, Path]:
         "availability.py",
         "node_health.py",
         "broker_health.py",
+        "broker_snapshot.py",
         "factory_health.py",
         "import_health.py",
         "operator_units.py",

scripts/release/operator_units.py

@@ -11,10 +11,10 @@ from .policy import DeliveryError, invalid_constant, unique_object
 
 PROFILES = {
     "node-health": ("root", 0, 128, 16, 50, 300),
-    "broker-health": ("rabbitmq", 1204, 256, 64, 50, 45),
-    "factory-health": ("kf-worker", 1202, 32, 4, 10, 30),
-    "import-health": ("kf-oj", 1200, 128, 16, 25, 60),
-    "outbox-health": ("kf-oj", 1200, 128, 16, 25, 60),
+    "broker-health": ("rabbitmq", 1204, 256, 64, 100, 45),
+    "factory-health": ("kf-worker", 1202, 32, 4, 100, 30),
+    "import-health": ("kf-oj", 1200, 128, 16, 100, 60),
+    "outbox-health": ("kf-oj", 1200, 128, 16, 100, 60),
     "node-maintenance": ("root", 0, 256, 32, 50, 900),
     "node-observation": ("root", 0, 128, 16, 25, 360),
 }