lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_commit_barrier_reply.c
Rongsong Shen 7519fa96dd fix(ha-commit-engine,persist): reserve WAL lsn 0 — first entry starts at lsn 1 
Root cause (found via live 3-node E2E + a verification test): lsn was 0-based
(hdr.lsn = current_lsn++ from current_lsn=0), and 0 was overloaded as a
sentinel, causing three cold-start defects on a freshly-promoted leader:
  - the first WAL entry (own-term promotion NOOP at lsn 0) was structurally
    uncommittable: commit_advance_locked only stores when cand>cur and
    committed_lsn starts at 0, so lsn 0 can never advance it;
  - commit_set_writable_lsn(noop_lsn=0) set writable_lsn=0, which wait_done
    reads as 'gate off' — the own-term safety gate was a no-op on the first leader;
  - bug-543 (WAL hole): ha_stream_wal_to_standby filters hdr.lsn>standby_lsn;
    a fresh empty standby reports standby_lsn=0 (ambiguous 'empty' vs 'has lsn 0'),
    so lsn 0 failed 0>0 and was dropped -> non-contiguous standby WAL.

Fix: a fresh persist store initializes current_lsn=1, reserving lsn 0 as the
none/empty sentinel. The first real entry is lsn 1 -> committable (0->1),
writable_lsn=1 -> gate effective, and standby_lsn=0 unambiguously means 'empty'
so catch-up streams from the first entry with no hole. persist_recover is
data-driven (current_lsn=max(hdr.lsn+1)), so existing on-disk WALs and non-fresh
recovery are unaffected; the change only sets the base for brand-new stores.
No wire/capnp change.

Verification: test_coldstart_lsn0 drives the real commit engine + real streamer
and proves H1 (first entry commits 0->1), H3 (gate effective + blocks when
committed<writable), H4 (empty-standby catch-up is contiguous [1,2], control
standby_lsn=1 gets only [2]). Full controller suite green under ASAN. Tests that
relied on the lsn-0 base were shifted (several strengthened); commit.c /
config_repl.c / ha.c / wire format untouched.

Owning specs: ha-commit-engine / ha-wal-replication / persist. Follow-up:
sweep stale lsn-0 comments (repl.c:429-436, persist.c:1055-1075/1240-1243/1296).
2026-06-14 10:46:16 +00:00

1894 lines
82 KiB
C
Raw Blame History

