lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_e2e_capnp.c
Rongsong Shen 5cd8687b32
feat(scheduler-core): add scheduler admin opcode schema (task 7.4) 
Adds SchedAdminAction (recordStart/recordStop/recordStatus),
SchedAdminRequest (action/recordFile), SchedAdminReply (ok/error/
recording/file/eventCount/startedAt) to controller.capnp, matching
requirements.md Req 5 AC 10 verbatim. Adds schedAdmin@42 to OPCode,
schedadminmsg@31 to Request.body, schedadminreplymsg@19 to
RequestReply.data -- the exact next-available ordinals, verified live
before implementation and re-verified byte-identical against every
existing value/member.

Deliberately schema-only, per this task's own scope -- the controller
dispatch logic (cluster.c's case OPCode_schedAdmin:) is task 10.1's
job. cluster.c, src/frontend/, src/xceadmin/, src/scheduler/ all
untouched.

Round-trip tests cover all 3 SchedAdminAction values and both
ok=true/ok=false SchedAdminReply cases, including an asymmetric
ok=false/recording=true combination that rules out a Bool-field
bit-packing mixup at the wire level.

Closes out Group 7 (Foundation: Replay Wire Format) entirely.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
2026-08-04 14:00:16 +00:00

1524 lines
61 KiB
C
Raw Blame History

/* P0-15e: Cap'n Proto partial merge semantics
*
* Verifies that the three persistence Cap'n Proto mapped types
* (capnp_persist_cfg_t, capnp_snap_hdr_t, capnp_repl_envelope_t)
* and the generated struct PersistConfig / SnapshotHeader / ReplEnvelope
* round-trip through their C struct representations correctly.
*
* These are structural/compile-time + value round-trip tests that span
* the Cap'n Proto schema generation (P0-C) and the typemap.h definitions.
*/
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include <capnp_c.h>
#include "config.h"
#include "controller.capnp.h"
#include "persist.h"
/* ------------------------------------------------------------------ */
/* Helpers: persist_config_t → capnp_persist_cfg_t round-trip */
/* ------------------------------------------------------------------ */
/* Manually map persist_config_t fields to capnp_persist_cfg_t, matching the
* field mapping in the persist.c checkpoint serialisation path. */
static capnp_persist_cfg_t cfg_to_capnp(const persist_config_t* c)
{
capnp_persist_cfg_t out;
memset(&out, 0, sizeof(out));
out.walSyncMs = c->wal_sync_ms;
out.snapIntervalS = c->snap_interval_s;
out.snapWalSizeMb = c->snap_wal_size_mb;
out.snapKeepN = c->snap_keep_n;
out.transientJobTimeoutS = c->transient_job_timeout_s;
out.finishedRetainS = c->finished_retain_s;
out.workerRetainS = c->worker_retain_s;
out.syncTimeoutMs = c->sync_timeout_ms;
return out;
}
/* ------------------------------------------------------------------ */
/* Test 1: persist_config_t → capnp_persist_cfg_t round-trip */
/* ------------------------------------------------------------------ */
static void test_persist_config_capnp_round_trip(void)
{
persist_config_t src = persist_config_defaults();
/* Override to known values for deterministic checking */
src.wal_sync_ms = 42u;
src.snap_interval_s = 120u;
src.snap_wal_size_mb = 128u;
src.snap_keep_n = 3u;
src.transient_job_timeout_s = 600u;
src.finished_retain_s = 7200u;
src.worker_retain_s = 172800u;
src.sync_timeout_ms = 100u;
capnp_persist_cfg_t out = cfg_to_capnp(&src);
TEST_ASSERT_EQUAL_UINT32(src.wal_sync_ms, out.walSyncMs);
TEST_ASSERT_EQUAL_UINT32(src.snap_interval_s, out.snapIntervalS);
TEST_ASSERT_EQUAL_UINT32(src.snap_wal_size_mb, out.snapWalSizeMb);
TEST_ASSERT_EQUAL_UINT32(src.snap_keep_n, out.snapKeepN);
TEST_ASSERT_EQUAL_UINT32(src.transient_job_timeout_s, out.transientJobTimeoutS);
TEST_ASSERT_EQUAL_UINT32(src.finished_retain_s, out.finishedRetainS);
TEST_ASSERT_EQUAL_UINT32(src.worker_retain_s, out.workerRetainS);
TEST_ASSERT_EQUAL_UINT32(src.sync_timeout_ms, out.syncTimeoutMs);
printf("test_persist_config_capnp_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 2: SnapshotHeader struct fields and version default */
/* ------------------------------------------------------------------ */
static void test_snapshot_header_capnp_fields(void)
{
/* Verify struct size constant matches the documented schema */
TEST_ASSERT_EQUAL_INT(48, (int)SnapshotHeader_struct_bytes_count);
/* Populate a capnp_snap_hdr_t with known values and check them */
capnp_snap_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = 12345ULL;
hdr.term = 2ULL;
hdr.timestamp = 999999ULL;
hdr.version = 1u;
hdr.nJobs = 10u;
hdr.nQueues = 3u;
hdr.nWorkers = 5u;
hdr.checksum = 0xDEADBEEFu;
hdr.headerChecksum = 0xCAFEBABEu;
TEST_ASSERT_EQUAL_UINT64(12345ULL, hdr.lsn);
TEST_ASSERT_EQUAL_UINT64(2ULL, hdr.term);
TEST_ASSERT_EQUAL_UINT64(999999ULL, hdr.timestamp);
TEST_ASSERT_EQUAL_UINT32(1u, hdr.version);
TEST_ASSERT_EQUAL_UINT32(10u, hdr.nJobs);
TEST_ASSERT_EQUAL_UINT32(3u, hdr.nQueues);
TEST_ASSERT_EQUAL_UINT32(5u, hdr.nWorkers);
TEST_ASSERT_EQUAL_UINT32(0xDEADBEEFu, hdr.checksum);
TEST_ASSERT_EQUAL_UINT32(0xCAFEBABEu, hdr.headerChecksum);
printf("test_snapshot_header_capnp_fields: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 3: ReplEnvelope struct fields accessible */
/* ------------------------------------------------------------------ */
static void test_repl_envelope_capnp_fields(void)
{
/* Verify struct size constant */
TEST_ASSERT_EQUAL_INT(40, (int)ReplEnvelope_struct_bytes_count);
capnp_repl_envelope_t env;
memset(&env, 0, sizeof(env));
env.term = 7ULL;
env.lsn = 100ULL;
env.entryType = (uint32_t)WAL_JOB_SUBMIT;
env.payload = NULL; /* optional data pointer */
env.crc = 0x12345678u;
env.prevCrc = WAL_CRC_NONE;
TEST_ASSERT_EQUAL_UINT64(7ULL, env.term);
TEST_ASSERT_EQUAL_UINT64(100ULL, env.lsn);
TEST_ASSERT_EQUAL_UINT32(WAL_JOB_SUBMIT, env.entryType);
TEST_ASSERT_EQUAL_UINT32(0x12345678u, env.crc);
TEST_ASSERT_EQUAL_UINT32(WAL_CRC_NONE, env.prevCrc);
printf("test_repl_envelope_capnp_fields: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 4: RequestVoteRequest carries the Raft (lastLogTerm, */
/* lastLogIndex) tuple through a real encode → wire → decode round-trip */
/* (ha-term-safety task 1.2, Req 2.3/2.4/2.6). */
/* ------------------------------------------------------------------ */
static void test_request_vote_last_log_tuple_round_trip(void)
{
request_vote_req_t src;
memset(&src, 0, sizeof(src));
src.term = 5ULL;
src.candidate_id = "node-1";
src.config_tag = 9ULL;
src.last_log_lsn = 100ULL;
src.last_log_term = 4ULL; /* NEW: candidate's last-log term */
src.last_log_index = 100ULL; /* NEW: candidate's last-log index */
/* Encode the typemap struct into a capnp message, serialize to a flat
* buffer, then decode it back — exercises BOTH the schema field and the
* typemap.h mirror mapping (encode_/decode_ go through the C struct). */
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
/* encode_*_ptr allocates a struct in the segment and writes the C struct
* fields into it (via the $C.mapname mapping). */
RequestVoteRequest_ptr p;
encode_RequestVoteRequest_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
RequestVoteRequest_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct RequestVoteRequest wire;
read_RequestVoteRequest(&wire, dp);
request_vote_req_t dst;
memset(&dst, 0, sizeof(dst));
decode_RequestVoteRequest(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(5ULL, dst.term);
TEST_ASSERT_EQUAL_UINT64(9ULL, dst.config_tag);
TEST_ASSERT_EQUAL_UINT64(100ULL, dst.last_log_lsn);
TEST_ASSERT_EQUAL_UINT64(4ULL, dst.last_log_term); /* survives the wire */
TEST_ASSERT_EQUAL_UINT64(100ULL, dst.last_log_index); /* survives the wire */
capn_free(&d);
free(buf);
printf("test_request_vote_last_log_tuple_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 5: WalCatchupEntry (replication envelope) round-trips the */
/* AppendEntries term fields (prev_log_index, prev_log_term, term) */
/* (ha-term-safety task 1.1, Req 1.3/1.4). */
/* ------------------------------------------------------------------ */
static void test_wal_catchup_append_term_fields_round_trip(void)
{
wal_catchup_entry_t src;
memset(&src, 0, sizeof(src));
src.lsn = 42ULL;
src.entry_type = (uint8_t)WAL_JOB_SUBMIT;
src.term = 7ULL; /* NEW: this entry's term stamp */
src.prev_log_index = 41ULL; /* NEW: lsn - 1 */
src.prev_log_term = 6ULL; /* NEW: term at prev_log_index */
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
WalCatchupEntry_ptr p;
encode_WalCatchupEntry_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
WalCatchupEntry_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct WalCatchupEntry wire;
read_WalCatchupEntry(&wire, dp);
wal_catchup_entry_t dst;
memset(&dst, 0, sizeof(dst));
decode_WalCatchupEntry(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(42ULL, dst.lsn);
TEST_ASSERT_EQUAL_UINT64(7ULL, dst.term); /* survives the wire */
TEST_ASSERT_EQUAL_UINT64(41ULL, dst.prev_log_index); /* survives the wire */
TEST_ASSERT_EQUAL_UINT64(6ULL, dst.prev_log_term); /* survives the wire */
capn_free(&d);
free(buf);
printf("test_wal_catchup_append_term_fields_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 6: Data union's new jobArray variant (job-array-single-wal-record */
/* task 1.2, Req 6.1/6.2) round-trips the compact (base jobid, array size, */
/* one shared spec) push descriptor through the wire -- reusing the client */
/* submit request's existing JobSubmit shape verbatim (no new struct). */
/* This only proves the additive schema variant exists and round-trips; */
/* nothing calls it yet (wiring into sched/apiserver push is task 5.1/5.2). */
/* ------------------------------------------------------------------ */
static void test_data_job_array_variant_round_trip(void)
{
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = "batch";
submit.user = "rshen";
submit.mintasks = 1u;
submit.maxtasks = 4u;
submit.start_info = NULL;
submit.job_name = "array-job";
submit.project_name = "proj";
submit.user_group = "grp";
submit.job_description = "job-array-single-wal-record round-trip test";
submit.resreq = "select=1";
xce_job_submit_t src;
memset(&src, 0, sizeof(src));
src.jobid = 5000ULL; /* base jobid */
src.array_size = 250ULL;
src.submit = &submit;
xce_data_t data;
memset(&data, 0, sizeof(data));
data.table = XceData_data_jobArray;
data.data.job_array = &src;
/* Encode -> flat wire buffer -> decode, exercising both the schema
* field (jobArray @6) and the typemap.h mirror mapping (xce_data_t's
* new job_array union member), the same pattern Test 4/5 use. */
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
XceData_ptr p;
encode_XceData_ptr(cs, &p, &data);
capn_setp(capn_root(&c), 0, p.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));
XceData_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct XceData wire;
read_XceData(&wire, dp);
xce_data_t dst;
memset(&dst, 0, sizeof(dst));
decode_XceData(&dst, &wire);
TEST_ASSERT_EQUAL_INT(XceData_data_jobArray, dst.table);
TEST_ASSERT_NOT_NULL(dst.data.job_array);
TEST_ASSERT_EQUAL_UINT64(5000ULL, dst.data.job_array->jobid);
TEST_ASSERT_EQUAL_UINT64(250ULL, dst.data.job_array->array_size);
TEST_ASSERT_NOT_NULL(dst.data.job_array->submit);
TEST_ASSERT_EQUAL_STRING("batch", dst.data.job_array->submit->queue);
TEST_ASSERT_EQUAL_STRING("rshen", dst.data.job_array->submit->user);
TEST_ASSERT_EQUAL_UINT32(1u, dst.data.job_array->submit->mintasks);
TEST_ASSERT_EQUAL_UINT32(4u, dst.data.job_array->submit->maxtasks);
TEST_ASSERT_EQUAL_STRING("array-job", dst.data.job_array->submit->job_name);
/* Untouched: the existing schedjobs/jobs/etc. wire layout is unaffected --
* spot-check the enum values these variants relied on before this change
* did not shift underneath them. */
TEST_ASSERT_EQUAL_INT(0, (int)XceData_data_schedjobs);
TEST_ASSERT_EQUAL_INT(2, (int)XceData_data_jobs);
TEST_ASSERT_EQUAL_INT(5, (int)XceData_data_resvals);
TEST_ASSERT_EQUAL_INT(6, (int)XceData_data_jobArray);
free_XceData(&dst);
capn_free(&d);
free(buf);
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");
}
/* ------------------------------------------------------------------ */
/* Test 9: JobStateChangeEvent round-trip (scheduler-core task 7.2, */
/* Req 5 AC 20) -- the first of the two genuinely new ReplayEvent.body */
/* payload types. Uses the encode_X/decode_X codec bridge (Test 6's */
/* pattern) since job_state_change_event_t is a real, fully-scalar C */
/* struct with no Data fields -- unlike the frame-only replay-envelope */
/* types in Tests 7-8, which have no live C struct yet. */
/* ------------------------------------------------------------------ */
static void test_job_state_change_event_round_trip(void)
{
job_state_change_event_t src;
memset(&src, 0, sizeof(src));
src.jobid = 42424ULL;
src.old_state = 2u;
src.new_state = 8u;
src.exit_code = -7;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
JobStateChangeEvent_ptr p;
encode_JobStateChangeEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
JobStateChangeEvent_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct JobStateChangeEvent wire;
read_JobStateChangeEvent(&wire, dp);
job_state_change_event_t dst;
memset(&dst, 0, sizeof(dst));
decode_JobStateChangeEvent(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(42424ULL, dst.jobid);
TEST_ASSERT_EQUAL_UINT32(2u, dst.old_state);
TEST_ASSERT_EQUAL_UINT32(8u, dst.new_state);
TEST_ASSERT_EQUAL_INT32(-7, dst.exit_code);
capn_free(&d);
free(buf);
printf("test_job_state_change_event_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 10: JobUpdateEvent round-trip (scheduler-core task 7.2, */
/* Req 5 AC 20) -- the second genuinely new ReplayEvent.body payload */
/* type. `job` embeds the FULL updated record by reusing the */
/* pre-existing Job/xce_job_t struct; populated non-trivially, including*/
/* nested submit/allocs/pidinfo/pend_reasons (incl. the HostAlloc/ */
/* ResAlloc list-of-pointers two levels down), to prove the whole tree */
/* round-trips through encode_Job_ptr/decode_Job_ptr -- the same nested-*/
/* pointer codec chain already proven for xce_job_submit_t->submit in */
/* Test 6, now exercised for Job's full field set. */
/* ------------------------------------------------------------------ */
static void test_job_update_event_round_trip(void)
{
/* Innermost: one HostAlloc with one ResAlloc. */
xce_alloc_t res_alloc;
memset(&res_alloc, 0, sizeof(res_alloc));
res_alloc.resname = "cpu";
res_alloc.reserve = "resv1";
res_alloc.instid = 42ULL;
res_alloc.value = 4.0;
res_alloc.from = 500ULL;
res_alloc.duration = 3600LL;
xce_alloc_t *res_allocs[1] = { &res_alloc };
xce_host_alloc_t host_alloc;
memset(&host_alloc, 0, sizeof(host_alloc));
host_alloc.host = "hostA";
host_alloc.nallocs = 1;
host_alloc.allocs = res_allocs;
xce_host_alloc_t *host_allocs[1] = { &host_alloc };
xce_job_alloc_t job_alloc;
memset(&job_alloc, 0, sizeof(job_alloc));
job_alloc.jobid = 8001ULL;
job_alloc.nhosts = 1;
job_alloc.allocs = host_allocs;
job_alloc.nres_shared = 0;
job_alloc.allocs_shared = NULL;
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = "normal";
submit.user = "alice";
submit.mintasks = 2u;
submit.maxtasks = 8u;
submit.start_info = NULL;
submit.job_name = "update-test-job";
submit.project_name = "proj-x";
submit.user_group = "grp-x";
submit.job_description = "job update event round-trip test";
submit.resreq = "select=2";
unsigned int pids[2] = { 111u, 222u };
unsigned int pgids[1] = { 333u };
job_pidinfo_t pidinfo;
memset(&pidinfo, 0, sizeof(pidinfo));
pidinfo.jobid = 8001ULL;
pidinfo.npids = 2;
pidinfo.pids = pids;
pidinfo.npgids = 1;
pidinfo.pgids = pgids;
pidinfo.exdata = "pid-exdata";
xce_job_preason_t pend_reasons;
memset(&pend_reasons, 0, sizeof(pend_reasons));
pend_reasons.main_reason = 7;
pend_reasons.nhost_reasons = 0;
pend_reasons.host_based_reasons = NULL;
pend_reasons.nres_reasons = 0;
pend_reasons.res_based_reasons = NULL;
xce_job_t job;
memset(&job, 0, sizeof(job));
job.jobid = 8001ULL;
job.submit = &submit;
job.sub_time = 1000;
job.queue_time = 2000;
job.start_time = 3000;
job.finish_time = 4000;
job.state = 5u;
job.allocs = &job_alloc;
job.run_time = 120;
job.pidinfo = &pidinfo;
job.pend_reasons = &pend_reasons;
job.transient_since = 9999ULL;
job_update_event_t src;
memset(&src, 0, sizeof(src));
src.jobid = 8001ULL;
src.changed_fields = 0x5u;
src.job = &job;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
JobUpdateEvent_ptr p;
encode_JobUpdateEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
JobUpdateEvent_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct JobUpdateEvent wire;
read_JobUpdateEvent(&wire, dp);
job_update_event_t dst;
memset(&dst, 0, sizeof(dst));
decode_JobUpdateEvent(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.jobid);
TEST_ASSERT_EQUAL_UINT32(0x5u, dst.changed_fields);
TEST_ASSERT_NOT_NULL(dst.job);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->jobid);
TEST_ASSERT_EQUAL_UINT64(1000ULL, (uint64_t)dst.job->sub_time);
TEST_ASSERT_EQUAL_UINT64(2000ULL, (uint64_t)dst.job->queue_time);
TEST_ASSERT_EQUAL_UINT64(3000ULL, (uint64_t)dst.job->start_time);
TEST_ASSERT_EQUAL_UINT64(4000ULL, (uint64_t)dst.job->finish_time);
TEST_ASSERT_EQUAL_UINT32(5u, dst.job->state);
TEST_ASSERT_EQUAL_INT(120, dst.job->run_time);
TEST_ASSERT_EQUAL_UINT64(9999ULL, dst.job->transient_since);
TEST_ASSERT_NOT_NULL(dst.job->submit);
TEST_ASSERT_EQUAL_STRING("normal", dst.job->submit->queue);
TEST_ASSERT_EQUAL_STRING("alice", dst.job->submit->user);
TEST_ASSERT_EQUAL_UINT32(2u, dst.job->submit->mintasks);
TEST_ASSERT_EQUAL_UINT32(8u, dst.job->submit->maxtasks);
TEST_ASSERT_EQUAL_STRING("update-test-job", dst.job->submit->job_name);
TEST_ASSERT_EQUAL_STRING("proj-x", dst.job->submit->project_name);
TEST_ASSERT_EQUAL_STRING("grp-x", dst.job->submit->user_group);
TEST_ASSERT_EQUAL_STRING("job update event round-trip test",
dst.job->submit->job_description);
TEST_ASSERT_EQUAL_STRING("select=2", dst.job->submit->resreq);
TEST_ASSERT_NOT_NULL(dst.job->allocs);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->allocs->jobid);
TEST_ASSERT_EQUAL_INT(1, dst.job->allocs->nhosts);
TEST_ASSERT_NOT_NULL(dst.job->allocs->allocs);
TEST_ASSERT_EQUAL_STRING("hostA", dst.job->allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_INT(1, dst.job->allocs->allocs[0]->nallocs);
TEST_ASSERT_NOT_NULL(dst.job->allocs->allocs[0]->allocs);
TEST_ASSERT_EQUAL_STRING("cpu",
dst.job->allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_STRING("resv1",
dst.job->allocs->allocs[0]->allocs[0]->reserve);
TEST_ASSERT_EQUAL_UINT64(42ULL,
dst.job->allocs->allocs[0]->allocs[0]->instid);
/* Unity is built without UNITY_INCLUDE_DOUBLE; compare via float cast
* (same pattern as scheduler/testsuite/test.c). */
TEST_ASSERT_EQUAL_FLOAT(4.0f,
(float)dst.job->allocs->allocs[0]->allocs[0]->value);
TEST_ASSERT_EQUAL_UINT64(500ULL,
dst.job->allocs->allocs[0]->allocs[0]->from);
TEST_ASSERT_EQUAL_INT64(3600LL,
dst.job->allocs->allocs[0]->allocs[0]->duration);
TEST_ASSERT_EQUAL_INT(0, dst.job->allocs->nres_shared);
TEST_ASSERT_NULL(dst.job->allocs->allocs_shared);
TEST_ASSERT_NOT_NULL(dst.job->pidinfo);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->pidinfo->jobid);
TEST_ASSERT_EQUAL_INT(2, dst.job->pidinfo->npids);
TEST_ASSERT_EQUAL_UINT32(111u, dst.job->pidinfo->pids[0]);
TEST_ASSERT_EQUAL_UINT32(222u, dst.job->pidinfo->pids[1]);
TEST_ASSERT_EQUAL_INT(1, dst.job->pidinfo->npgids);
TEST_ASSERT_EQUAL_UINT32(333u, dst.job->pidinfo->pgids[0]);
TEST_ASSERT_EQUAL_STRING("pid-exdata", dst.job->pidinfo->exdata);
TEST_ASSERT_NOT_NULL(dst.job->pend_reasons);
TEST_ASSERT_EQUAL_INT(7, dst.job->pend_reasons->main_reason);
TEST_ASSERT_EQUAL_INT(0, dst.job->pend_reasons->nhost_reasons);
TEST_ASSERT_NULL(dst.job->pend_reasons->host_based_reasons);
TEST_ASSERT_EQUAL_INT(0, dst.job->pend_reasons->nres_reasons);
TEST_ASSERT_NULL(dst.job->pend_reasons->res_based_reasons);
free_JobUpdateEvent(&dst);
capn_free(&d);
free(buf);
printf("test_job_update_event_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 11: DispatchDecision round-trip (scheduler-core task 7.3, Req 5 */
/* AC 5, AC 20) -- reuses JobAlloc verbatim (jobid lives at */
/* JobAlloc.jobid, no redundant top-level jobid field, per design */
/* review). Populates a real multi-level allocation tree (one host, one */
/* resource) via the JobAlloc/HostAlloc/ResAlloc codec chain already */
/* proven in Test 10. */
/* ------------------------------------------------------------------ */
static void test_dispatch_decision_round_trip(void)
{
xce_alloc_t res_alloc;
memset(&res_alloc, 0, sizeof(res_alloc));
res_alloc.resname = "gpu";
res_alloc.reserve = "resv2";
res_alloc.instid = 7ULL;
res_alloc.value = 2.0;
res_alloc.from = 100ULL;
res_alloc.duration = 1800LL;
xce_alloc_t *res_allocs[1] = { &res_alloc };
xce_host_alloc_t host_alloc;
memset(&host_alloc, 0, sizeof(host_alloc));
host_alloc.host = "hostB";
host_alloc.nallocs = 1;
host_alloc.allocs = res_allocs;
xce_host_alloc_t *host_allocs[1] = { &host_alloc };
xce_job_alloc_t job_alloc;
memset(&job_alloc, 0, sizeof(job_alloc));
job_alloc.jobid = 9001ULL;
job_alloc.nhosts = 1;
job_alloc.allocs = host_allocs;
job_alloc.nres_shared = 0;
job_alloc.allocs_shared = NULL;
sched_dispatch_decision_t src;
memset(&src, 0, sizeof(src));
src.alloc = &job_alloc;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
DispatchDecision_ptr p;
encode_DispatchDecision_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
DispatchDecision_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct DispatchDecision wire;
read_DispatchDecision(&wire, dp);
sched_dispatch_decision_t dst;
memset(&dst, 0, sizeof(dst));
decode_DispatchDecision(&dst, &wire);
TEST_ASSERT_NOT_NULL(dst.alloc);
TEST_ASSERT_EQUAL_UINT64(9001ULL, dst.alloc->jobid);
TEST_ASSERT_EQUAL_INT(1, dst.alloc->nhosts);
TEST_ASSERT_NOT_NULL(dst.alloc->allocs);
TEST_ASSERT_EQUAL_STRING("hostB", dst.alloc->allocs[0]->host);
TEST_ASSERT_EQUAL_INT(1, dst.alloc->allocs[0]->nallocs);
TEST_ASSERT_NOT_NULL(dst.alloc->allocs[0]->allocs);
TEST_ASSERT_EQUAL_STRING("gpu", dst.alloc->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_STRING("resv2", dst.alloc->allocs[0]->allocs[0]->reserve);
TEST_ASSERT_EQUAL_UINT64(7ULL, dst.alloc->allocs[0]->allocs[0]->instid);
TEST_ASSERT_EQUAL_FLOAT(2.0f, (float)dst.alloc->allocs[0]->allocs[0]->value);
TEST_ASSERT_EQUAL_UINT64(100ULL, dst.alloc->allocs[0]->allocs[0]->from);
TEST_ASSERT_EQUAL_INT64(1800LL, dst.alloc->allocs[0]->allocs[0]->duration);
TEST_ASSERT_EQUAL_INT(0, dst.alloc->nres_shared);
TEST_ASSERT_NULL(dst.alloc->allocs_shared);
free_DispatchDecision(&dst);
capn_free(&d);
free(buf);
printf("test_dispatch_decision_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 12: PendDecision round-trip (scheduler-core task 7.3, Req 5 */
/* AC 5, AC 20) -- reuses JobPendReason (xce_job_preason_t) verbatim. */
/* ------------------------------------------------------------------ */
static void test_pend_decision_round_trip(void)
{
xce_job_preason_t reason;
memset(&reason, 0, sizeof(reason));
reason.main_reason = 3;
reason.nhost_reasons = 0;
reason.host_based_reasons = NULL;
reason.nres_reasons = 0;
reason.res_based_reasons = NULL;
sched_pend_decision_t src;
memset(&src, 0, sizeof(src));
src.jobid = 4242ULL;
src.reason = &reason;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
PendDecision_ptr p;
encode_PendDecision_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
PendDecision_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct PendDecision wire;
read_PendDecision(&wire, dp);
sched_pend_decision_t dst;
memset(&dst, 0, sizeof(dst));
decode_PendDecision(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(4242ULL, dst.jobid);
TEST_ASSERT_NOT_NULL(dst.reason);
TEST_ASSERT_EQUAL_INT(3, dst.reason->main_reason);
TEST_ASSERT_EQUAL_INT(0, dst.reason->nhost_reasons);
TEST_ASSERT_NULL(dst.reason->host_based_reasons);
TEST_ASSERT_EQUAL_INT(0, dst.reason->nres_reasons);
TEST_ASSERT_NULL(dst.reason->res_based_reasons);
free_PendDecision(&dst);
capn_free(&d);
free(buf);
printf("test_pend_decision_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 13: PreemptDecision round-trip (scheduler-core task 7.3, Req 5 */
/* AC 5, AC 20) -- PreemptVictim is fresh (no pre-existing wire */
/* counterpart) and deliberately distinct from decision.h's */
/* core-internal xce_preempt_victim_t/xce_preempt_t (task 4.4's */
/* in-process live-decision representation). Populates a 2-entry victim */
/* list to exercise List(PreemptVictim) with real data. */
/* ------------------------------------------------------------------ */
static void test_preempt_decision_round_trip(void)
{
preempt_victim_t v1;
memset(&v1, 0, sizeof(v1));
v1.jobid = 111ULL;
v1.workerid = "hostX";
preempt_victim_t v2;
memset(&v2, 0, sizeof(v2));
v2.jobid = 222ULL;
v2.workerid = "hostY";
preempt_victim_t *victims[2] = { &v1, &v2 };
xce_job_preason_t reason;
memset(&reason, 0, sizeof(reason));
reason.main_reason = 9;
sched_preempt_decision_t src;
memset(&src, 0, sizeof(src));
src.jobid = 5555ULL;
src.victims = victims;
src.n_victims = 2;
src.reason = &reason;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
PreemptDecision_ptr p;
encode_PreemptDecision_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
PreemptDecision_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct PreemptDecision wire;
read_PreemptDecision(&wire, dp);
sched_preempt_decision_t dst;
memset(&dst, 0, sizeof(dst));
decode_PreemptDecision(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(5555ULL, dst.jobid);
TEST_ASSERT_EQUAL_INT(2, dst.n_victims);
TEST_ASSERT_NOT_NULL(dst.victims);
TEST_ASSERT_EQUAL_UINT64(111ULL, dst.victims[0]->jobid);
TEST_ASSERT_EQUAL_STRING("hostX", dst.victims[0]->workerid);
TEST_ASSERT_EQUAL_UINT64(222ULL, dst.victims[1]->jobid);
TEST_ASSERT_EQUAL_STRING("hostY", dst.victims[1]->workerid);
TEST_ASSERT_NOT_NULL(dst.reason);
TEST_ASSERT_EQUAL_INT(9, dst.reason->main_reason);
free_PreemptDecision(&dst);
capn_free(&d);
free(buf);
printf("test_preempt_decision_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 14: ReplayDecision's `body` union round-trips through the */
/* dispatch variant (scheduler-core task 7.3, Req 5 AC 20) -- proves */
/* the purely-additive union bolted onto task 7.1's ts@0-only stub */
/* actually wires up end to end (mirrors the JobControl/JoinData */
/* `:union` group precedent: the group name consumes no ordinal). */
/* ------------------------------------------------------------------ */
static void test_replay_decision_dispatch_union_round_trip(void)
{
xce_job_alloc_t job_alloc;
memset(&job_alloc, 0, sizeof(job_alloc));
job_alloc.jobid = 7777ULL;
job_alloc.nhosts = 0;
job_alloc.allocs = NULL;
job_alloc.nres_shared = 0;
job_alloc.allocs_shared = NULL;
sched_dispatch_decision_t dispatch;
memset(&dispatch, 0, sizeof(dispatch));
dispatch.alloc = &job_alloc;
replay_decision_t src;
memset(&src, 0, sizeof(src));
src.ts = 55555ULL;
src.body_which = ReplayDecision_body_dispatch;
src.body.dispatch = &dispatch;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
ReplayDecision_ptr p;
encode_ReplayDecision_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
ReplayDecision_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct ReplayDecision wire;
read_ReplayDecision(&wire, dp);
replay_decision_t dst;
memset(&dst, 0, sizeof(dst));
decode_ReplayDecision(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(55555ULL, dst.ts);
TEST_ASSERT_EQUAL_INT((int)ReplayDecision_body_dispatch, (int)dst.body_which);
TEST_ASSERT_NOT_NULL(dst.body.dispatch);
TEST_ASSERT_NOT_NULL(dst.body.dispatch->alloc);
TEST_ASSERT_EQUAL_UINT64(7777ULL, dst.body.dispatch->alloc->jobid);
free_ReplayDecision(&dst);
capn_free(&d);
free(buf);
printf("test_replay_decision_dispatch_union_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 15: SnapshotData round-trip (scheduler-core task 7.3, Req 5 */
/* AC 21, Req 6 AC 5) -- the full snapshot message, replacing task */
/* 7.1's zero-field placeholder wholesale. Populates all 11 fields with */
/* non-trivial data (including >=1 entry in each of the top-level */
/* List(...) fields: queues/users/projects) and round-trips them */
/* field-for-field through encode_SnapshotData_ptr/decode_SnapshotData */
/* -- EXCEPT internalState (a Data field), whose generated codec is a */
/* documented no-op (same limitation as every other Data field in this */
/* schema, e.g. ReplayHeader.configHash in Test 7) -- verified instead */
/* via the low-level SnapshotData_get/set_internalState accessor on the */
/* SAME struct pointer the codec bridge already populated, same hybrid */
/* technique used nowhere else in this file but safe because encode_/ */
/* set_ operate on independent field slots of one already-allocated */
/* capnp struct. */
/* */
/* Also demonstrates this task's internalVersion Observable: "a */
/* snapshot encoded with one internal-state version and decoded */
/* expecting a different version is rejected rather than */
/* misinterpreted." internalVersion is a schema-side format TAG only -- */
/* the real accept/reject decision is production logic for task 8.1+ to */
/* write; here we prove the tag survives the wire faithfully and that a */
/* simple equality check (the only sound way to interpret a version */
/* tag) distinguishes a matching reader from a mismatched one. The */
/* comparison below is TEST-ONLY logic illustrating the intended usage, */
/* not production code. */
/* ------------------------------------------------------------------ */
static void test_snapshot_data_round_trip(void)
{
/* One queue, reused both as SnapshotData's own top-level `queues`
* list and inside `config.queues` (Conf4Sched's own queue list) --
* proves both call sites of the pre-existing QueueConf codec. */
xce_queue_t queue;
memset(&queue, 0, sizeof(queue));
queue.queue = "batch";
queue.attrs = 1u;
queue.status = 0u;
queue.priority = 5;
queue.accept_interval = 10;
queue.sched_delay = 2;
queue.resreq = "select=1";
xce_queue_t *queues[1] = { &queue };
xce_conf4sched_t config;
memset(&config, 0, sizeof(config));
config.clustername = "xce-snap-cluster";
config.n_queues = 1;
config.queues = queues;
xce_job_t sjob;
memset(&sjob, 0, sizeof(sjob));
sjob.jobid = 8001ULL;
sjob.sub_time = 1000;
sjob.queue_time = 2000;
sjob.start_time = 3000;
sjob.finish_time = 4000;
sjob.state = 5u;
sjob.run_time = 120;
xce_job_t *sjobs[1] = { &sjob };
xce_jobs_data_t jobs_data;
memset(&jobs_data, 0, sizeof(jobs_data));
jobs_data.n_total = 1u;
jobs_data.pos = 0u;
jobs_data.count = 1;
jobs_data.jobs = sjobs;
xce_worker_t worker;
memset(&worker, 0, sizeof(worker));
worker.name = "workerA";
worker.status = XCE_WORKER_UNAVAIL;
xce_worker_t *workers_arr[1] = { &worker };
xce_workers_data_t workers_data;
memset(&workers_data, 0, sizeof(workers_data));
workers_data.n_total = 1u;
workers_data.pos = 0u;
workers_data.count = 1;
workers_data.workers = workers_arr;
xce_resource_t resource;
memset(&resource, 0, sizeof(resource));
resource.name = "cpu-res";
resource.desc = "CPU resource";
resource.idx = 0;
resource.valtype = 1;
resource.ordtype = XCE_ORD_INCR;
resource.flags = 0;
resource.interval = 60;
resource.last_report = 999;
xce_resource_t *resources_arr[1] = { &resource };
xce_resource_data_t resources_data;
memset(&resources_data, 0, sizeof(resources_data));
resources_data.n_total = 1u;
resources_data.pos = 0u;
resources_data.count = 1;
resources_data.resources = resources_arr;
xce_val_t val;
memset(&val, 0, sizeof(val));
val.which = 1; /* uvalue32 */
val.u.uvalue32 = 4096u;
xce_resval_t resval;
memset(&resval, 0, sizeof(resval));
resval.name = "mem";
resval.host = "hostA";
resval.val = &val;
xce_resval_t *resvals_arr[1] = { &resval };
xce_resval_list_t resval_list;
memset(&resval_list, 0, sizeof(resval_list));
resval_list.count = 1;
resval_list.vals = resvals_arr;
sched_snapshot_user_t user;
memset(&user, 0, sizeof(user));
user.name = "alice";
sched_snapshot_user_t *users_arr[1] = { &user };
sched_snapshot_project_t project;
memset(&project, 0, sizeof(project));
project.name = "proj1";
project.priority = 5;
sched_snapshot_project_t *projects_arr[1] = { &project };
uint8_t internal_bytes[8] = { 0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04 };
sched_snapshot_data_t src;
memset(&src, 0, sizeof(src));
src.config = &config;
src.jobs = &jobs_data;
src.queues = queues;
src.n_queues = 1;
src.workers = &workers_data;
src.resources = &resources_data;
src.resvals = &resval_list;
src.users = users_arr;
src.n_users = 1;
src.projects = projects_arr;
src.n_projects = 1;
src.prng_seed = 123456789ULL;
src.internal_state = internal_bytes;
src.internal_state_len = sizeof(internal_bytes);
src.internal_version = 2u;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
SnapshotData_ptr p;
encode_SnapshotData_ptr(cs, &p, &src);
/* internalState (Data field): codec no-op, set via low-level accessor
* on the same already-allocated struct pointer (see comment above). */
capn_list8 state_list = capn_new_list8(cs, (int)sizeof(internal_bytes));
capn_setv8(state_list, 0, internal_bytes, sizeof(internal_bytes));
capn_data state_data;
state_data.p = state_list.p;
SnapshotData_set_internalState(p, state_data);
capn_setp(capn_root(&c), 0, p.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));
SnapshotData_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct SnapshotData wire;
read_SnapshotData(&wire, dp);
sched_snapshot_data_t dst;
memset(&dst, 0, sizeof(dst));
decode_SnapshotData(&dst, &wire);
/* config (Conf4Sched) */
TEST_ASSERT_NOT_NULL(dst.config);
TEST_ASSERT_EQUAL_STRING("xce-snap-cluster", dst.config->clustername);
TEST_ASSERT_EQUAL_INT(1, dst.config->n_queues);
TEST_ASSERT_NOT_NULL(dst.config->queues);
TEST_ASSERT_EQUAL_STRING("batch", dst.config->queues[0]->queue);
TEST_ASSERT_EQUAL_UINT32(1u, dst.config->queues[0]->attrs);
TEST_ASSERT_EQUAL_INT32(5, dst.config->queues[0]->priority);
TEST_ASSERT_EQUAL_STRING("select=1", dst.config->queues[0]->resreq);
/* jobs (SchedJobData) */
TEST_ASSERT_NOT_NULL(dst.jobs);
TEST_ASSERT_EQUAL_UINT32(1u, dst.jobs->n_total);
TEST_ASSERT_EQUAL_INT(1, dst.jobs->count);
TEST_ASSERT_NOT_NULL(dst.jobs->jobs);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.jobs->jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT32(5u, dst.jobs->jobs[0]->state);
TEST_ASSERT_EQUAL_INT(120, dst.jobs->jobs[0]->run_time);
/* queues (top-level List(QueueConf)) */
TEST_ASSERT_EQUAL_INT(1, dst.n_queues);
TEST_ASSERT_NOT_NULL(dst.queues);
TEST_ASSERT_EQUAL_STRING("batch", dst.queues[0]->queue);
TEST_ASSERT_EQUAL_INT32(5, dst.queues[0]->priority);
/* workers (WorkerData) */
TEST_ASSERT_NOT_NULL(dst.workers);
TEST_ASSERT_EQUAL_UINT32(1u, dst.workers->n_total);
TEST_ASSERT_NOT_NULL(dst.workers->workers);
TEST_ASSERT_EQUAL_STRING("workerA", dst.workers->workers[0]->name);
TEST_ASSERT_EQUAL_UINT32((uint32_t)XCE_WORKER_UNAVAIL, dst.workers->workers[0]->status);
/* resources (ResourceData) */
TEST_ASSERT_NOT_NULL(dst.resources);
TEST_ASSERT_EQUAL_UINT32(1u, dst.resources->n_total);
TEST_ASSERT_NOT_NULL(dst.resources->resources);
TEST_ASSERT_EQUAL_STRING("cpu-res", dst.resources->resources[0]->name);
TEST_ASSERT_EQUAL_STRING("CPU resource", dst.resources->resources[0]->desc);
/* resvals (ResValList) */
TEST_ASSERT_NOT_NULL(dst.resvals);
TEST_ASSERT_EQUAL_INT(1, dst.resvals->count);
TEST_ASSERT_NOT_NULL(dst.resvals->vals);
TEST_ASSERT_EQUAL_STRING("mem", dst.resvals->vals[0]->name);
TEST_ASSERT_EQUAL_STRING("hostA", dst.resvals->vals[0]->host);
TEST_ASSERT_NOT_NULL(dst.resvals->vals[0]->val);
TEST_ASSERT_EQUAL_UINT32(4096u, dst.resvals->vals[0]->val->u.uvalue32);
/* users / projects (top-level List(SchedUser)/List(SchedProject)) */
TEST_ASSERT_EQUAL_INT(1, dst.n_users);
TEST_ASSERT_NOT_NULL(dst.users);
TEST_ASSERT_EQUAL_STRING("alice", dst.users[0]->name);
TEST_ASSERT_EQUAL_INT(1, dst.n_projects);
TEST_ASSERT_NOT_NULL(dst.projects);
TEST_ASSERT_EQUAL_STRING("proj1", dst.projects[0]->name);
TEST_ASSERT_EQUAL_INT32(5, dst.projects[0]->priority);
/* prngSeed / internalVersion (plain scalars, codec-bridge tested) */
TEST_ASSERT_EQUAL_UINT64(123456789ULL, dst.prng_seed);
TEST_ASSERT_EQUAL_UINT32(2u, dst.internal_version);
/* internalState -- low-level accessor read-back (Data field, codec
* no-op; see comment above). */
capn_data out_state = SnapshotData_get_internalState(dp);
TEST_ASSERT_EQUAL_INT((int)sizeof(internal_bytes),
capn_len((capn_list8){ out_state.p }));
uint8_t out_state_bytes[sizeof(internal_bytes)];
TEST_ASSERT_EQUAL_INT((int)sizeof(internal_bytes),
capn_getv8((capn_list8){ out_state.p }, 0,
out_state_bytes, sizeof(internal_bytes)));
TEST_ASSERT_EQUAL_MEMORY(internal_bytes, out_state_bytes, sizeof(internal_bytes));
/* Observable: version-mismatch rejection. A reader that expects the
* SAME internal_version accepts; a reader that expects a DIFFERENT
* version must reject rather than misinterpret the opaque bytes. This
* is the only sound reject/accept rule for an opaque versioned blob --
* demonstrated here with test-only comparison logic, not production
* code (task 8.1+ writes the real load_snapshot gate). */
const uint32_t matching_reader_expects = 2u;
const uint32_t mismatched_reader_expects = 3u;
TEST_ASSERT_TRUE(dst.internal_version == matching_reader_expects);
TEST_ASSERT_FALSE(dst.internal_version == mismatched_reader_expects);
free_SnapshotData(&dst);
capn_free(&d);
free(buf);
printf("test_snapshot_data_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 16: SchedAdminRequest round-trip (scheduler-core task 7.4, */
/* Req 5 AC 10) -- encodes and decodes a SchedAdminRequest for each of */
/* the three SchedAdminAction enum values (recordStart/recordStop/ */
/* recordStatus), proving the plain enum-typed `action` field (mapped */
/* to `int` in sched_admin_req_t, same convention as HostRole/ */
/* JoinData.role) and the `recordFile` Text field round-trip field-for- */
/* field. recordStart uses a non-empty recordFile; recordStop/ */
/* recordStatus use an empty one (auto-generate / pure query), covering */
/* both the populated and empty-string Text cases. Same encode_X_ptr/ */
/* decode_X codec-bridge pattern as Test 9 (job_state_change_event_t) */
/* since sched_admin_req_t is a fully-scalar struct plus one Text */
/* field, no Data fields. */
/* ------------------------------------------------------------------ */
static void sched_admin_request_round_trip_case(int action, const char *record_file)
{
sched_admin_req_t src;
memset(&src, 0, sizeof(src));
src.action = action;
src.recordFile = (char *)record_file;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
SchedAdminRequest_ptr p;
encode_SchedAdminRequest_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
SchedAdminRequest_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct SchedAdminRequest wire;
read_SchedAdminRequest(&wire, dp);
sched_admin_req_t dst;
memset(&dst, 0, sizeof(dst));
decode_SchedAdminRequest(&dst, &wire);
TEST_ASSERT_EQUAL_INT(action, (int)dst.action);
TEST_ASSERT_EQUAL_STRING(record_file, dst.recordFile);
free_SchedAdminRequest(&dst);
capn_free(&d);
free(buf);
}
static void test_sched_admin_request_round_trip(void)
{
/* recordStart with a non-empty recordFile. */
sched_admin_request_round_trip_case(SchedAdminAction_recordStart,
"manual-session.replay");
/* recordStop with an empty recordFile (stop takes no filename). */
sched_admin_request_round_trip_case(SchedAdminAction_recordStop, "");
/* recordStatus with an empty recordFile (status is a pure query). */
sched_admin_request_round_trip_case(SchedAdminAction_recordStatus, "");
printf("test_sched_admin_request_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 17: SchedAdminReply round-trip (scheduler-core task 7.4, */
/* Req 5 AC 10) -- encodes and decodes a SchedAdminReply for both an */
/* ok=true success case and an ok=false error case, each populating all */
/* six fields (ok, error, recording, file, eventCount, startedAt) with */
/* non-default values, proving every field -- including the two Bool */
/* fields (ok, recording) packed into the same wire byte -- round-trips */
/* independently regardless of the other's value. `error` is modelled */
/* as a general diagnostic-message field (populated with a non-fatal */
/* advisory even on success, an outright failure reason on error) so */
/* both cases can legitimately carry a non-empty value. Same codec- */
/* bridge pattern as Test 9/16. */
/* ------------------------------------------------------------------ */
static void sched_admin_reply_round_trip_case(int ok, const char *error,
int recording, const char *file,
uint64_t event_count,
uint64_t started_at)
{
sched_admin_reply_t src;
memset(&src, 0, sizeof(src));
src.ok = ok;
src.error = (char *)error;
src.recording = recording;
src.file = (char *)file;
src.eventCount = event_count;
src.startedAt = started_at;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
SchedAdminReply_ptr p;
encode_SchedAdminReply_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.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));
SchedAdminReply_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct SchedAdminReply wire;
read_SchedAdminReply(&wire, dp);
sched_admin_reply_t dst;
memset(&dst, 0, sizeof(dst));
decode_SchedAdminReply(&dst, &wire);
TEST_ASSERT_EQUAL_INT(ok ? 1 : 0, dst.ok ? 1 : 0);
TEST_ASSERT_EQUAL_STRING(error, dst.error);
TEST_ASSERT_EQUAL_INT(recording ? 1 : 0, dst.recording ? 1 : 0);
TEST_ASSERT_EQUAL_STRING(file, dst.file);
TEST_ASSERT_EQUAL_UINT64(event_count, dst.eventCount);
TEST_ASSERT_EQUAL_UINT64(started_at, dst.startedAt);
free_SchedAdminReply(&dst);
capn_free(&d);
free(buf);
}
static void test_sched_admin_reply_round_trip(void)
{
/* ok=true success case (e.g. a recordStatus reply): all six fields
* non-default. */
sched_admin_reply_round_trip_case(1, "warning: recording directory nearing disk quota",
1, "session-0007.replay",
12345ULL, 1700000000000000000ULL);
/* ok=false error case (e.g. a failed recordStart): all six fields
* still non-default, proving the wire carries every field regardless
* of `ok`. */
sched_admin_reply_round_trip_case(0, "recording directory not writable",
1, "session-0006.replay",
999ULL, 1699999999000000000ULL);
printf("test_sched_admin_reply_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
void test_e2e_capnp(void)
{
UnitySetTestFile(__FILE__);
printf("\n>> P0-15e: Cap'n Proto partial merge semantics\n\n");
RUN_TEST(test_persist_config_capnp_round_trip);
RUN_TEST(test_snapshot_header_capnp_fields);
RUN_TEST(test_repl_envelope_capnp_fields);
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);
RUN_TEST(test_job_state_change_event_round_trip);
RUN_TEST(test_job_update_event_round_trip);
RUN_TEST(test_dispatch_decision_round_trip);
RUN_TEST(test_pend_decision_round_trip);
RUN_TEST(test_preempt_decision_round_trip);
RUN_TEST(test_replay_decision_dispatch_union_round_trip);
RUN_TEST(test_snapshot_data_round_trip);
RUN_TEST(test_sched_admin_request_round_trip);
RUN_TEST(test_sched_admin_reply_round_trip);
}
Reference in a new issue View git blame Copy permalink