From 8034fb6e1ff11a6a6f0f746909b30c1ce8d6040a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 6 Aug 2026 05:29:15 +0000 Subject: [PATCH] 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). --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/xceadmin/src/xceadmin/scheduler.py | 367 ++++++++++++++++++++++++- src/xceadmin/src/xcecommon/errors.py | 22 ++ src/xceadmin/tests/test_scheduler.py | 300 +++++++++++++++++++- 4 files changed, 679 insertions(+), 13 deletions(-) diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 1e35010a..904a350b 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -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 diff --git a/src/xceadmin/src/xceadmin/scheduler.py b/src/xceadmin/src/xceadmin/scheduler.py index 8bae7673..87355ad5 100644 --- a/src/xceadmin/src/xceadmin/scheduler.py +++ b/src/xceadmin/src/xceadmin/scheduler.py @@ -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("> 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 diff --git a/src/xceadmin/src/xcecommon/errors.py b/src/xceadmin/src/xcecommon/errors.py index 6a5f6b2b..8c751979 100644 --- a/src/xceadmin/src/xcecommon/errors.py +++ b/src/xceadmin/src/xcecommon/errors.py @@ -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 -- diff --git a/src/xceadmin/tests/test_scheduler.py b/src/xceadmin/tests/test_scheduler.py index 209e8630..d789894f 100644 --- a/src/xceadmin/tests/test_scheduler.py +++ b/src/xceadmin/tests/test_scheduler.py @@ -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)