/*
* test_commit_barrier_reply.c — RED-phase integration tests for
* commit-first-then-reply per opcode (ha-commit-barrier-rollout task 3.1).
* Requirements 1.1 (success reply only after majority commit), 1.2 (commit
* timeout → -ETIMEDOUT error reply), 1.3 (barrier rejection → error reply,
* never success), 1.4 (no success-interpretable reply pre-commit).
*
* ALSO HOSTS (task 3.3, Req 2.3): the receipt-ACK verdict pins for
* handle_jobdispatch, do_set_preason / do_clr_preason, and the
* do_update_op rusage path — see the "Receipt-ACK verdict pins" section.
* These land GREEN by design (pins of current correct behavior, like the
* 3.1 validation pins); they live here, not in a new file, because they
* reuse this file's entire harness: cluster + real persist + wired engine
* + stub standby registry + reply-capture endpoint + the job.c link.
*
* Drives the four request-shaped handlers DIRECTLY on a worker pthread
* (vtask-context stand-in; production dispatch: cluster.c:325-363) against
* a real tmpdir persist_t embedded in a test cluster_t with registered
* persistence hooks, a wired commit engine (n_peers=3), a stub standby
* registry (2 dummy standbys), ACKs driven through repl_on_ack:
* do_jobcontrol_op — not-started-job branch → WAL_JOB_FAIL (SIGKILL)
* do_queuecontrol_op — QUEUE_CLOSE → WAL_QUEUE_CTRL
* do_hostcontrol_op — HOST_CLOSE → WAL_WORKER_CTRL/UPDATE
* handle_worker_join — first-time join → WAL_WORKER_CTRL/JOIN
* (via handle_join — handle_worker_join is static)
*
* Reply observation harness: there is no xce_reply test hook in src/lib
* (out of boundary), so replies are observed where the production send
* path deposits them — xce_reply() encodes the reply Request and
* endpoint_enq()s it onto ep->outq (request.c:334-368, netio.c:796). The
* requester endpoint is a zeroed static endpoint_t with no iomgr:
* endpoint_update_epoll() is a no-op and the zeroed pthread mutexes are
* valid locks on Linux, so packets stay queued. Tests drain via
* endpoint_deq() and parse with the raw capnp reader (capn_init_mem →
* read_Request → read_RequestReply), the raw-packet parse pattern of
* test_config_repl.c — least invasive, zero production changes.
*
* RED expectations against the current reply-before-commit handlers
* (anchors re-verified 2026-06-12: job.c:932, queue.c:130, worker.c:700,
* join.c:107-109):
* - *_withheld_acks_replies_error / *_success_reply_only_after_commit:
* the handlers reply immediately, before any covering ACK → the
* Req 1.1/1.2/1.4 ordering assertions fail. Additional RED finding:
* queue_ctrl/worker_ctrl/job_ctrl leak key_find's found rc (== 1), so
* queue/host control skip their AFTER_EVENT hook (NO WAL entry) and
* reply status 1 — pinned by the append assertion, 3.2 must close it.
* - *_error_reply_immediate: local validation failures already reply
* immediately with their existing codes — contract pins that stay
* green before AND after the task-3.2 restructure.
*
* Fixtures reused from test_commit_barrier_rollout.c (engine + stub
* registry + drop send hook + LSN-0 priming append) and test_worker_join.c
* (kvmgrs, resources, hostattrs, hostconfig; its collect_confvars stub
* also serves this file's join path).
*/
/* rtti.h must come first — it defines 'var' and 'typeinfo_t' used by kvmgr.h */
#include <rtti.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <capnp_c.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "cluster.h"
#include "clusterconfig.h"
#include "commit.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "myconf.h"
#include "netio.h"
#include "persist.h"
#include "qobj.h"
#include "repl.h"
#include "resobj.h"
#include "request.h"
#include "typemap.h"
#include "wal_types.h"
#include "wobj.h"
#include "xcealloc.h"
#include "xstring.h"
#include "zarray.h"
/* ── Link stubs ──────────────────────────────────────────────────────────
* job.c / join.c / opupd.c reference these cross-module functions (dataex.c,
* misc.c, sched.c, apiserver.c — not part of the unit-test link). For the
* 3.1 paths (jobctrl not-started branch, queue/host ctrl, worker join) all
* stubs are inert; for the 3.3 verdict pins two of them gained test
* controls with NULL/no-op defaults that preserve the 3.1 behavior:
* notify() — captures dispatch-outcome notifications (jobdispatch
* verdict pin observes notify_dispatch through it)
* get_ep_worker() — returns g_stub_ep_worker (rusage verdict pin needs a
* reporting worker; production impl is misc.c:49) */
static endpoint_t g_req_ep; /* requester endpoint (capture; set up below) */
/* notify() capture — records each xce_notify_data_t delivery. For
* jobDispatchFail events (the shared ok/fail event of notify_dispatch,
* job.c) jobid + code are unpacked; reply_already_enqueued snapshots
* whether the receipt ACK was already sitting on the requester outq when
* the outcome notification fired (single-threaded handler call → exact
* ordering observation, no race). */
#define CBRP_MAX_NOTIFY 8
static int g_notify_count;
static int g_notify_events[CBRP_MAX_NOTIFY];
static uint64_t g_notify_jobids[CBRP_MAX_NOTIFY];
static int g_notify_codes[CBRP_MAX_NOTIFY];
static int g_notify_reply_already_enqueued[CBRP_MAX_NOTIFY];
static void cbrp_notify_capture_reset(void)
{
g_notify_count = 0;
memset(g_notify_events, 0, sizeof(g_notify_events));
memset(g_notify_jobids, 0, sizeof(g_notify_jobids));
memset(g_notify_codes, 0, sizeof(g_notify_codes));
memset(g_notify_reply_already_enqueued, 0,
sizeof(g_notify_reply_already_enqueued));
}
static worker_t *g_stub_ep_worker; /* NULL by default (3.1 behavior) */
/* clang-format off */
int notify(cluster_t *c, endpoint_t *e, xce_notify_data_t *d)
{
(void)c; (void)e;
if (g_notify_count < CBRP_MAX_NOTIFY) {
g_notify_events[g_notify_count] = d->event;
if (d->event == NotifyData_data_jobDispatchFail) {
g_notify_jobids[g_notify_count] = d->data.dispatchfail->jobid;
g_notify_codes[g_notify_count] = d->data.dispatchfail->code;
}
g_notify_reply_already_enqueued[g_notify_count] =
!DATAQ_EMPTY(&g_req_ep.outq);
}
g_notify_count++;
return 0;
}
worker_t *get_ep_worker(cluster_t *c, endpoint_t *e)
{ (void)c; (void)e; return g_stub_ep_worker; }
int update_host_runusage(xce_job_runusage_t **r, char *h, int n,
xce_rusage_t **u)
{ (void)r; (void)h; (void)n; (void)u; return 0; }
int update_pids(job_pidinfo_t **i, int n, int *p)
{ (void)i; (void)n; (void)p; return 0; }
int update_pgids(job_pidinfo_t **i, int n, int *p)
{ (void)i; (void)n; (void)p; return 0; }
int on_scheduler_join(cluster_t *c) { (void)c; return 0; }
int add_apiserver(cluster_t *c, char *n, endpoint_t *e)
{ (void)c; (void)n; (void)e; return 0; }
int on_apiserver_join(cluster_t *c, endpoint_t *e)
{ (void)c; (void)e; return 0; }
/* clang-format on */
/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast
* from dereferencing the dummy standby endpoints. */
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int cbrp_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 */
}
static uint64_t cbrp_mono_ms(void)
{
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
}
static void cbrp_sleep_ms(uint32_t ms)
{
struct timespec ts;
ts.tv_sec = ms / 1000u;
ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
nanosleep(&ts, NULL);
}
/* ── Reply observation (see file header for the harness rationale) ───── */
typedef struct {
int n_replies; /* replymsg packets seen by this scan */
int32_t last_status; /* status of the last replymsg observed */
uint64_t committed_at_obs; /* committed_lsn sampled at observation time */
} cbrp_reply_obs_t;
static persist_t *g_obs_persist; /* sampled into committed_at_obs */
/* Drain every packet queued on ep->outq; record replymsg status. Non-reply
* packets (e.g. on_worker_join's config push) are parsed and discarded. */
static void cbrp_scan_replies(endpoint_t *ep, cbrp_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->committed_at_obs = atomic_load(&g_obs_persist->committed_lsn);
}
capn_free(&c);
free_packet(pkt);
free(pkt);
}
}
/* Poll until at least one replymsg is observed or the deadline expires. */
static void cbrp_poll_reply(endpoint_t *ep, uint32_t deadline_ms,
cbrp_reply_obs_t *obs)
{
uint64_t t0 = cbrp_mono_ms();
for (;;) {
cbrp_scan_replies(ep, obs);
if (obs->n_replies > 0 || cbrp_mono_ms() - t0 >= deadline_ms)
return;
cbrp_sleep_ms(5);
}
}
/* Poll until the WAL grows past lsn_before (the operation's append became
* visible) or the deadline expires; returns the appended watermark
* (current_lsn - 1) or 0 on timeout. */
static uint64_t cbrp_poll_append(persist_t *p, uint64_t lsn_before,
uint32_t deadline_ms)
{
uint64_t t0 = cbrp_mono_ms();
for (;;) {
uint64_t cur;
pthread_mutex_lock(&p->lock);
cur = p->current_lsn;
pthread_mutex_unlock(&p->lock);
if (cur > lsn_before)
return cur - 1;
if (cbrp_mono_ms() - t0 >= deadline_ms)
return 0;
cbrp_sleep_ms(5);
}
}
/* ── Handler-on-a-pthread runner (vtask-context stand-in) ───────────── */
typedef int (*cbrp_op_fn_t)(cluster_t *, endpoint_t *, xce_request_t *);
typedef struct {
cbrp_op_fn_t fn;
xce_request_t *req;
int rc;
} cbrp_op_arg_t;
/* Statics survive Unity's longjmp on assertion failure. */
static cbrp_op_arg_t g_op_arg;
static pthread_t g_op_tid;
static cluster_t g_hc; /* harness cluster (defined below) */
static endpoint_t g_req_ep; /* requester endpoint (capture, defined below) */
static void *cbrp_op_thread_fn(void *arg)
{
cbrp_op_arg_t *a = arg;
a->rc = a->fn(&g_hc, &g_req_ep, a->req);
return NULL;
}
/* ── Cluster harness ─────────────────────────────────────────────────── */
/* Resource names/types required by init_worker → apply_join_state
* (add_hostattrs_resvals, worker.c) — mirror of test_worker_join.c. */
#define CBRP_N_RES 9
static const char *cbrp_res_names[CBRP_N_RES] = {
"type", "model", "cpuf", "maxcpus", "maxmem",
"maxswap", "maxtmp", "ndisks", "slots"
};
static const int cbrp_res_valtypes[CBRP_N_RES] = {
XCE_VAL_STRING, XCE_VAL_STRING, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC
};
/* Statics survive Unity's longjmp on assertion failure.
* (g_hc / g_req_ep are tentatively defined above for the thread runner.) */
static commit_engine_t g_hce; /* wired commit engine */
static int g_hep_a, g_hep_b;/* dummy standby endpoints */
static ep_private_t g_req_priv;
static xce_peerinfo_t g_req_from;
static char g_hc_tmpdir[256];
/* myconf scaffolding for find_matched_hostconfig (worker join path) */
static myconf_t g_hc_myconf;
static cluster_config_t g_hc_clusterconf;
static xce_hostconfig_t g_hc_hconfig;
static xce_hostconfig_t *g_hc_hconfigs[1];
static void cbrp_setup(uint32_t sync_timeout_ms)
{
persist_config_t cfg;
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64, .delta = 16 };
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 64, .delta = 16 };
int i;
memset(&g_hc, 0, sizeof(g_hc));
memset(&g_req_ep, 0, sizeof(g_req_ep));
memset(&g_req_priv, 0, sizeof(g_req_priv));
memset(&g_req_from, 0, sizeof(g_req_from));
cbrp_notify_capture_reset();
g_stub_ep_worker = NULL;
/* Requester endpoint: zeroed, no iomgr — xce_reply packets stay on
* ep->outq (see file header). */
g_req_ep.fd = -1;
g_req_ep.data = &g_req_priv;
g_req_from.name = (char *)"requester";
g_req_from.key = (char *)"";
g_hc.myhost = &g_hc.myhost_info;
g_hc.myhost_info.name = (char *)"ctrl-test";
g_hc.myhost_info.key = (char *)"";
/* ha zero-init: CTRL_PRIMARY == 0 — handle_join's standby gates pass */
TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_hc.hooksys,
XCE_CONTROLLER_HOOK_MAX));
g_hc.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_hc.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_hc.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_hc.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_hc.jobmgr);
TEST_ASSERT_NOT_NULL(g_hc.qmgr);
TEST_ASSERT_NOT_NULL(g_hc.workermgr);
TEST_ASSERT_NOT_NULL(g_hc.resmgr);
zarray_init(&g_hc.sharedinst, 64, NULL);
for (i = 0; i < CBRP_N_RES; i++) {
xce_resource_t rdef;
memset(&rdef, 0, sizeof(rdef));
rdef.name = (char *)cbrp_res_names[i];
rdef.valtype = cbrp_res_valtypes[i];
rdef.idx = -1;
TEST_ASSERT_NOT_NULL_MESSAGE(makeResource(g_hc.resmgr, &rdef),
cbrp_res_names[i]);
}
/* hostconfig "default" so find_matched_hostconfig succeeds for joins */
memset(&g_hc_myconf, 0, sizeof(g_hc_myconf));
memset(&g_hc_clusterconf, 0, sizeof(g_hc_clusterconf));
memset(&g_hc_hconfig, 0, sizeof(g_hc_hconfig));
g_hc_hconfig.name = (char *)"default";
g_hc_hconfig.hosttype = (char *)"!";
g_hc_hconfig.hostmodel = (char *)"!";
g_hc_hconfig.mxj = 2;
g_hc_hconfigs[0] = &g_hc_hconfig;
g_hc_clusterconf.nhconfigs = 1;
g_hc_clusterconf.hconfigs = g_hc_hconfigs;
g_hc_myconf.cluster_config = &g_hc_clusterconf;
g_hc.myconf = &g_hc_myconf;
/* Real persist over a private tmpdir; cluster back-pointer set by
* persist_init so persist_start registers the production hooks. */
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = sync_timeout_ms;
snprintf(g_hc_tmpdir, sizeof(g_hc_tmpdir), "/tmp/xce_cbrp31_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_hc_tmpdir));
persist_set_workdir(&g_hc.persist, g_hc_tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_hc.persist, &cfg, &g_hc));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_hc.persist));
/* 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_hc.persist.repl, &g_hep_a, "sb_a", 7400) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_hc.persist.repl, &g_hep_b, "sb_b", 7401) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_hce, &g_hc.persist, 3, 8));
repl_set_commit_engine(&g_hc.persist.repl, &g_hce);
repl_broadcast_send_hook = cbrp_drop_send_hook;
g_obs_persist = &g_hc.persist;
/* Under the reserved-0 contract a fresh store starts current_lsn=1, so the
* first operation entry lands at LSN 1 with no priming append needed (LSN 0
* is the reserved sentinel — never a real entry; the engine rejects
* lsn == 0). Tests below derive watermarks relative to lsn_before, so the
* operation entries' absolute base does not affect the assertions. */
TEST_ASSERT_EQUAL_UINT64(1, g_hc.persist.current_lsn);
}
static void cbrp_teardown(void)
{
cbrp_reply_obs_t drain = { 0 };
cbrp_scan_replies(&g_req_ep, &drain); /* free any leftover packets */
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_hce);
persist_stop(&g_hc.persist);
persist_fini(&g_hc.persist);
zarray_fini(&g_hc.sharedinst);
if (g_hc.resmgr) dereference(g_hc.resmgr);
if (g_hc.workermgr) dereference(g_hc.workermgr);
if (g_hc.qmgr) dereference(g_hc.qmgr);
if (g_hc.jobmgr) dereference(g_hc.jobmgr);
xce_hook_fini(&g_hc.hooksys);
}
/* ── Per-opcode entity + request builders ────────────────────────────── */
/* job 42, submitted but never started (allocs == NULL): SIGKILL takes the
* not-started branch of job_ctrl → mark_job_fail → WAL_JOB_FAIL. */
static xce_signal_t g_sigkill = { .signo = SIGKILL, .force = 0 };
static xce_jobctrl_t g_jobctrl;
static xce_request_t g_jobctrl_req;
static job_submit_data_t g_job_submit; /* job_init requires submit != NULL */
static int g_jobobj_ready; /* jobj module tables — once per process */
/* Create a pending (never-started) job in g_hc.jobmgr. submit may carry
* queue/user/start_info (deep-copied by makeJob via makeJobsub). */
static void cbrp_make_job(uint64_t jobid, job_submit_data_t *submit)
{
xce_job_t jdef;
if (!g_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64));
g_jobobj_ready = 1;
}
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = submit;
TEST_ASSERT_NOT_NULL(makeJob(g_hc.jobmgr, &jdef));
}
static void cbrp_arrange_jobctrl(uint64_t jobid, int create_job)
{
if (create_job) {
memset(&g_job_submit, 0, sizeof(g_job_submit));
cbrp_make_job(jobid, &g_job_submit);
}
memset(&g_jobctrl, 0, sizeof(g_jobctrl));
g_jobctrl.action = JobControl_u_siginfo;
g_jobctrl.jobid = jobid;
g_jobctrl.u.siginfo = &g_sigkill;
memset(&g_jobctrl_req, 0, sizeof(g_jobctrl_req));
g_jobctrl_req.opcode = OPCode_jobcontrol;
g_jobctrl_req.seq = 11;
g_jobctrl_req.from = &g_req_from;
g_jobctrl_req.body_which = Request_body_jobctrlmsg;
g_jobctrl_req.body.jobctrlmsg = &g_jobctrl;
}
/* queue "q1": QUEUE_CLOSE → status |= QUEUE_STAT_CLOSED → WAL_QUEUE_CTRL */
static xce_queuectrl_t g_queuectrl;
static xce_request_t g_queuectrl_req;
static void cbrp_arrange_queuectrl(const char *qname, int create_queue)
{
if (create_queue) {
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char *)qname;
TEST_ASSERT_NOT_NULL(makeQueue(g_hc.qmgr, &qdef));
}
memset(&g_queuectrl, 0, sizeof(g_queuectrl));
g_queuectrl.action = QUEUE_CLOSE;
g_queuectrl.queue = (char *)qname;
memset(&g_queuectrl_req, 0, sizeof(g_queuectrl_req));
g_queuectrl_req.opcode = OPCode_queuecontrol;
g_queuectrl_req.seq = 12;
g_queuectrl_req.from = &g_req_from;
g_queuectrl_req.body_which = Request_body_queuectrlmsg;
g_queuectrl_req.body.queuectrlmsg = &g_queuectrl;
}
/* worker "hostA": HOST_CLOSE → status |= XCE_WORKER_CLOSED →
* WAL_WORKER_CTRL detail WORKER_UPDATE */
static xce_hostctrl_t g_hostctrl;
static xce_request_t g_hostctrl_req;
static void cbrp_arrange_hostctrl(const char *host, int create_worker)
{
if (create_worker) {
xce_worker_t wdef;
hostattrs_t *attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char *)host;
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
}
memset(&g_hostctrl, 0, sizeof(g_hostctrl));
g_hostctrl.action = HOST_CLOSE;
g_hostctrl.hostname = (char *)host;
memset(&g_hostctrl_req, 0, sizeof(g_hostctrl_req));
g_hostctrl_req.opcode = OPCode_hostcontrol;
g_hostctrl_req.seq = 13;
g_hostctrl_req.from = &g_req_from;
g_hostctrl_req.body_which = Request_body_hostctrlmsg;
g_hostctrl_req.body.hostctrlmsg = &g_hostctrl;
}
/* worker join: handle_join with role HostRole_worker reaches the static
* handle_worker_join; first-time join → init_worker → AFTER_EVENT hook →
* WAL_WORKER_CTRL detail WORKER_JOIN. */
static xce_joindata_t g_joindata;
static hostattrs_t g_join_attrs;
static xce_peerinfo_t g_join_from;
static xce_request_t g_join_req;
static void cbrp_arrange_worker_join(const char *host)
{
memset(&g_join_attrs, 0, sizeof(g_join_attrs));
g_join_attrs.hosttype = makeString("LINUX");
g_join_attrs.hostmodel = makeString("x86");
g_join_attrs.cpufactor = 1.0f;
g_join_attrs.maxcpus = 4;
g_join_attrs.maxmem = 16;
g_join_attrs.maxswap = 32;
g_join_attrs.maxtmp = 64;
g_join_attrs.ndisks = 1;
memset(&g_joindata, 0, sizeof(g_joindata));
g_joindata.role = HostRole_worker;
g_joindata.data_which = JoinData_data_attrs;
g_joindata.data.attrs = &g_join_attrs;
memset(&g_join_from, 0, sizeof(g_join_from));
g_join_from.name = (char *)host;
g_join_from.key = (char *)"";
memset(&g_join_req, 0, sizeof(g_join_req));
g_join_req.opcode = OPCode_join;
g_join_req.seq = 14;
g_join_req.from = &g_join_from;
g_join_req.body_which = Request_body_joinmsg;
g_join_req.body.joinmsg = &g_joindata;
}
/* ── Shared case drivers ─────────────────────────────────────────────── */
/* Withheld-ACK case (Req 1.2, 1.4): the requester reply must carry
* -ETIMEDOUT and no success reply may be observable while the operation's
* entry exceeds committed_lsn.
* RED: the current handlers reply SUCCESS immediately. */
static void cbrp_drive_withheld(cbrp_op_fn_t fn, xce_request_t *req,
uint32_t sync_timeout_ms)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, watermark, committed_final;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: handler on a worker pthread (post-3.2 it blocks at the barrier) */
g_op_arg.fn = fn;
g_op_arg.req = req;
g_op_arg.rc = -9999;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL,
cbrp_op_thread_fn, &g_op_arg));
/* The reply may legitimately take up to the commit timeout */
cbrp_poll_reply(&g_req_ep, sync_timeout_ms + 1500, &obs);
pthread_join(g_op_tid, NULL);
watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 1000);
committed_final = atomic_load(&g_hc.persist.committed_lsn);
/* Assert: the operation must produce its WAL entry — nothing to
* commit-confirm otherwise (Req 1.1). RED finding: queue_ctrl and
* worker_ctrl leak key_find's found rc (== 1) into the `if (rc == 0)`
* hook gate → NO WAL entry, status-1 reply (header; 3.2 closes it). */
TEST_ASSERT_TRUE_MESSAGE(watermark > 0,
"request-shaped control op must append its WAL entry (queue_ctrl/"
"worker_ctrl currently skip the AFTER_EVENT hook: rc==1 leak)");
/* ...which never committed (ACKs withheld) */
TEST_ASSERT_TRUE_MESSAGE(committed_final < watermark,
"committed_lsn must stay below the entry with ACKs withheld");
/* ...the requester got exactly an error reply */
TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0,
"the requester must receive a reply");
TEST_ASSERT_TRUE_MESSAGE(
!(obs.last_status == 0 && obs.committed_at_obs < watermark),
"no success reply may be observable while the operation's entry "
"exceeds committed_lsn (Req 1.4)");
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status,
"withheld ACKs past the commit timeout must surface -ETIMEDOUT "
"to the requester (Req 1.2)");
}
/* Covering-ACK case (Req 1.1, 1.4): the success reply must be observable
* only after committed_lsn covers the operation's entry.
* RED: the current handlers reply success before any ACK arrives. */
static void cbrp_drive_acked(cbrp_op_fn_t fn, xce_request_t *req)
{
cbrp_reply_obs_t early = { 0 };
cbrp_reply_obs_t final = { 0 };
uint64_t lsn_before, watermark;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
g_op_arg.fn = fn;
g_op_arg.req = req;
g_op_arg.rc = -9999;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL,
cbrp_op_thread_fn, &g_op_arg));
/* Wait until the operation's append is visible, then hold the handler
* blocked for a beat and sample the reply capture BEFORE any ACK. */
watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000);
cbrp_sleep_ms(150);
cbrp_scan_replies(&g_req_ep, &early);
/* Deliver the covering ACK, then collect the (now permitted) reply */
repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark);
cbrp_poll_reply(&g_req_ep, 2000, &final);
pthread_join(g_op_tid, NULL);
/* Assert: the operation must produce its WAL entry (see the rc==1
* leak note in cbrp_drive_withheld — same RED finding applies) */
TEST_ASSERT_TRUE_MESSAGE(watermark > 0,
"request-shaped control op must append its WAL entry (queue_ctrl/"
"worker_ctrl currently skip the AFTER_EVENT hook: rc==1 leak)");
/* THE ordering gate: nothing the requester can read as confirmation
* may exist before the covering ACK (Req 1.1, 1.4) */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, early.n_replies,
"no reply may be observable before committed_lsn covers the "
"operation's entry (Req 1.1, 1.4)");
/* After the ACK: success reply, entry committed */
TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0,
"the requester must receive the reply once the entry commits");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status,
"the post-commit reply must report success (Req 1.1)");
TEST_ASSERT_TRUE_MESSAGE(final.committed_at_obs >= watermark,
"the success reply must be observable only with committed_lsn "
"covering the operation's entry (Req 1.1)");
}
/* Local-validation-failure pin (Req 1.4 boundary, task brief): the
* existing error reply stays immediate — no barrier wait, no WAL entry.
* Green before AND after the task-3.2 restructure (latency-bounded). */
static void cbrp_drive_validation_error(cbrp_op_fn_t fn, xce_request_t *req,
int32_t expect_status,
int check_exact_status)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, lsn_after, t0, elapsed_ms;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: synchronous — a validation failure must never block */
t0 = cbrp_mono_ms();
(void)fn(&g_hc, &g_req_ep, req);
elapsed_ms = cbrp_mono_ms() - t0;
cbrp_scan_replies(&g_req_ep, &obs);
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Assert: immediate error reply with the existing code, no append */
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"local validation failures must reply without a barrier wait");
TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0,
"validation failure must still produce a requester reply");
if (check_exact_status)
TEST_ASSERT_EQUAL_INT_MESSAGE(expect_status, obs.last_status,
"validation failure must keep its existing error code");
else
TEST_ASSERT_TRUE_MESSAGE(obs.last_status < 0,
"validation failure must reply with an error status");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a rejected operation must append no WAL entry");
}
/* ── do_jobcontrol_op (not-started branch → WAL_JOB_FAIL) ────────────── */
static void test_jobctrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_jobctrl(42, 1);
cbrp_drive_withheld(do_jobcontrol_op, &g_jobctrl_req, 300);
cbrp_teardown();
printf("test_jobctrl_withheld_acks_replies_error: PASS\n");
}
static void test_jobctrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_jobctrl(43, 1);
cbrp_drive_acked(do_jobcontrol_op, &g_jobctrl_req);
cbrp_teardown();
printf("test_jobctrl_success_reply_only_after_commit: PASS\n");
}
static void test_jobctrl_unknown_job_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_jobctrl(9999, 0); /* job does not exist */
cbrp_drive_validation_error(do_jobcontrol_op, &g_jobctrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_jobctrl_unknown_job_error_reply_immediate: PASS\n");
}
/* ── do_queuecontrol_op (QUEUE_CLOSE → WAL_QUEUE_CTRL) ───────────────── */
static void test_queuectrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_queuectrl("q1", 1);
cbrp_drive_withheld(do_queuecontrol_op, &g_queuectrl_req, 300);
cbrp_teardown();
printf("test_queuectrl_withheld_acks_replies_error: PASS\n");
}
static void test_queuectrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_queuectrl("q2", 1);
cbrp_drive_acked(do_queuecontrol_op, &g_queuectrl_req);
cbrp_teardown();
printf("test_queuectrl_success_reply_only_after_commit: PASS\n");
}
static void test_queuectrl_unknown_queue_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_queuectrl("no-such-queue", 0);
cbrp_drive_validation_error(do_queuecontrol_op, &g_queuectrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_queuectrl_unknown_queue_error_reply_immediate: PASS\n");
}
/* ── do_hostcontrol_op (HOST_CLOSE → WAL_WORKER_CTRL/UPDATE) ─────────── */
static void test_hostctrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_hostctrl("hostA", 1);
cbrp_drive_withheld(do_hostcontrol_op, &g_hostctrl_req, 300);
cbrp_teardown();
printf("test_hostctrl_withheld_acks_replies_error: PASS\n");
}
static void test_hostctrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_hostctrl("hostB", 1);
cbrp_drive_acked(do_hostcontrol_op, &g_hostctrl_req);
cbrp_teardown();
printf("test_hostctrl_success_reply_only_after_commit: PASS\n");
}
static void test_hostctrl_unknown_host_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_hostctrl("no-such-host", 0);
cbrp_drive_validation_error(do_hostcontrol_op, &g_hostctrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_hostctrl_unknown_host_error_reply_immediate: PASS\n");
}
/* ── handle_worker_join via handle_join (→ WAL_WORKER_CTRL/JOIN) ─────── */
static void test_worker_join_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_worker_join("workerA");
cbrp_drive_withheld(handle_join, &g_join_req, 300);
cbrp_teardown();
printf("test_worker_join_withheld_acks_replies_error: PASS\n");
}
static void test_worker_join_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_worker_join("workerB");
cbrp_drive_acked(handle_join, &g_join_req);
cbrp_teardown();
printf("test_worker_join_success_reply_only_after_commit: PASS\n");
}
static void test_worker_join_no_hostconfig_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_worker_join("workerC");
/* Remove every hostconfig: find_matched_hostconfig returns NULL and
* handle_worker_join must reply its existing error immediately —
* the (future) barrier sits between the AFTER_EVENT hook and the
* SUCCESS reply only; error replies stay immediate. */
g_hc_clusterconf.nhconfigs = 0;
g_hc_clusterconf.hconfigs = NULL;
cbrp_drive_validation_error(handle_join, &g_join_req, 0, 0);
cbrp_teardown();
printf("test_worker_join_no_hostconfig_error_reply_immediate: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Receipt-ACK verdict pins (ha-commit-barrier-rollout task 3.3, Req 2.3)
*
* Umbrella Reply-Ordering Contract reference (ha-consistency-invariants
* design.md §Write-Path Classification ▸ Reply-Ordering Contract):
* handle_jobdispatch's early reply is an ACK-of-receipt and MAY STAND,
* because dispatch outcomes are fact-shaped — per-job results are
* delivered via notify_dispatch, and the resulting job-state changes
* arrive later as worker-reported facts. The same verdict covers
* do_set_preason / do_clr_preason (scheduler-origin WAL_JOB_PREASON
* facts, RETAIN) and the do_update_op rusage path (worker-origin
* WAL_JOB_UPDATE + WAL_JOB_RUSAGE facts, RETAIN): their replies
* acknowledge RECEIPT of the report, never commitment of any
* state-changing log entry.
*
* These are PINS of current correct behavior — green on landing, no RED
* phase exists (task-stated design, like the 3.1 validation pins). They
* run with ACKs withheld throughout, so any future regression that makes
* one of these handlers wait on the commit barrier trips the latency
* bound, and any future WAL append from dispatch processing trips either
* the unchanged-LSN pin or (for an unclassified type) the XCE_TESTING
* abort in wal_classify — the classification guard stays silent through
* a clean dispatch cycle.
* ════════════════════════════════════════════════════════════════════════ */
/* ── handle_jobdispatch fixture ──────────────────────────────────────── */
static endpoint_t g_wkr_disp_ep; /* dispatch worker endpoint */
static job_submit_data_t g_disp_submit;
static job_startinfo_t g_disp_startinfo;
static xce_job_dispatch_t g_dispatchmsg;
static xce_job_alloc_t *g_dispatch_allocs[2];
static xce_request_t g_dispatch_req;
#define CBRP_DISP_JOBID 60u
/* Full dispatch fixture: queue (get_queue_exec), pending job with a real
* submit (dispatch_a_job_ reads submit->queue and submit->start_info),
* worker with a live no-iomgr endpoint (the startjob request must be
* enqueueable), and a 2-job dispatch message exercising BOTH outcomes:
* allocs[0] — dispatchable to wkr-disp → notify_dispatch(ok, code 0)
* allocs[1] — nhosts == 0 → notify_dispatch(fail, -EINVAL)
* The allocs are heap-built via xcealloc helpers because the handler
* takes ownership (MOVE_PTR + free_JobAlloc_ptr per job). */
static void cbrp_arrange_jobdispatch(void)
{
xce_queue_t qdef;
xce_worker_t wdef;
hostattrs_t *attrs = NULL;
worker_t *w = NULL;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char *)"qd";
TEST_ASSERT_NOT_NULL(makeQueue(g_hc.qmgr, &qdef));
memset(&g_disp_startinfo, 0, sizeof(g_disp_startinfo));
g_disp_startinfo.cmdline = (char *)"/bin/true";
memset(&g_disp_submit, 0, sizeof(g_disp_submit));
g_disp_submit.queue = (char *)"qd";
g_disp_submit.user = (char *)"tester";
g_disp_submit.start_info = &g_disp_startinfo;
cbrp_make_job(CBRP_DISP_JOBID, &g_disp_submit);
attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char *)"wkr-disp";
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
/* dispatch gate: worker must have an endpoint and status 0 */
memset(&g_wkr_disp_ep, 0, sizeof(g_wkr_disp_ep));
g_wkr_disp_ep.fd = -1;
lock(g_hc.workermgr, RTTI_LOCK_READ);
key_find(g_hc.workermgr, (void *)"wkr-disp", (void **)&w);
unlock(g_hc.workermgr);
TEST_ASSERT_NOT_NULL(w);
w->ep = &g_wkr_disp_ep;
g_dispatch_allocs[0] = make_job_alloc(
CBRP_DISP_JOBID, 1,
make_host_allocs(1,
make_host_alloc((char *)"wkr-disp", 1,
make_allocs(1,
make_alloc((char *)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(g_dispatch_allocs[0]);
g_dispatch_allocs[1] = make_job_alloc(CBRP_DISP_JOBID, 0, NULL);
TEST_ASSERT_NOT_NULL(g_dispatch_allocs[1]);
memset(&g_dispatchmsg, 0, sizeof(g_dispatchmsg));
g_dispatchmsg.njobs = 2;
g_dispatchmsg.allocs = g_dispatch_allocs;
memset(&g_dispatch_req, 0, sizeof(g_dispatch_req));
g_dispatch_req.opcode = OPCode_jobdispatch;
g_dispatch_req.seq = 15;
g_dispatch_req.from = &g_req_from;
g_dispatch_req.body_which = Request_body_dispatchmsg;
g_dispatch_req.body.dispatchmsg = &g_dispatchmsg;
}
/* VERDICT PIN (Req 2.3): handle_jobdispatch (job.c:468-500, re-verified
* 2026-06-12) replies status=0 IMMEDIATELY — an ACK-of-receipt per the
* umbrella Reply-Ordering Contract (see section header) — and a full
* dispatch cycle (one successful + one failed allocation) appends NO WAL
* entries: allocation results return via notify_dispatch and job-state
* changes arrive later as worker facts. ACKs are withheld for the whole
* test, so a regression that adds a barrier wait trips the latency bound
* and any WAL append from dispatch processing trips the unchanged-LSN
* assert (an unclassified type would abort in wal_classify first). */
static void test_jobdispatch_receipt_ack_no_wal_append(void)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, lsn_after, t0, elapsed_ms;
int rc;
cbrp_setup(3000);
cbrp_arrange_jobdispatch();
cbrp_notify_capture_reset(); /* drop any arrange-time noise */
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: synchronous — the receipt ACK must never block on commit */
t0 = cbrp_mono_ms();
rc = handle_jobdispatch(&g_hc, &g_req_ep, &g_dispatch_req);
elapsed_ms = cbrp_mono_ms() - t0;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
cbrp_scan_replies(&g_req_ep, &obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handle_jobdispatch rc");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"receipt ACK must be latency-bounded — no commit barrier wait "
"(Req 2.3: the dispatch reply confirms receipt only)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
"scheduler must receive exactly the one receipt ACK");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status,
"the receipt ACK reports acceptance of the message (status 0)");
/* THE classification pin: dispatch processing appends no WAL entries,
* so there is nothing the receipt ACK could have falsely confirmed */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a dispatch cycle must append NO WAL entries (Req 2.3 — any "
"future append from dispatch processing must move behind the "
"barrier or be re-classified; the wal_classify XCE_TESTING "
"guard backstops unclassified types)");
/* notify_dispatch outcomes unaffected: one ok + one fail, in order */
TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_notify_count,
"both allocations must produce their per-job outcome notification");
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
g_notify_events[0]);
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
g_notify_events[1]);
TEST_ASSERT_EQUAL_UINT64(CBRP_DISP_JOBID, g_notify_jobids[0]);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_notify_codes[0],
"allocs[0] dispatch to the live worker must succeed (code 0)");
TEST_ASSERT_EQUAL_UINT64(CBRP_DISP_JOBID, g_notify_jobids[1]);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, g_notify_codes[1],
"allocs[1] (nhosts == 0) must fail with -EINVAL");
/* Ordering: the receipt ACK was already enqueued when the FIRST
* dispatch outcome fired — reply precedes processing (ACK-of-receipt,
* not ACK-of-outcome) */
TEST_ASSERT_TRUE_MESSAGE(g_notify_reply_already_enqueued[0],
"the receipt ACK must be enqueued before any dispatch outcome "
"notification (reply-before-processing is the verdict)");
/* drain the startjob request enqueued on the worker endpoint */
{
cbrp_reply_obs_t wdrain = { 0 };
cbrp_scan_replies(&g_wkr_disp_ep, &wdrain);
}
cbrp_teardown();
printf("test_jobdispatch_receipt_ack_no_wal_append: PASS\n");
}
/* ── do_set_preason / do_clr_preason fixtures ────────────────────────── */
static xce_pendreason_t g_pendreason;
static xce_pendreason_t *g_pendreasons[1];
static xce_preason_set_t g_preason_set;
static xce_request_t g_preason_req;
static uint64_t g_clr_jobs[1];
static xce_clear_reason_t g_clrreason;
static xce_request_t g_clrpreason_req;
/* set_preason MOVEs the inner xce_job_preason_t into the job (job.c
* MOVE_PTR(job->pend_reasons, ...)) and the job/do_clr_preason frees it
* later — so it must be heap-allocated; the carrier structs stay static. */
static void cbrp_arrange_set_preason(uint64_t jobid)
{
xce_job_preason_t *jpr = calloc(1, sizeof(*jpr));
TEST_ASSERT_NOT_NULL(jpr);
jpr->main_reason = 7; /* arbitrary scheduler pend reason */
memset(&g_pendreason, 0, sizeof(g_pendreason));
g_pendreason.jobid = jobid;
g_pendreason.reasons = jpr;
g_pendreasons[0] = &g_pendreason;
memset(&g_preason_set, 0, sizeof(g_preason_set));
g_preason_set.nreasons = 1;
g_preason_set.reasons = g_pendreasons;
memset(&g_preason_req, 0, sizeof(g_preason_req));
g_preason_req.opcode = OPCode_setpreason;
g_preason_req.seq = 16;
g_preason_req.from = &g_req_from;
g_preason_req.body_which = Request_body_preasonmsg;
g_preason_req.body.preasonmsg = &g_preason_set;
}
static void cbrp_arrange_clr_preason(uint64_t jobid)
{
g_clr_jobs[0] = jobid;
memset(&g_clrreason, 0, sizeof(g_clrreason));
g_clrreason.njobs = 1;
g_clrreason.jobs = g_clr_jobs;
memset(&g_clrpreason_req, 0, sizeof(g_clrpreason_req));
g_clrpreason_req.opcode = OPCode_clrpreason;
g_clrpreason_req.seq = 17;
g_clrpreason_req.from = &g_req_from;
g_clrpreason_req.body_which = Request_body_clrpreasonmsg;
g_clrpreason_req.body.clrpreasonmsg = &g_clrreason;
}
/* Shared receipt-style fact-handler driver: with ACKs withheld the
* handler must (a) reply status 0 latency-bounded — receipt only, no
* barrier wait — (b) append exactly expect_appends fact entries that
* remain PENDING (lsn > committed_lsn: the reply confirmed nothing about
* commitment), and (c) once a covering ACK arrives the retained suffix
* commits with no redelivery (the RETAIN contract observable per spec
* Implementation Notes 1.1: registration-free, pending purely by
* lsn > committed_lsn, ACK-driven advance). */
static void cbrp_drive_receipt_fact(cbrp_op_fn_t fn, xce_request_t *req,
uint64_t expect_appends)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, lsn_after, watermark, t0, elapsed_ms;
uint64_t deadline;
int rc;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: synchronous — receipt-style handlers must never block */
t0 = cbrp_mono_ms();
rc = fn(&g_hc, &g_req_ep, req);
elapsed_ms = cbrp_mono_ms() - t0;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
watermark = lsn_after - 1; /* last appended entry */
cbrp_scan_replies(&g_req_ep, &obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "fact handler rc");
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"receipt-style reply must be latency-bounded — no commit barrier "
"wait with ACKs withheld (facts have no commit timeout, Req 2.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
"the reporter must receive exactly one receipt reply");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status,
"the receipt reply reports acceptance (status 0)");
/* the fact appends happened ... */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + expect_appends, lsn_after,
"the handler must append exactly its fact entries");
/* ... and are RETAIN-pending: the receipt reply coexists with the
* uncommitted suffix, proving it confirmed receipt, not commitment */
TEST_ASSERT_TRUE_MESSAGE(
atomic_load(&g_hc.persist.committed_lsn) < watermark,
"fact entries must remain pending (lsn > committed_lsn) — the "
"receipt reply must not imply commitment (Req 2.3)");
TEST_ASSERT_TRUE_MESSAGE(obs.committed_at_obs < watermark,
"the receipt reply was observable while the fact entries were "
"still uncommitted — receipt-only semantics pinned");
/* RETAIN completion: a covering ACK (self + 1 standby = majority of
* n_peers=3) advances committed_lsn over the retained suffix with no
* redelivery from the reporter */
repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark);
deadline = cbrp_mono_ms() + 2000;
while (atomic_load(&g_hc.persist.committed_lsn) < watermark
&& cbrp_mono_ms() < deadline)
cbrp_sleep_ms(5);
TEST_ASSERT_TRUE_MESSAGE(
atomic_load(&g_hc.persist.committed_lsn) >= watermark,
"a covering ACK must commit the retained fact entries "
"(RETAIN registration contract)");
}
/* VERDICT PIN (Req 2.3): do_set_preason (job.c:1062) replies status=0
* receipt-style BEFORE applying the scheduler-reported pend reasons; the
* subsequent WAL_JOB_PREASON fact append (HOOK_DETAIL_SETPREASON →
* persist_job_hook_fun, RETAIN per wal_classify) stays pending until
* ACK-covered. Umbrella Reply-Ordering Contract: receipt-only ACK for a
* fact-shaped delivery (see section header). */
static void test_set_preason_receipt_reply_fact_retained(void)
{
cbrp_setup(3000);
cbrp_arrange_jobctrl(70, 1); /* job 70 exists; jobctrl req unused */
cbrp_arrange_set_preason(70);
cbrp_drive_receipt_fact(do_set_preason, &g_preason_req, 1);
cbrp_teardown();
printf("test_set_preason_receipt_reply_fact_retained: PASS\n");
}
/* VERDICT PIN (Req 2.3): do_clr_preason (job.c:1090) — same receipt-style
* verdict as do_set_preason; the clear is also a scheduler-origin
* WAL_JOB_PREASON fact (HOOK_DETAIL_CLRPREASON, RETAIN). */
static void test_clr_preason_receipt_reply_fact_retained(void)
{
cbrp_reply_obs_t drain = { 0 };
cbrp_setup(3000);
cbrp_arrange_jobctrl(71, 1);
/* seed pend reasons so the clear exercises the real erase path */
cbrp_arrange_set_preason(71);
TEST_ASSERT_EQUAL_INT(0, do_set_preason(&g_hc, &g_req_ep,
&g_preason_req));
cbrp_scan_replies(&g_req_ep, &drain); /* drop the seeding reply */
cbrp_arrange_clr_preason(71);
cbrp_drive_receipt_fact(do_clr_preason, &g_clrpreason_req, 1);
cbrp_teardown();
printf("test_clr_preason_receipt_reply_fact_retained: PASS\n");
}
/* ── do_update_op rusage path fixture ────────────────────────────────── */
/* KNOWN PRE-EXISTING MISMATCH (discovered while pinning, reported in the
* task evidence — NOT fixed here, production is out of boundary):
* update_job_rusages fires the AFTER_EVENT HOOK_DETAIL_JOB_RUSAGE_UPDATE
* hook with (cluster_t*, xce_job_rusages_t*) [job.c:1015-1016], but
* persist_job_hook_fun's default arm reads the second vararg as a job_t*
* [persist.c:1354-1359] and dereferences j->job.state at offset 56 —
* exactly sizeof(xce_job_rusages_t) — an out-of-bounds read on an
* exactly-sized allocation (the WAL_JOB_RUSAGE payload's state field is
* garbage). The padded box below keeps THIS pin's read in-bounds under
* ASAN without masking the report; the jobid field (offset 0 in both
* types) lines up by accident. */
static union {
xce_job_rusages_t r;
unsigned char pad[sizeof(xce_job_rusages_t) + 64];
} g_rusages_box;
static xce_update_data_t g_updatemsg;
static xce_request_t g_update_req;
static void cbrp_arrange_update_rusage(uint64_t jobid)
{
xce_worker_t wdef;
hostattrs_t *attrs = NULL;
worker_t *w = NULL;
/* reporting worker — get_ep_worker (stubbed) must return it */
attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char *)"wkr-rus";
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
lock(g_hc.workermgr, RTTI_LOCK_READ);
key_find(g_hc.workermgr, (void *)"wkr-rus", (void **)&w);
unlock(g_hc.workermgr);
TEST_ASSERT_NOT_NULL(w);
g_stub_ep_worker = w;
memset(&g_rusages_box, 0, sizeof(g_rusages_box));
g_rusages_box.r.jobid = jobid; /* count/npids/npgids 0 — stubs no-op */
memset(&g_updatemsg, 0, sizeof(g_updatemsg));
g_updatemsg.table = Update_data_jobrusages;
g_updatemsg.data.jobrusages = &g_rusages_box.r;
memset(&g_update_req, 0, sizeof(g_update_req));
g_update_req.opcode = OPCode_update;
g_update_req.seq = 18;
g_update_req.from = &g_req_from;
g_update_req.body_which = Request_body_updatemsg;
g_update_req.body.updatemsg = &g_updatemsg;
}
/* VERDICT PIN (Req 2.3): the do_update_op rusage path (opupd.c:69 →
* do_update_jobrusage → update_job_rusages) replies receipt-style — the
* reply carries the LOCAL application result, never a commit barrier
* result — and its two fact appends (HOOK_DETAIL_JOB_UPDATE →
* WAL_JOB_UPDATE, HOOK_DETAIL_JOB_RUSAGE_UPDATE → WAL_JOB_RUSAGE, both
* RETAIN per wal_classify) stay pending until ACK-covered. Umbrella
* Reply-Ordering Contract: receipt-only ACK for fact-shaped delivery. */
static void test_update_op_rusage_receipt_reply_facts_retained(void)
{
cbrp_setup(3000);
cbrp_arrange_jobctrl(72, 1); /* job 72 exists; jobctrl req unused */
cbrp_arrange_update_rusage(72);
cbrp_drive_receipt_fact(do_update_op, &g_update_req, 2);
g_stub_ep_worker = NULL;
cbrp_teardown();
printf("test_update_op_rusage_receipt_reply_facts_retained: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Fact retention across quorum loss (task 4.3, Req 3.1, 3.3, 3.4)
*
* GREEN-on-landing integration proof of the RETAIN persistence semantics
* (umbrella Req 2.8 end-to-end evidence; completes the bug-396 fix):
* with quorum LOST (both standby registry slots disconnected via
* repl_remove_standby) the worker-fact paths still append, with no error
* and no commit timeout; the entries stay pending (lsn > committed_lsn)
* and nothing is dropped (each fact on disk exactly once); after
* reconnect + a covering ACK, committed_lsn advances over the retained
* suffix with NO re-delivery of the facts (the log itself is the queue).
*
* It lives HERE (not test_commit_barrier_rollout.c) because it needs this
* file's full harness: cluster_t with jobmgr + real tmpdir persist with
* registered production hooks + wired engine (n_peers=3) + the stub
* standby registry + the job.c link (cbrp_make_job).
*
* Facts are driven via on_job_start / on_job_finish (job.c) DIRECTLY:
* do_notify lives in cluster.c, which is NOT in the unit-test link
* (meson controller_unit_test_src), and on_job_* is exactly do_notify's
* fact path minus the opcode decode. Gate admission of OPCode_notify
* during quorum loss is task 4.1/4.2's coverage (test_commit_barrier_
* rollout.c fact-gate cases); this test's focus is RETAIN persistence.
*
* mark_job_* is currently UNGUARDED (idempotency guards land in task 6),
* so only distinct forward transitions are driven: job 80 start→finish,
* job 81 start→finish — four distinct (job, transition) facts, no
* duplicates, independent of future guard behavior.
* ════════════════════════════════════════════════════════════════════════ */
/* Count WAL entries with (wal_type, wal_jobstate_t.jobid) by direct scan
* of the active on-disk segment (persist_wal_async pwrites synchronously,
* so same-process reads see every append without an fsync). The priming
* entry (payload "p", 1 byte) and any non-jobstate entries are skipped by
* the payload-size check. Single segment is guaranteed here: rotation
* needs 64 MB (WAL_SEGMENT_SIZE_BYTES). Returns the match count;
* *state_out (optional) receives the LAST matching entry's state. */
static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type,
uint64_t jobid, uint32_t *state_out)
{
char seg_path[sizeof(p->seg_path)];
uint64_t seg_end, off = 0;
int fd, count = 0;
pthread_mutex_lock(&p->lock);
memcpy(seg_path, p->seg_path, sizeof(seg_path));
seg_end = p->seg_offset;
pthread_mutex_unlock(&p->lock);
fd = open(seg_path, O_RDONLY);
TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "active WAL segment must be openable");
while (off + sizeof(wal_entry_hdr_t) <= seg_end) {
wal_entry_hdr_t hdr;
TEST_ASSERT_EQUAL_INT_MESSAGE(
(int)sizeof(hdr),
(int)pread(fd, &hdr, sizeof(hdr), (off_t)off),
"WAL segment scan: short header read");
if (hdr.type == wal_type
&& hdr.payload_len == sizeof(wal_jobstate_t)) {
wal_jobstate_t js;
TEST_ASSERT_EQUAL_INT_MESSAGE(
(int)sizeof(js),
(int)pread(fd, &js, sizeof(js), (off_t)(off + sizeof(hdr))),
"WAL segment scan: short payload read");
if (js.jobid == jobid) {
count++;
if (state_out != NULL)
*state_out = js.state;
}
}
off += sizeof(hdr) + hdr.payload_len;
}
close(fd);
return count;
}
/* Drive one fact call: it must succeed (rc 0 — in particular NEVER
* -ETIMEDOUT) and return latency-bounded. sync_timeout_ms is 5000 in this
* test, so a fact path that wrongly ran a WAIT barrier with ACKs
* unobtainable would block ~5000 ms and trip the 1000 ms bound — the
* "no timeout was ever applied" assertion (Req 3.1: facts have NO commit
* timeout; RETAIN returns immediately). */
#define CBRP_DRIVE_FACT(call, what) \
do { \
uint64_t t0_ = cbrp_mono_ms(); \
int rc_ = (call); \
uint64_t el_ = cbrp_mono_ms() - t0_; \
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_, \
what ": fact append must succeed with no error during quorum " \
"loss — never -ETIMEDOUT (Req 3.1, 3.2)"); \
TEST_ASSERT_TRUE_MESSAGE(el_ < 1000, \
what ": no commit timeout may ever be applied to a fact entry " \
"— RETAIN returns immediately (Req 3.1)"); \
} while (0)
/* makeJob deep-copies submit; statics survive Unity's longjmp. */
static job_submit_data_t g_fr_submit80;
static job_submit_data_t g_fr_submit81;
static void test_fact_retention_across_quorum_loss(void)
{
static unsigned int fr_pids[2] = { 8001, 8101 };
job_pidinfo_t start80, start81;
xce_jobfinish_t fin80, fin81;
uint64_t lsn_before, lsn_after, first_fact_lsn, watermark;
uint64_t committed, deadline;
uint32_t st_start80 = 0, st_fin80 = 0;
uint32_t st_start81 = 0, st_fin81 = 0;
/* Arrange: 3-member cluster, two running jobs-to-be. The large
* sync_timeout (5000 ms) gives the CBRP_DRIVE_FACT latency bound its
* discriminating power (see macro comment). */
cbrp_setup(5000);
memset(&g_fr_submit80, 0, sizeof(g_fr_submit80));
memset(&g_fr_submit81, 0, sizeof(g_fr_submit81));
cbrp_make_job(80, &g_fr_submit80);
cbrp_make_job(81, &g_fr_submit81);
/* Act 1 — QUORUM LOST: disconnect both standby registry slots
* (repl_remove_standby = the RDHUP disconnect path's registry effect).
* n_peers stays 3 in the engine: self alone is 1 of 3, no majority. */
repl_remove_standby(&g_hc.persist.repl, &g_hep_a);
repl_remove_standby(&g_hc.persist.repl, &g_hep_b);
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
first_fact_lsn = lsn_before; /* persist_wal_async assigns current_lsn */
/* Act 2 — drive four distinct worker-fact transitions with zero
* quorum: job 80 start/finish, job 81 start/finish (do_notify-level
* fact paths; see section header for why on_job_* directly). */
memset(&start80, 0, sizeof(start80));
start80.jobid = 80;
start80.npids = 1;
start80.pids = &fr_pids[0];
CBRP_DRIVE_FACT(on_job_start(&g_hc, &start80), "on_job_start(80)");
memset(&start81, 0, sizeof(start81));
start81.jobid = 81;
start81.npids = 1;
start81.pids = &fr_pids[1];
CBRP_DRIVE_FACT(on_job_start(&g_hc, &start81), "on_job_start(81)");
memset(&fin80, 0, sizeof(fin80));
fin80.jobid = 80;
fin80.exit_code = 0; /* → JOB_STAT_DONE */
CBRP_DRIVE_FACT(on_job_finish(&g_hc, &fin80), "on_job_finish(80)");
memset(&fin81, 0, sizeof(fin81));
fin81.jobid = 81;
fin81.exit_code = 0;
CBRP_DRIVE_FACT(on_job_finish(&g_hc, &fin81), "on_job_finish(81)");
/* Assert — pending, nothing dropped (Req 3.1, 3.3): all four facts
* appended exactly once, none committed (committed_lsn untouched by
* the membership shrink: zero countable standbys → candidate 0). */
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
watermark = lsn_after - 1;
committed = atomic_load(&g_hc.persist.committed_lsn);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 4, lsn_after,
"each of the four facts must append exactly one WAL entry "
"(accepted, not rejected, during quorum loss — Req 3.2)");
TEST_ASSERT_TRUE_MESSAGE(committed < first_fact_lsn,
"every fact entry must remain pending (lsn > committed_lsn) "
"while quorum is lost (Req 3.1)");
/* Retained-depth observability counter (design §Observability:
* committed-vs-appended lag), visible in the test log. */
printf("test_fact_retention_across_quorum_loss: retained-depth=%lu "
"(appended watermark=%lu, committed_lsn=%lu)\n",
(unsigned long)(watermark - committed),
(unsigned long)watermark, (unsigned long)committed);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, watermark - committed,
"retained depth must equal the four uncommitted fact entries");
/* Nothing dropped: each fact is on disk exactly once, with the
* post-transition state recorded (start → RUN, finish(0) → DONE). */
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 80,
&st_start80),
"job 80 START fact must be retained exactly once (Req 3.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 81,
&st_start81),
"job 81 START fact must be retained exactly once (Req 3.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 80,
&st_fin80),
"job 80 FINISH fact must be retained exactly once (Req 3.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 81,
&st_fin81),
"job 81 FINISH fact must be retained exactly once (Req 3.3)");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, st_start80);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, st_start81);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_fin80);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_fin81);
/* Act 3 — RECONNECT + covering ACK (Req 3.4): re-register both
* standbys (slots were vacated; repl_add_standby resets their
* watermarks to 0) and deliver one ACK covering the retained suffix —
* self + sb_a = 2 of n_peers=3, a majority. */
TEST_ASSERT_TRUE(
repl_add_standby(&g_hc.persist.repl, &g_hep_a, "sb_a", 7400) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_hc.persist.repl, &g_hep_b, "sb_b", 7401) >= 0);
repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark);
deadline = cbrp_mono_ms() + 2000;
while (atomic_load(&g_hc.persist.committed_lsn) < watermark
&& cbrp_mono_ms() < deadline)
cbrp_sleep_ms(5);
/* Assert — committed without re-delivery (Req 3.4): committed_lsn
* covers the retained suffix; the WAL still holds each fact exactly
* once (no duplicate appends) and current_lsn never moved — the
* RETAINED entries committed, not re-delivered copies. */
TEST_ASSERT_TRUE_MESSAGE(
atomic_load(&g_hc.persist.committed_lsn) >= watermark,
"a covering ACK after reconnect must advance committed_lsn over "
"the retained fact suffix (Req 3.4)");
pthread_mutex_lock(&g_hc.persist.lock);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_after, g_hc.persist.current_lsn,
"commitment must consume the retained entries — no re-delivery, "
"no duplicate appends (Req 3.4)");
pthread_mutex_unlock(&g_hc.persist.lock);
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 80, NULL),
"job 80 START must still be on disk exactly once after commit");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 81, NULL),
"job 81 START must still be on disk exactly once after commit");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 80, NULL),
"job 80 FINISH must still be on disk exactly once after commit");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 81, NULL),
"job 81 FINISH must still be on disk exactly once after commit");
cbrp_teardown();
printf("test_fact_retention_across_quorum_loss: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Idempotency / ordering guards for mark_job_* (task 6.1, Req 5.1/5.2/5.3)
* ════════════════════════════════════════════════════════════════════════
*
* RED-phase tests for the transition guards that task 6.2 adds at the top
* of mark_job_start / mark_job_finish / mark_job_fail / mark_job_update
* (job.c:583/638/696/726 pre-task anchors). Design §Idempotency Guards:
* - duplicate (already-reflected transition) → INFO log, NO-OP: state
* unchanged, BEFORE/AFTER_EVENT hooks NOT fired → NO WAL entry appended
* (current_lsn unmoved), caller-visible rc treated as success-no-op (5.1)
* - backward (terminal current state, non-terminal requested: START on a
* finished job, UPDATE clearing terminal bits) → WARN log, REJECT,
* terminal state preserved (5.2)
* - replay: re-applying the same transition converges to single-
* application state — no second WAL entry, identical final state (5.3)
* - mark_job_fail's existing fail-after-finish PERR-merge (job.c:702-716)
* is forward-compatible and must be PRESERVED by the guard (pin)
*
* Drivers: the static mark_job_* are reached through their non-static
* callers on_job_start / on_job_finish / on_job_fail (job.c) — exactly the
* fact path used by test_fact_retention_across_quorum_loss, minus the
* opcode decode. These wrappers each return 0 unconditionally today, so the
* "success-no-op rc" observable (every caller treats a guarded-out
* transition as success) is asserted directly on the wrapper rc.
*
* Observability of state: the per-job WAL append count and the LAST
* recorded post-transition state come from cbrp_count_jobstate_entries
* (direct on-disk scan, defined above); the job's LIVE in-memory state is
* read via cbrp_job_live_state (key_find on g_hc.jobmgr). current_lsn is
* the "did a hook fire / WAL entry append" oracle (a guarded-out no-op runs
* no hooks → no persist_wal_async → current_lsn frozen).
*
* RED expectation against the UNGUARDED mark_job_* (no guard yet):
* - duplicate FINISH re-fires the AFTER_EVENT hook → a 2nd WAL_JOB_FINISH
* append, current_lsn advances → the "current_lsn unmoved / count==1"
* assertions FAIL.
* - START on a DONE job unconditionally sets state = JOB_STAT_RUN and
* appends WAL_JOB_START → the "state preserved DONE / no new append"
* assertions FAIL.
* - replay FINISH appends a duplicate entry → the "single append"
* assertion FAILS.
* The fail-after-finish PERR-merge pin and the success-no-op rc pins lock
* behavior that must hold both before and after the guard (preservation
* pins, like the task-3.1 validation pins); they ride this group so 6.2
* cannot silently regress the merge or the wrapper return contract.
*
* NOTE on the UPDATE-clears-terminal arm (5.2): mark_job_update is static
* with no public caller that produces a terminal→non-terminal state
* (on_job_postdone's update is forward — it OR's PDONE, preserving terminal;
* job_ctrl SIGCONT only clears suspend bits). mark_job_start IS the
* canonical "set state to the non-terminal JOB_STAT_RUN" overwrite, so the
* START-on-terminal case below is the faithful, publicly-reachable witness
* of the shared backward rule the design lists for START and UPDATE
* together. test_guard_forward_update_on_terminal_preserved additionally
* pins that a FORWARD update on a terminal job (on_job_postdone) is NOT
* over-rejected — the boundary the guard must respect.
*/
/* Read a job's live in-memory state from g_hc.jobmgr (post-transition).
* Returns the state flags, or -1 if the job is absent. */
static int cbrp_job_live_state(uint64_t jobid)
{
job_t *j = NULL;
int st = -1;
lock(g_hc.jobmgr, RTTI_LOCK_READ);
key_find(g_hc.jobmgr, (void *)jobid, (void **)&j);
if (j != NULL) {
reference(j);
}
unlock(g_hc.jobmgr);
if (j != NULL) {
lock(j, RTTI_LOCK_READ);
st = (int)j->job.state;
unlock(j);
dereference(j);
}
return st;
}
/* Drive a job to terminal DONE via start(jobid) → finish(jobid, exit 0).
* Returns current_lsn after the two forward facts (the watermark the
* duplicate/backward cases must NOT advance). */
static uint64_t cbrp_drive_to_done(uint64_t jobid)
{
static unsigned int gd_pids[1] = { 9999 };
job_pidinfo_t start;
xce_jobfinish_t fin;
uint64_t lsn;
memset(&start, 0, sizeof(start));
start.jobid = jobid;
start.npids = 1;
start.pids = &gd_pids[0];
TEST_ASSERT_EQUAL_INT(0, on_job_start(&g_hc, &start));
memset(&fin, 0, sizeof(fin));
fin.jobid = jobid;
fin.exit_code = 0; /* → JOB_STAT_DONE */
TEST_ASSERT_EQUAL_INT(0, on_job_finish(&g_hc, &fin));
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_DONE, cbrp_job_live_state(jobid),
"precondition: job must be DONE after start→finish(0)");
pthread_mutex_lock(&g_hc.persist.lock);
lsn = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
return lsn;
}
/* 5.1 — duplicate FINISH on a DONE job is a logged no-op: state unchanged,
* NO AFTER_EVENT hook fired (NO new WAL entry → current_lsn frozen), exactly
* one FINISH entry on disk, and on_job_finish returns success-no-op (0). */
static void test_guard_duplicate_finish_on_done_is_noop(void)
{
xce_jobfinish_t fin;
uint64_t lsn_after_done, lsn_after_dup;
uint32_t st_finish = 0;
int rc;
cbrp_setup(3000);
cbrp_make_job(70, &g_job_submit);
memset(&g_job_submit, 0, sizeof(g_job_submit));
lsn_after_done = cbrp_drive_to_done(70);
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 70, NULL),
"exactly one FINISH entry must exist after the first finish");
/* Act: a duplicate FINISH on the already-DONE job (retransmit / replay) */
memset(&fin, 0, sizeof(fin));
fin.jobid = 70;
fin.exit_code = 0;
rc = on_job_finish(&g_hc, &fin);
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after_dup = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Assert: success-no-op rc; state unchanged; NO hook fired (no append) */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a guarded-out duplicate FINISH must return success-no-op (5.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_DONE, cbrp_job_live_state(70),
"duplicate FINISH must leave the job's state unchanged (5.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_after_done, lsn_after_dup,
"duplicate FINISH must fire NO AFTER_EVENT hook → NO WAL entry "
"appended → current_lsn unmoved (5.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 70,
&st_finish),
"duplicate FINISH must not append a second FINISH entry (5.1)");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_finish);
cbrp_teardown();
printf("test_guard_duplicate_finish_on_done_is_noop: PASS\n");
}
/* 5.2 — START on a finished (DONE) job is rejected: the terminal state is
* preserved (never overwritten with JOB_STAT_RUN), NO WAL_JOB_START entry is
* appended, and on_job_start returns a success-no-op code that the caller
* (do_notify) treats as success. mark_job_start is the canonical
* terminal→non-terminal overwrite, so this also witnesses the UPDATE-clears-
* terminal arm of the shared backward rule (5.2). */
static void test_guard_start_on_terminal_rejected_state_preserved(void)
{
static unsigned int rs_pids[1] = { 7001 };
job_pidinfo_t restart;
uint64_t lsn_after_done, lsn_after_start;
int rc;
cbrp_setup(3000);
cbrp_make_job(71, &g_job_submit);
memset(&g_job_submit, 0, sizeof(g_job_submit));
lsn_after_done = cbrp_drive_to_done(71);
/* Precondition: drive_to_done's start→finish appended exactly one
* WAL_JOB_START (the forward start). The rejected backward START below
* must not add a second one. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 71, NULL),
"precondition: exactly one START entry from the forward start");
/* Act: a stale/duplicate START arrives for the already-finished job */
memset(&restart, 0, sizeof(restart));
restart.jobid = 71;
restart.npids = 1;
restart.pids = &rs_pids[0];
rc = on_job_start(&g_hc, &restart);
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after_start = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Assert: terminal preserved, no append, success-no-op rc */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a rejected backward START must return success-no-op to the caller "
"(5.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_DONE, cbrp_job_live_state(71),
"START on a DONE job must preserve the terminal state — never "
"overwrite it with JOB_STAT_RUN (5.2)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_after_done, lsn_after_start,
"a rejected backward START must append NO WAL entry (5.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 71, NULL),
"a rejected backward START must not append a second START entry "
"(5.2)");
cbrp_teardown();
printf("test_guard_start_on_terminal_rejected_state_preserved: PASS\n");
}
/* 5.3 — replay equivalence: applying the same FINISH transition twice yields
* state identical to a single application AND a single WAL entry. The second
* application is the guarded-out duplicate; the on-disk FINISH count staying
* at 1 with current_lsn frozen is the replay-once invariant. */
static void test_guard_replay_finish_equals_single_application(void)
{
xce_jobfinish_t fin;
uint64_t lsn_single, lsn_replay;
int state_single, state_replay;
uint32_t st_disk = 0;
cbrp_setup(3000);
cbrp_make_job(72, &g_job_submit);
memset(&g_job_submit, 0, sizeof(g_job_submit));
/* Single application baseline */
lsn_single = cbrp_drive_to_done(72);
state_single = cbrp_job_live_state(72);
TEST_ASSERT_EQUAL_INT(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 72, NULL));
/* Replay the FINISH (retry / WAL replay re-application) */
memset(&fin, 0, sizeof(fin));
fin.jobid = 72;
fin.exit_code = 0;
TEST_ASSERT_EQUAL_INT_MESSAGE(0, on_job_finish(&g_hc, &fin),
"replayed FINISH must return success-no-op (5.3)");
pthread_mutex_lock(&g_hc.persist.lock);
lsn_replay = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
state_replay = cbrp_job_live_state(72);
/* Assert: state and the WAL log are identical to single application */
TEST_ASSERT_EQUAL_INT_MESSAGE(state_single, state_replay,
"replayed transition must converge to single-application state "
"(5.3)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_single, lsn_replay,
"replaying a committed transition must append no second WAL entry "
"(5.3)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 72,
&st_disk),
"the FINISH fact must remain on disk exactly once after replay "
"(5.3)");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_disk);
cbrp_teardown();
printf("test_guard_replay_finish_equals_single_application: PASS\n");
}
/* 5.2 boundary pin — a FORWARD update on a terminal job must NOT be
* over-rejected by the guard. on_job_postdone is the reachable mark_job_update
* caller: on a DONE job it OR's JOB_STAT_PDONE (terminal preserved, strictly
* forward). This must still apply (state gains PDONE, an entry appends) after
* the guard lands — the guard rejects backward, never a forward merge.
* Preservation pin: green before and after task 6.2. */
static void test_guard_forward_update_on_terminal_preserved(void)
{
uint64_t lsn_after_done, lsn_after_pdone;
int st;
cbrp_setup(3000);
cbrp_make_job(73, &g_job_submit);
memset(&g_job_submit, 0, sizeof(g_job_submit));
lsn_after_done = cbrp_drive_to_done(73);
/* Act: post-done arrives — forward update merging PDONE onto DONE */
TEST_ASSERT_EQUAL_INT(0, on_job_postdone(&g_hc, &g_req_ep, 73));
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after_pdone = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
st = cbrp_job_live_state(73);
/* Assert: terminal preserved, PDONE merged forward, the update appended */
TEST_ASSERT_TRUE_MESSAGE((st & JOB_STAT_DONE) != 0,
"a forward update on a terminal job must preserve the terminal bit");
TEST_ASSERT_TRUE_MESSAGE((st & JOB_STAT_PDONE) != 0,
"the forward update must merge JOB_STAT_PDONE (not be rejected)");
TEST_ASSERT_TRUE_MESSAGE(lsn_after_pdone > lsn_after_done,
"a forward update on a terminal job must still append its WAL entry "
"— the guard rejects backward transitions, never forward merges");
cbrp_teardown();
printf("test_guard_forward_update_on_terminal_preserved: PASS\n");
}
/* 5.2 pin — mark_job_fail's existing fail-after-finish PERR-merge
* (job.c:702-716) must be PRESERVED by the guard: a POST_EXEC-stage fail on a
* DONE job keeps DONE and OR's JOB_STAT_PERR (it does NOT reset the job to a
* bare JOB_STAT_EXIT). Forward-compatible terminal→terminal merge — the guard
* must not reject it. Preservation pin: green before and after task 6.2. */
static void test_guard_fail_after_finish_preserves_perr_merge(void)
{
job_failinfo_t failinfo;
uint64_t lsn_after_done, lsn_after_fail;
int st;
cbrp_setup(3000);
cbrp_make_job(74, &g_job_submit);
memset(&g_job_submit, 0, sizeof(g_job_submit));
lsn_after_done = cbrp_drive_to_done(74);
/* Act: a post-exec-stage failure for the already-finished job */
memset(&failinfo, 0, sizeof(failinfo));
failinfo.jobid = 74;
failinfo.stage = JOB_EXEC_STAGE_POST_EXEC;
failinfo.code = 1;
failinfo.detail = (char *)"post-exec fail after finish";
TEST_ASSERT_EQUAL_INT(0, on_job_fail(&g_hc, &failinfo));
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after_fail = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
st = cbrp_job_live_state(74);
/* Assert: DONE preserved, PERR merged in — not reset to bare EXIT */
TEST_ASSERT_TRUE_MESSAGE((st & JOB_STAT_DONE) != 0,
"fail-after-finish must keep the DONE terminal bit (PERR-merge "
"behavior, job.c:702-716)");
TEST_ASSERT_TRUE_MESSAGE((st & JOB_STAT_PERR) != 0,
"fail-after-finish must OR JOB_STAT_PERR onto the existing terminal "
"state (PERR-merge behavior, job.c:702-716)");
TEST_ASSERT_TRUE_MESSAGE(lsn_after_fail > lsn_after_done,
"the PERR-merge fail must still append its WAL entry (forward "
"terminal→terminal merge — not a guarded-out no-op)");
cbrp_teardown();
printf("test_guard_fail_after_finish_preserves_perr_merge: PASS\n");
}
/* ── Test group entry point ──────────────────────────────────────────── */
void test_commit_barrier_reply(void)
{
/* makeString needs the global xstring table. In the full suite,
* test_worker_join already initialised it (unguarded, its line 520);
* standalone (-t test_commit_barrier_reply) it is uninitialised and
* makeString returns NULL. Probe instead of double-initialising. */
if (makeString((char *)"cbrp-probe") == NULL)
xstring_table_init(139);
UNITY_BEGIN();
/* do_jobcontrol_op — Req 1.1, 1.2, 1.4 */
RUN_TEST(test_jobctrl_withheld_acks_replies_error);
RUN_TEST(test_jobctrl_success_reply_only_after_commit);
RUN_TEST(test_jobctrl_unknown_job_error_reply_immediate);
/* do_queuecontrol_op */
RUN_TEST(test_queuectrl_withheld_acks_replies_error);
RUN_TEST(test_queuectrl_success_reply_only_after_commit);
RUN_TEST(test_queuectrl_unknown_queue_error_reply_immediate);
/* do_hostcontrol_op */
RUN_TEST(test_hostctrl_withheld_acks_replies_error);
RUN_TEST(test_hostctrl_success_reply_only_after_commit);
RUN_TEST(test_hostctrl_unknown_host_error_reply_immediate);
/* handle_worker_join (via handle_join) */
RUN_TEST(test_worker_join_withheld_acks_replies_error);
RUN_TEST(test_worker_join_success_reply_only_after_commit);
RUN_TEST(test_worker_join_no_hostconfig_error_reply_immediate);
/* Receipt-ACK verdict pins (task 3.3, Req 2.3) — green on landing */
RUN_TEST(test_jobdispatch_receipt_ack_no_wal_append);
RUN_TEST(test_set_preason_receipt_reply_fact_retained);
RUN_TEST(test_clr_preason_receipt_reply_fact_retained);
RUN_TEST(test_update_op_rusage_receipt_reply_facts_retained);
/* Fact retention across quorum loss (task 4.3, Req 3.1, 3.3, 3.4) —
* green on landing against tasks 1.2 + 4.2 output */
RUN_TEST(test_fact_retention_across_quorum_loss);
/* mark_job_* idempotency / ordering guards (task 6.1, Req 5.1/5.2/5.3)
* — the no-op / backward-reject / replay cases FAIL (red) against the
* unguarded mark_job_*; the forward-merge and PERR-merge cases are
* preservation pins (green before and after task 6.2) */
RUN_TEST(test_guard_duplicate_finish_on_done_is_noop);
RUN_TEST(test_guard_start_on_terminal_rejected_state_preserved);
RUN_TEST(test_guard_replay_finish_equals_single_application);
RUN_TEST(test_guard_forward_update_on_terminal_preserved);
RUN_TEST(test_guard_fail_after_finish_preserves_perr_merge);
UNITY_END();
}
Reference in a new issue View git blame Copy permalink