/* * 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 #include #include #include #include #include #include #include #include #include #include #include #include #include #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 */ int32_t last_data_which; /* RequestReply_data_which of the last * replymsg observed (raft-apply-decoupling * task 6.2: distinguishes a jobid-bearing * success reply from a nodata one) */ uint64_t last_jobid; /* rr.data.jobid of the last replymsg observed * -- meaningful only when last_data_which == * RequestReply_data_jobid */ } 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->last_data_which = rr.data_which; if (rr.data_which == RequestReply_data_jobid) obs->last_jobid = rr.data.jobid; 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; /* bug-1419: commit_engine_fini frees any still-pending continuation * without invoking its callback, so a test that deliberately leaves a * submission unresolved (no ack/timeout/truncation) leaks its jobid * range in job.c's process-lifetime in-flight registry. Clear it here * so it never outlives this fixture. */ inflight_submits_reset_for_test(); 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"); } /* ── handle_jobsubmit (raft-apply-decoupling task 3.1/3.2) ─────────────── * * task 3.1 (Req 2.1/2.3): a job submission whose commit fails or times out * must leave NO job object in c->jobmgr — construction is deferred until * after the commit barrier succeeds. * * task 3.2 (Req 1.1, 1.4, 2.1, 2.3): the synchronous commit call from 3.1 is * replaced with a registration against the commit engine's continuation * mechanism (commit_submit_async, via persist_submit_job_payloads_async). * handle_jobsubmit now returns BEFORE any commit outcome is known. A * failed/timed-out outcome still replies failure directly, with no job ever * constructed (same observable as 3.1). A COMMITTED outcome is handed off to * the not-yet-built ordered apply pump (task 6.1/6.2) via * persist_register_pending_apply() instead of applying or replying directly * — so a successful submission has NO independently observable behavior * (no reply, no job in c->jobmgr) until task 6 exists; the test below * asserts only that the hand-off itself was registered for the correct lsn. */ static job_startinfo_t g_jobsub_startinfo; static job_submit_data_t g_jobsub_data; static xce_job_submit_t g_jobsub_msg; static xce_request_t g_jobsub_req; static xce_peerinfo_t g_jobsub_from; static void cbrp_arrange_jobsubmit(uint64_t jobid, uint64_t array_size) { memset(&g_jobsub_startinfo, 0, sizeof(g_jobsub_startinfo)); g_jobsub_startinfo.cmdline = (char *)"/bin/true"; memset(&g_jobsub_data, 0, sizeof(g_jobsub_data)); g_jobsub_data.queue = (char *)"normal"; g_jobsub_data.user = (char *)"tester"; g_jobsub_data.start_info = &g_jobsub_startinfo; memset(&g_jobsub_msg, 0, sizeof(g_jobsub_msg)); g_jobsub_msg.jobid = jobid; g_jobsub_msg.array_size = array_size; g_jobsub_msg.submit = &g_jobsub_data; g_jobsub_from.name = (char *)"requester"; g_jobsub_from.key = (char *)""; memset(&g_jobsub_req, 0, sizeof(g_jobsub_req)); g_jobsub_req.opcode = OPCode_jobsub; g_jobsub_req.seq = 15; g_jobsub_req.from = &g_jobsub_from; g_jobsub_req.body_which = Request_body_jobsubmsg; g_jobsub_req.body.jobsubmsg = &g_jobsub_msg; } /* Direct c->jobmgr lookup — the task 3.1 observable is about the in-memory * job TABLE, not the wire reply, so it is checked independently of * cbrp_scan_replies/cbrp_reply_obs_t (which do not carry a jobid). */ static bool cbrp_job_exists(uint64_t jobid) { job_t *j = NULL; int rc; lock(g_hc.jobmgr, RTTI_LOCK_READ); rc = key_find(g_hc.jobmgr, (void *)jobid, (void **)&j); unlock(g_hc.jobmgr); return (rc > 0 && j != NULL); } static void test_jobsubmit_withheld_acks_no_phantom_job(void) { cbrp_setup(300); cbrp_arrange_jobsubmit(9100, 1); cbrp_drive_withheld(handle_jobsubmit, &g_jobsub_req, 300); TEST_ASSERT_FALSE_MESSAGE(cbrp_job_exists(9100), "a job submission whose commit times out must leave no job object " "in the in-memory job table (raft-apply-decoupling task 3.1, " "Req 2.1/2.3)"); cbrp_teardown(); printf("test_jobsubmit_withheld_acks_no_phantom_job: PASS\n"); } /* raft-apply-decoupling task 3.2 (original intent, superseded by task 6.1/ * 6.2 -- see raft-apply-decoupling task 7.3, Req 11.1): supersedes the * pre-3.2 test_jobsubmit_success_reply_and_job_visible_after_commit (which * drove handle_jobsubmit through cbrp_drive_acked, a helper that asserts * the handler blocks until its reply is observable — no longer true here). * This test still asserts task 3.2's own contract (the calling thread * returns before the commit resolves, and the WAL entry is appended before * that return); it no longer asserts task 3.2's INTERIM "no reply / no job * yet" gap on the committed-outcome path, because the ordered apply pump * built by task 6.1/6.2 now closes that gap: the pump consumes the * pending-apply hand-off, applies the entry via apply_committed_entry(), * and fires the client's success reply — all strictly after the entry * commits, never before or during the calling thread's earlier return. * (Catching persist_has_pending_apply()'s transient true window from a * second thread is inherently racy against the pump's own consumption of * it — proven flaky empirically across repeated runs of this test — so * this test asserts the deterministic settled state instead, the same * poll-for-reply idiom test_jobsubmit_success_reply_after_apply_carries_ * jobid already uses for the identical ordering proof.) */ static void test_jobsubmit_returns_before_commit_and_registers_pending_apply(void) { uint64_t lsn_before, watermark, t0, join_elapsed_ms; cbrp_reply_obs_t obs = { 0 }; bool job_exists; cbrp_setup(2000); /* generous timeout: must never be needed */ cbrp_arrange_jobsubmit(9101, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); /* No ACK is delivered before this join. If handle_jobsubmit still * blocked on the majority-commit barrier (the pre-3.2 behavior), this * join could not return before the 2000ms sync_timeout elapses; a fast * return proves registration replaced blocking (Req 1.1, 1.4). */ t0 = cbrp_mono_ms(); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); join_elapsed_ms = cbrp_mono_ms() - t0; watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); /* Deliver the covering ACK, then poll for the settled, end-to-end * committed-outcome effect: the ordered apply pump applies the entry * and fires the client's success reply (raft-apply-decoupling task * 6.1/6.2). */ repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); cbrp_poll_reply(&g_req_ep, 2000, &obs); job_exists = cbrp_job_exists(9101); cbrp_teardown(); TEST_ASSERT_TRUE_MESSAGE(join_elapsed_ms < 500, "handle_jobsubmit must return before any commit outcome is known " "(raft-apply-decoupling task 3.2, Req 1.1/1.4) — no ACK was ever " "delivered before this join"); TEST_ASSERT_TRUE_MESSAGE(watermark > 0, "job submission must still append its WAL entry"); TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, "a committed entry must reach the client with a reply once the " "ordered apply pump consumes its pending-apply hand-off " "(raft-apply-decoupling task 6.1/6.2, Req 2.2)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status, "a committed, applied job submission must reply success"); TEST_ASSERT_TRUE_MESSAGE(job_exists, "the job must be visible in the in-memory job table once the " "ordered apply pump applies its committed entry " "(raft-apply-decoupling task 6.1/6.2, Req 2.2)"); printf("test_jobsubmit_returns_before_commit_and_registers_pending_apply:" " PASS\n"); } /* raft-apply-decoupling task 3.2: the failure/timeout path must ALSO prove * the handler returns immediately — cbrp_drive_withheld (used by * test_jobsubmit_withheld_acks_no_phantom_job below) only proves the * EVENTUAL reply/no-phantom-job observable, not that the calling thread * stopped blocking well before the commit timeout elapses. */ static void test_jobsubmit_returns_before_commit_on_failure_path_too(void) { uint64_t lsn_before, watermark, t0, join_elapsed_ms; cbrp_reply_obs_t obs = { 0 }; bool job_exists; cbrp_setup(1000); cbrp_arrange_jobsubmit(9104, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); /* No ACK is EVER delivered in this test. The 1000ms sync_timeout_ms is * how long the ENTRY takes to resolve as FAILED — the HANDLER itself * must return immediately, well before that timeout. */ t0 = cbrp_mono_ms(); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); join_elapsed_ms = cbrp_mono_ms() - t0; watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 500); /* Let the timeout sweep resolve the entry FAILED and confirm the * (eventual) failure reply + no-phantom-job observable — same * contract test_jobsubmit_withheld_acks_no_phantom_job already pins, * verified here alongside the non-blocking-return proof. */ cbrp_poll_reply(&g_req_ep, 1000 + 1500, &obs); job_exists = cbrp_job_exists(9104); cbrp_teardown(); TEST_ASSERT_TRUE_MESSAGE(join_elapsed_ms < 500, "handle_jobsubmit must return immediately after registering its " "continuation, without waiting for the commit timeout to elapse " "(raft-apply-decoupling task 3.2, Req 1.1/1.4)"); TEST_ASSERT_TRUE_MESSAGE(watermark > 0, "job submission must still append its WAL entry before the " "handler returns"); TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, "a forced timeout must still eventually reply to the client"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status, "a timed-out commit must reply with -ETIMEDOUT, the same " "vocabulary the synchronous path already used (Req 2.3)"); TEST_ASSERT_FALSE_MESSAGE(job_exists, "no job may ever be constructed for an entry whose commit timed " "out (raft-apply-decoupling task 3.2, Req 2.3)"); printf("test_jobsubmit_returns_before_commit_on_failure_path_too: PASS\n"); } /* ════════════════════════════════════════════════════════════════════════ * raft-apply-decoupling task 3.3 (Req 2.1, 2.3): the failure-path guarantee * task 3.2 built, and the single-timeout tests above already proved, now * needs proof that it (a) also holds for an explicit, non-timeout FAILED * outcome, and (b) holds across REPEATED forced failures, not just once. * Every test below follows the task-3.1-round-3 lesson: assertion-relevant * observables are captured into locals, cbrp_teardown() runs exactly once * and unconditionally, and every TEST_ASSERT fires only after teardown * returns. * * Demotion is deliberately NOT used to force the non-timeout FAILED case * here: raft_cb_become_follower's demotion branch resolves continuations * on `c->commit` (the cluster_t-embedded commit_engine_t), but this file's * cbrp_setup fixture wires job submission's continuations through a * SEPARATE standalone `g_hce` (via repl_set_commit_engine) exactly like * every other handler in this file -- calling the demotion seam against * g_hc would touch the wrong, never-initialized engine and prove nothing. * Truncation resolves through persist_truncate_wal_above -> * p->on_truncate -> the SAME g_hce commit_engine_init wires * (raft-apply-decoupling task 1.4), so it is the correct and only * deterministic way to force an explicit FAILED outcome for job * submission in this harness. Generic exactly-once demotion-resolve * behavior already has dedicated coverage at the commit-engine level in * test_demotion_convergence.c (task 1.4/1.5). * ════════════════════════════════════════════════════════════════════════ */ /* Single-shot: an explicit FAILED outcome (rc=-ENOENT, forced via * log-tail truncation) must leave no phantom job, exactly like the * -ETIMEDOUT path -- but the reply status must carry the DISTINCT * truncation vocabulary, not the timeout one, proving the continuation * genuinely observed a different cause. */ static void test_jobsubmit_truncation_forces_failed_no_phantom_job(void) { uint64_t lsn_before, watermark; int create_rc, join_rc, trunc_rc; cbrp_reply_obs_t obs = { 0 }; bool job_exists; cbrp_setup(5000); /* generous: truncation, not the timeout sweep, must * be what resolves this entry */ cbrp_arrange_jobsubmit(9110, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; create_rc = pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg); join_rc = (create_rc == 0) ? pthread_join(g_op_tid, NULL) : -1; watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); /* Force an explicit FAILED outcome (rc=-ENOENT), NOT a timeout: * truncate the entry that was just appended. Exercises * commit_on_truncate -> commit_resolve_pending_above(ce, cutoff, * -ENOENT) (task 1.4) through the real job-submit continuation * (task 3.2), distinct from the -ETIMEDOUT path already covered by * test_jobsubmit_returns_before_commit_on_failure_path_too. */ trunc_rc = (watermark > 0) ? persist_truncate_wal_above(&g_hc.persist, watermark - 1) : -1; cbrp_poll_reply(&g_req_ep, 2000, &obs); job_exists = cbrp_job_exists(9110); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(0, create_rc, "pthread_create must succeed"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, join_rc, "pthread_join must succeed"); TEST_ASSERT_TRUE_MESSAGE(watermark > 0, "job submission must append its WAL entry before truncation can " "force its continuation FAILED"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, trunc_rc, "forcing truncation must succeed"); TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, "a truncated commit must still reply to the client"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, obs.last_status, "truncation must reply with -ENOENT, the existing " "commit_resolve_pending_above error vocabulary (raft-apply-" "decoupling task 1.4) -- distinct from the -ETIMEDOUT timeout " "path already covered elsewhere in this file"); TEST_ASSERT_FALSE_MESSAGE(job_exists, "no job may ever be constructed for an entry whose commit failed " "via truncation, not just via timeout (raft-apply-decoupling task " "3.3, Req 2.1/2.3)"); printf("test_jobsubmit_truncation_forces_failed_no_phantom_job: PASS\n"); } #define CBRP_JOBSUBMIT_REPEAT_N 5 /* Proves the failure-path guarantee holds across MULTIPLE forced timeouts * in sequence, not just once -- and that nothing about the fixture (the * jobmgr, the per-request jobsubmit_async_ctx_t lifecycle) leaks or * accumulates state across repeated failures. */ static void test_jobsubmit_repeated_timeout_failures_no_phantom_jobs(void) { const uint64_t base_jobid = 9120; int create_rc[CBRP_JOBSUBMIT_REPEAT_N]; int join_rc[CBRP_JOBSUBMIT_REPEAT_N]; uint64_t watermark[CBRP_JOBSUBMIT_REPEAT_N]; int n_replies[CBRP_JOBSUBMIT_REPEAT_N]; int32_t last_status[CBRP_JOBSUBMIT_REPEAT_N]; bool job_exists[CBRP_JOBSUBMIT_REPEAT_N]; int i; cbrp_setup(200); /* short timeout: bound the loop's wall time */ for (i = 0; i < CBRP_JOBSUBMIT_REPEAT_N; i++) { uint64_t jobid = base_jobid + (uint64_t)i; uint64_t lsn_before; cbrp_reply_obs_t obs = { 0 }; cbrp_arrange_jobsubmit(jobid, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; create_rc[i] = pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg); join_rc[i] = (create_rc[i] == 0) ? pthread_join(g_op_tid, NULL) : -1; watermark[i] = cbrp_poll_append(&g_hc.persist, lsn_before, 1000); /* No ACK is ever delivered: this iteration's entry must resolve * FAILED via the timeout sweep (task 1.3), not majority ack. */ cbrp_poll_reply(&g_req_ep, 200 + 1500, &obs); n_replies[i] = obs.n_replies; last_status[i] = obs.last_status; job_exists[i] = cbrp_job_exists(jobid); } cbrp_teardown(); for (i = 0; i < CBRP_JOBSUBMIT_REPEAT_N; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE(0, create_rc[i], "pthread_create must succeed for every repeated iteration"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, join_rc[i], "pthread_join must succeed for every repeated iteration"); TEST_ASSERT_TRUE_MESSAGE(watermark[i] > 0, "each repeated submission must append its own WAL entry"); TEST_ASSERT_TRUE_MESSAGE(n_replies[i] > 0, "every repeated forced-timeout submission must still reply"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, last_status[i], "every repeated forced-timeout submission must reply " "-ETIMEDOUT"); TEST_ASSERT_FALSE_MESSAGE(job_exists[i], "no phantom job may exist for any repeated forced-timeout " "failure (raft-apply-decoupling task 3.3, Req 2.1/2.3) -- " "proves the guarantee is not a one-shot fluke"); } printf("test_jobsubmit_repeated_timeout_failures_no_phantom_jobs: PASS\n"); } /* bug-1419: a client retry that resends the SAME jobid before it has seen * the first attempt's outcome must be rejected as a duplicate, not append * a second WAL_JOB_SUBMIT entry. * * Root cause this proves fixed: jobid_range_exists()'s c->jobmgr check is * airtight against a genuinely completed prior submission (the job is * there), but raft-apply-decoupling task 3.1 deliberately defers job * construction until AFTER commit -- so a submission that is durably * appended and awaiting its commit outcome does NOT yet exist in * c->jobmgr, and jobid_range_exists() alone cannot see it. libxce's * client-side failover retry wrapper (XCECLI_FAILOVER_CALL, * client_failover.c) resends the identical jobid on a transport error or * failover-mid-flight, exactly the shape this test drives directly against * the handler. * * Withholds ACKs (mirrors cbrp_drive_withheld) so the first call's commit * stays genuinely unresolved when the second call arrives -- the real * race window, not a timing approximation. */ static void test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry(void) { const uint64_t jobid = 9139; uint64_t lsn_before, lsn_before_second, watermark_first, watermark_second; int rc_second; cbrp_setup(2000); /* generous: must never fire -- this test resolves * via explicit teardown, not the timeout sweep */ cbrp_arrange_jobsubmit(jobid, 1); pthread_mutex_lock(&g_hc.persist.lock); lsn_before = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); /* First attempt: durably appends and registers its commit continuation, * then returns immediately (task 3.2 -- handle_jobsubmit never blocks). * No ACK is ever delivered, so its outcome stays unresolved for the * rest of this test. */ g_op_arg.fn = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); watermark_first = cbrp_poll_append(&g_hc.persist, lsn_before, 1000); TEST_ASSERT_TRUE_MESSAGE(watermark_first > 0, "the first attempt must append its WAL entry"); /* Second attempt: the retry, same jobid, while the first is still * in flight (no ACK delivered, no timeout elapsed). Called directly -- * handle_jobsubmit never blocks, so no second thread is needed. */ cbrp_arrange_jobsubmit(jobid, 1); /* fresh xce_request_t (same seq/jobid) */ pthread_mutex_lock(&g_hc.persist.lock); lsn_before_second = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); rc_second = handle_jobsubmit(&g_hc, &g_req_ep, &g_jobsub_req); watermark_second = cbrp_poll_append(&g_hc.persist, lsn_before_second, 300); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc_second, "bug-1419: a resubmit for a jobid still in flight (durably " "appended, commit outcome unresolved) must be rejected -EEXIST, " "identically to a resubmit for a jobid already visible in " "c->jobmgr"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, (int)watermark_second, "bug-1419: a rejected resubmit must NOT append a second " "WAL_JOB_SUBMIT entry for the same jobid"); printf("test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry: PASS\n"); } /* bug-1419 residual: does the SAME dedup guard still hold AFTER the first * attempt has fully resolved (committed, applied, replied, in-flight claim * released via jobsubmit_async_ctx_release) -- not just DURING the * still-unresolved window the test above proves. A live chaotic-soak run * (2026-07-08) showed a genuine WAL-level jobid duplicate (two committed * WAL_JOB_SUBMIT entries, same term, same primary) that neither known guard * should have let through if the retry arrived strictly before OR strictly * after resolution -- this test pins down whether the "after" window is as * airtight as jobid_range_exists()'s own doc comment claims, by driving the * first attempt all the way to a fully-applied, fully-replied, * fully-released state before firing the retry. */ static void test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job(void) { const uint64_t jobid = 9150; uint64_t lsn_before, lsn_before_second, watermark, watermark_second; cbrp_reply_obs_t obs = { 0 }; int rc_second; cbrp_setup(2000); cbrp_arrange_jobsubmit(jobid, 1); pthread_mutex_lock(&g_hc.persist.lock); lsn_before = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); /* First attempt: append, commit (ACK delivered), apply, and reply -- * drive it all the way to the fully-resolved, fully-released state. */ g_op_arg.fn = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 1000); TEST_ASSERT_TRUE_MESSAGE(watermark > 0, "the first attempt must append its WAL entry"); repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); cbrp_poll_reply(&g_req_ep, 2000, &obs); TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, "the first attempt must reach a replied, fully-applied state " "before the retry below fires"); TEST_ASSERT_TRUE_MESSAGE(cbrp_job_exists(jobid), "the first attempt's job must be visible in c->jobmgr before the " "retry below fires -- otherwise this test does not actually " "exercise the AFTER-resolution window"); /* Second attempt: the retry arrives strictly AFTER the first has fully * committed, applied, replied, and released its in-flight claim -- the * window jobid_range_exists()'s own doc comment claims is "airtight * against a genuinely completed prior submission". */ cbrp_arrange_jobsubmit(jobid, 1); pthread_mutex_lock(&g_hc.persist.lock); lsn_before_second = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); rc_second = handle_jobsubmit(&g_hc, &g_req_ep, &g_jobsub_req); watermark_second = cbrp_poll_append(&g_hc.persist, lsn_before_second, 300); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc_second, "a resubmit for a jobid already fully committed+applied must be " "rejected -EEXIST via jobid_range_exists()"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, (int)watermark_second, "a rejected post-resolution resubmit must NOT append a second " "WAL_JOB_SUBMIT entry for the same jobid"); printf("test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job:" " PASS\n"); } /* bug-1419 layer 3: a jobid claimed by something OTHER than this node's own * handle_jobsubmit call -- simulating the apply pump processing a * WAL_JOB_SUBMIT entry received via replication/replay, never locally * submitted -- must still block a same-jobid submit arriving through * handle_jobsubmit. Before this fix, job_wal_commit_claim did not exist and * apply_committed_entry() never registered an in-flight claim for an * externally-sourced entry, so jobid_range_exists() (c->jobmgr, not yet * applied) and the in-flight registry (only ever populated by * handle_jobsubmit itself) both missed it -- exactly the cross-primary * retry gap a live chaotic soak reproduced (waldump showed two committed * WAL_JOB_SUBMIT entries for one jobid, same term, no fresh idgen collision * -- the same generated jobid resent by the client after a connection drop, * landing on a newly elected primary that never saw the original). */ static void test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit(void) { const uint64_t jobid = 9160; int rc; cbrp_setup(2000); cbrp_arrange_jobsubmit(jobid, 1); /* Simulate: the apply pump is mid-dispatch on a WAL_JOB_SUBMIT entry for * this jobid that this node never itself submitted (e.g. replicated * from a prior primary that committed it before dying). */ TEST_ASSERT_TRUE_MESSAGE(job_wal_commit_claim(jobid), "fixture sanity: claiming a fresh jobid must succeed"); rc = handle_jobsubmit(&g_hc, &g_req_ep, &g_jobsub_req); job_wal_commit_release(jobid); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc, "bug-1419 layer 3: a submit for a jobid claimed by a foreign " "(non-self-originated) commit must be rejected -EEXIST, closing " "the cross-primary retry gap"); printf("test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit:" " PASS\n"); } /* The explicit-FAILED (non-timeout) sibling of the repeated-timeout test * above: repeats the truncation-forced FAILED path * CBRP_JOBSUBMIT_REPEAT_N times in the same fixture, proving the * no-phantom-job guarantee holds for "repeated commit failures" in the * literal sense the task text names (not just "repeated timeouts"), and * that repeated FAILED resolutions release their jobsubmit_async_ctx_t * cleanly each time (a fresh entry always starts from the same * current_lsn==1 baseline the previous iteration's truncation left * behind -- if a context or WAL state leaked, later iterations would * observe a growing lsn_before/current_lsn or a stuck reply queue). */ static void test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs(void) { const uint64_t base_jobid = 9130; int create_rc[CBRP_JOBSUBMIT_REPEAT_N]; int join_rc[CBRP_JOBSUBMIT_REPEAT_N]; uint64_t watermark[CBRP_JOBSUBMIT_REPEAT_N]; int trunc_rc[CBRP_JOBSUBMIT_REPEAT_N]; int n_replies[CBRP_JOBSUBMIT_REPEAT_N]; int32_t last_status[CBRP_JOBSUBMIT_REPEAT_N]; bool job_exists[CBRP_JOBSUBMIT_REPEAT_N]; int i; cbrp_setup(5000); /* generous: truncation, not the timeout sweep, must * be what resolves each iteration */ for (i = 0; i < CBRP_JOBSUBMIT_REPEAT_N; i++) { uint64_t jobid = base_jobid + (uint64_t)i; uint64_t lsn_before; cbrp_reply_obs_t obs = { 0 }; cbrp_arrange_jobsubmit(jobid, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; create_rc[i] = pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg); join_rc[i] = (create_rc[i] == 0) ? pthread_join(g_op_tid, NULL) : -1; watermark[i] = cbrp_poll_append(&g_hc.persist, lsn_before, 1000); trunc_rc[i] = (watermark[i] > 0) ? persist_truncate_wal_above(&g_hc.persist, watermark[i] - 1) : -1; cbrp_poll_reply(&g_req_ep, 2000, &obs); n_replies[i] = obs.n_replies; last_status[i] = obs.last_status; job_exists[i] = cbrp_job_exists(jobid); } cbrp_teardown(); for (i = 0; i < CBRP_JOBSUBMIT_REPEAT_N; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE(0, create_rc[i], "pthread_create must succeed for every repeated iteration"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, join_rc[i], "pthread_join must succeed for every repeated iteration"); TEST_ASSERT_TRUE_MESSAGE(watermark[i] > 0, "each repeated submission must append its own WAL entry"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, trunc_rc[i], "forcing truncation must succeed for every repeated iteration"); TEST_ASSERT_TRUE_MESSAGE(n_replies[i] > 0, "every repeated forced-truncation submission must still " "reply"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, last_status[i], "every repeated forced-truncation submission must reply " "-ENOENT, the explicit non-timeout FAILED vocabulary " "(raft-apply-decoupling task 1.4)"); TEST_ASSERT_FALSE_MESSAGE(job_exists[i], "no phantom job may exist for any repeated forced-truncation " "failure (raft-apply-decoupling task 3.3, Req 2.1/2.3)"); } printf("test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs:" " PASS\n"); } /* Closes task 3.2's flagged gap ("no array_size > 1 mixed-outcome test * exists yet"): element 1 commits (covering ACK) while elements 2 and 3 * are left to time out FAILED. Confirms the failed elements produce zero * phantom jobs and that the committed element's independent resolution * is not blocked or skipped by its failing siblings -- task 3.2's * per-element independence, proven by a real test rather than by code * inspection alone. * * raft-apply-decoupling task 7.3 (Req 11.1) update: this test originally * asserted (task 3.2, before task 6.1/6.2 existed) that the committed * element had NO job yet, only a pending-apply hand-off -- catching that * hand-off's transient window from a second thread is inherently racy * against the ordered apply pump's own consumption of it (proven flaky * empirically across repeated runs of this test: sometimes the window * was caught, sometimes it was already consumed by the time this test * polled). Now that the pump exists, the deterministic, settled effect is * that element 1's mutation applies (job1 becomes visible in c->jobmgr) * independently of its failing siblings, exactly like any other committed * entry -- the per-element independence this test exists to prove now * extends from "resolution" to "apply", per the design's own stated * intent that a command's apply is per-entry and never gated on sibling * array elements (collapsing that behavior is job-array-single-wal- * record's job, not this feature's). The client-visible batch reply is * unaffected: it is still the single failure reply from the timed-out * siblings, because jobsubmit_reply_success_once() is gated on the same * actx->replied flag as jobsubmit_reply_failure_once() (job.c), and the * failure reply always wins that race here. */ static void test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements(void) { const uint64_t base_jobid = 9140; uint64_t lsn_before, watermark, lsn1; int create_rc, join_rc; cbrp_reply_obs_t obs = { 0 }; bool job1_exists, job2_exists, job3_exists; cbrp_setup(300); /* short: elements 2/3 must resolve via the timeout * sweep within a bounded window */ cbrp_arrange_jobsubmit(base_jobid, 3); pthread_mutex_lock(&g_hc.persist.lock); lsn_before = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); lsn1 = lsn_before; g_op_arg.fn = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; create_rc = pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg); join_rc = (create_rc == 0) ? pthread_join(g_op_tid, NULL) : -1; /* All three array elements append back-to-back (task 3.2's * non-blocking, no per-element ACK gate) before handle_jobsubmit * returns -- by the time pthread_join above returns, every element's * WAL entry already exists. */ watermark = cbrp_poll_append(&g_hc.persist, lsn_before + 2, 2000); /* Cover ONLY the first element with a majority ACK; elements 2 and 3 * are left pending -- they must resolve FAILED via the timeout sweep, * independently of element 1's success. */ repl_on_ack(&g_hc.persist.repl, &g_hep_a, lsn1); /* Poll for the batch's single settled reply (the failing siblings' * timeout), then read the deterministic end state of every element's * job existence -- not a transient hand-off window (see doc comment * above). */ cbrp_poll_reply(&g_req_ep, 300 + 1500, &obs); job1_exists = cbrp_job_exists(base_jobid); job2_exists = cbrp_job_exists(base_jobid + 1); job3_exists = cbrp_job_exists(base_jobid + 2); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(0, create_rc, "pthread_create must succeed"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, join_rc, "pthread_join must succeed"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 2, watermark, "all three array elements must append before this poll returns"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, "a partially-failed array batch must produce exactly one failure " "reply for the whole request (jobsubmit_reply_failure_once)"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status, "the batch's single reply must carry the failing elements' " "-ETIMEDOUT outcome"); TEST_ASSERT_TRUE_MESSAGE(job1_exists, "the committed array element's mutation must apply independently " "of its failing siblings, exactly like any other committed entry, " "now that the ordered apply pump exists (raft-apply-decoupling " "task 6.1/6.2) -- the client's batch reply still reports failure " "because a failing sibling wins jobsubmit_async_ctx_t's " "single-reply race, but that reply does not gate this element's " "independent apply"); TEST_ASSERT_FALSE_MESSAGE(job2_exists, "a failed array element must never have a phantom job"); TEST_ASSERT_FALSE_MESSAGE(job3_exists, "a failed array element must never have a phantom job"); printf("test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements:" " PASS\n"); } /* ── handle_jobsubmit success path via the ordered apply pump * (raft-apply-decoupling task 6.2, Req 1.1, 2.1-2.3) ──────────────────── * * Everything above this point pins the COMMITTED-outcome path's gap left by * task 3.2: a committed job submission is handed off to * persist_register_pending_apply() and then has NO further observable * effect (no reply, no job in c->jobmgr) — by design, until task 6 exists. * Task 6.1 built the ordered apply pump that finally calls * apply_committed_entry() live; task 6.2 (this section) wires the pump to * consume that hand-off via persist_take_pending_apply() and fire the * registered continuation (jobsubmit_apply_complete -> the client's success * reply), strictly after the entry's mutation, never before or during it. * * Ordering proof mechanism: apply_committed_entry()'s WAL_JOB_SUBMIT case * (persist.c) fires HOOK_DETAIL_JOB_NEW as an AFTER_EVENT, synchronously, * immediately after inserting the job into c->jobmgr — and, by * persist_apply_pump_run()'s own strict sequencing (task 6.2), strictly * BEFORE the pump reaches its persist_take_pending_apply()/invoke step for * that same lsn. A spy hook on this exact (situation, detail) pair can * therefore observe, synchronously, whether a reply has already been * enqueued on the requester's outq at the moment the mutation completes — * mirroring this file's own pre-existing g_notify_reply_already_enqueued * idiom (see notify() capture above) for the identical class of ordering * assertion. */ static bool g_jobsubmit_order_spy_fired; static bool g_jobsubmit_order_spy_reply_seen_early; static uint64_t g_jobsubmit_order_spy_last_jobid; static int g_jobsubmit_order_spy_fire_count; static int cbrp_jobsubmit_order_spy(int type, int situation, int detail, void *data, va_list ap) { (void)type; (void)data; if (situation != HOOK_AFTER_EVENT || detail != HOOK_DETAIL_JOB_NEW) return 0; /* apply_committed_entry's WAL_JOB_SUBMIT branch fires this hook with * (cluster_t*, xlist_t*) — mirroring test_apply_pump.c's tap_spy_hook, * which decodes the identical detail the same way. */ (void)va_arg(ap, cluster_t *); xlist_t *jlist = va_arg(ap, xlist_t *); g_jobsubmit_order_spy_fired = true; g_jobsubmit_order_spy_fire_count++; if (jlist && jlist->count > 0) { job_t *j = (job_t *)((void **)jlist->data)[0]; g_jobsubmit_order_spy_last_jobid = j->job.jobid; } /* THE ordering proof, latched sticky (OR'd) across every firing: once * true, a reply was observably enqueued no later than one of these * post-mutation notifications — a real ordering violation, regardless * of which array element it happened on. */ if (!DATAQ_EMPTY(&g_req_ep.outq)) g_jobsubmit_order_spy_reply_seen_early = true; return 0; } static void cbrp_jobsubmit_order_spy_reset(void) { g_jobsubmit_order_spy_fired = false; g_jobsubmit_order_spy_reply_seen_early = false; g_jobsubmit_order_spy_last_jobid = 0; g_jobsubmit_order_spy_fire_count = 0; } /* Single-element submission: the success reply must carry the job id, and * must be observable only after the job is visible in c->jobmgr. */ static void test_jobsubmit_success_reply_after_apply_carries_jobid(void) { const uint64_t jobid = 9200; uint64_t lsn_before, watermark; cbrp_reply_obs_t early = { 0 }; cbrp_reply_obs_t final = { 0 }; bool job_exists_before_ack, job_exists_at_reply; cbrp_setup(2000); TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, cbrp_jobsubmit_order_spy, &g_hc)); cbrp_jobsubmit_order_spy_reset(); cbrp_arrange_jobsubmit(jobid, 1); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); /* Before any covering ACK: no reply, no job -- the pre-existing * task-3.2 contract this task's own change must not disturb. */ cbrp_sleep_ms(150); cbrp_scan_replies(&g_req_ep, &early); job_exists_before_ack = cbrp_job_exists(jobid); repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); cbrp_poll_reply(&g_req_ep, 2000, &final); job_exists_at_reply = cbrp_job_exists(jobid); cbrp_teardown(); TEST_ASSERT_TRUE_MESSAGE(watermark > 0, "job submission must append its WAL entry"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, early.n_replies, "no reply may be observable before the entry commits (Req 1.1)"); TEST_ASSERT_FALSE_MESSAGE(job_exists_before_ack, "the job must not exist in the in-memory table before commit"); TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0, "a successful job submission must reply once the ordered apply " "pump consumes the commit hand-off (raft-apply-decoupling task " "6.2 -- closes the standing gap left open since task 3.2)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status, "a committed, applied job submission must reply success"); TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, final.last_data_which, "the success reply must carry the job id (RequestReply_data_jobid)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(jobid, final.last_jobid, "the success reply's job id must match the submitted job"); TEST_ASSERT_TRUE_MESSAGE(job_exists_at_reply, "the job must be visible in the in-memory job table by the time " "the success reply is observed"); TEST_ASSERT_TRUE_MESSAGE(g_jobsubmit_order_spy_fired, "apply_committed_entry's post-mutation HOOK_DETAIL_JOB_NEW " "notification must fire for this job submission"); TEST_ASSERT_EQUAL_UINT64(jobid, g_jobsubmit_order_spy_last_jobid); TEST_ASSERT_FALSE_MESSAGE(g_jobsubmit_order_spy_reply_seen_early, "the client reply must be sent strictly AFTER the job's mutation " "(apply_committed_entry's own post-mutation hook), never before or " "concurrently with it -- task 6.2's own named observable " "(Req 1.1, 2.1-2.3)"); printf("test_jobsubmit_success_reply_after_apply_carries_jobid: PASS\n"); } /* Array submission (3 elements, base jobid 9210): a single covering ACK * commits all three entries at once (mirroring test_apply_pump.c's * burst-order coverage), and the pump applies + hands off each one * independently, in ascending LSN order. Confirms task 3.2's per-element * independent registration is correctly drained by task 6.2's per-element * consumption: every element's own post-mutation notification fires before * the single batch-level reply is ever observable, and every element is * visible in c->jobmgr by the time that reply arrives. */ static void test_jobsubmit_array_success_all_elements_apply_before_single_reply( void) { const uint64_t base_jobid = 9210; const uint64_t array_size = 3; uint64_t lsn_before, watermark; cbrp_reply_obs_t final = { 0 }; bool job_exists[3]; cbrp_setup(2000); TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, cbrp_jobsubmit_order_spy, &g_hc)); cbrp_jobsubmit_order_spy_reset(); cbrp_arrange_jobsubmit(base_jobid, array_size); 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 = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; g_op_arg.rc = -9999; TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); /* All three elements append back-to-back before handle_jobsubmit * returns (task 3.2) -- one covering ACK past the last of them commits * the whole batch in a single jump. */ watermark = cbrp_poll_append(&g_hc.persist, lsn_before + (array_size - 1), 2000); repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); cbrp_poll_reply(&g_req_ep, 2000, &final); job_exists[0] = cbrp_job_exists(base_jobid); job_exists[1] = cbrp_job_exists(base_jobid + 1); job_exists[2] = cbrp_job_exists(base_jobid + 2); cbrp_teardown(); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + (array_size - 1), watermark, "all three array elements must append before this poll returns"); TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0, "the batch must eventually reply once every element applies"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, final.n_replies, "an all-succeeding array batch must produce exactly ONE reply for " "the whole request, not one per element (Req 9.3)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status, "the batch reply must report success"); TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, final.last_data_which, "the batch success reply must carry a job id"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, final.last_jobid, "the batch success reply must echo the BASE job id, matching " "pre-task-3.1 semantics (git history commit 6f1fa97), never a " "per-element id"); TEST_ASSERT_TRUE_MESSAGE(job_exists[0], "array element 0 must be visible in c->jobmgr by reply time"); TEST_ASSERT_TRUE_MESSAGE(job_exists[1], "array element 1 must be visible in c->jobmgr by reply time"); TEST_ASSERT_TRUE_MESSAGE(job_exists[2], "array element 2 must be visible in c->jobmgr by reply time"); TEST_ASSERT_EQUAL_INT_MESSAGE(3, g_jobsubmit_order_spy_fire_count, "each of the three array elements must independently fire its own " "post-mutation notification (task 3.2's per-element registration, " "drained one at a time by task 6.2)"); TEST_ASSERT_FALSE_MESSAGE(g_jobsubmit_order_spy_reply_seen_early, "no array element's post-mutation notification may ever observe " "the batch's single reply already enqueued -- the reply can only " "follow the LAST element's own mutation, and every earlier " "element's mutation must equally precede it"); printf("test_jobsubmit_array_success_all_elements_apply_before_single_" "reply: PASS\n"); } /* REMOVED (raft-apply-decoupling task 3.2): the array-partial-construction- * failure remediation test (task 3.1 round-2 fix, bug-1408) that lived here * drove handle_jobsubmit's OLD sequential, per-element BLOCKING append loop * — it ACKed array element 0 and only then observed element 1's WAL entry * appear, exploiting that strict ordering to inject a deterministic jobid * collision between the two elements' commits. Task 3.2 appends every array * element back-to-back, non-blocking, with no per-element ACK gate between * them, so that ordering assumption no longer holds — the test's whole * premise is gone, not just its expected outcome. * * More importantly, the CODE this test exercised (handle_jobsubmit's own * post-commit create_job()-loop + jlist-build + xce_run_hooks(AFTER_EVENT) * block) no longer exists in job.c at all (see the "REMOVED" comment above * handle_jobsubmit's async registration block) — it is superseded, not * relocated, by apply_committed_entry()'s existing WAL_JOB_SUBMIT case * (persist.c, task 2.2), which already does restore_job() + fires * HOOK_DETAIL_JOB_NEW per entry, independently per array element, and is * therefore structurally immune to the bug-1408 class already. Coverage for * that invariant now belongs to persist.c's own apply-seam test suite * (test_apply_seam.c), not here. See this task's status report for the * full reasoning; git history (commit 6f1fa97) preserves the removed test * verbatim. */ /* ════════════════════════════════════════════════════════════════════════ * 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/pad_out (both optional pointers) receive the LAST matching * entry's state/_pad. raft-apply-decoupling task 6.1 round 2: WAL_JOB_FAIL and a * postdone WAL_JOB_UPDATE now carry their real information in `_pad` (see * wal_types.h's doc comment on wal_jobstate_t._pad) rather than a * precomputed `state` — pad_out lets a test observe that. */ static int cbrp_count_jobstate_entries_ex(persist_t *p, uint32_t wal_type, uint64_t jobid, uint32_t *state_out, uint32_t *pad_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; if (pad_out != NULL) *pad_out = js._pad; } } off += sizeof(hdr) + hdr.payload_len; } close(fd); return count; } /* Thin wrapper preserving every pre-existing caller's signature/contract * unchanged (pad = 0, "legacy absolute state" — see wal_types.h). */ static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, uint64_t jobid, uint32_t *state_out) { return cbrp_count_jobstate_entries_ex(p, wal_type, jobid, state_out, NULL); } /* 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 the fact-transition receipt handlers * (guard extraction: task 2.1, Req 5.1/5.2/5.3 — landed and green) * ════════════════════════════════════════════════════════════════════════ * * Pins the transition guards task 2.1 extracted into apply_job_start_ * transition / apply_job_finish_transition / apply_job_fail_transition / * apply_job_update_transition, and peeked at the top of the receipt-side * wrappers (record_job_start_fact / record_job_finish_fact — renamed from * mark_job_start / mark_job_finish by task 4.1 — and mark_job_fail / * mark_job_update, still used by job_ctrl's separate command path). Design * §Idempotency Guards: * - duplicate (already-reflected transition) → INFO log, NO-OP: 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, no * WAL entry appended (5.2) * - replay: re-applying the same transition appends no second WAL entry * (5.3) * - apply_job_fail_transition's existing fail-after-finish PERR-merge is * forward-compatible and must be PRESERVED by the guard (pin) * * Drivers: 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, so the * "success-no-op rc" observable (every caller treats a guarded-out * transition as success) is asserted directly on the wrapper rc. * * Observability: 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) — current_lsn is the "did a WAL entry * append" oracle (a guarded-out no-op never calls persist_wal_async → * current_lsn frozen). cbrp_job_live_state remains meaningful in these * tests only because cbrp_drive_to_done (task 4.1) now reaches its * terminal precondition by applying the guard cores directly, as a * test-only stand-in for the not-yet-built live apply pump — since task 4.1, * the receipt handlers themselves never mutate live state on ANY path * (guarded-out or accepted), so a passing guard test no longer implies * "the guard blocked a live mutation," only "the guard blocked a WAL * append." test_guard_forward_update_on_terminal_preserved and * test_guard_fail_after_finish_preserves_perr_merge were updated by task 4.1 * to assert the forward-merge/PERR-merge outcome against the WAL-recorded * fact instead of live state, plus an explicit live-state-unchanged pin. * * NOTE on the UPDATE-clears-terminal arm (5.2): mark_job_update/ * record_job_update_fact have 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). record_job_start_fact 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). * * raft-apply-decoupling task 4.1: on_job_start/on_job_finish are now * durable-append-only — they no longer mutate the job's LIVE state (that * moves to apply_committed_entry(), driven by the not-yet-built live apply * pump, task 6). This helper's callers still need the job's LIVE state to * reach a genuine terminal DONE so they can exercise backward-reject / * duplicate-noop / forward-merge guard behavior against a real terminal * job, so — after driving the two facts through the real receipt path to * verify the durable append — this also applies the identical two * transitions directly through the task-2.1 guard cores * (apply_job_start_transition/apply_job_finish_transition), a test-only * stand-in for what the future apply pump will do live once an entry * commits. This is not a production code path. */ 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; job_t* job = NULL; 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)); lock(g_hc.jobmgr, RTTI_LOCK_READ); key_find(g_hc.jobmgr, (void *)jobid, (void **)&job); if (job != NULL) { reference(job); } unlock(g_hc.jobmgr); TEST_ASSERT_NOT_NULL_MESSAGE(job, "job must exist in jobmgr after cbrp_make_job + on_job_start"); lock(job, RTTI_LOCK_WRITE); apply_job_start_transition(job, &start); apply_job_finish_transition(job, &fin); unlock(job); dereference(job); 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. record_job_start_fact (renamed from * mark_job_start by task 4.1) 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 * record_job_postdone_fact caller: on a DONE job it OR's JOB_STAT_PDONE * (terminal preserved, strictly forward). This must still durably append * (an entry appends, marked for the correct apply-time merge) — the guard * rejects backward, never a forward merge. Preservation pin: green before * and after task 6.2. * * raft-apply-decoupling task 4.1: on_job_postdone is now durable-append-only. * task 6.1 round 2 (merge-staleness fix, Req 11.3): the WAL-recorded fact no * longer carries a precomputed `state` at all — postdone's merge depends on * whatever state holds at APPLY time (which may differ from receipt time * once apply is deferred), so it is recomputed then by * apply_job_postdone_transition(), never baked into the WAL entry. The * oracle for "was this fact durably recorded correctly" is therefore the * WAL entry's `_pad` marker (cbrp_count_jobstate_entries_ex); the oracle for * "does the forward merge itself compute correctly" is a direct call to * apply_job_postdone_transition() below, mirroring how * test_guard_fail_after_finish_preserves_perr_merge pins apply_job_fail_ * transition(). The live-state-unchanged assertion is task 4.1's own pin * (Req 5.1): live state must still show only DONE (from cbrp_drive_to_done's * direct apply), never PDONE, since nothing applies this fact live yet. */ static void test_guard_forward_update_on_terminal_preserved(void) { uint64_t lsn_after_done, lsn_after_pdone; uint32_t pad_disk = 0; int st_live; job_t scratch; 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); /* Assert: the update appended, marked for a correct apply-time forward * merge (terminal DONE preserved, PDONE merged in — not 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"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_UPDATE, 73, NULL, &pad_disk), "exactly one UPDATE fact must be durably recorded (Req 5.1)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(WAL_JOB_UPDATE_POSTDONE_MARK, pad_disk, "the recorded UPDATE fact must be marked for apply-time postdone " "recomputation (task 6.1 round 2) — not carry a state precomputed " "at receipt time, which would go stale under deferred apply"); /* Assert: the actual forward-merge guarantee this test exists to pin — * DONE preserved, PDONE merged in, not rejected — verified directly * against the guard core that performs it at apply time, on an * independent scratch job seeded at the same terminal state job 73 was * driven to above. */ memset(&scratch, 0, sizeof(scratch)); scratch.job.jobid = 73; scratch.job.state = JOB_STAT_DONE; TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, apply_job_postdone_transition(&scratch)); TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_DONE) != 0, "the recorded UPDATE fact must preserve the terminal bit"); TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_PDONE) != 0, "the recorded UPDATE fact must merge JOB_STAT_PDONE (not be " "rejected)"); /* Assert (Req 5.1, task 4.1's own observable): live state is UNCHANGED * by the receipt handler — still just DONE, no PDONE — since nothing * applies this fact live until the apply pump (task 6) exists. */ st_live = cbrp_job_live_state(73); TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_DONE, st_live, "on_job_postdone must leave live job state unchanged (Req 5.1) — " "the PDONE merge above is recorded in the WAL only, not applied " "live, until the apply pump exists"); cbrp_teardown(); printf("test_guard_forward_update_on_terminal_preserved: PASS\n"); } /* 5.2 pin — apply_job_fail_transition's existing fail-after-finish PERR-merge * must be PRESERVED: 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. * * raft-apply-decoupling task 4.1: on_job_fail (via the new * record_job_fail_fact) is now durable-append-only. task 6.1 round 2 * (merge-staleness fix, Req 11.3): the WAL-recorded fact no longer carries a * precomputed target state at all — the PERR merge genuinely depends on * whatever state holds at APPLY time (a receipt-time snapshot goes stale * the moment another fact for the same job, e.g. a FINISH, is still pending * apply — the exact bug this round fixed), so it is recomputed then by the * REAL apply_job_fail_transition(), never baked into the WAL entry at * receipt time. The oracle for "was this fact durably recorded correctly" * is therefore the WAL entry's `_pad`-carried stage * (cbrp_count_jobstate_entries_ex); the oracle for "does the PERR merge * itself compute correctly" is a direct call to apply_job_fail_transition() * below (its own logic is unchanged by this round — only WHEN it runs * changed). The live-state-unchanged assertion is task 4.1's own pin * (Req 5.1). */ static void test_guard_fail_after_finish_preserves_perr_merge(void) { job_failinfo_t failinfo; uint64_t lsn_after_done, lsn_after_fail; uint32_t pad_disk = 0; int st_live; job_t scratch; 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); /* Assert: the FAIL fact appended, marked for a correct apply-time * PERR-merge — DONE preserved, PERR merged in, not reset to bare EXIT. */ 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)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_FAIL, 74, NULL, &pad_disk), "exactly one FAIL fact must be durably recorded (Req 5.1)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE( WAL_JOB_FAIL_STAGE_FLAG | (uint32_t)JOB_EXEC_STAGE_POST_EXEC, pad_disk, "the recorded FAIL fact must carry its stage marked for apply-time " "recomputation (task 6.1 round 2) — not a state precomputed at " "receipt time, which would go stale under deferred apply"); /* Assert: the actual PERR-merge guarantee this test exists to pin — * DONE preserved, PERR merged in, not reset to bare EXIT — verified * directly against the guard core that performs it at apply time, on an * independent scratch job seeded at the same terminal state job 74 was * driven to above. */ memset(&scratch, 0, sizeof(scratch)); scratch.job.jobid = 74; scratch.job.state = JOB_STAT_DONE; TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, apply_job_fail_transition(&scratch, &failinfo)); TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_DONE) != 0, "the recorded FAIL fact must keep the DONE terminal bit " "(PERR-merge behavior)"); TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_PERR) != 0, "the recorded FAIL fact must OR JOB_STAT_PERR onto the existing " "terminal state (PERR-merge behavior)"); /* Assert (Req 5.1, task 4.1's own observable): live state is UNCHANGED * by the receipt handler — still just DONE, no PERR — since nothing * applies this fact live until the apply pump (task 6) exists. */ st_live = cbrp_job_live_state(74); TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_DONE, st_live, "on_job_fail must leave live job state unchanged (Req 5.1) — the " "PERR merge above is recorded in the WAL only, not applied live, " "until the apply pump exists"); cbrp_teardown(); printf("test_guard_fail_after_finish_preserves_perr_merge: PASS\n"); } /* ════════════════════════════════════════════════════════════════════════ * raft-apply-decoupling task 4.1 (Req 5.1) — receipt handlers durably * append only, no live mutation, no AFTER_EVENT firing * ════════════════════════════════════════════════════════════════════════ * * Spy hook counting BEFORE_EVENT vs AFTER_EVENT firings for * XCE_CONTROLLER_HOOK_JOB — registered alongside persist_start's production * hooks (persist_job_hook_fun etc., a fan-out, so both subscribers coexist). * Used to pin: the pre-mutation validation hook (BEFORE_EVENT) still fires * exactly once per accepted fact, unchanged; AFTER_EVENT never fires at * receipt time any more (it now only fires from apply_committed_entry(), * which nothing drives yet — task 6). */ static int g_cbrp_before_fires; static int g_cbrp_after_fires; static int cbrp_job_hook_spy(int type, int situation, int detail, void *data, va_list ap) { (void)type; (void)detail; (void)data; (void)ap; if (situation == HOOK_BEFORE_EVENT) g_cbrp_before_fires++; else if (situation == HOOK_AFTER_EVENT) g_cbrp_after_fires++; return 0; } /* The task's own RED/GREEN observable: after each of on_job_start/ * on_job_finish/on_job_fail/on_job_postdone returns, the job's LIVE * in-memory state is unchanged — even though every one of the four facts * durably appends with the correct target state, and the pre-mutation * BEFORE_EVENT hook still fires exactly once per accepted fact. Against the * pre-task-4.1 code (mutate-then-notify at receipt) this test's live-state- * unchanged assertions fail on the very first fact (START moves live state * PEND -> RUN immediately) — confirmed by running this test against a * revert of job.c/persist.c/persist.h before applying this task's fix. */ static void test_receipt_facts_durably_append_without_live_mutation(void) { static unsigned int rf_pids[1] = { 6001 }; job_pidinfo_t start; xce_jobfinish_t fin; job_failinfo_t failinfo; uint32_t st_disk = 0; int live_before, live_after; cbrp_setup(3000); cbrp_make_job(90, &g_job_submit); memset(&g_job_submit, 0, sizeof(g_job_submit)); g_cbrp_before_fires = 0; g_cbrp_after_fires = 0; TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, cbrp_job_hook_spy, NULL)); /* START */ live_before = cbrp_job_live_state(90); memset(&start, 0, sizeof(start)); start.jobid = 90; start.npids = 1; start.pids = &rf_pids[0]; TEST_ASSERT_EQUAL_INT(0, on_job_start(&g_hc, &start)); live_after = cbrp_job_live_state(90); TEST_ASSERT_EQUAL_INT_MESSAGE(live_before, live_after, "on_job_start must leave live job state unchanged (Req 5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 90, &st_disk), "on_job_start must durably append exactly one START fact (Req 5.1)"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, st_disk); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_cbrp_before_fires, "the pre-mutation validation hook (BEFORE_EVENT) must still fire " "unchanged at receipt time"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must NOT fire at receipt time any more (Req 5.1, 5.3)"); /* FINISH */ live_before = cbrp_job_live_state(90); memset(&fin, 0, sizeof(fin)); fin.jobid = 90; fin.exit_code = 0; TEST_ASSERT_EQUAL_INT(0, on_job_finish(&g_hc, &fin)); live_after = cbrp_job_live_state(90); TEST_ASSERT_EQUAL_INT_MESSAGE(live_before, live_after, "on_job_finish must leave live job state unchanged (Req 5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 90, &st_disk), "on_job_finish must durably append exactly one FINISH fact (Req " "5.1)"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_disk); TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_cbrp_before_fires, "BEFORE_EVENT must fire again for the FINISH fact"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must still never fire at receipt time"); /* FAIL */ live_before = cbrp_job_live_state(90); memset(&failinfo, 0, sizeof(failinfo)); failinfo.jobid = 90; failinfo.stage = JOB_EXEC_STAGE_POST_EXEC; failinfo.code = 1; failinfo.detail = (char *)"synthetic fail for no-mutation pin"; TEST_ASSERT_EQUAL_INT(0, on_job_fail(&g_hc, &failinfo)); live_after = cbrp_job_live_state(90); TEST_ASSERT_EQUAL_INT_MESSAGE(live_before, live_after, "on_job_fail must leave live job state unchanged (Req 5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FAIL, 90, &st_disk), "on_job_fail must durably append exactly one FAIL fact (Req 5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(3, g_cbrp_before_fires, "BEFORE_EVENT must fire again for the FAIL fact — mark_job_fail has " "no guard, so it always fires"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must still never fire at receipt time"); /* UPDATE (via on_job_postdone) */ live_before = cbrp_job_live_state(90); TEST_ASSERT_EQUAL_INT(0, on_job_postdone(&g_hc, &g_req_ep, 90)); live_after = cbrp_job_live_state(90); TEST_ASSERT_EQUAL_INT_MESSAGE(live_before, live_after, "on_job_postdone must leave live job state unchanged (Req 5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_UPDATE, 90, &st_disk), "on_job_postdone must durably append exactly one UPDATE fact (Req " "5.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(4, g_cbrp_before_fires, "BEFORE_EVENT must fire again for the UPDATE fact"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must NEVER fire at receipt time for any of the four " "fact types (Req 5.1, 5.3) — it now only fires from " "apply_committed_entry(), which nothing drives live until task 6"); cbrp_teardown(); printf("test_receipt_facts_durably_append_without_live_mutation: " "PASS\n"); } /* ════════════════════════════════════════════════════════════════════════ * Task 4.2 (Req 6.1, 6.2): durable retention of facts submitted while * majority is unreachable, held for the SUSTAINED DURATION of the outage — * not just a single immediate check. * * This closes a gap neither existing test covers on its own: * - test_receipt_facts_durably_append_without_live_mutation (task 4.1, * above) proves live-state-unchanged for all four fact types, but * under ordinary (majority-reachable) conditions, and only checks * state once, immediately after each receipt call returns. * - test_fact_retention_across_quorum_loss (pre-existing, from * ha-commit-barrier-rollout task 4.3, further above) drives a genuine * quorum-loss window and proves WAL-level durability/no-duplication, * but never inspects live job state at all — it predates task 4.1's * receipt/apply split, when receipt-time mutation was still the norm * for the behavior it was pinning. * * This test combines both: a genuine simulated quorum-loss window (same * repl_remove_standby mechanism as test_fact_retention_across_quorum_loss) * driving all four real fact-receipt handlers from task 4.1, with the * job's live in-memory state polled at MULTIPLE points spread across the * outage — not merely before/after a single call — to prove the guarantee * holds for the outage's entire duration, sealed by one final check after * the last poll. * * Per persist.c's wal_hook_append_classified (task 4.1's persist_submit_ * job_fact routes through it): COMMIT_MODE_RETAIN "never honors the * timeout" and commit_submit() returns 0 immediately for RETAIN entries — * facts have NO commit-timeout/failure concept at the receipt layer, * unlike commands (task 3.2's commit_submit_async). A fact's entry simply * sits durably appended, uncommitted, for as long as majority stays * unreachable; nothing ever resolves it as FAILED the way a command's * registered continuation would. This test's "outage" is therefore * genuinely open-ended by construction (committed_lsn structurally cannot * advance past it while both standbys are disconnected), not a race * against a timeout that might independently fire. */ static void test_fact_state_unchanged_throughout_quorum_loss_outage(void) { static unsigned int qo_pids[1] = { 8501 }; job_pidinfo_t start; xce_jobfinish_t fin; job_failinfo_t failinfo; uint64_t lsn_before, lsn_after; uint64_t committed_at_poll; int initial_state, poll_state, final_state; uint32_t st_start = 0, st_fin = 0, pad_fail = 0, pad_upd = 0; int cnt_start, cnt_fin, cnt_fail, cnt_upd; int i; enum { OUTAGE_POLL_COUNT = 5, OUTAGE_POLL_INTERVAL_MS = 20 }; cbrp_setup(5000); memset(&g_job_submit, 0, sizeof(g_job_submit)); cbrp_make_job(85, &g_job_submit); initial_state = cbrp_job_live_state(85); TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_STAT_PEND, initial_state, "precondition: a freshly created job starts PEND"); g_cbrp_before_fires = 0; g_cbrp_after_fires = 0; TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, cbrp_job_hook_spy, NULL)); /* Act 1 — QUORUM LOST: disconnect both standby registry slots * (repl_remove_standby mirrors the RDHUP-disconnect registry effect, * same mechanism test_fact_retention_across_quorum_loss uses). n_peers * stays 3 in the engine: self alone is 1 of 3, never a majority, so * committed_lsn structurally cannot advance past this point for as * long as both standbys remain disconnected. */ 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); /* Act 2 — submit all four fact types for the SAME job while majority * is unreachable, through the real receipt handlers from task 4.1 * (identical call sequence to test_receipt_facts_durably_append_ * without_live_mutation above, now under a genuine quorum-loss * window instead of ordinary conditions). */ memset(&start, 0, sizeof(start)); start.jobid = 85; start.npids = 1; start.pids = &qo_pids[0]; CBRP_DRIVE_FACT(on_job_start(&g_hc, &start), "on_job_start(85)"); memset(&fin, 0, sizeof(fin)); fin.jobid = 85; fin.exit_code = 0; /* -> JOB_STAT_DONE */ CBRP_DRIVE_FACT(on_job_finish(&g_hc, &fin), "on_job_finish(85)"); memset(&failinfo, 0, sizeof(failinfo)); failinfo.jobid = 85; failinfo.stage = JOB_EXEC_STAGE_POST_EXEC; failinfo.code = 1; failinfo.detail = (char *)"synthetic fail during simulated quorum loss"; CBRP_DRIVE_FACT(on_job_fail(&g_hc, &failinfo), "on_job_fail(85)"); CBRP_DRIVE_FACT(on_job_postdone(&g_hc, &g_req_ep, 85), "on_job_postdone(85)"); pthread_mutex_lock(&g_hc.persist.lock); lsn_after = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 4, lsn_after, "each of the four facts must append exactly one WAL entry while " "majority is unreachable (Req 6.1)"); TEST_ASSERT_TRUE_MESSAGE( atomic_load(&g_hc.persist.committed_lsn) < lsn_before, "no fact entry may commit while majority is unreachable (Req 6.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must never fire for an uncommitted fact — no " "reaction (slot free, dispatch, notify) may escape the outage " "(Req 6.1, 6.2)"); /* Act 3 — the core observable this task adds: poll live job state AND * the commit watermark repeatedly across a simulated outage window, * not just once immediately after receipt. Every single poll must see * the job unchanged and the outage still genuinely in effect — proving * the no-mutation guarantee holds for the outage's sustained duration, * not merely at one instant. */ for (i = 0; i < OUTAGE_POLL_COUNT; i++) { cbrp_sleep_ms(OUTAGE_POLL_INTERVAL_MS); poll_state = cbrp_job_live_state(85); committed_at_poll = atomic_load(&g_hc.persist.committed_lsn); TEST_ASSERT_EQUAL_INT_MESSAGE(initial_state, poll_state, "job state must remain unchanged at every point during the " "quorum-loss outage (Req 6.1, 6.2) — there is no local-only " "apply path that could mutate it while majority is " "unreachable"); TEST_ASSERT_TRUE_MESSAGE(committed_at_poll < lsn_before, "the outage must still be genuinely in effect at this poll — " "committed_lsn must not have advanced while both standbys " "remain disconnected"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must still not have fired at this poll point — " "no reaction may fire mid-outage (Req 6.1, 6.2)"); } /* Assert — nothing dropped, nothing duplicated: all four facts are on * disk exactly once with the correct post-transition state, confirmed * AFTER the full polled outage window (not merely right after Act 2). */ cnt_start = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_START, 85, &st_start); cnt_fin = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 85, &st_fin); cnt_fail = cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_FAIL, 85, NULL, &pad_fail); cnt_upd = cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_UPDATE, 85, NULL, &pad_upd); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cnt_start, "job 85 START fact must remain durably retained exactly once " "throughout the whole outage window (Req 6.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cnt_fin, "job 85 FINISH fact must remain durably retained exactly once " "throughout the whole outage window (Req 6.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cnt_fail, "job 85 FAIL fact must remain durably retained exactly once " "throughout the whole outage window (Req 6.1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cnt_upd, "job 85 UPDATE fact must remain durably retained exactly once " "throughout the whole outage window (Req 6.1)"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, st_start); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_fin); /* task 6.1 round 2 (Req 11.3): FAIL's and postdone's UPDATE's merge * results depend on current state at the (never-reached, in this * outage-only test) apply step — they are no longer precomputed at * receipt time at all (the merge-staleness fix), so only their durable * retention count and their apply-time-recomputation marker are this * test's concern. */ TEST_ASSERT_EQUAL_UINT32_MESSAGE( WAL_JOB_FAIL_STAGE_FLAG | (uint32_t)JOB_EXEC_STAGE_POST_EXEC, pad_fail, "the recorded FAIL fact must carry its stage marked for apply-time " "recomputation"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(WAL_JOB_UPDATE_POSTDONE_MARK, pad_upd, "the recorded UPDATE fact must be marked for apply-time postdone " "recomputation"); /* Final check: after the ENTIRE outage window, job state is STILL * unchanged and no reaction has fired — sealing the "for the duration * of the outage" observable end-to-end. */ final_state = cbrp_job_live_state(85); TEST_ASSERT_EQUAL_INT_MESSAGE(initial_state, final_state, "job state must still be unchanged at the end of the outage " "window (Req 6.1, 6.2)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_cbrp_after_fires, "AFTER_EVENT must never have fired across the entire outage " "window (Req 6.1, 6.2)"); cbrp_teardown(); printf("test_fact_state_unchanged_throughout_quorum_loss_outage: " "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); /* handle_jobsubmit (raft-apply-decoupling task 3.1, Req 2.1/2.2/2.3) */ RUN_TEST(test_jobsubmit_withheld_acks_no_phantom_job); RUN_TEST(test_jobsubmit_returns_before_commit_and_registers_pending_apply); RUN_TEST(test_jobsubmit_returns_before_commit_on_failure_path_too); /* raft-apply-decoupling task 3.3 (Req 2.1, 2.3): repeated/explicit- * FAILED failure-path coverage beyond the single-timeout tests above */ RUN_TEST(test_jobsubmit_truncation_forces_failed_no_phantom_job); RUN_TEST(test_jobsubmit_repeated_timeout_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry); RUN_TEST(test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job); RUN_TEST(test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit); RUN_TEST(test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements); /* handle_jobsubmit success path via the ordered apply pump * (raft-apply-decoupling task 6.2, Req 1.1, 2.1-2.3) */ RUN_TEST(test_jobsubmit_success_reply_after_apply_carries_jobid); RUN_TEST(test_jobsubmit_array_success_all_elements_apply_before_single_reply); /* Receipt-ACK verdict pins (ha-commit-barrier-rollout 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); /* Fact-transition idempotency / ordering guards (extracted by task 2.1, * already landed and green — the stale "RED-phase for task 6.2" framing * this comment block used to carry predates that landing) */ 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); /* Fact receipt handlers durably append only (task 4.1, Req 5.1) */ RUN_TEST(test_receipt_facts_durably_append_without_live_mutation); /* Durable retention held for the outage's sustained duration under * genuine quorum loss (task 4.2, Req 6.1, 6.2) */ RUN_TEST(test_fact_state_unchanged_throughout_quorum_loss_outage); UNITY_END(); }