lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_commit_barrier_rollout.c
Rongsong Shen a81f2a249c fix(controller-tests): make Unity test-file attribution and counters consistent suite-wide 
Only 21 of 86 test_*.c group functions called their own nested
UNITY_BEGIN()/UNITY_END(). Each nested call reset Unity's global
NumberOfTests/TestFailures counters, so controller_test.c's final
`return UNITY_END();` -- the value meson test actually gates on --
only ever reflected whichever small window of tests ran since the
LAST nested reset, silently discarding pass/fail evidence for
everything before it (measured: 16 of 972 total tests in one run).
The same nested-BEGIN calls also caused PASS/FAIL banners to show
the wrong source file for tests in files that don't reset it
themselves (Unity.TestFile just carries over from whichever group
last called UNITY_BEGIN()).

Fix: replace every nested UNITY_BEGIN() with UnitySetTestFile(__FILE__)
(sets file attribution only, no counter reset) and drop the matching
UNITY_END(); add the same UnitySetTestFile(__FILE__) as the first
statement to the other group functions that had neither. The 2
genuinely standalone binaries (test_job_array_sched_push.c,
test_job_array_apiserver_push.c -- each has its own int main(), not
registered via controller_test.c's TEST() macro) are untouched.

Verified: single consolidated summary now reads
"972 Tests 0 Failures 0 Ignored OK", exactly matching the
ground-truth RUN_TEST call-site count across all 83 registered
group files (976 total minus 4 belonging to the 2 standalone
binaries). Every PASS/FAIL banner now names its own true source
file. Full xce-build.sh all regression pass clean (Ok: 9, Fail: 0).

No production code touched -- test-harness plumbing only.
2026-07-17 14:49:12 +00:00

2369 lines
100 KiB
C
Raw Blame History

/*
* test_commit_barrier_rollout.c — RED-phase tests for write-path commit
* classification (ha-commit-barrier-rollout task 1.1).
*
* Covers Requirements 2.1 (every state-changing WAL type explicitly
* classified WAIT/RETAIN/LOCAL), 2.2 (unclassified type fails fast under
* XCE_TESTING), and 3.1 (fact-shaped appends are retained — appended,
* non-blocking, pending purely by lsn > committed_lsn, committed later
* by covering ACKs without redelivery).
*
* The classification table under test is the VERIFIED table from
* .kiro/specs/ha-commit-barrier-rollout/design.md §Write-Path
* Classification (it supersedes the umbrella draft table — research.md §2
* records the call-site-verified deviations: WAL_JOB_STOP/RESUME are
* fact-origin only; WAL_JOB_UPDATE/WAL_JOB_FAIL are dual-origin and
* classify RETAIN at the seam, with the request-shaped guarantee realized
* by the handler barrier in do_jobcontrol_op — tasks 2.x/3.x):
*
* WAL_JOB_SUBMIT request WAIT
* WAL_QUEUE_CTRL request WAIT
* WAL_WORKER_CTRL + JOIN/UPDATE request WAIT
* WAL_WORKER_CTRL + DEAD fact RETAIN
* WAL_JOB_START/FINISH/FAIL/STOP/
* RESUME/UPDATE/RUSAGE/CKPT/
* PREASON fact RETAIN
* WAL_CHECKPOINT local LOCAL (never replicated)
* unknown — XCE_TESTING assert
*
* These cases FAIL (RED) against the task-1.1 inert wal_classify stub in
* persist.c (returns an invalid sentinel mode) and go green with the
* task-1.2 implementation.
*
* Honest-observable note for the RETAIN-registration case (task brief):
* commit_submit(COMMIT_MODE_RETAIN) is registration-free bookkeeping
* (commit.c — returns 0 immediately; commit state is derived entirely
* from registry watermarks at ACK time), so RETAIN routing adds NO
* engine-side observable distinct from the append itself. The fact-hook
* test therefore asserts the full observable retention contract —
* classified RETAIN (the RED carrier), hook returns only the local-append
* rc without blocking, entry in the WAL with lsn > committed_lsn (pending,
* never confirmed early), and committed_lsn advancing over the entry once
* a covering ACK arrives (no redelivery) — rather than inventing a fake
* registration observable.
*
* Fixture patterns from test_commit_engine.c (stub persist_t + replicator
* with dummy endpoints; real tmpdir persist_t for the hook-driven append;
* repl_broadcast_send_hook to drop broadcasts) and test_persist_hooks.c
* (variadic trampoline driving persist_job_hook_fun with a real va_list).
*
* Task 2.1 adds the persist_commit_barrier(p, timeout_ms) semantics cases
* (Req 1.1, 1.2, 1.3): the handler-callable WAIT barrier returns 0 without
* blocking when committed_lsn already covers the appended watermark
* (current_lsn - 1, the same snapshot persist_wal_sync_append reads — and
* trivially when nothing was ever appended); -ETIMEDOUT when ACKs are
* withheld past the deadline with committed_lsn unmoved; 0 once a covering
* watermark ACK advances the commit index; timeout_ms = 0 falls back to
* p->config.sync_timeout_ms; n_peers=1 succeeds immediately (engine
* self-quorum, standalone preserved); and the engine's admission-cap
* -XCE_ERR_NOT_READY passes through unmodified. RED against the task-2.1
* inert stub in persist.c (returns -ENOSYS — every case fails on its
* assertions, not linking); green with the task-2.2 implementation.
*
* Task 4.1 adds the dispatch-gate fact-allowlist cases (Req 3.2 — facts
* accepted during quorum loss, the bug-396 silent-loss fix): with
* CTRL_PRIMARY and write quorum ABSENT, the fact-bearing opcodes
* OPCode_notify / OPCode_update / OPCode_setpreason / OPCode_clrpreason
* must pass ha_dispatch_gate (return 0) so reported facts reach the
* RETAIN path; the request-shaped opcodes (jobsub, jobcontrol,
* queuecontrol, hostcontrol, jobdispatch, join) stay rejected with
* -XCE_ERR_NOT_READY; with quorum present everything passes as today.
* The four fact cases FAIL (RED) against the current allowlist
* (ha.c CTRL_PRIMARY no-quorum switch — facts hit the default reject);
* the request-rejection and quorum-present cases are PINS that must stay
* green through task 4.2. Fixture copied from test_commit_engine.c
* gate_setup (ha_init + legacy standby_endpoints counting, repl
* back-pointer NULL; quorum-absent = empty endpoint registry).
* No transitional leader sub-state is tested: the single leader role is
* CTRL_PRIMARY (the former transitional state was deleted wholesale by
* ha-single-leader-role).
*
* Task 5.1 adds the config-barrier cases (Req 4.1-4.4) — configAck
* per-push arithmetic, push_config_to_standbys barrier-result contract,
* and the upgradeConfig requester failure/success paths; see the task-5.1
* section banner below for the full RED rationale.
*
* Requirements: 1.1, 1.2, 1.3 (task 2.1); 2.1, 2.2, 3.1 (task 1.1);
* 3.2 (task 4.1); 4.1, 4.2, 4.3, 4.4 (task 5.1)
*/
#include <errno.h>
#include <fcntl.h>
#include <openssl/sha.h> /* SHA256 (task 5.1 config blob hashing) */
#include <pthread.h>
#include <signal.h>
#include <stdarg.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h> /* mkdir (task 5.1 upgrade fixture) */
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#include <capnp_c.h> /* raw reply parsing (task 5.1) */
#include "xce/cluster_def.h" /* HOOK_DETAIL_*, XCE_CONTROLLER_HOOK_JOB */
#include "xce/errnox.h" /* XCE_ERR_NOT_READY (task 2.1 passthrough) */
#include "xce/types.h" /* job_t */
#include "commit.h"
#include "config.h" /* MAX_ARRAY_IDX (task 2.3 bounded-record test) */
#include "config_repl.h" /* push_config_to_standbys, handle_upgrade_config
* (+ netio.h, request.h, myconf.h, clusterconfig.h) */
#include "controller.capnp.h" /* OPCode_* (task 4.1 gate cases) */
#include "ha.h" /* ha_state_t, ha_config_t, zarray (task 4.1) */
#include "hook.h" /* HOOK_AFTER_EVENT */
#include "persist.h" /* wal_commit_mode_t, wal_classify (task 1.1 decls) */
#include "repl.h"
#include "typemap.h" /* wal_catchup_entry_t (send hook signature) */
#include "wal_types.h"
/* ha_dispatch_gate — defined in ha.c (pattern from test_commit_engine.c) */
int ha_dispatch_gate(ha_state_t *ha, int opcode);
/* Test-injectable send hook (defined in repl.c under XCE_TESTING) — keeps
* repl_broadcast from dereferencing the dummy endpoints. */
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int cbr_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
(void)ep;
(void)entries;
(void)n_entries;
return 0; /* pretend success, drop the data */
}
/* ── Shared helpers ──────────────────────────────────────────────────────── */
static uint64_t mono_ms(void)
{
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
}
/* Valid-mode predicate for the totality sweep (Req 2.1). */
static int is_valid_mode(wal_commit_mode_t m)
{
return m == WAL_MODE_WAIT || m == WAL_MODE_RETAIN || m == WAL_MODE_LOCAL;
}
/* ── Classification table cases (Req 2.1) ─────────────────────────────────
* All RED against the inert stub: it returns an out-of-enum sentinel, so
* every equality below fails on assertion, not linking. */
static void test_classify_job_submit_is_wait(void)
{
/* Arrange / Act */
wal_commit_mode_t mode = wal_classify(WAL_JOB_SUBMIT, HOOK_DETAIL_JOB_NEW);
/* Assert: client job submit is request-shaped — commit barrier before
* the requester reply (reference impl keeps the in-hook sync path) */
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_WAIT, mode,
"WAL_JOB_SUBMIT must classify WAIT (request-shaped)");
printf("test_classify_job_submit_is_wait: PASS\n");
}
static void test_classify_job_array_submit_is_wait(void)
{
/* Arrange / Act */
wal_commit_mode_t mode = wal_classify(WAL_JOB_ARRAY_SUBMIT,
HOOK_DETAIL_JOB_NEW);
/* Assert: a job-array submission is request-shaped exactly like a
* single-job submission — one requester awaiting the commit barrier
* before reply (job-array-single-wal-record task 1.1) */
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_WAIT, mode,
"WAL_JOB_ARRAY_SUBMIT must classify WAIT (request-shaped, same as "
"WAL_JOB_SUBMIT)");
printf("test_classify_job_array_submit_is_wait: PASS\n");
}
static void test_classify_queue_ctrl_is_wait(void)
{
/* Arrange / Act */
wal_commit_mode_t mode = wal_classify(WAL_QUEUE_CTRL,
HOOK_DETAIL_QUEUE_UPDATE);
/* Assert: queue control is client-request-shaped */
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_WAIT, mode,
"WAL_QUEUE_CTRL must classify WAIT (request-shaped)");
printf("test_classify_queue_ctrl_is_wait: PASS\n");
}
static void test_classify_worker_ctrl_detail_split(void)
{
/* Arrange / Act: WAL_WORKER_CTRL is the only detail-disambiguated type */
wal_commit_mode_t join_mode =
wal_classify(WAL_WORKER_CTRL, HOOK_DETAIL_WORKER_JOIN);
wal_commit_mode_t update_mode =
wal_classify(WAL_WORKER_CTRL, HOOK_DETAIL_WORKER_UPDATE);
wal_commit_mode_t dead_mode =
wal_classify(WAL_WORKER_CTRL, HOOK_DETAIL_WORKER_DEAD);
/* Assert: join/update have a requester awaiting confirmation (WAIT);
* worker death is controller-detected — a reported fact (RETAIN) */
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_WAIT, join_mode,
"WAL_WORKER_CTRL + WORKER_JOIN must classify WAIT");
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_WAIT, update_mode,
"WAL_WORKER_CTRL + WORKER_UPDATE must classify WAIT");
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_RETAIN, dead_mode,
"WAL_WORKER_CTRL + WORKER_DEAD must classify RETAIN (fact)");
printf("test_classify_worker_ctrl_detail_split: PASS\n");
}
static void test_classify_all_job_state_types_are_retain(void)
{
/* Arrange: every WAL_JOB_* state-transition type with its natural hook
* detail (verified table: all fact-shaped — STOP/RESUME are fact-origin
* only; FAIL/UPDATE dual-origin, request side covered by the handler
* barrier in later tasks) */
static const struct {
uint32_t wal_type;
int detail;
const char *name;
} facts[] = {
{ WAL_JOB_START, HOOK_DETAIL_JOB_START, "WAL_JOB_START" },
{ WAL_JOB_FINISH, HOOK_DETAIL_JOB_EXIT, "WAL_JOB_FINISH" },
{ WAL_JOB_FAIL, HOOK_DETAIL_JOB_FAIL, "WAL_JOB_FAIL" },
{ WAL_JOB_STOP, HOOK_DETAIL_JOB_STOP, "WAL_JOB_STOP" },
{ WAL_JOB_RESUME, HOOK_DETAIL_JOB_RESUME, "WAL_JOB_RESUME" },
{ WAL_JOB_UPDATE, HOOK_DETAIL_JOB_UPDATE, "WAL_JOB_UPDATE" },
{ WAL_JOB_RUSAGE, HOOK_DETAIL_JOB_RUSAGE_UPDATE, "WAL_JOB_RUSAGE" },
{ WAL_JOB_CKPT, HOOK_DETAIL_JOB_CKPT, "WAL_JOB_CKPT" },
{ WAL_JOB_PREASON, HOOK_DETAIL_SETPREASON, "WAL_JOB_PREASON" },
};
size_t i;
/* Act + Assert: each is a reported fact — retained until committed */
for (i = 0; i < sizeof(facts) / sizeof(facts[0]); i++) {
char msg[96];
snprintf(msg, sizeof(msg), "%s must classify RETAIN (fact-shaped)",
facts[i].name);
TEST_ASSERT_EQUAL_INT_MESSAGE(
WAL_MODE_RETAIN, wal_classify(facts[i].wal_type, facts[i].detail),
msg);
}
printf("test_classify_all_job_state_types_are_retain: PASS\n");
}
static void test_classify_checkpoint_is_local(void)
{
/* Arrange / Act: WAL_CHECKPOINT is a local WAL marker, never replicated */
wal_commit_mode_t mode = wal_classify(WAL_CHECKPOINT, 0);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(WAL_MODE_LOCAL, mode,
"WAL_CHECKPOINT must classify LOCAL (never replicated)");
printf("test_classify_checkpoint_is_local: PASS\n");
}
static void test_classify_is_total_over_wal_entry_type(void)
{
/* Arrange: EVERY value of wal_entry_type (wal_types.h:44-58, verified
* 2026-06-11: 13 values, no WAL_NOOP yet). When a new type is added,
* this sweep plus the unknown-type guard force it to be classified. */
static const struct {
uint32_t wal_type;
int detail;
const char *name;
} all_types[] = {
{ WAL_JOB_SUBMIT, HOOK_DETAIL_JOB_NEW, "WAL_JOB_SUBMIT" },
{ WAL_JOB_ARRAY_SUBMIT, HOOK_DETAIL_JOB_NEW, "WAL_JOB_ARRAY_SUBMIT" },
{ WAL_JOB_START, HOOK_DETAIL_JOB_START, "WAL_JOB_START" },
{ WAL_JOB_FINISH, HOOK_DETAIL_JOB_EXIT, "WAL_JOB_FINISH" },
{ WAL_JOB_FAIL, HOOK_DETAIL_JOB_FAIL, "WAL_JOB_FAIL" },
{ WAL_JOB_STOP, HOOK_DETAIL_JOB_STOP, "WAL_JOB_STOP" },
{ WAL_JOB_RESUME, HOOK_DETAIL_JOB_RESUME, "WAL_JOB_RESUME" },
{ WAL_JOB_UPDATE, HOOK_DETAIL_JOB_UPDATE, "WAL_JOB_UPDATE" },
{ WAL_JOB_RUSAGE, HOOK_DETAIL_JOB_RUSAGE_UPDATE, "WAL_JOB_RUSAGE" },
{ WAL_JOB_CKPT, HOOK_DETAIL_JOB_CKPT, "WAL_JOB_CKPT" },
{ WAL_JOB_PREASON, HOOK_DETAIL_CLRPREASON, "WAL_JOB_PREASON" },
{ WAL_QUEUE_CTRL, HOOK_DETAIL_QUEUE_UPDATE, "WAL_QUEUE_CTRL" },
{ WAL_WORKER_CTRL, HOOK_DETAIL_WORKER_DEAD, "WAL_WORKER_CTRL" },
{ WAL_CONFIG_MARKER, 0, "WAL_CONFIG_MARKER" },
{ WAL_CHECKPOINT, 0, "WAL_CHECKPOINT" },
};
size_t i;
/* Act + Assert: classification is total — every enum value maps to a
* valid mode (no silent barrier bypass for any state-changing type) */
for (i = 0; i < sizeof(all_types) / sizeof(all_types[0]); i++) {
char msg[96];
snprintf(msg, sizeof(msg),
"%s must map to a valid commit mode (totality)",
all_types[i].name);
TEST_ASSERT_TRUE_MESSAGE(
is_valid_mode(wal_classify(all_types[i].wal_type,
all_types[i].detail)),
msg);
}
printf("test_classify_is_total_over_wal_entry_type: PASS\n");
}
/* ── Unknown type fails fast under XCE_TESTING (Req 2.2) ──────────────────
* The guard contract is assert(0) in testing builds (abort), production
* ERROR + RETAIN fallback. The abort is observed from a forked child so
* the test runner survives (fork pattern precedent: lib testsuite test.c).
* RED: the inert stub returns its sentinel without asserting, so the child
* exits normally and the SIGABRT assertion fails. */
static void test_classify_unknown_type_fails_fast_in_testing(void)
{
pid_t pid;
int status = 0;
/* Arrange: a WAL type far outside wal_entry_type */
const uint32_t unknown_type = 777u;
/* Act: evaluate the unknown type in a child process */
pid = fork();
TEST_ASSERT_TRUE_MESSAGE(pid >= 0, "fork must succeed");
if (pid == 0) {
/* Child: silence the expected assert/abort noise, then classify.
* Reaching _exit(0) means the fail-fast path did NOT trigger. */
int devnull = open("/dev/null", O_WRONLY);
if (devnull >= 0) {
dup2(devnull, STDERR_FILENO);
dup2(devnull, STDOUT_FILENO);
}
(void)wal_classify(unknown_type, 0);
_exit(0);
}
/* Assert: the child must die on the XCE_TESTING fail-fast path */
TEST_ASSERT_EQUAL_INT_MESSAGE(pid, waitpid(pid, &status, 0),
"waitpid must reap the classifier child");
TEST_ASSERT_TRUE_MESSAGE(
WIFSIGNALED(status) && WTERMSIG(status) == SIGABRT,
"unknown WAL type must trip the XCE_TESTING fail-fast (abort)");
printf("test_classify_unknown_type_fails_fast_in_testing: PASS\n");
}
/* ── Fact-type append through the hook seam is RETAIN-routed (Req 3.1) ────
* Drives the production persist_job_hook_fun with a worker-reported job
* finish (fact) against a real tmpdir persist_t with a wired commit engine
* (n_peers=3) and ACKs withheld. Observable retention contract:
* - wal_classify routes the type RETAIN (RED carrier — fails against the
* inert stub; the behavioral assertions below pass on unmodified
* persist.c and pin the contract task 1.2 must preserve)
* - the hook returns the local-append rc (0) WITHOUT blocking on commit
* progress (no timeout applies to facts)
* - the entry is in the WAL and pending purely by lsn > committed_lsn
* (never confirmed early)
* - a later covering ACK advances committed_lsn over the entry with no
* redelivery of the fact
* commit_submit(RETAIN) itself is registration-free (commit.c) — see the
* file header note; there is deliberately no fake "registered" observable. */
/* Variadic trampoline so the test passes typed args through a real va_list
* (pattern from test_persist_hooks.c). */
static int call_job_hook(int type, int sit, int detail, persist_t *p, ...)
{
va_list ap;
int rc;
va_start(ap, p);
rc = persist_job_hook_fun(type, sit, detail, p, ap);
va_end(ap);
return rc;
}
/* Statics survive Unity's longjmp on assertion failure. */
static persist_t *g_fp;
static commit_engine_t g_fce;
static job_t g_fact_job;
static int g_fep_a, g_fep_b; /* dummy endpoints — replicator compares pointers */
static void test_fact_hook_append_is_retained_and_nonblocking(void)
{
persist_config_t cfg;
char tmpdir[256];
uint64_t t0, elapsed_ms, fact_lsn;
int rc;
/* Arrange: real persist over a private tmpdir; sync timeout large so a
* misrouted blocking wait is detectable as elapsed time */
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = 3000;
snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cbr11_fact_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
g_fp = calloc(1, sizeof(persist_t));
TEST_ASSERT_NOT_NULL(g_fp);
persist_set_workdir(g_fp, tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(g_fp, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(g_fp));
/* 3-member cluster: 2 standbys registered, engine wired, ACKs withheld */
TEST_ASSERT_TRUE(repl_add_standby(&g_fp->repl, &g_fep_a, "sb_a", 7200) >= 0);
TEST_ASSERT_TRUE(repl_add_standby(&g_fp->repl, &g_fep_b, "sb_b", 7201) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_fce, g_fp, 3, 8));
repl_set_commit_engine(&g_fp->repl, &g_fce);
repl_broadcast_send_hook = cbr_drop_send_hook;
/* Under the reserved-0 contract a fresh store starts current_lsn=1, so the
* fact entry lands at LSN 1 with no priming append (LSN 0 is the reserved
* sentinel — never a real entry; the engine rejects lsn == 0). */
fact_lsn = g_fp->current_lsn; /* the LSN the fact append will get */
TEST_ASSERT_EQUAL_UINT64(1, fact_lsn);
memset(&g_fact_job, 0, sizeof(g_fact_job));
g_fact_job.job.jobid = 42;
g_fact_job.job.state = 0x10;
/* Act: worker-reported job finish (fact) through the production hook.
* JOB_EXIT AFTER_EVENT args: cluster_t*, xce_jobfinish_t*, job_t* */
t0 = mono_ms();
rc = call_job_hook(XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_EXIT, g_fp,
(void *)NULL, (void *)NULL, &g_fact_job);
elapsed_ms = mono_ms() - t0;
/* Assert: the seam classifies the fact RETAIN (RED carrier — the routing
* contract the behavioral assertions below depend on) */
TEST_ASSERT_EQUAL_INT_MESSAGE(
WAL_MODE_RETAIN, wal_classify(WAL_JOB_FINISH, HOOK_DETAIL_JOB_EXIT),
"fact-type WAL_JOB_FINISH must be RETAIN-routed at the hook seam");
/* ...the hook propagated only the local-append rc, without blocking on
* commit progress (facts have no commit timeout) */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"fact hook must return the local-append rc only (no commit error)");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"fact append must not block on commit progress (no timeout applies)");
/* ...the entry is in the WAL, pending purely by lsn > committed_lsn */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(fact_lsn + 1, g_fp->current_lsn,
"the fact entry must be appended to the WAL");
TEST_ASSERT_TRUE_MESSAGE(
atomic_load(&g_fp->committed_lsn) < fact_lsn,
"the uncommitted fact must be pending (lsn > committed_lsn), "
"never confirmed early");
/* ha-commit-durability 2.2: the primary's self-contribution to the commit
* majority is now capped at durable_lsn, which the demand-driven
* group-commit flusher advances asynchronously after fdatasync. Force the
* fact durable here (synchronous fdatasync) so the covering ACK below
* deterministically completes the majority (durable self + standby ACK);
* a bare synchronous read would otherwise race the flusher and observe
* durable_lsn == 0 → self-cap 0 → committed_lsn unchanged. This mirrors the
* durable_lsn priming done for the in-memory stub fixtures in
* barrier_setup_vcpus. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, persist_flush(g_fp),
"fact must fdatasync (durable) before the covering ACK");
TEST_ASSERT_TRUE_MESSAGE(persist_durable_lsn(g_fp) >= fact_lsn,
"durable_lsn must cover the fact after persist_flush");
/* ...and a covering ACK then commits the now-durable retained entry —
* repl_on_ack drives commit_on_ack synchronously, advancing committed_lsn
* over it, no redelivery needed */
repl_on_ack(&g_fp->repl, &g_fep_a, fact_lsn);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
fact_lsn, atomic_load(&g_fp->committed_lsn),
"a covering ACK must advance committed_lsn over the retained fact");
/* Teardown */
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_fce);
persist_stop(g_fp);
persist_fini(g_fp);
free(g_fp);
g_fp = NULL;
printf("test_fact_hook_append_is_retained_and_nonblocking: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 2.1 — persist_commit_barrier(p, timeout_ms) semantics
* (Req 1.1, 1.2, 1.3). RED against the inert -ENOSYS stub in persist.c.
*
* Fixture: stub persist_t + replicator + wired commit engine (pattern from
* test_commit_engine.c engine_setup_vcpus). The barrier reaches the engine
* via p->repl.ce (the same wiring the RETAIN hook path uses), so
* repl_set_commit_engine is part of the fixture. ACKs are driven through
* repl_on_ack (registry watermark + engine evaluation, the production
* path). current_lsn semantics: highest appended LSN = current_lsn - 1
* (persist.h invariant; persist.c:1021); current_lsn == 0 = nothing ever
* appended. n_peers=3 unless noted.
* ════════════════════════════════════════════════════════════════════════ */
/* Statics survive Unity's longjmp on assertion failure. */
static persist_t g_bp; /* stub persist for the barrier cases */
static commit_engine_t g_bce;
static int g_bep_a, g_bep_b; /* dummy endpoints (pointer compares) */
static void barrier_setup_vcpus(int n_peers, int n_standbys,
uint64_t current_lsn, int n_vcpus,
uint32_t sync_timeout_ms)
{
memset(&g_bp, 0, sizeof(g_bp));
TEST_ASSERT_EQUAL_INT(0, pthread_mutex_init(&g_bp.lock, NULL));
g_bp.config = persist_config_defaults();
g_bp.config.sync_timeout_ms = sync_timeout_ms;
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_bp.repl));
if (n_standbys >= 1)
TEST_ASSERT_TRUE(
repl_add_standby(&g_bp.repl, &g_bep_a, "bsb_a", 7300) >= 0);
if (n_standbys >= 2)
TEST_ASSERT_TRUE(
repl_add_standby(&g_bp.repl, &g_bep_b, "bsb_b", 7301) >= 0);
g_bp.current_lsn = current_lsn;
atomic_store(&g_bp.committed_lsn, 0);
/* ha-commit-durability 2.2: self caps at durable_lsn. This stub has no
* flusher, so treat the whole appended log (current_lsn - 1) as durable
* up front, preserving the pre-durability "self covers every appended
* entry" model. (current_lsn == 0 → nothing appended → durable 0.) */
atomic_store(&g_bp.durable_lsn, current_lsn > 0 ? current_lsn - 1 : 0);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&g_bce, &g_bp, n_peers, n_vcpus));
repl_set_commit_engine(&g_bp.repl, &g_bce);
}
static void barrier_setup(int n_peers, int n_standbys, uint64_t current_lsn)
{
barrier_setup_vcpus(n_peers, n_standbys, current_lsn, 8, 3000);
}
static void barrier_teardown(void)
{
commit_engine_fini(&g_bce);
repl_fini(&g_bp.repl);
pthread_mutex_destroy(&g_bp.lock);
}
static void cbr_sleep_ms(uint32_t ms)
{
struct timespec ts;
ts.tv_sec = ms / 1000u;
ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
nanosleep(&ts, NULL);
}
/* Barrier-caller thread context (covering-ACK case). */
typedef struct {
uint32_t timeout_ms;
int rc;
uint64_t elapsed_ms;
} barrier_arg_t;
static barrier_arg_t g_barg;
static pthread_t g_btid;
static void *barrier_thread_fn(void *arg)
{
barrier_arg_t *a = arg;
uint64_t t0 = mono_ms();
a->rc = persist_commit_barrier(&g_bp, a->timeout_ms);
a->elapsed_ms = mono_ms() - t0;
return NULL;
}
/* Direct engine waiter — saturates the WAIT admission slot for the
* NOT_READY passthrough case (test_commit_engine.c test-13 pattern). */
typedef struct {
uint64_t lsn;
uint32_t timeout_ms;
int rc;
} engine_wait_arg_t;
static engine_wait_arg_t g_ewarg;
static pthread_t g_ewtid;
static void *engine_wait_thread_fn(void *arg)
{
engine_wait_arg_t *a = arg;
a->rc = commit_submit(&g_bce, a->lsn, COMMIT_MODE_WAIT, a->timeout_ms);
return NULL;
}
/* Poll until `want` waiters are blocked inside the engine or the deadline
* expires; returns the last observed inflight count. */
static int barrier_poll_inflight(int want, uint32_t deadline_ms)
{
uint64_t t0 = mono_ms();
int v;
for (;;) {
pthread_mutex_lock(&g_bce.lock);
v = g_bce.inflight_waits;
pthread_mutex_unlock(&g_bce.lock);
if (v >= want || mono_ms() - t0 >= deadline_ms)
return v;
cbr_sleep_ms(5);
}
}
/* ── Barrier short-circuits when already covered (Req 1.1) ──────────────── */
static void test_barrier_returns_zero_when_nothing_appended(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: 3-member cluster, NOTHING ever appended (current_lsn == 0,
* fresh persist) — there are no pending entries to wait for; ACKs
* withheld so any misrouted engine wait would block visibly */
barrier_setup(3, 2, 0);
/* Act */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: immediate success without consuming the timeout */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"barrier over an empty log must return 0 (nothing to commit)");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"barrier over an empty log must not block");
barrier_teardown();
printf("test_barrier_returns_zero_when_nothing_appended: PASS\n");
}
static void test_barrier_returns_zero_when_committed_covers_appended(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: entries 1..2 appended (current_lsn = 3); a standby watermark
* ACK has already committed both — committed_lsn covers the appended
* watermark current_lsn - 1 before the barrier is called */
barrier_setup(3, 2, 3);
repl_on_ack(&g_bp.repl, &g_bep_a, 2);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(2, atomic_load(&g_bp.committed_lsn),
"fixture: covering ACK must have committed the appended entries");
/* Act */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: already covered → 0 immediately, no blocking */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"barrier must return 0 when committed_lsn already covers "
"current_lsn - 1");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"already-covered barrier must not block");
barrier_teardown();
printf("test_barrier_returns_zero_when_committed_covers_appended: PASS\n");
}
/* ── Withheld ACKs → -ETIMEDOUT, committed_lsn unmoved (Req 1.2) ────────── */
static void test_barrier_times_out_when_acks_withheld(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: entry 1 pending (current_lsn = 2), ACKs withheld forever */
barrier_setup(3, 2, 2);
/* Act: short deadline, no ACK will ever arrive */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 250);
elapsed_ms = mono_ms() - t0;
/* Assert: timeout surfaced, deadline honored, commit index unmoved */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"withheld ACKs past the deadline must surface -ETIMEDOUT");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms >= 200,
"barrier must hold until its deadline before timing out");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, atomic_load(&g_bp.committed_lsn),
"committed_lsn must stay unmoved when no ACK ever covers the entry");
barrier_teardown();
printf("test_barrier_times_out_when_acks_withheld: PASS\n");
}
/* ── Covering ACK while blocked → 0 (Req 1.1) ───────────────────────────── */
static void test_barrier_returns_zero_once_covering_ack_arrives(void)
{
/* Arrange: entry 1 pending (current_lsn = 2); the barrier call runs in
* a thread so this thread can deliver the covering ACK */
barrier_setup(3, 2, 2);
g_barg.timeout_ms = 2000;
g_barg.rc = -9999;
g_barg.elapsed_ms = 0;
TEST_ASSERT_EQUAL_INT(
0, pthread_create(&g_btid, NULL, barrier_thread_fn, &g_barg));
/* Act: hold the barrier blocked >= 120 ms, then a covering watermark
* ACK advances the commit index; join BEFORE any assertion */
cbr_sleep_ms(120);
repl_on_ack(&g_bp.repl, &g_bep_a, 1);
pthread_join(g_btid, NULL);
/* Assert: success only after the ACK — the barrier really blocked */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_barg.rc,
"barrier must return 0 once a covering ACK advances the commit index");
TEST_ASSERT_TRUE_MESSAGE(g_barg.elapsed_ms >= 100,
"barrier must block until the covering ACK arrives, not before");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, atomic_load(&g_bp.committed_lsn),
"the covering ACK must have advanced committed_lsn over the entry");
barrier_teardown();
printf("test_barrier_returns_zero_once_covering_ack_arrives: PASS\n");
}
/* ── timeout_ms = 0 falls back to p->config.sync_timeout_ms (Req 1.2) ───── */
static void test_barrier_zero_timeout_falls_back_to_config_sync_timeout(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: entry 1 pending, ACKs withheld; config sync_timeout_ms set
* to a short, recognizable deadline (250 ms) */
barrier_setup_vcpus(3, 2, 2, 8, 250);
/* Act: explicit timeout 0 → must borrow the config deadline */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 0);
elapsed_ms = mono_ms() - t0;
/* Assert: timed out after ~the CONFIG deadline — neither an instant
* zero-timeout reject nor an unbounded wait */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"timeout_ms=0 with withheld ACKs must time out via the config value");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms >= 200,
"timeout_ms=0 must fall back to config.sync_timeout_ms, not 0");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1500,
"timeout_ms=0 must be bounded by config.sync_timeout_ms");
barrier_teardown();
printf("test_barrier_zero_timeout_falls_back_to_config_sync_timeout: "
"PASS\n");
}
/* ── n_peers=1 → immediate success (standalone preserved, Req 1.1) ──────── */
static void test_barrier_single_member_returns_immediately(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: single-member cluster, entry 1 pending (current_lsn = 2) —
* engine self-quorum: the primary's own append is the majority */
barrier_setup(1, 0, 2);
/* Act: large timeout that must NOT be consumed */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: immediate local confirmation, standalone behavior preserved */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"n_peers=1 barrier must return 0 immediately (self-quorum)");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"n_peers=1 barrier must not block");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, atomic_load(&g_bp.committed_lsn),
"self-quorum must advance committed_lsn over the entry");
barrier_teardown();
printf("test_barrier_single_member_returns_immediately: PASS\n");
}
/* ── Engine -XCE_ERR_NOT_READY passes through unmodified (Req 1.3) ────────
* n_vcpus=2, COMMIT_WAIT_RESERVED=2 → admission cap = max(1, 0) = 1
* (test_commit_engine.c test-13 pattern): one direct engine WAIT saturates
* the only slot, so the barrier's own commit_submit(WAIT) is rejected fast
* with the retriable admission error — which the barrier must return
* UNMODIFIED (no remapping, no success). */
static void test_barrier_not_ready_passes_through_unmodified(void)
{
int rc, observed_inflight;
uint64_t t0, elapsed_ms;
/* Arrange: entry 1 pending, cap = 1; a direct waiter occupies the slot */
barrier_setup_vcpus(3, 2, 2, 2, 3000);
g_ewarg.lsn = 1;
g_ewarg.timeout_ms = 5000;
g_ewarg.rc = -9999;
TEST_ASSERT_EQUAL_INT(
0, pthread_create(&g_ewtid, NULL, engine_wait_thread_fn, &g_ewarg));
observed_inflight = barrier_poll_inflight(1, 2000);
/* Act: barrier call while the admission cap is saturated */
t0 = mono_ms();
rc = persist_commit_barrier(&g_bp, 5000);
elapsed_ms = mono_ms() - t0;
/* Release the blocked waiter, then join BEFORE any assertion */
repl_on_ack(&g_bp.repl, &g_bep_a, 1);
pthread_join(g_ewtid, NULL);
/* Assert: the slot was really saturated when the barrier ran... */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, observed_inflight,
"the direct engine waiter must occupy the only admission slot");
/* ...the engine's admission rejection passed through unmodified... */
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc,
"barrier must return the engine's -XCE_ERR_NOT_READY unmodified");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 500,
"admission-cap rejection must not block");
/* ...and the admitted waiter still completed via the covering ACK */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_ewarg.rc,
"the admitted direct waiter must complete with success");
barrier_teardown();
printf("test_barrier_not_ready_passes_through_unmodified: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 4.1 — dispatch-gate fact allowlist (Req 3.2).
* (design.md §Fact Gate Admission — facts admitted during quorum loss)
*
* Fixture copied from test_commit_engine.c gate_setup: ha_init zeroes the
* struct (repl back-pointer NULL → legacy standby_endpoints counting),
* role forced to CTRL_PRIMARY, quorum driven purely by the number of
* endpoints in standby_endpoints (n_peers=3: 0 endpoints = quorum absent,
* 1 endpoint = quorum present).
* ════════════════════════════════════════════════════════════════════════ */
/* Statics survive Unity's longjmp on assertion failure. */
static ha_state_t g_fg_ha;
static int g_fg_ep; /* dummy endpoint (zarray stores the pointer) */
static ha_config_t g_fg_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
};
/* Arrange a CTRL_PRIMARY ha_state_t with n_endpoints connected standby
* endpoints (what raft_cb_become_leader / on_standby_connected populate). */
static void fact_gate_setup(int n_peers, int n_endpoints)
{
int i;
TEST_ASSERT_EQUAL_INT(0, ha_init(&g_fg_ha, &g_fg_cfg, 0));
g_fg_ha.n_peers = n_peers;
atomic_store_explicit(&g_fg_ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
g_fg_ha.standby_endpoints = calloc(1, sizeof(zarray_t));
TEST_ASSERT_NOT_NULL(g_fg_ha.standby_endpoints);
zarray_init(g_fg_ha.standby_endpoints, 4, NULL);
for (i = 0; i < n_endpoints; i++) {
zarray_store(g_fg_ha.standby_endpoints, &g_fg_ep);
}
}
static void fact_gate_teardown(void)
{
if (g_fg_ha.standby_endpoints) {
zarray_fini(g_fg_ha.standby_endpoints);
free(g_fg_ha.standby_endpoints);
g_fg_ha.standby_endpoints = NULL;
}
ha_fini(&g_fg_ha);
}
/* ── OPCode_notify admitted without write quorum (Req 3.2) ────────────────
* Worker-reported job state transitions arrive as OPCode_notify with no
* requester awaiting confirmation. Rejecting them during quorum loss is
* the bug-396 silent-loss hole: the worker never redelivers, so the fact
* is gone forever. The gate must admit it so the RETAIN path can append
* and retry until committed. RED against the current allowlist. */
static void test_gate_no_quorum_admits_notify_fact(void)
{
int rc;
/* Arrange: 3-member primary, zero connected standbys — quorum absent */
fact_gate_setup(3, 0);
/* Act */
rc = ha_dispatch_gate(&g_fg_ha, OPCode_notify);
/* Assert: fact-bearing opcode must pass the gate during quorum loss */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_notify (worker fact) must be admitted without write quorum");
fact_gate_teardown();
printf("test_gate_no_quorum_admits_notify_fact: PASS\n");
}
/* ── OPCode_update admitted without write quorum (Req 3.2) ────────────────
* Worker-reported resource usage (do_update_op rusage path) is a fact:
* receipt-style reply, RETAIN-registered append. RED. */
static void test_gate_no_quorum_admits_update_fact(void)
{
int rc;
/* Arrange: quorum absent */
fact_gate_setup(3, 0);
/* Act */
rc = ha_dispatch_gate(&g_fg_ha, OPCode_update);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_update (worker rusage fact) must be admitted without write quorum");
fact_gate_teardown();
printf("test_gate_no_quorum_admits_update_fact: PASS\n");
}
/* ── OPCode_setpreason admitted without write quorum (Req 3.2) ────────────
* Scheduler-reported pending reasons are facts (do_set_preason,
* receipt-style reply). RED. */
static void test_gate_no_quorum_admits_setpreason_fact(void)
{
int rc;
/* Arrange: quorum absent */
fact_gate_setup(3, 0);
/* Act */
rc = ha_dispatch_gate(&g_fg_ha, OPCode_setpreason);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_setpreason (scheduler fact) must be admitted without write quorum");
fact_gate_teardown();
printf("test_gate_no_quorum_admits_setpreason_fact: PASS\n");
}
/* ── OPCode_clrpreason admitted without write quorum (Req 3.2) ────────────
* Clearing a pending reason is the same scheduler-fact shape. RED. */
static void test_gate_no_quorum_admits_clrpreason_fact(void)
{
int rc;
/* Arrange: quorum absent */
fact_gate_setup(3, 0);
/* Act */
rc = ha_dispatch_gate(&g_fg_ha, OPCode_clrpreason);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_clrpreason (scheduler fact) must be admitted without write quorum");
fact_gate_teardown();
printf("test_gate_no_quorum_admits_clrpreason_fact: PASS\n");
}
/* ── PIN: request-shaped opcodes stay rejected without quorum (Req 3.2) ───
* The allowlist is FACTS ONLY: every request-shaped opcode (a requester
* awaits confirmation, served by the task-3 handler barrier) must keep
* the -XCE_ERR_NOT_READY fast rejection. Green today; must remain green
* after task 4.2 widens the allowlist. */
static void test_gate_no_quorum_still_rejects_request_opcodes(void)
{
static const struct { int opcode; const char *name; } reqs[] = {
{ OPCode_jobsub, "OPCode_jobsub" },
{ OPCode_jobcontrol, "OPCode_jobcontrol" },
{ OPCode_queuecontrol, "OPCode_queuecontrol" },
{ OPCode_hostcontrol, "OPCode_hostcontrol" },
{ OPCode_jobdispatch, "OPCode_jobdispatch" },
{ OPCode_join, "OPCode_join" },
};
size_t i;
/* Arrange: quorum absent */
fact_gate_setup(3, 0);
/* Act + Assert: every request-shaped opcode keeps the fast reject */
for (i = 0; i < sizeof(reqs) / sizeof(reqs[0]); i++) {
int rc = ha_dispatch_gate(&g_fg_ha, reqs[i].opcode);
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc, reqs[i].name);
}
fact_gate_teardown();
printf("test_gate_no_quorum_still_rejects_request_opcodes: PASS\n");
}
/* ── PIN: with write quorum present everything passes as today (Req 3.2) ──
* The quorum-present path is `return 0` before any opcode inspection —
* facts AND requests alike. Green today; must remain green after 4.2. */
static void test_gate_with_quorum_admits_all_opcodes(void)
{
static const struct { int opcode; const char *name; } ops[] = {
{ OPCode_notify, "OPCode_notify" },
{ OPCode_update, "OPCode_update" },
{ OPCode_setpreason, "OPCode_setpreason" },
{ OPCode_clrpreason, "OPCode_clrpreason" },
{ OPCode_jobsub, "OPCode_jobsub" },
{ OPCode_jobcontrol, "OPCode_jobcontrol" },
{ OPCode_queuecontrol, "OPCode_queuecontrol" },
{ OPCode_hostcontrol, "OPCode_hostcontrol" },
{ OPCode_jobdispatch, "OPCode_jobdispatch" },
{ OPCode_join, "OPCode_join" },
};
size_t i;
/* Arrange: 3-member primary, one connected standby — quorum present */
fact_gate_setup(3, 1);
/* Act + Assert: quorum-present primary admits everything */
for (i = 0; i < sizeof(ops) / sizeof(ops[0]); i++) {
int rc = ha_dispatch_gate(&g_fg_ha, ops[i].opcode);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, ops[i].name);
}
fact_gate_teardown();
printf("test_gate_with_quorum_admits_all_opcodes: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 5.1 — config barrier: configAck per-push arithmetic + upgrade
* failure path (Req 4.1, 4.2, 4.3, 4.4; design §Config Barrier).
*
* UNIT half (repl.c configAck branch via repl_wait_majority_ack): the
* hardened branch must take the in-flight push tag through the EXISTING
* entry_lsn parameter (added by ha-commit-engine; no second parameter)
* and count standbys with acked_config_lsn >= push_tag; majority =
* 1 (self) + count > n_peers / 2 over configured members (engine
* n_peers, the fixture's wired commit_engine_init value); n_peers <= 1 →
* immediate success. RED carriers against the current branch
* (repl.c:444-510, re-verified this session): it ignores entry_lsn,
* counts acked_config_lsn > 0 ("ACKed anything ever"), uses denominator
* n_active, and short-circuits n_active == 0 → success.
*
* INTEGRATION half (config_repl.c): push_config_to_standbys must return
* the barrier result (-ETIMEDOUT with acks withheld / 0 with majority)
* with the transport push count via the acked_out out-param (the task-5.1
* behavior-preserving signature change keeps the legacy push-count RETURN
* until task 5.2 — both rc assertions are RED against it), and the
* upgradeConfig requester must observe the barrier outcome: on failure an
* error reply in the send_upgrade_error shape (status < 0, nodata) with
* committed_config_tag unmoved and restart_requested clear (RED against
* the unconditional success ACK + restart, config_repl.c:1073-1106
* post-edit anchors); on success the existing reply + restart behavior is
* unchanged (PIN — both standbys pre-ack with UINT64_MAX so the case is
* green under both the legacy counting and the per-push arithmetic).
*
* Fixtures: the unit cases reuse barrier_setup (stub persist_t + repl
* registry + wired engine — the engine carries n_peers for the configAck
* denominator). The integration cases use a minimal cluster_t with stack
* standby endpoints (zeroed, no iomgr — pushes and replies stay on
* ep->outq, the test_config_repl.c:2210 / test_commit_barrier_reply.c
* capture precedent) and parse replies with the raw capnp reader.
* ════════════════════════════════════════════════════════════════════════ */
/* ── Unit: configAck per-push arithmetic (Req 4.1, 4.2) ─────────────────── */
/* A standby's acked_config_lsn from a PREVIOUS push tag must not count
* toward the current push: both standbys acked tag 0x50, the in-flight
* push is 0x100 → no covering ack → -ETIMEDOUT. RED: the current branch
* counts acked_config_lsn > 0, sees 2 >= quorum 2 and returns 0. */
static void test_configack_previous_tag_ack_does_not_count(void)
{
int rc;
/* Arrange: n_peers=3, both standbys acked an OLDER push tag */
barrier_setup(3, 2, 0);
repl_on_config_ack(&g_bp.repl, &g_bep_a, 0x50);
repl_on_config_ack(&g_bp.repl, &g_bep_b, 0x50);
/* Act: wait on push tag 0x100 (entry_lsn carries the push tag) */
rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 150,
(uint8_t)OPCode_upgradeConfigAck);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"a previous push's config ack must not count toward the current tag");
barrier_teardown();
printf("test_configack_previous_tag_ack_does_not_count: PASS\n");
}
/* n_peers=3 with ZERO covering acks (here: zero connected standbys) must
* time out — a 3-member cluster can never reach majority on self alone.
* RED: the current `n_active == 0 → success` short-circuit returns 0. */
static void test_configack_no_active_standbys_times_out(void)
{
int rc;
/* Arrange: n_peers=3, no standby connected at all */
barrier_setup(3, 0, 0);
/* Act */
rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 150,
(uint8_t)OPCode_upgradeConfigAck);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"n_peers=3 with zero covering config acks must never succeed "
"(the n_active == 0 success short-circuit must die on this branch)");
barrier_teardown();
printf("test_configack_no_active_standbys_times_out: PASS\n");
}
/* Self + one covering standby is a majority of n_peers=3 (1 + 1 = 2 > 1).
* RED: the current branch uses denominator n_active with NO self-count —
* quorum (2+2)/2 = 2 vs acked 1 → it times out. */
static void test_configack_self_plus_one_covering_is_majority_of_three(void)
{
int rc;
/* Arrange: n_peers=3, two standbys, one acks the in-flight tag */
barrier_setup(3, 2, 0);
repl_on_config_ack(&g_bp.repl, &g_bep_a, 0x100);
/* Act */
rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 400,
(uint8_t)OPCode_upgradeConfigAck);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"self + one covering standby must satisfy majority of n_peers=3");
barrier_teardown();
printf("test_configack_self_plus_one_covering_is_majority_of_three: PASS\n");
}
/* n_peers=1: single-member cluster — immediate success without consuming
* the timeout. PIN: green today (via the n_active == 0 short-circuit) and
* green after task 5.2 (explicit n_peers <= 1 rule). */
static void test_configack_single_member_immediate_success(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange */
barrier_setup(1, 0, 0);
/* Act */
t0 = mono_ms();
rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 500,
(uint8_t)OPCode_upgradeConfigAck);
elapsed_ms = mono_ms() - t0;
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"single-member cluster must confirm a config push immediately");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 200,
"single-member success must not consume the ack timeout");
barrier_teardown();
printf("test_configack_single_member_immediate_success: PASS\n");
}
/* ── Integration fixture: minimal cluster_t for the config push path ────── */
/* Statics survive Unity's longjmp on assertion failure. */
static cluster_t g_cb5_c;
static commit_engine_t g_cb5_eng;
static endpoint_t g_cb5_sb_ep[2]; /* standby push endpoints */
static ep_private_t g_cb5_sb_priv[2];
static endpoint_t g_cb5_req_ep; /* upgradeConfig requester */
static ep_private_t g_cb5_req_priv;
static zarray_t g_cb5_standbys;
static myconf_t g_cb5_myconf;
static char g_cb5_tmpdir[256];
static int g_cb5_have_tmpdir;
static char g_cb5_waldir[256]; /* started-persist WAL dir (task 4.2) */
static int g_cb5_have_waldir;
static void cb5_setup(void)
{
int i;
memset(&g_cb5_c, 0, sizeof(g_cb5_c));
memset(g_cb5_sb_ep, 0, sizeof(g_cb5_sb_ep));
memset(g_cb5_sb_priv, 0, sizeof(g_cb5_sb_priv));
memset(&g_cb5_req_ep, 0, sizeof(g_cb5_req_ep));
memset(&g_cb5_req_priv, 0, sizeof(g_cb5_req_priv));
memset(&g_cb5_myconf, 0, sizeof(g_cb5_myconf));
g_cb5_have_tmpdir = 0;
g_cb5_have_waldir = 0;
g_cb5_c.myhost = &g_cb5_c.myhost_info;
g_cb5_c.myhost_info.name = (char *)"ctrl-cb5";
g_cb5_c.myhost_info.key = (char *)"";
/* Zeroed stack endpoints, no iomgr: sends stay queued on ep->outq
* (endpoint_update_epoll no-ops; zeroed pthread mutexes are valid). */
g_cb5_req_ep.fd = -1;
g_cb5_req_ep.data = &g_cb5_req_priv;
TEST_ASSERT_EQUAL_INT(0, zarray_init(&g_cb5_standbys, 4, NULL));
for (i = 0; i < 2; i++) {
g_cb5_sb_ep[i].fd = -1;
g_cb5_sb_ep[i].data = &g_cb5_sb_priv[i];
zarray_store(&g_cb5_standbys, &g_cb5_sb_ep[i]);
}
g_cb5_c.ha.standby_endpoints = &g_cb5_standbys;
/* Started persist so the live config-change path (config_commit_change →
* config_append_marker → ha_primary_log_append → persist_wal_async) can
* append the WAL_CONFIG_MARKER consensus record (ha-term-safety task 4.2).
* persist_init creates the replicator registry; we then mirror the push
* endpoints and wire the n_peers=3 engine (configAck majority denominator).
* A dedicated tmpdir keeps the WAL isolated from the myconf workdir. */
{
persist_config_t pcfg = persist_config_defaults();
pcfg.wal_sync_ms = 5000;
snprintf(g_cb5_waldir, sizeof(g_cb5_waldir), "/tmp/xce_cb5wal_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_cb5_waldir));
g_cb5_have_waldir = 1;
persist_set_workdir(&g_cb5_c.persist, g_cb5_waldir, "ctrl-cb5");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_cb5_c.persist, &pcfg, &g_cb5_c));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_cb5_c.persist));
}
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
&g_cb5_sb_ep[0], "cb5_a", 7500) >= 0);
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
&g_cb5_sb_ep[1], "cb5_b", 7501) >= 0);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&g_cb5_eng, &g_cb5_c.persist, 3, 8));
repl_set_commit_engine(&g_cb5_c.persist.repl, &g_cb5_eng);
/* The current do_upgrade_config_with_bytes restart path does
* cbuf_put(&c->wbuf, …) — must be a real cbuf or it blocks forever. */
TEST_ASSERT_EQUAL_INT(0, cbuf_init(&g_cb5_c.wbuf, 4, sizeof(int)));
}
/* Upgrade-path extension: tmpdir-backed myconf so conf_save_config works
* (it writes to <workdir>/<name>/conf — the parent must exist). */
static void cb5_setup_upgrade(void)
{
char inst[300];
cb5_setup();
snprintf(g_cb5_tmpdir, sizeof(g_cb5_tmpdir), "/tmp/xce_cb51_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_cb5_tmpdir));
g_cb5_have_tmpdir = 1;
snprintf(inst, sizeof(inst), "%s/ctrl", g_cb5_tmpdir);
TEST_ASSERT_EQUAL_INT(0, mkdir(inst, 0700));
g_cb5_myconf.workdir = g_cb5_tmpdir;
g_cb5_myconf.name = (char *)"ctrl";
g_cb5_myconf.conf_keep_n = 2;
g_cb5_c.myconf = &g_cb5_myconf;
}
/* Drain and free every packet queued on an endpoint's outq. */
static void cb5_drain_ep(endpoint_t *ep)
{
data_item_t *it;
while ((it = endpoint_deq(ep)) != NULL) {
io_packet_t *pkt = ITPTR_TO_PACKET(it);
free_packet(pkt);
free(pkt);
}
}
static void cb5_teardown(void)
{
cb5_drain_ep(&g_cb5_sb_ep[0]);
cb5_drain_ep(&g_cb5_sb_ep[1]);
cb5_drain_ep(&g_cb5_req_ep);
commit_engine_fini(&g_cb5_eng);
/* persist_fini calls repl_fini internally (started-persist fixture). */
persist_stop(&g_cb5_c.persist);
persist_fini(&g_cb5_c.persist);
zarray_fini(&g_cb5_standbys);
cbuf_fini(&g_cb5_c.wbuf);
free(g_cb5_myconf.conffile);
g_cb5_myconf.conffile = NULL;
if (g_cb5_have_tmpdir) {
char cmd[300];
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_cb5_tmpdir);
system(cmd);
g_cb5_have_tmpdir = 0;
}
if (g_cb5_have_waldir) {
char cmd[300];
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_cb5_waldir);
system(cmd);
g_cb5_have_waldir = 0;
}
}
/* ── Reply observation (raw capnp parse of ep->outq packets) ────────────── */
typedef struct {
int n_replies; /* replymsg packets seen */
int32_t last_status; /* status of the last replymsg observed */
int last_data_which; /* RequestReply_data_which discriminant */
} cb5_reply_obs_t;
static void cb5_scan_replies(endpoint_t *ep, cb5_reply_obs_t *obs)
{
data_item_t *it;
while ((it = endpoint_deq(ep)) != NULL) {
io_packet_t *pkt = ITPTR_TO_PACKET(it);
struct capn c;
struct Request rq;
Request_ptr rp;
capn_init_mem(&c, (const uint8_t *)pkt->data, pkt->sz, 0);
rp.p = capn_getp(capn_root(&c), 0, 1);
read_Request(&rq, rp);
if (rq.body_which == Request_body_replymsg) {
struct RequestReply rr;
read_RequestReply(&rr, rq.body.replymsg);
obs->n_replies += 1;
obs->last_status = rr.status;
obs->last_data_which = (int)rr.data_which;
}
capn_free(&c);
free_packet(pkt);
free(pkt);
}
}
/* Build a minimal valid serialised ClusterConfig blob (caller frees). */
static uint8_t *cb5_make_config_blob(size_t *len_out)
{
cluster_config_t cfg;
struct capn cap;
struct capn_segment *cs;
ClusterConfig_ptr p;
uint8_t *buf;
size_t sz;
memset(&cfg, 0, sizeof(cfg));
cfg.clustername = (char *)"cb5-upgrade";
cfg.port = 9990;
cfg.config_generation = 1;
capn_init_malloc(&cap);
cs = capn_root(&cap).seg;
encode_ClusterConfig_ptr(cs, &p, &cfg);
capn_setp(capn_root(&cap), 0, p.p);
sz = (size_t)capn_size(&cap);
buf = (uint8_t *)malloc(sz);
TEST_ASSERT_NOT_NULL(buf);
capn_write_mem(&cap, buf, sz, 0);
capn_free(&cap);
*len_out = sz;
return buf;
}
/* upgradeConfig request scaffolding (statics survive Unity longjmp).
* handle_upgrade_config reads the raw_* fields xce_decode would have
* populated; the test owns the buffers, so xce_free_request is NOT used. */
static upgrade_config_req_t g_cb5_ucr;
static xce_request_t g_cb5_req;
static void cb5_arrange_upgrade_req(uint8_t *blob, size_t blen,
uint8_t sha[32])
{
memset(&g_cb5_ucr, 0, sizeof(g_cb5_ucr));
memset(&g_cb5_req, 0, sizeof(g_cb5_req));
g_cb5_ucr.raw_sha256 = sha;
g_cb5_ucr.raw_bytes = blob;
g_cb5_ucr.raw_bytes_len = blen;
g_cb5_req.opcode = OPCode_upgradeConfig;
g_cb5_req.seq = 1;
g_cb5_req.body_which = Request_body_upgradeconfigrequestmsg;
g_cb5_req.body.upgradeconfigrequestmsg = &g_cb5_ucr;
}
/* ── push_config_to_standbys: barrier result + push count (Req 4.2, 4.3) ── */
/* ACKs withheld: the push barrier must report -ETIMEDOUT (return value),
* the transport push count via acked_out, and the tag must never commit.
* RED: the current return value IS the push count (2), and the internal
* majority wait only logs its timeout. */
static void test_push_config_withheld_acks_returns_etimedout(void)
{
static const uint8_t blob[] = "cb5-push-blob";
uint8_t sha[32];
int pushed = -1;
int rc;
/* Arrange */
cb5_setup();
SHA256(blob, sizeof(blob) - 1, sha);
/* Act: no standby ever acks tag 0xAB12 */
rc = push_config_to_standbys(&g_cb5_c, blob, sizeof(blob) - 1, sha,
0xAB12, &pushed);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"withheld config ACKs must surface as -ETIMEDOUT from the push (Req 4.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE(2, pushed,
"acked_out must report the per-standby transport push count");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0,
atomic_load_explicit(&g_cb5_c.ha.committed_config_tag,
memory_order_acquire),
"an unacked config must never become committed (Req 4.3)");
cb5_teardown();
printf("test_push_config_withheld_acks_returns_etimedout: PASS\n");
}
/* Majority acks (self + 1 covering standby of n_peers=3): the push must
* return 0 and commit the tag. RED twice over: the return value is the
* push count (2 != 0), and the legacy internal wait (quorum 2 over
* n_active, no self-count) times out so the tag never advances. */
static void test_push_config_majority_acks_returns_zero_and_commits(void)
{
static const uint8_t blob[] = "cb5-push-blob";
uint8_t sha[32];
int pushed = -1;
int rc;
/* Arrange: one standby has already acked the in-flight tag */
cb5_setup();
SHA256(blob, sizeof(blob) - 1, sha);
repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[0], 0xAB12);
/* Act */
rc = push_config_to_standbys(&g_cb5_c, blob, sizeof(blob) - 1, sha,
0xAB12, &pushed);
/* Assert */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"majority config ACK must surface as success from the push (Req 4.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(2, pushed,
"acked_out must report the per-standby transport push count");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0xAB12,
atomic_load_explicit(&g_cb5_c.ha.committed_config_tag,
memory_order_acquire),
"majority-acked config must advance committed_config_tag (Req 4.1)");
cb5_teardown();
printf("test_push_config_majority_acks_returns_zero_and_commits: PASS\n");
}
/* ── upgradeConfig requester: failure and success paths (Req 4.1-4.4) ───── */
/* Barrier failure (ACKs withheld) must reach the requester as an error
* reply in the send_upgrade_error shape (status < 0, nodata), leave
* committed_config_tag unmoved, and keep restart_requested clear.
* RED: today the handler unconditionally ACKs status=0 with the
* upgradeConfigReply payload and requests the restart. */
static void test_upgrade_config_barrier_failure_error_reply_no_restart(void)
{
uint8_t *blob;
size_t blen = 0;
uint8_t sha[32];
cb5_reply_obs_t obs = { 0 };
/* Arrange */
cb5_setup_upgrade();
blob = cb5_make_config_blob(&blen);
SHA256(blob, blen, sha);
cb5_arrange_upgrade_req(blob, blen, sha);
/* Act: no standby ever acks the new tag */
(void)handle_upgrade_config(&g_cb5_c, &g_cb5_req_ep, &g_cb5_req);
/* Assert: error reply, no commit, no restart */
cb5_scan_replies(&g_cb5_req_ep, &obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
"the requester must receive exactly one reply");
TEST_ASSERT_TRUE_MESSAGE(obs.last_status < 0,
"a failed config commit must reach the requester as an error reply"
" (Req 4.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)RequestReply_data_nodata,
obs.last_data_which,
"the failure reply must use the send_upgrade_error shape (nodata)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0,
atomic_load_explicit(&g_cb5_c.ha.committed_config_tag,
memory_order_acquire),
"a failed upgrade must leave committed_config_tag unmoved (Req 4.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_cb5_c.restart_requested),
"a failed upgrade must not request the activation restart (Req 4.4)");
free(blob);
cb5_teardown();
printf("test_upgrade_config_barrier_failure_error_reply_no_restart: PASS\n");
}
/* Success path PIN: with a full majority of covering ACKs the existing
* behavior is unchanged — status-0 reply with the upgradeConfigReply
* payload, committed_config_tag at the saved tag, restart requested.
* Both standbys pre-ack UINT64_MAX (covers any tag — the new tag is only
* known after conf_save_config hashes the re-encoded blob), so the case
* is green under both the legacy counting and the per-push arithmetic. */
static void test_upgrade_config_majority_ack_success_unchanged(void)
{
uint8_t *blob;
size_t blen = 0;
uint8_t sha[32];
int rc;
cb5_reply_obs_t obs = { 0 };
/* Arrange: full covering pre-ack from both standbys */
cb5_setup_upgrade();
blob = cb5_make_config_blob(&blen);
SHA256(blob, blen, sha);
cb5_arrange_upgrade_req(blob, blen, sha);
repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[0], UINT64_MAX);
repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[1], UINT64_MAX);
/* Act */
rc = handle_upgrade_config(&g_cb5_c, &g_cb5_req_ep, &g_cb5_req);
/* Assert: existing success reply + restart behavior unchanged */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a majority-acked upgrade must succeed");
cb5_scan_replies(&g_cb5_req_ep, &obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
"the requester must receive exactly one reply");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status,
"the success ACK must carry status 0 (Req 4.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)RequestReply_data_upgradeconfigreplymsg,
obs.last_data_which,
"the success ACK must carry the upgradeConfigReply payload");
TEST_ASSERT_TRUE_MESSAGE(g_cb5_myconf.config_version_tag != 0,
"conf_save_config must have derived the new version tag");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(g_cb5_myconf.config_version_tag,
atomic_load_explicit(&g_cb5_c.ha.committed_config_tag,
memory_order_acquire),
"a majority-acked config must be committed at the saved tag (Req 4.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cb5_c.restart_requested),
"a successful upgrade must request the activation restart");
free(blob);
cb5_teardown();
printf("test_upgrade_config_majority_ack_success_unchanged: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 5.2 remediation r1 — wire-path config ACK ingestion (Req 4.1, 4.2).
*
* The barrier consumes acked_config_lsn, which only repl_on_config_ack
* advances — so the standby's OP_replConfigPush reply
* (replconfigpushreplymsg, config_repl.c:744-752) must be decoded on the
* primary and fed to repl_on_config_ack. This case drives the ACK over
* the WIRE: it builds the standby's capnp reply bytes and pushes them
* through peer_ctrl_msg_handler → repl_ack_decode_and_dispatch (the
* production decode entry for promoted peer endpoints), asserting the
* decoded tag lands in the replicator slot and completes a pending
* barrier. This also completes the unlanded half of ha-config-bootstrap
* task 8.1 (production ingestion of the config-push ACK), which until
* now was only exercised by direct repl_on_config_ack injection.
* ════════════════════════════════════════════════════════════════════════ */
/* ha.c test seams (test_ha_connection_reuse.c precedent) */
extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep,
ha_state_t *ha,
ha_elect_ctx_t *ctx,
int peer_idx,
uint64_t my_term,
uint64_t my_tag,
uint64_t my_lsn,
const char *my_name);
extern void vote_ep_ctx_set_cluster_for_test(endpoint_t *ep, cluster_t *c);
/* Build the standby's config-push ACK exactly as xce_reply() puts it on
* the wire (config_repl.c:744-752): Request{opcode=reply,
* body.replymsg=RequestReply{data.replconfigpushreplymsg=
* ReplConfigPushReply{accepted, versionTag}}}. The handler takes
* ownership (free_packet + free). */
static io_packet_t *cb5_build_config_ack_pkt(int accepted, uint64_t tag)
{
struct capn cs;
struct capn_segment *seg;
struct ReplConfigPushReply ack_capn;
ReplConfigPushReply_ptr ack_ptr;
struct RequestReply rr_capn;
RequestReply_ptr rr_ptr;
struct Request rq_capn;
Request_ptr rq_ptr;
io_packet_t *pkt;
capn_init_malloc(&cs);
seg = capn_root(&cs).seg;
memset(&ack_capn, 0, sizeof(ack_capn));
ack_capn.accepted = accepted ? 1 : 0;
ack_capn.versionTag = tag;
ack_ptr = new_ReplConfigPushReply(seg);
write_ReplConfigPushReply(&ack_capn, ack_ptr);
memset(&rr_capn, 0, sizeof(rr_capn));
rr_capn.seq = 1;
rr_capn.opcode = OPCode_replConfigPush;
rr_capn.status = 0;
rr_capn.data_which = RequestReply_data_replconfigpushreplymsg;
rr_capn.data.replconfigpushreplymsg = ack_ptr;
rr_ptr = new_RequestReply(seg);
write_RequestReply(&rr_capn, rr_ptr);
memset(&rq_capn, 0, sizeof(rq_capn));
rq_capn.opcode = OPCode_reply;
rq_capn.seq = 1;
rq_capn.body_which = Request_body_replymsg;
rq_capn.body.replymsg = rr_ptr;
rq_ptr = new_Request(seg);
write_Request(&rq_capn, rq_ptr);
capn_setp(capn_root(&cs), 0, rq_ptr.p);
pkt = calloc(1, sizeof(*pkt));
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "config ACK pkt alloc");
pkt->sz = capn_size(&cs);
TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, "config ACK pkt data alloc");
capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0);
capn_free(&cs);
return pkt;
}
typedef struct {
endpoint_t *ep;
io_packet_t *pkt;
unsigned delay_ms;
} cb5_wire_ack_args_t;
/* Deliver the wire ACK to the promoted-peer message handler after a delay,
* while the main thread is parked inside the barrier wait. */
static void *cb5_wire_ack_thread(void *arg)
{
cb5_wire_ack_args_t *a = (cb5_wire_ack_args_t *)arg;
usleep(a->delay_ms * 1000);
peer_ctrl_msg_handler_for_test(a->ep, a->pkt);
return NULL;
}
/* statics survive Unity longjmp */
static endpoint_t g_cb5_wire_ep;
static cb5_wire_ack_args_t g_cb5_wire_args;
/* A config ACK delivered as WIRE BYTES through the production decode entry
* (peer_ctrl_msg_handler → repl_ack_decode_and_dispatch) must reach
* repl_on_config_ack: the replicator slot's acked_config_lsn advances to
* the decoded tag and a pending configAck barrier completes (Req 4.1).
* RED (remediation r1): repl_ack_decode_and_dispatch has no
* replconfigpushreplymsg branch — the reply is silently dropped, the
* barrier times out, rc=-ETIMEDOUT. */
static void test_config_ack_wire_path_completes_barrier(void)
{
const uint64_t push_tag = 0xC0FFEE;
pthread_t th;
int rc;
uint64_t acked = 0;
int found = 0;
/* Arrange: cb5 cluster (wired engine n_peers=3, two never-acking
* standbys) + a third, promoted-peer endpoint whose vote_ep_ctx routes
* peer_ctrl_msg_handler to repl_ack_decode_and_dispatch (role PRIMARY). */
cb5_setup();
memset(&g_cb5_wire_ep, 0, sizeof(g_cb5_wire_ep));
g_cb5_wire_ep.fd = -1;
vote_ep_ctx_build_for_test(&g_cb5_wire_ep, &g_cb5_c.ha, NULL, 0,
1, 0, 0, "ctrl-cb5");
vote_ep_ctx_set_cluster_for_test(&g_cb5_wire_ep, &g_cb5_c);
atomic_store_explicit(&g_cb5_c.ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
&g_cb5_wire_ep, "cb5_wire", 7502) >= 0);
/* Act: park in the barrier (self + 1 covering = 2 > 3/2 quorum), then
* deliver the standby's ACK as raw capnp bytes through the handler. */
g_cb5_wire_args.ep = &g_cb5_wire_ep;
g_cb5_wire_args.pkt = cb5_build_config_ack_pkt(1, push_tag);
g_cb5_wire_args.delay_ms = 50;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, cb5_wire_ack_thread,
&g_cb5_wire_args));
rc = repl_wait_majority_ack(&g_cb5_c.persist.repl, push_tag, 1000,
(uint8_t)OPCode_upgradeConfigAck);
pthread_join(th, NULL);
/* Assert: barrier completed off the wire-decoded ACK … */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a wire-delivered config ACK must complete a pending barrier"
" (production ingestion path — Req 4.1)");
/* … because repl_on_config_ack advanced the slot to the decoded tag */
for (int i = 0; i < MAX_STANDBYS; i++) {
if (g_cb5_c.persist.repl.standbys[i].ep == (void *)&g_cb5_wire_ep) {
acked = atomic_load_explicit(
&g_cb5_c.persist.repl.standbys[i].acked_config_lsn,
memory_order_acquire);
found = 1;
break;
}
}
TEST_ASSERT_TRUE_MESSAGE(found, "wire ep must be registered in the replicator");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(push_tag, acked,
"the decoded versionTag must land in acked_config_lsn via"
" repl_on_config_ack");
free(g_cb5_wire_ep.data); /* vctx from vote_ep_ctx_build_for_test */
g_cb5_wire_ep.data = NULL;
cb5_teardown();
printf("test_config_ack_wire_path_completes_barrier: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* job-array-single-wal-record task 2.1 — persist_submit_job_array_async:
* the single-record array append and commit path (Req 1.1, 1.2, 1.3, 4.2,
* 9.2). Fixture mirrors test_fact_hook_append_is_retained_and_nonblocking
* above (standalone persist_t, no cluster_t -- p->cluster stays NULL so
* the append routes through the persist_wal_async fallback branch inside
* append_job_array_wal_entry_local; real tmpdir WAL; wired commit engine
* with 2 stub standbys; ACKs driven through repl_on_ack). commit_engine_init
* here never wires a vtaskmgr pool (ce->pool stays NULL), so
* commit_resolve_pending's continuation dispatch is the synchronous,
* same-thread inline fallback (cerebrum precedent: "the NO-POOL
* inline-dispatch fallback calls fn() synchronously ON THE CALLING
* THREAD") -- a covering repl_on_ack call resolves a registered
* continuation before it returns, so no polling is needed anywhere below.
*
* Deliberately calls persist_submit_job_array_async() DIRECTLY -- job.c's
* handle_jobsubmit is not wired to this function yet (that is task 2.2's
* boundary, not this one's); nothing in job.c/handle_jobsubmit is touched
* here.
* ════════════════════════════════════════════════════════════════════════ */
static persist_t g_jap; /* stub persist for array-submit tests */
static commit_engine_t g_jace;
static int g_jaep_a, g_jaep_b; /* dummy standby endpoints */
static char g_jap_tmpdir[256];
static void jap_setup(uint32_t sync_timeout_ms)
{
persist_config_t cfg;
memset(&g_jap, 0, sizeof(g_jap));
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = sync_timeout_ms;
snprintf(g_jap_tmpdir, sizeof(g_jap_tmpdir), "/tmp/xce_jap21_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_jap_tmpdir));
persist_set_workdir(&g_jap, g_jap_tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_jap, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_jap));
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until
* a test explicitly delivers them via repl_on_ack. */
TEST_ASSERT_TRUE(
repl_add_standby(&g_jap.repl, &g_jaep_a, "jasb_a", 7420) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_jap.repl, &g_jaep_b, "jasb_b", 7421) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_jace, &g_jap, 3, 8));
repl_set_commit_engine(&g_jap.repl, &g_jace);
repl_broadcast_send_hook = cbr_drop_send_hook;
/* Reserved-0 contract: a fresh store starts current_lsn=1 -- the first
* array record lands at LSN 1 with no priming append needed. */
TEST_ASSERT_EQUAL_UINT64(1, g_jap.current_lsn);
}
static void jap_teardown(void)
{
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_jace);
persist_stop(&g_jap);
persist_fini(&g_jap);
/* tmpdir left on disk — /tmp is ephemeral */
}
/* Minimal job_submit_data_t + xce_job_submit_t base spec -- same shape as
* cbrp_arrange_jobsubmit (test_commit_barrier_reply.c) and
* test_data_job_array_variant_round_trip (test_e2e_capnp.c); start_info
* left NULL, which that round-trip test already proves
* encode_JobStartInfo_ptr handles safely. */
static job_submit_data_t g_jap_submit;
static xce_job_submit_t g_jap_spec;
static void jap_arrange_base_spec(void)
{
memset(&g_jap_submit, 0, sizeof(g_jap_submit));
g_jap_submit.queue = (char *)"normal";
g_jap_submit.user = (char *)"tester";
g_jap_submit.mintasks = 1;
g_jap_submit.maxtasks = 1;
g_jap_submit.start_info = NULL;
g_jap_submit.job_name = (char *)"array-job";
memset(&g_jap_spec, 0, sizeof(g_jap_spec));
/* Deliberately NOT set to a test's real base_jobid/array_size: proves
* persist_submit_job_array_async uses its own explicit base_jobid/
* array_size PARAMETERS, never these same-shaped struct fields. */
g_jap_spec.jobid = 0;
g_jap_spec.array_size = 0;
g_jap_spec.submit = &g_jap_submit;
}
/* Read the WAL entry header located at explicit segment offset `at` back
* from the active segment file. Unlike read_last_wal_hdr in
* test_ha_role_lifecycle.c (which back-computes `seg_offset -
* sizeof(hdr)` and is only ever used for zero-payload WAL_NOOP entries,
* where that arithmetic happens to land on the header), this helper takes
* the header's start offset EXPLICITLY -- the caller snapshots
* p->seg_offset immediately before the append it wants to read back.
* Back-computing from the post-append p->seg_offset is only valid when
* payload_len == 0; every WAL_JOB_ARRAY_SUBMIT entry here carries a
* nonzero capnp payload, so `seg_offset - sizeof(hdr)` would land inside
* the payload bytes, not the header (caught by task 2.1 RED: read garbage
* decoded as a bogus type). */
static int jap_read_wal_hdr_at(persist_t *p, off_t at, wal_entry_hdr_t *out)
{
if (at < 0 || p->seg_offset < (uint64_t)at + sizeof(wal_entry_hdr_t))
return -ENOENT;
int fd = open(p->seg_path, O_RDONLY);
if (fd < 0)
return -errno;
ssize_t r = pread(fd, out, sizeof(*out), at);
close(fd);
if (r != (ssize_t)sizeof(*out))
return -EIO;
return 0;
}
/* Read the header AND payload bytes of the WAL entry located at explicit
* segment offset `at` back from the active segment file. Caller must
* free(*out_payload) (NULL/0-len payload leaves *out_payload NULL). */
static int jap_read_wal_entry_at(persist_t *p, off_t at,
wal_entry_hdr_t *out_hdr,
uint8_t **out_payload)
{
int rc = jap_read_wal_hdr_at(p, at, out_hdr);
if (rc != 0)
return rc;
if (out_hdr->payload_len == 0) {
*out_payload = NULL;
return 0;
}
int fd = open(p->seg_path, O_RDONLY);
if (fd < 0)
return -errno;
off_t payload_at = at + (off_t)sizeof(wal_entry_hdr_t);
uint8_t *buf = (uint8_t *)malloc(out_hdr->payload_len);
TEST_ASSERT_NOT_NULL(buf);
ssize_t r = pread(fd, buf, out_hdr->payload_len, payload_at);
close(fd);
if (r != (ssize_t)out_hdr->payload_len) {
free(buf);
return -EIO;
}
*out_payload = buf;
return 0;
}
/* Notify-callback capture (persist_job_commit_notify_fn). */
typedef struct {
int call_count;
uint64_t last_lsn;
int last_rc;
} jap_notify_obs_t;
static jap_notify_obs_t g_jap_obs;
static void jap_notify_reset(void)
{
memset(&g_jap_obs, 0, sizeof(g_jap_obs));
}
static void jap_notify_fn(void *ctx, uint64_t lsn, int rc)
{
jap_notify_obs_t *obs = (jap_notify_obs_t *)ctx;
obs->call_count++;
obs->last_lsn = lsn;
obs->last_rc = rc;
}
/* ── One WAL append, one commit continuation, decodable back to the same
* (base_jobid, array_size, submit spec) — Req 1.1, 1.2, 1.3, 4.2, 9.2 ── */
static void test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation(void)
{
const uint64_t base_jobid = 9500;
const uint64_t array_size = 5;
uint64_t lsn_before, out_lsn = 0;
int rc;
off_t entry_off;
wal_entry_hdr_t hdr;
uint8_t *payload = NULL;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
lsn_before = g_jap.current_lsn;
entry_off = (off_t)g_jap.seg_offset;
/* Act: append the array record; the continuation is registered but not
* yet resolved (no covering ACK delivered yet). */
rc = persist_submit_job_array_async(&g_jap, base_jobid, array_size,
&g_jap_spec, &out_lsn,
jap_notify_fn, &g_jap_obs);
/* Assert: exactly one WAL entry was appended -- current_lsn advanced by
* exactly 1, regardless of array_size=5 */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a valid array submission must append and register successfully");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, out_lsn,
"the array record must land at the pre-append current_lsn");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn,
"exactly ONE WAL entry must be appended for the whole array, "
"not one per element");
/* ...and it is the ONE entry on disk, typed WAL_JOB_ARRAY_SUBMIT,
* decodable back to the same (base_jobid, array_size, submit spec) --
* NOT g_jap_spec's own (deliberately mismatched) jobid/array_size
* fields, proving the function used its explicit parameters */
TEST_ASSERT_EQUAL_INT(0,
jap_read_wal_entry_at(&g_jap, entry_off, &hdr, &payload));
TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)WAL_JOB_ARRAY_SUBMIT, hdr.type,
"the array record's WAL type must be WAL_JOB_ARRAY_SUBMIT");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(out_lsn, hdr.lsn,
"the on-disk entry's lsn must match the returned out_lsn");
{
struct capn cctx;
TEST_ASSERT_EQUAL_INT(0,
capn_init_mem(&cctx, payload, hdr.payload_len, 0));
JobSubmit_ptr jp;
jp.p = capn_getp(capn_root(&cctx), 0, 1);
TEST_ASSERT_TRUE(jp.p.type != CAPN_NULL);
struct JobSubmit wire;
memset(&wire, 0, sizeof(wire));
read_JobSubmit(&wire, jp);
xce_job_submit_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_JobSubmit(&decoded, &wire);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, decoded.jobid,
"decoded base jobid must match the base_jobid PARAMETER");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(array_size, decoded.array_size,
"decoded array_size must match the array_size PARAMETER");
TEST_ASSERT_NOT_NULL(decoded.submit);
TEST_ASSERT_EQUAL_STRING("normal", decoded.submit->queue);
TEST_ASSERT_EQUAL_STRING("tester", decoded.submit->user);
free_JobSubmit(&decoded);
capn_free(&cctx);
}
free(payload);
/* ...exactly one continuation is registered -- not yet resolved before
* any ACK arrives */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count,
"the notify callback must not fire before a covering commit");
/* ...and fires EXACTLY ONCE, with rc==0, once a covering majority ACK
* resolves the single registered continuation */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count,
"the notify callback must fire EXACTLY ONCE for the whole array, "
"not once per element");
TEST_ASSERT_EQUAL_UINT64(out_lsn, g_jap_obs.last_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc,
"a majority-committed array record must resolve rc=0");
jap_teardown();
printf("test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation: PASS\n");
}
/* ── Encoded payload size does not depend on array_size (Req 7.1, 7.2) ──── */
static void test_job_array_submit_payload_size_independent_of_array_size(void)
{
wal_entry_hdr_t hdr_small, hdr_large;
uint64_t out_lsn = 0;
int rc;
off_t off_small, off_large;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
/* Act: array_size=2, then array_size=10000, SAME base spec -- both
* entries encoded through the identical code path in the same fixture
* so their on-disk payload lengths are directly comparable. */
off_small = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 9600, 2, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_small, &hdr_small));
jap_notify_reset();
off_large = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 20000, 10000, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_large, &hdr_large));
/* Assert: the encoded payload's size does not depend on array_size --
* arraySize is a fixed-width uint64 field, not a per-element list */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len,
hdr_large.payload_len,
"the WAL_JOB_ARRAY_SUBMIT payload size must be IDENTICAL for "
"array_size=2 and array_size=10000 -- only the fixed-width "
"arraySize field differs, never the encoded byte length");
/* Cleanup: resolve both pending continuations with one covering ACK
* before teardown (hygiene — avoids leaving unresolved slots for
* commit_engine_fini to silently drop). */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, hdr_large.lsn);
jap_teardown();
printf("test_job_array_submit_payload_size_independent_of_array_size: PASS\n");
}
/* ── Invalid-argument rejection (defensive input validation) ────────────── */
static void test_job_array_submit_rejects_array_size_one_or_less(void)
{
uint64_t out_lsn = 0;
int rc;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
/* Act / Assert: array_size<=1 is a single-job submission's shape, not
* this function's precondition (design.md: "array_size > 1") -- reject
* before any WAL append, no continuation registered, no callback fired */
rc = persist_submit_job_array_async(&g_jap, 9700, 1, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc,
"array_size == 1 must be rejected -- not this function's job");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn,
"a rejected submission must never append a WAL entry");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count,
"a rejected submission's callback contract is 'never called', not "
"'called with an error' — the caller never registered anything");
rc = persist_submit_job_array_async(&g_jap, 9700, 0, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc,
"array_size == 0 must be rejected");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn,
"a rejected submission must never append a WAL entry");
jap_teardown();
printf("test_job_array_submit_rejects_array_size_one_or_less: PASS\n");
}
/* ── One entry / one continuation holds across several array sizes
* (task 2.3, Req 1.1-1.3) ────────────────────────────────────────────────
* test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation
* above pins the full one-append/one-continuation observable at exactly
* one size (5). This sweep repeats that SAME observable (append succeeds,
* current_lsn advances by exactly one, the continuation does not resolve
* before a covering ACK, and resolves exactly once after one) at several
* materially different sizes in a single test, proving the property is not
* size-specific. */
static void test_job_array_submit_one_entry_one_continuation_across_sizes(void)
{
static const uint64_t sizes[] = { 2, 3, 100, 1000 };
size_t i;
jap_setup(2000);
jap_arrange_base_spec();
for (i = 0; i < sizeof(sizes) / sizeof(sizes[0]); i++) {
const uint64_t array_size = sizes[i];
const uint64_t base_jobid = 30000 + (uint64_t)i * 100000;
uint64_t lsn_before, out_lsn = 0;
int rc;
char msg[128];
jap_notify_reset();
lsn_before = g_jap.current_lsn;
rc = persist_submit_job_array_async(&g_jap, base_jobid, array_size,
&g_jap_spec, &out_lsn,
jap_notify_fn, &g_jap_obs);
snprintf(msg, sizeof(msg),
"array_size=%llu: append must succeed",
(unsigned long long)array_size);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, msg);
snprintf(msg, sizeof(msg),
"array_size=%llu: exactly ONE WAL entry must be appended, "
"not one per element",
(unsigned long long)array_size);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn,
msg);
snprintf(msg, sizeof(msg),
"array_size=%llu: continuation must not resolve before a "
"covering ACK", (unsigned long long)array_size);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count, msg);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn);
snprintf(msg, sizeof(msg),
"array_size=%llu: exactly ONE commit continuation must "
"resolve for the whole array, not one per element",
(unsigned long long)array_size);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, msg);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, msg);
}
jap_teardown();
printf("test_job_array_submit_one_entry_one_continuation_across_sizes: "
"PASS\n");
}
/* ── Payload size stays bounded at the actual upstream array-size ceiling
* (task 2.3, Req 7.1, 7.2) ───────────────────────────────────────────────
* test_job_array_submit_payload_size_independent_of_array_size above
* compares an arbitrary "large" sample (10000) against a small one. This
* test additionally exercises the REAL structural ceiling design.md relies
* on to satisfy Requirement 7 without a chunking policy: idgen() rejects
* arraySize > MAX_ARRAY_IDX + 1, so MAX_ARRAY_IDX + 1 (32768, config.h) is
* the largest array_size that can ever legally reach this function in
* production. Confirms the encoded payload at that exact ceiling is still
* identical in size to a tiny array's payload AND stays well under a sane
* absolute byte bound -- Requirement 7.1's "no single record exceeds a
* bounded size" is an absolute claim, not only a relative one. */
static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void)
{
const uint64_t ceiling_array_size = MAX_ARRAY_IDX + 1; /* 32768 */
wal_entry_hdr_t hdr_small, hdr_ceiling;
uint64_t out_lsn = 0;
int rc;
off_t off_small, off_ceiling;
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
off_small = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 40000, 2, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0,
jap_read_wal_hdr_at(&g_jap, off_small, &hdr_small));
jap_notify_reset();
off_ceiling = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 50000, ceiling_array_size,
&g_jap_spec, &out_lsn,
jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0,
jap_read_wal_hdr_at(&g_jap, off_ceiling, &hdr_ceiling));
TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len,
hdr_ceiling.payload_len,
"the payload size at the actual upstream array_size ceiling "
"(MAX_ARRAY_IDX + 1 = 32768) must be identical to a tiny array's "
"payload size -- proving Requirement 7's bound holds at the real "
"cap, not just at an arbitrary large sample");
TEST_ASSERT_TRUE_MESSAGE(hdr_ceiling.payload_len < 4096,
"the array record's encoded payload must stay well under a sane "
"absolute bound even at the maximum legal array_size (Req 7.1)");
/* Cleanup: resolve both pending continuations before teardown. */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, hdr_ceiling.lsn);
jap_teardown();
printf("test_job_array_submit_payload_size_bounded_at_max_array_idx: "
"PASS\n");
}
/* ── Non-array (array_size == 1) baseline: unmodified WAL_JOB_SUBMIT shape
* (task 2.3, Req 8.1) ────────────────────────────────────────────────────
* Drives persist_submit_job_payloads_async() -- the pre-existing,
* completely UNMODIFIED single-job append function job.c's array_size == 1
* path still calls -- directly with exactly the one-payload xlist_t shape
* build_job_submit_payloads() constructs for a non-array submission.
* Confirms the concrete regression sentinel this feature must never
* disturb: the on-disk WAL entry type stays WAL_JOB_SUBMIT (never
* WAL_JOB_ARRAY_SUBMIT), the payload decodes via struct Job/xce_job_t
* (never struct JobSubmit/xce_job_submit_t), exactly one entry is
* appended, and exactly one commit continuation resolves -- the same four
* properties a pre-feature build already guaranteed, now pinned so a
* future change cannot silently route array_size == 1 into the
* array-shaped path or otherwise perturb the single-job wire format. */
static void test_job_submit_single_payload_wal_type_matches_pre_feature_baseline(
void)
{
const uint64_t jobid = 60000;
time_t now = time(NULL);
xce_job_t *jp;
xlist_t plist;
uint64_t lsns[1] = { 0 };
uint64_t lsn_before;
off_t entry_off;
wal_entry_hdr_t hdr;
uint8_t *payload = NULL;
int rc;
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
/* One payload -- the exact shape build_job_submit_payloads() (job.c)
* produces for a single-element (non-array) submission. */
TEST_ASSERT_EQUAL_INT(0, xlist_init(&plist, 1));
jp = calloc(1, sizeof(*jp));
TEST_ASSERT_NOT_NULL(jp);
jp->submit = &g_jap_submit;
jp->jobid = jobid;
jp->sub_time = now;
jp->queue_time = now;
jp->state = JOB_STAT_PEND;
TEST_ASSERT_TRUE(xlist_store(&plist, jp) >= 0);
lsn_before = g_jap.current_lsn;
entry_off = (off_t)g_jap.seg_offset;
rc = persist_submit_job_payloads_async(&g_jap, &plist, lsns,
jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"the unmodified single-job append path must still succeed");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn,
"a single (non-array) submission must append exactly one WAL "
"entry, same as before this feature");
TEST_ASSERT_EQUAL_INT(0,
jap_read_wal_entry_at(&g_jap, entry_off, &hdr, &payload));
TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)WAL_JOB_SUBMIT, hdr.type,
"a non-array submission's WAL entry must still be typed "
"WAL_JOB_SUBMIT, never WAL_JOB_ARRAY_SUBMIT (Req 8.1)");
{
struct capn cctx;
TEST_ASSERT_EQUAL_INT(0,
capn_init_mem(&cctx, payload, hdr.payload_len, 0));
Job_ptr jobp;
jobp.p = capn_getp(capn_root(&cctx), 0, 1);
TEST_ASSERT_TRUE(jobp.p.type != CAPN_NULL);
struct Job wire;
memset(&wire, 0, sizeof(wire));
read_Job(&wire, jobp);
xce_job_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_Job(&decoded, &wire);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(jobid, decoded.jobid,
"the decoded jobid must round-trip through the unmodified "
"struct Job/xce_job_t wire shape, never struct JobSubmit");
TEST_ASSERT_NOT_NULL(decoded.submit);
TEST_ASSERT_EQUAL_STRING("normal", decoded.submit->queue);
TEST_ASSERT_EQUAL_STRING("tester", decoded.submit->user);
free_Job(&decoded);
capn_free(&cctx);
}
free(payload);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count,
"the continuation must not resolve before a covering ACK");
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, lsns[0]);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count,
"exactly ONE commit continuation must resolve for the single "
"payload, same contract as before this feature");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, 0);
xlist_fini(&plist);
free(jp);
jap_teardown();
printf("test_job_submit_single_payload_wal_type_matches_pre_feature_"
"baseline: PASS\n");
}
/* ── Test group entry point ──────────────────────────────────────────────── */
void test_commit_barrier_rollout(void)
{
UnitySetTestFile(__FILE__);
/* Task 1.1 — classification totality + RETAIN routing (Req 2.1, 2.2, 3.1) */
RUN_TEST(test_classify_job_submit_is_wait);
RUN_TEST(test_classify_job_array_submit_is_wait);
RUN_TEST(test_classify_queue_ctrl_is_wait);
RUN_TEST(test_classify_worker_ctrl_detail_split);
RUN_TEST(test_classify_all_job_state_types_are_retain);
RUN_TEST(test_classify_checkpoint_is_local);
RUN_TEST(test_classify_is_total_over_wal_entry_type);
RUN_TEST(test_classify_unknown_type_fails_fast_in_testing);
RUN_TEST(test_fact_hook_append_is_retained_and_nonblocking);
/* Task 2.1 — persist_commit_barrier semantics (Req 1.1, 1.2, 1.3) */
RUN_TEST(test_barrier_returns_zero_when_nothing_appended);
RUN_TEST(test_barrier_returns_zero_when_committed_covers_appended);
RUN_TEST(test_barrier_times_out_when_acks_withheld);
RUN_TEST(test_barrier_returns_zero_once_covering_ack_arrives);
RUN_TEST(test_barrier_zero_timeout_falls_back_to_config_sync_timeout);
RUN_TEST(test_barrier_single_member_returns_immediately);
RUN_TEST(test_barrier_not_ready_passes_through_unmodified);
/* Task 4.1 — dispatch-gate fact allowlist (Req 3.2) */
RUN_TEST(test_gate_no_quorum_admits_notify_fact);
RUN_TEST(test_gate_no_quorum_admits_update_fact);
RUN_TEST(test_gate_no_quorum_admits_setpreason_fact);
RUN_TEST(test_gate_no_quorum_admits_clrpreason_fact);
RUN_TEST(test_gate_no_quorum_still_rejects_request_opcodes);
RUN_TEST(test_gate_with_quorum_admits_all_opcodes);
/* Task 5.1 — config barrier: configAck arithmetic + upgrade failure
* path (Req 4.1, 4.2, 4.3, 4.4) */
RUN_TEST(test_configack_previous_tag_ack_does_not_count);
RUN_TEST(test_configack_no_active_standbys_times_out);
RUN_TEST(test_configack_self_plus_one_covering_is_majority_of_three);
RUN_TEST(test_configack_single_member_immediate_success);
RUN_TEST(test_push_config_withheld_acks_returns_etimedout);
RUN_TEST(test_push_config_majority_acks_returns_zero_and_commits);
RUN_TEST(test_upgrade_config_barrier_failure_error_reply_no_restart);
RUN_TEST(test_upgrade_config_majority_ack_success_unchanged);
/* Task 5.2 remediation r1 — production wire-path ACK ingestion
* (completes the unlanded half of ha-config-bootstrap task 8.1) */
RUN_TEST(test_config_ack_wire_path_completes_barrier);
/* job-array-single-wal-record task 2.1 — persist_submit_job_array_async:
* single-record array append/commit path (Req 1.1, 1.2, 1.3, 4.2, 9.2) */
RUN_TEST(test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation);
RUN_TEST(test_job_array_submit_payload_size_independent_of_array_size);
RUN_TEST(test_job_array_submit_rejects_array_size_one_or_less);
/* job-array-single-wal-record task 2.3 — unit tests for the array
* append/commit path (Req 1.1-1.3, 7.1, 7.2, 8.1) */
RUN_TEST(test_job_array_submit_one_entry_one_continuation_across_sizes);
RUN_TEST(test_job_array_submit_payload_size_bounded_at_max_array_idx);
RUN_TEST(test_job_submit_single_payload_wal_type_matches_pre_feature_baseline);
}
Reference in a new issue View git blame Copy permalink