feat(scheduler-core): define replay stream envelope schema (task 7.1)
Adds ReplayCluster/ReplayHeader/SnapshotEntry/ReplayEvent/
SessionSummary/ReplaySession to controller.capnp, additively, first
task of Phase 3 (Foundation: Replay Wire Format).
Requirements.md's AC 20 capnp sketch isn't self-contained -- ReplayEvent's
union references JobStateChangeEvent/JobUpdateEvent (task 7.2's job),
SnapshotEntry.data/ReplayDecision's union reference SnapshotData/
DispatchDecision/PendDecision/PreemptDecision (task 7.3's job), none
of which exist yet. Resolved by verifying capnp's real constraints
against the toolchain directly: SnapshotData is an empty {} placeholder
(task 7.3 replaces its mapname'd C struct wholesale when adding the
real ~11 fields), ReplayDecision holds only {ts} (task 7.3 appends a
body union at new ordinals), and ReplayEvent's union holds only the 9
members with pre-existing backing types (Conf4Sched/SchedJobData/
WorkerData/ResourceData/ResValList/QueueConf/SchedJobRusages/Void),
deferring jobStateChange/jobUpdate to task 7.2 as new ordinals --
capnp requires strictly sequential ordinals with no holes, so this
split is compiler-mandated, not stylistic.
Round-trip test added to test_e2e_capnp.c. Full project build (all
binaries sharing this schema) and 1075/1075 controller_unit_test
verified clean.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
parent
d39ffc1b4e
commit
68a6196517
SSH key fingerprint:
SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
4 changed files with 407 additions and 1 deletions
|
|
@ -331,6 +331,220 @@ static void test_data_job_array_variant_round_trip(void)
|
|||
printf("test_data_job_array_variant_round_trip: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Test 7: ReplayHeader + SnapshotEntry (replay stream envelope schema, */
|
||||
/* scheduler-core task 7.1, Req 5 AC 20) round-trip through the wire. */
|
||||
/* */
|
||||
/* Built entirely through the low-level capnp accessor API (new_X / */
|
||||
/* X_get_* / X_set_*, all generated by $C.fieldgetset) instead of the */
|
||||
/* encode_X/decode_X "codec" bridge to a mapname'd C struct. The codec */
|
||||
/* bridge is a documented no-op for Data fields in this codebase's */
|
||||
/* capnpc-c setup (see decode_UpgradeConfigRequest's empty body above, */
|
||||
/* and request.c's xce_replconfigpush_req_raw comment) -- encoding */
|
||||
/* ReplayHeader.configHash (a Data field) through it would silently */
|
||||
/* write whatever garbage sits in an uninitialized native `struct */
|
||||
/* ReplayHeader` local. Going through the accessors avoids that pitfall */
|
||||
/* entirely and, since no "real" application C struct exists yet for */
|
||||
/* these frame-only replay-envelope types (task 9+ builds the actual */
|
||||
/* reader/writer), is the correct level to prove the SCHEMA/wire format */
|
||||
/* itself round-trips -- which is all task 7.1 owns. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static void test_replay_header_snapshot_entry_round_trip(void)
|
||||
{
|
||||
/* Arrange + Act (message 1: ReplayHeader) */
|
||||
struct capn c1;
|
||||
capn_init_malloc(&c1);
|
||||
struct capn_segment *cs1 = capn_root(&c1).seg;
|
||||
|
||||
ReplayCluster_ptr cluster = new_ReplayCluster(cs1);
|
||||
capn_text cname = { 7, "xce-lab", NULL };
|
||||
ReplayCluster_set_name(cluster, cname);
|
||||
capn_ptr ctrl_list = capn_new_ptr_list(cs1, 2);
|
||||
capn_setp(ctrl_list, 0, capn_new_string(cs1, "host1:9991", 10));
|
||||
capn_setp(ctrl_list, 1, capn_new_string(cs1, "host2:9992", 10));
|
||||
ReplayCluster_set_controllers(cluster, ctrl_list);
|
||||
capn_text rhost = { 5, "host1", NULL };
|
||||
ReplayCluster_set_recordingHost(cluster, rhost);
|
||||
|
||||
uint8_t hash[32];
|
||||
for (int i = 0; i < 32; i++)
|
||||
hash[i] = (uint8_t)(i * 3 + 1);
|
||||
capn_list8 hash_list = capn_new_list8(cs1, 32);
|
||||
capn_setv8(hash_list, 0, hash, 32);
|
||||
capn_data hash_data;
|
||||
hash_data.p = hash_list.p;
|
||||
|
||||
ReplayHeader_ptr hdr = new_ReplayHeader(cs1);
|
||||
capn_text rv = { 5, "1.0.0", NULL };
|
||||
ReplayHeader_set_recorderVersion(hdr, rv);
|
||||
ReplayHeader_set_configHash(hdr, hash_data);
|
||||
ReplayHeader_set_cluster(hdr, cluster);
|
||||
ReplayHeader_set_recordingStartTs(hdr, 123456789ULL);
|
||||
capn_text wall = { 20, "2026-08-04T00:00:00Z", NULL };
|
||||
ReplayHeader_set_recordingStartWall(hdr, wall);
|
||||
capn_text policy = { 12, "builtin_fcfs", NULL };
|
||||
ReplayHeader_set_schedPolicy(hdr, policy);
|
||||
|
||||
capn_setp(capn_root(&c1), 0, hdr.p);
|
||||
int64_t sz1 = capn_size(&c1);
|
||||
TEST_ASSERT_TRUE(sz1 > 0);
|
||||
uint8_t *buf1 = (uint8_t *)malloc((size_t)sz1);
|
||||
TEST_ASSERT_NOT_NULL(buf1);
|
||||
capn_write_mem(&c1, buf1, (size_t)sz1, 0);
|
||||
capn_free(&c1);
|
||||
|
||||
/* Arrange + Act (message 2: SnapshotEntry -- MUST be the first data
|
||||
* message after ReplayHeader per Req 5 AC 20's stream framing rule;
|
||||
* encoded here as an independent message buffer since the file-level
|
||||
* multi-message concatenation is task 9+ reader/writer logic, not
|
||||
* this schema-only task's concern). */
|
||||
struct capn c2;
|
||||
capn_init_malloc(&c2);
|
||||
struct capn_segment *cs2 = capn_root(&c2).seg;
|
||||
|
||||
SnapshotData_ptr sdata = new_SnapshotData(cs2);
|
||||
SnapshotEntry_ptr snap = new_SnapshotEntry(cs2);
|
||||
SnapshotEntry_set_ts(snap, 99999ULL);
|
||||
SnapshotEntry_set_data(snap, sdata);
|
||||
|
||||
capn_setp(capn_root(&c2), 0, snap.p);
|
||||
int64_t sz2 = capn_size(&c2);
|
||||
TEST_ASSERT_TRUE(sz2 > 0);
|
||||
uint8_t *buf2 = (uint8_t *)malloc((size_t)sz2);
|
||||
TEST_ASSERT_NOT_NULL(buf2);
|
||||
capn_write_mem(&c2, buf2, (size_t)sz2, 0);
|
||||
capn_free(&c2);
|
||||
|
||||
/* Act: decode both messages back */
|
||||
struct capn d1;
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d1, buf1, (size_t)sz1, 0));
|
||||
ReplayHeader_ptr rhdr;
|
||||
rhdr.p = capn_getp(capn_root(&d1), 0, 1);
|
||||
|
||||
struct capn d2;
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d2, buf2, (size_t)sz2, 0));
|
||||
SnapshotEntry_ptr rsnap;
|
||||
rsnap.p = capn_getp(capn_root(&d2), 0, 1);
|
||||
|
||||
/* Assert: ReplayHeader fields survive field-for-field */
|
||||
capn_text out_rv = ReplayHeader_get_recorderVersion(rhdr);
|
||||
TEST_ASSERT_EQUAL_INT(5, out_rv.len);
|
||||
TEST_ASSERT_EQUAL_STRING("1.0.0", out_rv.str);
|
||||
|
||||
capn_data out_hash = ReplayHeader_get_configHash(rhdr);
|
||||
TEST_ASSERT_EQUAL_INT(32, capn_len((capn_list8){ out_hash.p }));
|
||||
uint8_t out_hash_bytes[32];
|
||||
TEST_ASSERT_EQUAL_INT(32, capn_getv8((capn_list8){ out_hash.p }, 0,
|
||||
out_hash_bytes, 32));
|
||||
TEST_ASSERT_EQUAL_MEMORY(hash, out_hash_bytes, 32);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(123456789ULL,
|
||||
ReplayHeader_get_recordingStartTs(rhdr));
|
||||
capn_text out_wall = ReplayHeader_get_recordingStartWall(rhdr);
|
||||
TEST_ASSERT_EQUAL_STRING("2026-08-04T00:00:00Z", out_wall.str);
|
||||
capn_text out_policy = ReplayHeader_get_schedPolicy(rhdr);
|
||||
TEST_ASSERT_EQUAL_STRING("builtin_fcfs", out_policy.str);
|
||||
|
||||
ReplayCluster_ptr out_cluster = ReplayHeader_get_cluster(rhdr);
|
||||
/* A nested-struct field getter (capn_getp(..., resolve=0)) can return an
|
||||
* unresolved far pointer; per-field accessors that index raw struct
|
||||
* offsets (capn_read*) do NOT auto-resolve it (unlike capn_get_text/
|
||||
* capn_get_data, which resolve internally), so any nested struct
|
||||
* pointer must be explicitly resolved before its own field accessors
|
||||
* are used. */
|
||||
capn_resolve(&out_cluster.p);
|
||||
capn_text out_cname = ReplayCluster_get_name(out_cluster);
|
||||
TEST_ASSERT_EQUAL_STRING("xce-lab", out_cname.str);
|
||||
capn_ptr out_ctrl_list = ReplayCluster_get_controllers(out_cluster);
|
||||
capn_resolve(&out_ctrl_list);
|
||||
TEST_ASSERT_EQUAL_INT(2, out_ctrl_list.len);
|
||||
capn_text text_default = { 0, "", NULL };
|
||||
capn_text c0 = capn_get_text(out_ctrl_list, 0, text_default);
|
||||
capn_text c1t = capn_get_text(out_ctrl_list, 1, text_default);
|
||||
TEST_ASSERT_EQUAL_STRING("host1:9991", c0.str);
|
||||
TEST_ASSERT_EQUAL_STRING("host2:9992", c1t.str);
|
||||
capn_text out_rhost = ReplayCluster_get_recordingHost(out_cluster);
|
||||
TEST_ASSERT_EQUAL_STRING("host1", out_rhost.str);
|
||||
|
||||
/* Assert: SnapshotEntry -- the required Observable of task 7.1
|
||||
* (Req 5 AC 20): a minimal header-plus-snapshot-entry byte stream
|
||||
* round-trips through encode/decode. */
|
||||
TEST_ASSERT_EQUAL_UINT64(99999ULL, SnapshotEntry_get_ts(rsnap));
|
||||
SnapshotData_ptr out_sdata = SnapshotEntry_get_data(rsnap);
|
||||
TEST_ASSERT_FALSE(out_sdata.p.type == CAPN_NULL);
|
||||
|
||||
capn_free(&d1);
|
||||
capn_free(&d2);
|
||||
free(buf1);
|
||||
free(buf2);
|
||||
printf("test_replay_header_snapshot_entry_round_trip: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Test 8: SessionSummary (fully-real, all-scalar struct) round-trips. */
|
||||
/* Bonus coverage beyond task 7.1's required Observable -- proves the */
|
||||
/* rest of the new schema surface (ReplaySession/ReplayDecision/ */
|
||||
/* ReplayEvent) actually compiles and links, not just SnapshotEntry. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static void test_session_summary_round_trip(void)
|
||||
{
|
||||
struct capn c;
|
||||
capn_init_malloc(&c);
|
||||
struct capn_segment *cs = capn_root(&c).seg;
|
||||
|
||||
SessionSummary_ptr summary = new_SessionSummary(cs);
|
||||
SessionSummary_set_durationNs(summary, 5000000000ULL);
|
||||
SessionSummary_set_nDecisions(summary, 42u);
|
||||
SessionSummary_set_nDispatched(summary, 40u);
|
||||
SessionSummary_set_nPended(summary, 2u);
|
||||
|
||||
ReplayDecision_list decisions = new_ReplayDecision_list(cs, 1);
|
||||
set_ReplayDecision(&(struct ReplayDecision){ .ts = 111ULL }, decisions, 0);
|
||||
|
||||
ReplaySession_ptr sess = new_ReplaySession(cs);
|
||||
ReplaySession_set_beginRound(sess, 7ULL);
|
||||
ReplaySession_set_decisions(sess, decisions);
|
||||
ReplaySession_set_summary(sess, summary);
|
||||
|
||||
capn_setp(capn_root(&c), 0, sess.p);
|
||||
int64_t sz = capn_size(&c);
|
||||
TEST_ASSERT_TRUE(sz > 0);
|
||||
uint8_t *buf = (uint8_t *)malloc((size_t)sz);
|
||||
TEST_ASSERT_NOT_NULL(buf);
|
||||
capn_write_mem(&c, buf, (size_t)sz, 0);
|
||||
capn_free(&c);
|
||||
|
||||
struct capn d;
|
||||
TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0));
|
||||
ReplaySession_ptr rsess;
|
||||
rsess.p = capn_getp(capn_root(&d), 0, 1);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(7ULL, ReplaySession_get_beginRound(rsess));
|
||||
SessionSummary_ptr out_summary = ReplaySession_get_summary(rsess);
|
||||
/* See the matching comment in the ReplayHeader test above: a nested
|
||||
* struct pointer returned from a generated X_get_Y() accessor may be an
|
||||
* unresolved far pointer and must be resolved before its own per-field
|
||||
* accessors (capn_read*-based) are used. */
|
||||
capn_resolve(&out_summary.p);
|
||||
TEST_ASSERT_EQUAL_UINT64(5000000000ULL,
|
||||
SessionSummary_get_durationNs(out_summary));
|
||||
TEST_ASSERT_EQUAL_UINT32(42u, SessionSummary_get_nDecisions(out_summary));
|
||||
TEST_ASSERT_EQUAL_UINT32(40u, SessionSummary_get_nDispatched(out_summary));
|
||||
TEST_ASSERT_EQUAL_UINT32(2u, SessionSummary_get_nPended(out_summary));
|
||||
|
||||
ReplayDecision_list out_decisions = ReplaySession_get_decisions(rsess);
|
||||
TEST_ASSERT_EQUAL_INT(1, capn_len(out_decisions));
|
||||
struct ReplayDecision out_dec;
|
||||
get_ReplayDecision(&out_dec, out_decisions, 0);
|
||||
TEST_ASSERT_EQUAL_UINT64(111ULL, out_dec.ts);
|
||||
|
||||
capn_free(&d);
|
||||
free(buf);
|
||||
printf("test_session_summary_round_trip: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Suite entry point */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
|
@ -346,4 +560,6 @@ void test_e2e_capnp(void)
|
|||
RUN_TEST(test_request_vote_last_log_tuple_round_trip);
|
||||
RUN_TEST(test_wal_catchup_append_term_fields_round_trip);
|
||||
RUN_TEST(test_data_job_array_variant_round_trip);
|
||||
RUN_TEST(test_replay_header_snapshot_entry_round_trip);
|
||||
RUN_TEST(test_session_summary_round_trip);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue