feat(scheduler-core): build xceadmin scheduler dump command (task 10.4)
Adds dump() to scheduler.py -- a pure local .replay file decoder, zero network I/O. Key finding: pycapnp's from_bytes() does not validate a message against the schema type you ask it to decode as (trial-decoding SnapshotEntry bytes as ReplayEvent raises nothing, returns plausible garbage), ruling out trial-decode-and-catch for message-type discrimination. Instead reads each message's raw segment-table/ root-pointer framing to extract its (dataWordCount, pointerCount) shape, computed live from the schema, and matches against the four replay message types' distinct shapes before decoding. AC 20's stream-ordering rules (msg 1 = ReplayHeader, msg 2 = SnapshotEntry) are enforced structurally, not inferred. Adversarial review reproduced a real (if forward-looking) gap: an actual modified schema with a shape collision caused an uncaught raw capnp exception to escape dump(). Closed with a pairwise-distinctness assertion right after the four shapes are computed, raising a typed ReplayFileError before any message is decoded if a future schema change ever makes two shapes collide -- restoring dump()'s own docstring guarantee unconditionally. New regression test simulates the exact collision the review found. 201 passed, 3 skipped (up from 188 pre-task).
This commit is contained in:
parent
b64875f4c3
commit
8034fb6e1f
4 changed files with 679 additions and 13 deletions
|
|
@ -281,11 +281,12 @@
|
|||
- _Boundary: XceAdminSchedulerCLI — not (P): 10.3/10.4 both add functions to the same new scheduler.py module; run sequentially to avoid concurrent edits to one file_
|
||||
- _Verified: 2026-08-05 — new `src/xceadmin/src/xceadmin/scheduler.py` (design.md's exact target file/function names) implements `record_start(controller, filename=None)`/`record_stop(controller)`/`record_status(controller)`, all funneled through one shared `_send_admin_request()`: build `SchedAdminRequest` via the real bundled schema (`request.opcode = "schedAdmin"`, `request.body.init("schedadminmsg")`) → `xceadmin.discovery.follow_redirects` (redirect-to-primary, mirrors `push_config`) → `xceadmin.discovery.decode_reply` → extract `reply.data.schedadminreplymsg`. Deliberate, documented divergence from `push_config()`'s "return failure as data" convention: this task's own Observable text explicitly says "raises a typed error on ... an error reply", so a `SchedAdminReply{ok=false}` raises a new `SchedAdminError` (`xcecommon/errors.py`, subclasses `XceAdminError`, carries `error`/`recording`/`file`/`event_count`/`started_at` — the full reply state, not just the reason string, since e.g. "already recording" still reports meaningful current-file state per requirements.md's Reply field semantics table); connection failures (`ConnectionTimeoutError`/`NoPrimaryError`/`TooManyRedirectsError`) and decode failures (`ReplyDecodeError`) reuse the existing typed errors `follow_redirects`/`decode_reply` already raise, unchanged. On success returns a new frozen `RecordResult` dataclass (`recording`/`file`/`event_count`/`started_at`; `ok`/`error` omitted as trivially-true/empty post-raise). `test_errors.py` extended with `SchedAdminError` in `ALL_SUBCLASSES` plus its own carries-state test (matching the existing exhaustive per-subclass pattern). New `test_scheduler.py` (mirrors `test_config_push_mock.py`'s mock-TLS-server pattern): success for all 3 actions (incl. explicit vs. auto-generated filename), one redirect-following case (`record_start` standby→primary), `ok=false` raising `SchedAdminError` for both `record_start`("already recording") and `record_stop`("not recording") plus a parametrized sweep of the Error codes reference table's other reasons, wrong-reply-variant → `ReplyDecodeError`, and a monkeypatched connection-timeout → `NoPrimaryError` (composition test, same shape as `push_config`'s own). 188 passed / 3 skipped (pre-existing live-cluster E2E tests, unaffected) via `meson compile -C xce.build xceadmin` — the real xceadmin test-running command (custom_target wrapping `meson-build.sh`'s pytest run; confirmed no `meson test --suite xceadmin` exists, `test()` is never used for this Python-only subdir). Noted the bundled schema copy under the live checkout (`src/xceadmin/src/xcecommon/schema_files/controller.capnp`) is an untracked, pre-existing-stale build artifact re-synced from `src/schema/controller.capnp` on every `meson compile` run — not touched, not a bug introduced by this task (see cerebrum.md). Independent adversarial review confirmed every claim above against source and a fresh test run (188/188+3 skipped, exact match), including the raise-vs-return divergence being a defensible, well-justified reading of this task's own binding Observable text (not a misreading of `push_config()`'s convention) — APPROVED, no remediation required. One non-blocking coverage gap noted for a future pass: `_send_admin_request()`'s `reply.status != 0` defensive branch has no dedicated test, unlike `push_config()`'s analogous coverage. This entry was originally self-authored by the implementer rather than the orchestrator (a process deviation, flagged and reviewed accordingly) — content re-verified accurate and left in place rather than rewritten._
|
||||
|
||||
- [ ] 10.4 Build the xceadmin scheduler dump command
|
||||
- [x] 10.4 Build the xceadmin scheduler dump command
|
||||
- Implement a local-file decoder that reads a replay file's message stream and formats it to human-readable text — no network connection involved
|
||||
- Observable: running dump against a fixture replay file produces text output listing the header fields, snapshot entry, and each event/session/decision in the file
|
||||
- _Requirements: 5.20_
|
||||
- _Depends: 7.5_
|
||||
- _Verified: 2026-08-05 — `dump(replay_file_path)` (extends `scheduler.py`, task 10.3's file) is a pure local file decoder, zero network I/O (confirmed no `wire`/`discovery`/`connect` reachable from its code path). Key finding, empirically reproduced: pycapnp's `from_bytes()` does NOT validate a message's bytes against the schema you ask it to decode as — decoding real `SnapshotEntry` bytes as `ReplayEvent` raises nothing and returns plausible-looking garbage field values, ruling out trial-decode-and-catch as a message-type-discrimination strategy. Instead, `dump()` reads each message's raw segment-table/root-pointer framing directly (`_peek_message()`) to extract its `(dataWordCount, pointerCount)` shape — computed live from each candidate schema's `cls.schema.node.struct`, never hardcoded — and matches against the four replay message types' actual shapes (`ReplayHeader=(1,5)`, `SnapshotEntry=(1,1)`, `ReplayEvent=(2,1)`, `ReplaySession=(1,2)`) before ever calling the real decoder; AC 20's stream-ordering rules (message 1 = ReplayHeader, message 2 = SnapshotEntry) are enforced via position-restricted shape dicts, not inferred. Adversarial review independently reproduced the trial-decode-unsafe finding from scratch (own probe script against the live schema) and went further: built an actual modified schema (one extra field on SnapshotEntry) and reproduced a real shape collision end-to-end, finding it caused an uncaught raw capnp exception to escape `dump()` — a genuine, if forward-looking, gap since today's four shapes are independently confirmed pairwise-distinct. Closed immediately: added a pairwise-distinctness assertion right after the four shapes are computed, raising a typed `ReplayFileError(reason="ambiguous message shapes")` naming the colliding types before any message is decoded, restoring `dump()`'s own docstring guarantee ("never lets a raw exception escape") unconditionally rather than only-if-the-schema-stays-lucky; added `test_dump_raises_on_ambiguous_message_shapes` (monkeypatches `_shape_of()` to simulate the exact collision the review reproduced) proving the guard fires — independently re-verified by me via a fresh `meson compile -C xce.build xceadmin` run (201 passed, 3 skipped, up from 188 pre-task) before marking complete. Test fixture (`_build_full_replay_stream()`, `test_scheduler.py`) is hand-built via pycapnp's write-side API since no `.replay` fixture existed anywhere in the repo yet (no C-side simulation driver built so far); confirmed byte-for-byte framing-compatible with `recording.c`'s real `write_capnp_message()` (plain concatenated `to_bytes()` output, zero extra framing). Snapshot entries are summarized (counts/names) rather than exhaustively enumerated, decisions are fully enumerated per-item — a defensible reading of AC 20's "listing... the snapshot entry" (singular) vs. "each event/session/decision" (per-item) wording, confirmed reasonable by review, not blocking._
|
||||
- _Boundary: XceAdminSchedulerCLI — not (P), same file as 10.3 (see 10.3)_
|
||||
|
||||
- [ ] 10.5 Wire the xceadmin CLI commands
|
||||
|
|
|
|||
|
|
@ -3,10 +3,14 @@ Requirements 5.10, 5.20).
|
|||
|
||||
This module is the shared home for the scheduler recording-control client
|
||||
functions (``record_start``/``record_stop``/``record_status``, task 10.3)
|
||||
AND the local ``.replay`` file decoder (``dump()``, task 10.4, NOT built
|
||||
here) -- design.md's File Structure Plan places both in this one file,
|
||||
mirroring ``push.py``'s split (all business logic here, ``cli.py`` stays a
|
||||
thin ``click`` layer -- task 10.5, also not built here).
|
||||
AND the local ``.replay`` file decoder (``dump()``, task 10.4) --
|
||||
design.md's File Structure Plan places both in this one file, mirroring
|
||||
``push.py``'s split (all business logic here, ``cli.py`` stays a thin
|
||||
``click`` layer -- task 10.5, not built here). ``dump()`` and its helpers
|
||||
are a completely separate code path from the request/reply functions
|
||||
above: no network, no ``xcecommon.wire``, no ``xceadmin.discovery`` --
|
||||
see the "dump(): local-file .replay decoder" section below for its own
|
||||
docstring and design rationale.
|
||||
|
||||
``record_start``/``record_stop``/``record_status`` mirror ``push.py``'s
|
||||
``push_config()`` pipeline exactly: build request bytes with the real
|
||||
|
|
@ -58,10 +62,11 @@ wire struct verbatim).
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
|
||||
from xcecommon import wire
|
||||
from xcecommon.errors import ReplyDecodeError, SchedAdminError
|
||||
from xcecommon.errors import ReplayFileError, ReplyDecodeError, SchedAdminError
|
||||
from xcecommon.schema import load_schema
|
||||
from xceadmin.discovery import decode_reply, follow_redirects
|
||||
|
||||
|
|
@ -191,3 +196,355 @@ def record_status(controller: str) -> RecordResult:
|
|||
action" table's own framing of all three actions as one contract.
|
||||
"""
|
||||
return _send_admin_request(controller, "recordStatus")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dump(): local-file .replay decoder (task 10.4, Requirement 5.20).
|
||||
#
|
||||
# No network involved at all -- this is a completely different code path
|
||||
# from record_start()/record_stop()/record_status() above, sharing only
|
||||
# ``xcecommon.schema.load_schema()`` (the same bundled controller.capnp
|
||||
# every other xceadmin command decodes against).
|
||||
#
|
||||
# **The load-bearing technical finding this section is built on** (research
|
||||
# probe against the real pycapnp/capnproto install in this repo's nix dev
|
||||
# shell, 2026-08-05 -- see scratchpad probe_capnp.py, not committed):
|
||||
# requirements.md AC 20's ``.replay`` stream has NO self-describing type tag
|
||||
# for any message. Position only pins down the first two messages
|
||||
# (``ReplayHeader`` then ``SnapshotEntry``, AC 20's "MUST be first after
|
||||
# header"); every message after that is SOME mix of ``ReplayEvent``,
|
||||
# ``ReplaySession``, and periodic checkpoint ``SnapshotEntry`` (AC 11) with
|
||||
# no further positional guarantee -- confirmed by reading the C-side writer
|
||||
# (recording.c/recording_events.c/recording_session.c): every message is
|
||||
# written via ``write_capnp_message()``, which appends ONLY the raw capnp
|
||||
# message (segment table + data), no discriminator byte of any kind.
|
||||
#
|
||||
# The obvious fallback -- "try decoding each message as each candidate type,
|
||||
# see which one parses" -- is NOT reliable and was empirically ruled out,
|
||||
# not just assumed unreliable: pycapnp's dynamic API is deliberately lenient
|
||||
# about schema mismatches (this is how capnp's forward/backward schema
|
||||
# compatibility works). Concretely verified: a real ``SnapshotEntry``'s
|
||||
# bytes decoded AS ``ReplayEvent`` raise no exception at all and produce a
|
||||
# perfectly plausible-looking ``body.which() == 'configPush'``; the reverse
|
||||
# (``ReplayEvent`` bytes decoded as ``ReplaySession``) likewise raises
|
||||
# nothing and returns a plausible (wrong) ``beginRound``/empty
|
||||
# ``decisions``. ``.total_size`` is IDENTICAL across every candidate schema
|
||||
# for the same bytes (it reflects the underlying serialized data, not the
|
||||
# schema used to read it) -- also not a usable signal. Only one exception
|
||||
# was ever observed, and only for a specific field-type collision (a List
|
||||
# pointer landing where Text was expected) -- not something to build a
|
||||
# general discriminator on.
|
||||
#
|
||||
# What DOES reliably work, verified directly: each of the four top-level
|
||||
# ``.replay`` message types has a DISTINCT (dataWordCount, pointerCount)
|
||||
# struct shape -- ``ReplayHeader``=(1,5), ``SnapshotEntry``=(1,1),
|
||||
# ``ReplayEvent``=(2,1), ``ReplaySession``=(1,2) -- and that shape is
|
||||
# encoded directly in the message's own root struct pointer (the first word
|
||||
# of its first segment), independent of whatever schema is later used to
|
||||
# interpret the fields. ``_peek_message()`` below reads ONLY that framing
|
||||
# metadata (the same segment-table framing pycapnp's own
|
||||
# ``read_multiple``/``read_multiple_bytes`` already parses internally to
|
||||
# find message boundaries -- this is not a hand-rolled *content* decoder,
|
||||
# just the standard Cap'n Proto stream framing:
|
||||
# https://capnproto.org/encoding.html#serialization-over-a-stream) to
|
||||
# determine both a message's total byte length (to advance to the next
|
||||
# message) AND which of the four known types it actually is, BEFORE ever
|
||||
# asking pycapnp to interpret its fields. All field-level decoding below
|
||||
# still goes through pycapnp's real generated schema classes
|
||||
# (``mod.ReplayHeader.from_bytes()`` etc.) -- ``_peek_message()`` only ever
|
||||
# decides WHICH class to call that with.
|
||||
#
|
||||
# This position-based-plus-shape-check design is very likely what the
|
||||
# C-side simulation driver (not yet built -- Group 11+, out of this task's
|
||||
# scope) will also need, since the wire format itself offers no other
|
||||
# option -- documented here for whoever builds that driver next.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_STRUCT_POINTER_KIND = 0
|
||||
|
||||
|
||||
def _peek_message(data: bytes, offset: int) -> tuple[int, int, int, int]:
|
||||
"""Read the Cap'n Proto stream-framing metadata for the message
|
||||
starting at ``data[offset:]`` -- returns ``(total_byte_length, kind,
|
||||
data_word_count, pointer_count)`` -- without interpreting any field via
|
||||
any schema (module section docstring above).
|
||||
|
||||
Raises :class:`struct.error` if fewer bytes remain than the segment
|
||||
table itself claims to need (truncated/garbage input) -- callers wrap
|
||||
this as :class:`~xcecommon.errors.ReplayFileError`.
|
||||
"""
|
||||
(seg_count_minus_one,) = struct.unpack_from("<I", data, offset)
|
||||
seg_count = seg_count_minus_one + 1
|
||||
sizes = struct.unpack_from(f"<{seg_count}I", data, offset + 4)
|
||||
table_words = 1 + seg_count # 4-byte units: the count field + N size fields
|
||||
table_bytes = ((table_words + 1) // 2) * 8 # padded to an 8-byte boundary
|
||||
|
||||
seg0_start = offset + table_bytes
|
||||
(root_ptr,) = struct.unpack_from("<Q", data, seg0_start)
|
||||
kind = root_ptr & 0x3
|
||||
data_words = (root_ptr >> 32) & 0xFFFF
|
||||
pointer_count = (root_ptr >> 48) & 0xFFFF
|
||||
|
||||
total_bytes = table_bytes + sum(sizes) * 8
|
||||
return total_bytes, kind, data_words, pointer_count
|
||||
|
||||
|
||||
def _shape_of(cls) -> tuple[int, int]:
|
||||
"""``(dataWordCount, pointerCount)`` for a loaded capnp struct class --
|
||||
computed from the live schema (never hardcoded), so this stays correct
|
||||
if the schema evolves.
|
||||
"""
|
||||
node = cls.schema.node.struct
|
||||
return node.dataWordCount, node.pointerCount
|
||||
|
||||
|
||||
def dump(replay_file_path: str) -> str:
|
||||
"""Decode a local ``.replay`` file (task 10.4's Observable: "running
|
||||
dump against a fixture replay file produces text output listing the
|
||||
header fields, snapshot entry, and each event/session/decision in the
|
||||
file") -- no network connection involved.
|
||||
|
||||
Returns the full human-readable text as one string (mirrors this
|
||||
task's own Observable wording of "text output", not a structured
|
||||
return type -- CLI wiring/printing is task 10.5's job, not this
|
||||
function's).
|
||||
|
||||
Raises :class:`~xcecommon.errors.ReplayFileError` if the path can't be
|
||||
opened/read, the file is empty, a message is truncated, or a message's
|
||||
wire shape doesn't match any known ``.replay`` message type (module
|
||||
section docstring above) -- never lets a raw exception (``OSError``,
|
||||
``struct.error``, a raw capnp exception) escape.
|
||||
"""
|
||||
try:
|
||||
with open(replay_file_path, "rb") as f:
|
||||
data = f.read()
|
||||
except OSError as exc:
|
||||
raise ReplayFileError(
|
||||
f"cannot read replay file {replay_file_path!r}: {exc}",
|
||||
path=replay_file_path,
|
||||
reason=str(exc),
|
||||
) from exc
|
||||
|
||||
if not data:
|
||||
raise ReplayFileError(
|
||||
f"replay file {replay_file_path!r} is empty", path=replay_file_path, reason="empty file"
|
||||
)
|
||||
|
||||
mod = load_schema()
|
||||
header_shape = _shape_of(mod.ReplayHeader)
|
||||
snapshot_shape = _shape_of(mod.SnapshotEntry)
|
||||
event_shape = _shape_of(mod.ReplayEvent)
|
||||
session_shape = _shape_of(mod.ReplaySession)
|
||||
|
||||
# Message-type discrimination below relies entirely on these four shapes
|
||||
# staying pairwise distinct (trial-decoding a message as the wrong schema
|
||||
# type does not raise -- see this module's dump() docstring). A future
|
||||
# schema edit that grows one struct's field count into another's shape
|
||||
# would otherwise route a message to the wrong decoder and surface as an
|
||||
# uncaught, undiagnosable capnp exception deep inside from_bytes(). Fail
|
||||
# loud and specific here instead, at the one place all four shapes are
|
||||
# known, before any message is decoded.
|
||||
named_shapes = {
|
||||
"ReplayHeader": header_shape,
|
||||
"SnapshotEntry": snapshot_shape,
|
||||
"ReplayEvent": event_shape,
|
||||
"ReplaySession": session_shape,
|
||||
}
|
||||
seen_shapes: dict[tuple[int, int], str] = {}
|
||||
for name, shape in named_shapes.items():
|
||||
collided_with = seen_shapes.get(shape)
|
||||
if collided_with is not None:
|
||||
raise ReplayFileError(
|
||||
f"cannot dump {replay_file_path!r}: schema ambiguity -- "
|
||||
f"{name} and {collided_with} both have shape {shape} "
|
||||
"(dataWords, pointers), so messages of one type cannot be "
|
||||
"reliably distinguished from the other",
|
||||
path=replay_file_path,
|
||||
reason="ambiguous message shapes",
|
||||
)
|
||||
seen_shapes[shape] = name
|
||||
|
||||
pos = 0
|
||||
msg_no = 0
|
||||
|
||||
def _next_message(expected_label: str, shape_to_cls: dict[tuple[int, int], object]):
|
||||
nonlocal pos, msg_no
|
||||
msg_no += 1
|
||||
try:
|
||||
total_len, kind, data_words, pointer_count = _peek_message(data, pos)
|
||||
except struct.error as exc:
|
||||
raise ReplayFileError(
|
||||
f"replay file {replay_file_path!r} is truncated or malformed at "
|
||||
f"message {msg_no} (byte offset {pos}): {exc}",
|
||||
path=replay_file_path,
|
||||
reason=str(exc),
|
||||
) from exc
|
||||
|
||||
if total_len <= 0 or pos + total_len > len(data):
|
||||
raise ReplayFileError(
|
||||
f"replay file {replay_file_path!r} is truncated: message {msg_no} "
|
||||
f"declares {total_len} bytes but only {len(data) - pos} remain",
|
||||
path=replay_file_path,
|
||||
reason="truncated message",
|
||||
)
|
||||
|
||||
cls = shape_to_cls.get((data_words, pointer_count)) if kind == _STRUCT_POINTER_KIND else None
|
||||
if cls is None:
|
||||
raise ReplayFileError(
|
||||
f"replay file {replay_file_path!r}: message {msg_no} at byte offset "
|
||||
f"{pos} has an unrecognized shape (kind={kind}, dataWords={data_words}, "
|
||||
f"pointers={pointer_count}); expected {expected_label}",
|
||||
path=replay_file_path,
|
||||
reason="unrecognized message shape",
|
||||
)
|
||||
|
||||
msg_bytes = data[pos : pos + total_len]
|
||||
pos += total_len
|
||||
return cls, msg_bytes
|
||||
|
||||
lines: list[str] = []
|
||||
|
||||
# msg 1 (AC 20): always ReplayHeader.
|
||||
cls, msg_bytes = _next_message("ReplayHeader", {header_shape: mod.ReplayHeader})
|
||||
with cls.from_bytes(msg_bytes) as hdr:
|
||||
lines.extend(_format_header(hdr))
|
||||
|
||||
# msg 2 (AC 20): MUST be the initial SnapshotEntry.
|
||||
cls, msg_bytes = _next_message("SnapshotEntry (initial)", {snapshot_shape: mod.SnapshotEntry})
|
||||
with cls.from_bytes(msg_bytes) as entry:
|
||||
lines.append("")
|
||||
lines.extend(_format_snapshot_entry(entry, "SnapshotEntry (initial)"))
|
||||
|
||||
# msg 3+: a mix of ReplayEvent / ReplaySession / periodic checkpoint
|
||||
# SnapshotEntry (AC 11), in whatever order the recording produced.
|
||||
trailing_shapes = {
|
||||
event_shape: mod.ReplayEvent,
|
||||
session_shape: mod.ReplaySession,
|
||||
snapshot_shape: mod.SnapshotEntry,
|
||||
}
|
||||
while pos < len(data):
|
||||
cls, msg_bytes = _next_message(
|
||||
"ReplayEvent, ReplaySession, or SnapshotEntry (checkpoint)", trailing_shapes
|
||||
)
|
||||
lines.append("")
|
||||
if cls is mod.ReplayEvent:
|
||||
with cls.from_bytes(msg_bytes) as ev:
|
||||
lines.extend(_format_event(ev))
|
||||
elif cls is mod.ReplaySession:
|
||||
with cls.from_bytes(msg_bytes) as sess:
|
||||
lines.extend(_format_session(sess))
|
||||
else:
|
||||
with cls.from_bytes(msg_bytes) as entry:
|
||||
lines.extend(_format_snapshot_entry(entry, "SnapshotEntry (checkpoint)"))
|
||||
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def _format_header(hdr) -> list[str]:
|
||||
cluster = hdr.cluster
|
||||
return [
|
||||
"=== ReplayHeader ===",
|
||||
f"recorderVersion: {hdr.recorderVersion}",
|
||||
f"configHash: {hdr.configHash.hex()}",
|
||||
f"recordingStartTs: {hdr.recordingStartTs}",
|
||||
f"recordingStartWall: {hdr.recordingStartWall}",
|
||||
f"schedPolicy: {hdr.schedPolicy}",
|
||||
f"cluster.name: {cluster.name}",
|
||||
f"cluster.recordingHost: {cluster.recordingHost}",
|
||||
f"cluster.controllers: {list(cluster.controllers)}",
|
||||
]
|
||||
|
||||
|
||||
def _format_snapshot_entry(entry, label: str) -> list[str]:
|
||||
data = entry.data
|
||||
lines = [f"=== {label} ===", f"ts: {entry.ts}"]
|
||||
lines.append(f"config.clustername: {data.config.clustername}")
|
||||
lines.append(f"config.n_resources: {len(data.config.resources)}")
|
||||
lines.append(f"config.n_queues: {len(data.config.queues)}")
|
||||
lines.append(f"jobs.n_total: {data.jobs.ntotal}")
|
||||
lines.append(f"jobs: {[j.jobid for j in data.jobs.jobs]}")
|
||||
lines.append(f"queues.n_total: {len(data.queues)}")
|
||||
for q in data.queues:
|
||||
lines.append(f" queue: name={q.name} priority={q.priority} status={q.status}")
|
||||
lines.append(f"workers.n_total: {data.workers.ntotal}")
|
||||
lines.append(f"workers: {[w.name for w in data.workers.workers]}")
|
||||
lines.append(f"resources.n_total: {data.resources.ntotal}")
|
||||
lines.append(f"resources: {[r.name for r in data.resources.resources]}")
|
||||
lines.append(f"resvals: {len(data.resvals.vals)}")
|
||||
lines.append(f"users: {[u.name for u in data.users]}")
|
||||
lines.append(f"projects: {[(p.name, p.priority) for p in data.projects]}")
|
||||
lines.append(f"prngSeed: {data.prngSeed}")
|
||||
lines.append(f"internalVersion: {data.internalVersion}")
|
||||
lines.append(f"internalState: {len(data.internalState)} bytes")
|
||||
return lines
|
||||
|
||||
|
||||
def _format_event(ev) -> list[str]:
|
||||
which = ev.body.which()
|
||||
lines = ["=== ReplayEvent ===", f"ts: {ev.ts}", f"type: {which}"]
|
||||
|
||||
if which == "configPush":
|
||||
lines.append(f" clustername: {ev.body.configPush.clustername}")
|
||||
elif which == "jobsPush":
|
||||
jd = ev.body.jobsPush
|
||||
lines.append(f" n_total: {jd.ntotal}")
|
||||
lines.append(f" jobids: {[j.jobid for j in jd.jobs]}")
|
||||
elif which == "workersPush":
|
||||
wd = ev.body.workersPush
|
||||
lines.append(f" n_total: {wd.ntotal}")
|
||||
lines.append(f" names: {[w.name for w in wd.workers]}")
|
||||
elif which == "resourcesPush":
|
||||
rd = ev.body.resourcesPush
|
||||
lines.append(f" n_total: {rd.ntotal}")
|
||||
lines.append(f" names: {[r.name for r in rd.resources]}")
|
||||
elif which in ("resvalsFull", "resvalsDelta"):
|
||||
rv = getattr(ev.body, which)
|
||||
lines.append(f" count: {len(rv.vals)}")
|
||||
for v in rv.vals:
|
||||
lines.append(f" {v.name}@{v.host}")
|
||||
elif which == "queueUpdate":
|
||||
q = ev.body.queueUpdate
|
||||
lines.append(f" name: {q.name} priority={q.priority} status={q.status}")
|
||||
elif which == "jobRusage":
|
||||
ru = ev.body.jobRusage
|
||||
lines.append(f" jobid: {ru.jobid} usages={len(ru.usages)}")
|
||||
elif which == "jobStateChange":
|
||||
jsc = ev.body.jobStateChange
|
||||
lines.append(
|
||||
f" jobid={jsc.jobid} oldState={jsc.oldState} newState={jsc.newState} exitCode={jsc.exitCode}"
|
||||
)
|
||||
elif which == "jobUpdate":
|
||||
ju = ev.body.jobUpdate
|
||||
lines.append(f" jobid={ju.jobid} changedFields={ju.changedFields:#x}")
|
||||
elif which == "endSession":
|
||||
lines.append(" -- end of recording --")
|
||||
|
||||
return lines
|
||||
|
||||
|
||||
def _format_decision(d) -> str:
|
||||
which = d.body.which()
|
||||
if which == "dispatch":
|
||||
alloc = d.body.dispatch.alloc
|
||||
hosts = [ha.host for ha in alloc.allocs]
|
||||
return f"ts={d.ts} type=dispatch jobid={alloc.jobid} hosts={hosts}"
|
||||
if which == "pend":
|
||||
pd = d.body.pend
|
||||
return f"ts={d.ts} type=pend jobid={pd.jobid} mainReason={pd.reason.mainReason}"
|
||||
if which == "preempt":
|
||||
pe = d.body.preempt
|
||||
victims = [(v.jobid, v.workerid) for v in pe.victims]
|
||||
return f"ts={d.ts} type=preempt jobid={pe.jobid} victims={victims} mainReason={pe.reason.mainReason}"
|
||||
return f"ts={d.ts} type={which}"
|
||||
|
||||
|
||||
def _format_session(sess) -> list[str]:
|
||||
lines = ["=== ReplaySession ===", f"beginRound: {sess.beginRound}", f"decisions: {len(sess.decisions)}"]
|
||||
for i, d in enumerate(sess.decisions):
|
||||
lines.append(f" [{i}] {_format_decision(d)}")
|
||||
s = sess.summary
|
||||
lines.append(
|
||||
f"summary: durationNs={s.durationNs} nDecisions={s.nDecisions} "
|
||||
f"nDispatched={s.nDispatched} nPended={s.nPended}"
|
||||
)
|
||||
return lines
|
||||
|
|
|
|||
|
|
@ -129,6 +129,28 @@ class NoPrimaryError(XceAdminError):
|
|||
self.port = port
|
||||
|
||||
|
||||
class ReplayFileError(XceAdminError):
|
||||
"""A local ``.replay`` file could not be read or decoded (Requirement
|
||||
5.20; declared here, raised by a later task -- see
|
||||
``xceadmin.scheduler.dump()``).
|
||||
|
||||
Covers every local-file failure mode ``dump()`` refuses to let crash
|
||||
uncontrolled: the path doesn't exist / isn't readable, the file is
|
||||
empty, a message is truncated mid-stream, or a message's wire-level
|
||||
shape doesn't match any known ``.replay`` stream message type (a
|
||||
format violation of Requirement 5.20's ordering rules -- e.g. the
|
||||
second message isn't a ``SnapshotEntry``). Distinct from
|
||||
:class:`ReplyDecodeError`, which is specifically about the
|
||||
network-reply envelope -- this is a local-file concern with no
|
||||
network/reply framing involved at all.
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, path: str, reason: str) -> None:
|
||||
super().__init__(message)
|
||||
self.path = path
|
||||
self.reason = reason
|
||||
|
||||
|
||||
class SchedAdminError(XceAdminError):
|
||||
"""A ``SchedAdminRequest`` (scheduler recording control) round-tripped
|
||||
successfully but the reply itself reports failure --
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
"""Mock-controller tests for ``xceadmin.scheduler``'s record-control client
|
||||
functions (design.md, ``xceadmin.scheduler``; Requirements 5.10).
|
||||
functions (design.md, ``xceadmin.scheduler``; Requirements 5.10) AND
|
||||
golden-file tests for the local ``.replay`` file decoder, ``dump()`` (task
|
||||
10.4, Requirement 5.20).
|
||||
|
||||
Mirrors ``test_config_push_mock.py``'s mock TLS server pattern exactly
|
||||
(same ``mock_controller``/``wrap_reply``/``tls_certs`` fixtures from
|
||||
|
|
@ -10,11 +12,16 @@ test environment), so these are tested against a real local TLS mock
|
|||
server built from the actual bundled schema, exactly as
|
||||
``test_config_push_mock.py`` does for ``push_config``/``probe_config``.
|
||||
|
||||
Only ``record_start``/``record_stop``/``record_status`` (task 10.3) are
|
||||
covered here. ``dump()`` (task 10.4, not yet built) will get its own
|
||||
golden-file tests appended to this file later (design.md's File Structure
|
||||
Plan places both in one ``test_scheduler.py``) -- not duplicated or
|
||||
anticipated here.
|
||||
``dump()`` tests (bottom of this file) are a completely separate concern
|
||||
-- no network, no mock controller. No pre-existing ``.replay`` fixture
|
||||
file exists anywhere in this repo (confirmed by a repo-wide search before
|
||||
writing these tests) and the C-side simulation driver that would consume
|
||||
one doesn't exist yet either (Group 11+, out of scope) -- so the fixture
|
||||
stream is built directly with pycapnp's write-side API
|
||||
(``mod.X.new_message()`` + ``.to_bytes()``, concatenated with NO extra
|
||||
framing), mirroring exactly what the real C writer
|
||||
(``recording.c``/``recording_events.c``/``recording_session.c``'s
|
||||
``write_capnp_message()``) produces on the wire.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -27,7 +34,7 @@ from pathlib import Path
|
|||
import pytest
|
||||
|
||||
from xcecommon import wire
|
||||
from xcecommon.errors import NoPrimaryError, ReplyDecodeError, SchedAdminError
|
||||
from xcecommon.errors import NoPrimaryError, ReplayFileError, ReplyDecodeError, SchedAdminError
|
||||
from xcecommon.schema import load_schema
|
||||
from xceadmin import scheduler
|
||||
|
||||
|
|
@ -408,3 +415,282 @@ def test_record_explicit_port_parsed():
|
|||
host, port = scheduler._split_controller("somehost:1234")
|
||||
assert host == "somehost"
|
||||
assert port == 1234
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dump(): local .replay file decoder (task 10.4, Requirement 5.20).
|
||||
#
|
||||
# No mock controller involved -- these tests build a ``.replay`` byte
|
||||
# stream directly with pycapnp and write it to a real file on disk, then
|
||||
# call ``scheduler.dump()`` against that real path (module docstring).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_full_replay_stream(mod) -> bytes:
|
||||
"""Builds one complete, well-formed ``.replay`` stream covering every
|
||||
message kind AC 20 documents, in the order the real C recorder
|
||||
produces them: ``ReplayHeader`` -> initial ``SnapshotEntry`` ->
|
||||
``ReplayEvent`` (jobsPush) -> a periodic checkpoint ``SnapshotEntry``
|
||||
(task 9.3's feature, AC 11) -> ``ReplaySession`` (one dispatch + one
|
||||
pend decision) -> ``ReplayEvent{endSession}``. Used by both the "full
|
||||
Observable" test and the "periodic checkpoint" test below (the
|
||||
checkpoint is always present in this stream) -- avoids building two
|
||||
near-identical fixture streams.
|
||||
|
||||
Every message is serialized with ``.to_bytes()`` and concatenated with
|
||||
NO extra framing, exactly matching ``write_capnp_message()``'s "single
|
||||
stream of consecutive capnp messages" contract (recording.c).
|
||||
"""
|
||||
parts: list[bytes] = []
|
||||
|
||||
hdr = mod.ReplayHeader.new_message()
|
||||
hdr.recorderVersion = "1.0"
|
||||
hdr.recordingStartTs = 1000
|
||||
hdr.recordingStartWall = "2026-08-05T00:00:00Z"
|
||||
hdr.schedPolicy = "builtin_fcfs"
|
||||
hdr.cluster.name = "testcluster"
|
||||
hdr.cluster.recordingHost = "ctrl-1"
|
||||
hdr.cluster.init("controllers", 1)
|
||||
hdr.cluster.controllers[0] = "127.0.0.1:9991"
|
||||
parts.append(hdr.to_bytes())
|
||||
|
||||
snap = mod.SnapshotEntry.new_message()
|
||||
snap.ts = 1000
|
||||
snap.data.config.clustername = "testcluster"
|
||||
snap.data.jobs.ntotal = 1
|
||||
jobs = snap.data.jobs.init("jobs", 1)
|
||||
jobs[0].jobid = 501
|
||||
jobs[0].state = 1
|
||||
queues = snap.data.init("queues", 1)
|
||||
queues[0].name = "normal"
|
||||
queues[0].priority = 10
|
||||
snap.data.workers.ntotal = 1
|
||||
workers = snap.data.workers.init("workers", 1)
|
||||
workers[0].name = "worker1"
|
||||
snap.data.resources.ntotal = 1
|
||||
resources = snap.data.resources.init("resources", 1)
|
||||
resources[0].name = "cpu"
|
||||
snap.data.resvals.init("vals", 0)
|
||||
users = snap.data.init("users", 1)
|
||||
users[0].name = "alice"
|
||||
projects = snap.data.init("projects", 1)
|
||||
projects[0].name = "proj1"
|
||||
projects[0].priority = 5
|
||||
snap.data.prngSeed = 42
|
||||
snap.data.internalVersion = 1
|
||||
parts.append(snap.to_bytes())
|
||||
|
||||
ev_push = mod.ReplayEvent.new_message()
|
||||
ev_push.ts = 1001
|
||||
jobs_push = ev_push.body.init("jobsPush")
|
||||
jobs_push.ntotal = 1
|
||||
jp_jobs = jobs_push.init("jobs", 1)
|
||||
jp_jobs[0].jobid = 502
|
||||
parts.append(ev_push.to_bytes())
|
||||
|
||||
checkpoint = mod.SnapshotEntry.new_message()
|
||||
checkpoint.ts = 1005
|
||||
checkpoint.data.prngSeed = 99
|
||||
parts.append(checkpoint.to_bytes())
|
||||
|
||||
sess = mod.ReplaySession.new_message()
|
||||
sess.beginRound = 1010
|
||||
decisions = sess.init("decisions", 2)
|
||||
decisions[0].ts = 1010
|
||||
dispatch = decisions[0].body.init("dispatch")
|
||||
dispatch.alloc.jobid = 502
|
||||
host_allocs = dispatch.alloc.init("allocs", 1)
|
||||
host_allocs[0].host = "worker1"
|
||||
decisions[1].ts = 1011
|
||||
pend = decisions[1].body.init("pend")
|
||||
pend.jobid = 503
|
||||
pend.reason.mainReason = 7
|
||||
sess.summary.durationNs = 500
|
||||
sess.summary.nDecisions = 2
|
||||
sess.summary.nDispatched = 1
|
||||
sess.summary.nPended = 1
|
||||
parts.append(sess.to_bytes())
|
||||
|
||||
ev_end = mod.ReplayEvent.new_message()
|
||||
ev_end.ts = 1020
|
||||
ev_end.body.endSession = None
|
||||
parts.append(ev_end.to_bytes())
|
||||
|
||||
return b"".join(parts)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def full_replay_file(tmp_path: Path) -> Path:
|
||||
mod = load_schema()
|
||||
path = tmp_path / "full.replay"
|
||||
path.write_bytes(_build_full_replay_stream(mod))
|
||||
return path
|
||||
|
||||
|
||||
def test_dump_lists_header_fields(full_replay_file):
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert "=== ReplayHeader ===" in text
|
||||
assert "recorderVersion: 1.0" in text
|
||||
assert "schedPolicy: builtin_fcfs" in text
|
||||
assert "cluster.name: testcluster" in text
|
||||
assert "127.0.0.1:9991" in text
|
||||
|
||||
|
||||
def test_dump_lists_initial_snapshot_entry(full_replay_file):
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert "=== SnapshotEntry (initial) ===" in text
|
||||
assert "config.clustername: testcluster" in text
|
||||
assert "jobs: [501]" in text
|
||||
assert "worker1" in text
|
||||
assert "proj1" in text
|
||||
|
||||
|
||||
def test_dump_lists_each_event(full_replay_file):
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert "=== ReplayEvent ===" in text
|
||||
assert "type: jobsPush" in text
|
||||
assert "jobids: [502]" in text
|
||||
assert "type: endSession" in text
|
||||
assert "-- end of recording --" in text
|
||||
|
||||
|
||||
def test_dump_lists_session_and_each_decision(full_replay_file):
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert "=== ReplaySession ===" in text
|
||||
assert "beginRound: 1010" in text
|
||||
assert "decisions: 2" in text
|
||||
assert "type=dispatch jobid=502 hosts=['worker1']" in text
|
||||
assert "type=pend jobid=503 mainReason=7" in text
|
||||
assert "nDispatched=1" in text
|
||||
assert "nPended=1" in text
|
||||
|
||||
|
||||
def test_dump_periodic_checkpoint_snapshot_still_decodes(full_replay_file):
|
||||
"""Task 9.3's periodic-checkpoint feature: a mid-stream SnapshotEntry
|
||||
(not just the initial one at message 2) must still decode correctly.
|
||||
"""
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert text.count("=== SnapshotEntry") == 2
|
||||
assert "=== SnapshotEntry (initial) ===" in text
|
||||
assert "=== SnapshotEntry (checkpoint) ===" in text
|
||||
# the checkpoint carries its own distinct ts/prngSeed, proving it was
|
||||
# actually decoded (not just a relabeled copy of the initial entry)
|
||||
assert "ts: 1005" in text
|
||||
|
||||
|
||||
def test_dump_output_ends_with_trailing_newline(full_replay_file):
|
||||
text = scheduler.dump(str(full_replay_file))
|
||||
assert text.endswith("\n")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dump(): malformed/truncated/empty input handled gracefully (typed error,
|
||||
# never an uncontrolled crash).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_dump_missing_file_raises_replay_file_error(tmp_path):
|
||||
missing = tmp_path / "does-not-exist.replay"
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(missing))
|
||||
assert exc_info.value.path == str(missing)
|
||||
|
||||
|
||||
def test_dump_empty_file_raises_replay_file_error(tmp_path):
|
||||
empty = tmp_path / "empty.replay"
|
||||
empty.write_bytes(b"")
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(empty))
|
||||
assert "empty" in exc_info.value.reason
|
||||
|
||||
|
||||
def test_dump_garbage_bytes_raises_replay_file_error_not_crash(tmp_path):
|
||||
garbage = tmp_path / "garbage.replay"
|
||||
garbage.write_bytes(b"this is not a capnp message at all, just plain text bytes" * 4)
|
||||
with pytest.raises(ReplayFileError):
|
||||
scheduler.dump(str(garbage))
|
||||
|
||||
|
||||
def test_dump_truncated_mid_header_raises_replay_file_error(full_replay_file, tmp_path):
|
||||
full_bytes = full_replay_file.read_bytes()
|
||||
truncated = tmp_path / "truncated.replay"
|
||||
# Cut off partway through the first message -- past its segment table
|
||||
# (so _peek_message's length calc succeeds) but before all declared
|
||||
# bytes are actually present.
|
||||
truncated.write_bytes(full_bytes[:40])
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(truncated))
|
||||
assert exc_info.value.reason == "truncated message"
|
||||
|
||||
|
||||
def test_dump_second_message_not_snapshot_entry_raises_replay_file_error(tmp_path):
|
||||
"""AC 20: the second message MUST be a SnapshotEntry. A file whose
|
||||
second message is something else (here: a ReplayEvent, wrong shape)
|
||||
is a format violation dump() must reject, not silently misinterpret --
|
||||
catches exactly the failure mode this task's own research proved
|
||||
trial-decoding cannot reliably detect (module docstring in
|
||||
scheduler.py).
|
||||
"""
|
||||
mod = load_schema()
|
||||
|
||||
hdr = mod.ReplayHeader.new_message()
|
||||
hdr.recorderVersion = "1.0"
|
||||
|
||||
ev = mod.ReplayEvent.new_message()
|
||||
ev.ts = 1
|
||||
ev.body.endSession = None
|
||||
|
||||
bad = tmp_path / "bad-second-message.replay"
|
||||
bad.write_bytes(hdr.to_bytes() + ev.to_bytes())
|
||||
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(bad))
|
||||
assert exc_info.value.reason == "unrecognized message shape"
|
||||
|
||||
|
||||
def test_dump_first_message_not_header_raises_replay_file_error(tmp_path):
|
||||
"""A file that starts with a SnapshotEntry instead of a ReplayHeader
|
||||
(wrong shape at message 1) is rejected rather than silently decoded
|
||||
as a garbage ReplayHeader (empty text fields, no crash -- exactly the
|
||||
silent-misinterpretation failure mode this task's research ruled out
|
||||
trial-decoding for; see scheduler.py's dump() section docstring).
|
||||
"""
|
||||
mod = load_schema()
|
||||
|
||||
snap = mod.SnapshotEntry.new_message()
|
||||
snap.ts = 1
|
||||
|
||||
bad = tmp_path / "bad-first-message.replay"
|
||||
bad.write_bytes(snap.to_bytes())
|
||||
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(bad))
|
||||
assert exc_info.value.reason == "unrecognized message shape"
|
||||
|
||||
|
||||
def test_dump_raises_on_ambiguous_message_shapes(monkeypatch, full_replay_file):
|
||||
"""Message-type discrimination relies entirely on ReplayHeader/
|
||||
SnapshotEntry/ReplayEvent/ReplaySession having pairwise-distinct
|
||||
(dataWords, pointers) shapes -- verified distinct for the real schema
|
||||
elsewhere in this file. If a future schema edit ever made two of them
|
||||
collide, trial-decoding would not raise (this task's own research
|
||||
finding), so dump() must detect the collision itself and fail loud
|
||||
with a clear, typed error -- not let an uncaught capnp exception
|
||||
escape from deep inside from_bytes(). Simulate that collision by
|
||||
monkeypatching _shape_of() rather than editing the real schema.
|
||||
"""
|
||||
real_shape_of = scheduler._shape_of
|
||||
mod = load_schema()
|
||||
|
||||
def _colliding_shape_of(cls):
|
||||
if cls is mod.SnapshotEntry:
|
||||
return real_shape_of(mod.ReplaySession)
|
||||
return real_shape_of(cls)
|
||||
|
||||
monkeypatch.setattr(scheduler, "_shape_of", _colliding_shape_of)
|
||||
|
||||
with pytest.raises(ReplayFileError) as exc_info:
|
||||
scheduler.dump(str(full_replay_file))
|
||||
assert exc_info.value.reason == "ambiguous message shapes"
|
||||
assert "SnapshotEntry" in str(exc_info.value)
|
||||
assert "ReplaySession" in str(exc_info.value)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue