/* test_apply_pump_integration.c — raft-apply-decoupling tasks 6.3 and 6.4 * * Integration coverage for command AND fact apply/notify timing through the * SAME full pump (task 6.1's ordered walk + task 6.2's continuation * hand-off), superseding the earlier synchronous-path check from task 3.1 * (which predated the pump and could only pin "no phantom job", not * "notified only after commit"). * * Tests 1-2 are task 6.3's (command+fact interleave through one commit; fact * reactions withheld/restored across a genuine outage). Tests 3-4 are task * 6.4's: a genuine quorum-loss-then-restore catch-up over a MIX of commands * AND facts (Test 2 above is fact-only, by design — see task 6.1's * Implementation Notes calling that out as non-duplicative of 6.4's * charter), plus a real-multi-vcpu-pool burst variant that stress-tests task * 6.2's bug-1415 self-heal (persist_register_pending_apply) under several * concurrent command continuations racing the walk at once, not just the * one forced interleaving test_apply_pump.c's own regression test covers. * See Test 3 and Test 4's own header comments below for the full scope * split and why NULL-pool determinism vs a real pool each earn their own * test rather than one doing both jobs. * * Two things this file does NOT re-prove, because they are already covered * elsewhere in this exact spec's own suite: * - Pure command reply-after-visibility ordering for job submit — * test_commit_barrier_reply.c's test_jobsubmit_success_reply_after_ * apply_carries_jobid (task 6.2) already drives this through the real * pump end to end. * - Pure fact no-live-mutation-before-commit for a single fact type — * test_apply_pump.c (task 6.1) and test_commit_barrier_reply.c's * test_fact_state_unchanged_throughout_quorum_loss_outage (task 4.2) * already cover this for state, across a genuine outage window. * * What is genuinely new ground, and this file's actual job: * 1. A command (job submit) and a fact (job-state report) interleaved * through the ONE SAME apply-pump pass, proving commit-order * application and independent gating hold when they are mixed, not * just individually (task boundary: "this needs BOTH command and fact * coverage through the SAME full pump"). * 2. The actual REACTION hooks a real sched.c/apiserver.c subscriber * would receive for a fact — not merely "did live state change" — * proving worker slot release / scheduler dispatch triggering / * client-apiserver notification (all downstream of the identical * AFTER_EVENT firing, see below) never fire during a delayed-commit * window, then DO fire once the entry commits and applies. * 3. That every such firing carries proof, inline, that the entry was * ALREADY committed at fire time (committed_lsn_at_fire >= the * entry's own lsn) — the structural reason a consumer of the * notification never needs to separately re-check a commit watermark * (Req 4.3): the notification's mere occurrence already implies it. * * Why ONE reaction spy stands in for THREE named reactions (worker slot * release, scheduler dispatch triggering, client/apiserver notification): * apply_committed_entry() (persist.c) fires every job-state fact as the * SAME generic HOOK_DETAIL_JOB_UPDATE — a real, documented deviation from a * literal per-type hook fan-out (persist.h's apply_committed_entry doc * comment, task 2.2), because the on-disk fact payload (wal_jobstate_t * {jobid,state}) cannot supply what the type-specific hooks * (HOOK_DETAIL_JOB_START/_EXIT/_FAIL/_RUSAGE_UPDATE) require. Concretely, * this means sched.c's sched_job_update_fun / apiserver.c's * apiserver_job_update_fun (both HOOK_DETAIL_JOB_UPDATE subscribers) are * the ONLY reachable reaction path for a live-applied fact — sched_job_ * start_fun/_exit_fun/_fail_fun and apiserver's equivalents are, by this * same documented deviation, structurally unreachable for a fact applied * through the live pump. A spy that listened for the type-specific details * instead would silently test something that can never happen in * production and would never catch a real timing violation. This file * deliberately mirrors that reality (HOOK_DETAIL_JOB_NEW for submit, * HOOK_DETAIL_JOB_UPDATE for every fact type) rather than an idealized * per-type routing table the wire format cannot support. The explicit * per-task note (persist.h, tasks.md's task 2.2 Implementation Notes) * states no task through 6.3 closes this hook-granularity gap — this file * verifies TIMING correctness of whatever hook actually fires, which is * this task's boundary, not the routing granularity itself. * * "Did we miss a second call site" audit (task 6.1 already found one * undocumented double-append/double-notify hazard in queue/worker hooks — * this file's own pass, via `grep -n xce_run_hooks job.c persist.c` * cross-referenced against HOOK_DETAIL_JOB_NEW/_UPDATE): the only AFTER_ * EVENT firers for these two details are (a) apply_committed_entry() * itself (persist.c, guarded against double-appending via the thread-local * t_apply_seam_notifying flag) and (b) job_ctrl's mark_job_update * (job.c:1268, SIGCONT/SIGSTOP live command path, legitimately outside the * apply seam and already covered by test_apply_pump.c's guard-precision * test). No orphaned third call site was found for either detail. * * Fixture mirrors test_apply_pump.c's tap_setup (real persist_init + * persist_start over a private tmpdir, a real 3-member commit_engine_t with * two stub standbys, ACKs delivered explicitly via repl_on_ack) plus * test_commit_barrier_reply.c's reply-observation harness (a zeroed, * no-iomgr requester endpoint; xce_reply's packets stay queued on ep->outq * and are parsed with the raw capnp reader). tpi_setup() takes a * pool_n_vcpus parameter (task 6.4): 0 (Tests 1-3) leaves ce->pool == NULL * so the apply pump's inline fallback applies, exactly as tasks 6.3's own * tests always ran; > 0 (Test 4 only) wires a REAL vtaskmgr_t, mirroring * test_apply_pump.c's task 6.2 tpr_setup. * * All four tests below follow this series' established disciplines: capture * every assertion-relevant observable into locals, then teardown * unconditionally, THEN assert (a failed TEST_ASSERT's longjmp must never * skip teardown); poll for eventual apply-pump consistency rather than * assume synchronous completion (persist_start's fsync_worker thread * independently kicks the pump on every durable_lsn advance, racing this * test's own repl_on_ack-driven kick — see tap_wait_applied's lineage in * test_apply_pump.c for the full mechanism); and poll ACROSS a real time * window to prove an absence, not just check once. */ #include #include #include #include #include #include #include #include #include #include #include #include #include "cluster.h" #include "commit.h" #include "controller.capnp.h" #include "hook.h" #include "jobj.h" #include "netio.h" #include "persist.h" #include "repl.h" #include "request.h" #include "vtaskmgr.h" #include "wal_types.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xstring.h" /* jobj.c's jobsub table and the process-global xstring table are shared * across every test file in the controller_unit_test binary — guard * idempotently, mirroring every other raft-apply-decoupling test file. */ static int g_tpi_jobobj_ready; static cluster_t g_tpi_c; static commit_engine_t g_tpi_hce; static int g_tpi_hep_a, g_tpi_hep_b; /* dummy standby endpoints */ static char g_tpi_tmpdir[256]; /* task 6.4: a REAL multi-vcpu pool, wired only when a test explicitly asks * for one via tpi_setup's pool_n_vcpus parameter -- every fixture through * task 6.3 keeps ce->pool == NULL (see test_apply_pump.c's task 6.2 bug-1415 * regression test header for why that trivially serializes commit_resolve_ * pending()'s two independent dispatches and so cannot reproduce that race * class). g_tpi_pool_active tracks whether tpi_teardown owes a * vtaskmgr_fini this run. */ static vtaskmgr_t g_tpi_pool; static int g_tpi_pool_active; /* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast * from dereferencing the dummy standby endpoints above. */ extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, int n_entries); static int tpi_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 */ } /* ── Requester endpoint + reply observation (mirrors test_commit_barrier_ * reply.c's cbrp harness — see that file's header for the full rationale: * xce_reply()'s packets stay queued on a zeroed, no-iomgr endpoint's outq, * parsed with the raw capnp reader). ── */ static endpoint_t g_tpi_req_ep; static ep_private_t g_tpi_req_priv; /* task 6.4: bounded cap on how many individual replies tpi_reply_obs_t can * distinguish by jobid/status -- separate from TPI_SPY_CAP below since the * two arrays serve different structs; sized generously above the largest * batch this file's tests submit. */ #define TPI_REPLY_CAP 8 typedef struct { int n_replies; int32_t last_status; int32_t last_data_which; uint64_t last_jobid; /* task 6.4: per-reply jobid/status, in receipt order, so a mixed-batch * test can confirm EVERY submitted command got its own exactly-one * reply rather than only inspecting the last one observed. Unused by * tasks 6.3's tests, which only ever check n_replies/last_*. */ uint64_t jobids[TPI_REPLY_CAP]; int32_t statuses[TPI_REPLY_CAP]; } tpi_reply_obs_t; static void tpi_scan_replies(endpoint_t *ep, tpi_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); if (obs->n_replies < TPI_REPLY_CAP) { obs->jobids[obs->n_replies] = (rr.data_which == RequestReply_data_jobid) ? rr.data.jobid : 0; obs->statuses[obs->n_replies] = rr.status; } 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; } capn_free(&c); free_packet(pkt); free(pkt); } } /* task 6.4: true iff `jobid` appears in obs's captured replies exactly once, * with a success (0) status -- catches both a hang (0 occurrences, since a * missing reply never appears) and a bug-1415-class double-fire (>1 * occurrence) with one call, for the mixed-batch tests below. */ static bool tpi_reply_has_exactly_one_success(const tpi_reply_obs_t *obs, uint64_t jobid) { int count = 0; int cap = obs->n_replies < TPI_REPLY_CAP ? obs->n_replies : TPI_REPLY_CAP; for (int i = 0; i < cap; i++) { if (obs->jobids[i] == jobid) { if (obs->statuses[i] != 0) return false; count++; } } return count == 1; } /* ── Reaction spy: stands in for sched.c's sched_job_update_fun / * apiserver.c's apiserver_job_update_fun (the real, only-reachable-for- * facts subscribers — see file header). Captures, at the EXACT moment the * AFTER_EVENT fires: which detail, which job, the job's live state (for * HOOK_DETAIL_JOB_UPDATE), committed_lsn sampled at that instant (the * no-watermark-recheck-needed proof), and whether the job-submit reply was * already enqueued (the reply-never-precedes-notification order proof, * mirroring test_commit_barrier_reply.c's cbrp_jobsubmit_order_spy). ── */ #define TPI_SPY_CAP 16 typedef struct { int detail; uint64_t jobid; uint32_t state_at_fire; uint64_t committed_lsn_at_fire; bool reply_already_enqueued; } tpi_spy_entry_t; static tpi_spy_entry_t g_tpi_spy[TPI_SPY_CAP]; static int g_tpi_spy_n; static int tpi_reaction_spy(int type, int situation, int detail, void *data, va_list ap) { (void)type; (void)data; if (situation != HOOK_AFTER_EVENT) return 0; uint64_t jobid = 0; uint32_t state = 0; switch (detail) { case HOOK_DETAIL_JOB_NEW: { /* args: cluster_t*, xlist_t* (apply_committed_entry's WAL_JOB_SUBMIT * branch) — mirrors test_apply_pump.c's tap_spy_hook decode. */ (void)va_arg(ap, cluster_t *); xlist_t *jlist = va_arg(ap, xlist_t *); if (jlist && jlist->count > 0) { job_t *j = (job_t *)((void **)jlist->data)[0]; jobid = j->job.jobid; state = j->job.state; } break; } case HOOK_DETAIL_JOB_UPDATE: { /* args: cluster_t*, job_t* — the ONLY detail a live-applied fact of * any type (START/FINISH/FAIL/UPDATE/...) ever fires, per the * documented generic-routing deviation (see file header). */ (void)va_arg(ap, cluster_t *); job_t *j = va_arg(ap, job_t *); if (j) { jobid = j->job.jobid; state = j->job.state; } break; } default: /* HOOK_DETAIL_JOB_START/_EXIT/_FAIL/_RUSAGE_UPDATE: structurally * unreachable for a live-applied fact (see file header) — not * relevant to this file's assertions. */ return 0; } if (g_tpi_spy_n < TPI_SPY_CAP) { g_tpi_spy[g_tpi_spy_n].detail = detail; g_tpi_spy[g_tpi_spy_n].jobid = jobid; g_tpi_spy[g_tpi_spy_n].state_at_fire = state; g_tpi_spy[g_tpi_spy_n].committed_lsn_at_fire = atomic_load(&g_tpi_c.persist.committed_lsn); g_tpi_spy[g_tpi_spy_n].reply_already_enqueued = !DATAQ_EMPTY(&g_tpi_req_ep.outq); g_tpi_spy_n++; } return 0; } static void tpi_spy_reset(void) { g_tpi_spy_n = 0; memset(g_tpi_spy, 0, sizeof(g_tpi_spy)); } /* pool_n_vcpus == 0 (every task-6.3 test, and task 6.4's deterministic * ordering test) keeps ce->pool == NULL exactly as before -- no behavior * change for any existing caller. pool_n_vcpus > 0 (task 6.4's burst test * only) wires a REAL vtaskmgr_t, mirroring test_apply_pump.c's task 6.2 * tpr_setup, so commit_resolve_pending()'s two independent dispatches run * as genuine concurrent vtasks instead of the inline fallback. */ static void tpi_setup(int pool_n_vcpus) { kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, .nslots = 32, .delta = 16 }; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; if (makeString((char *)"tpi-probe") == NULL) xstring_table_init(64); if (!g_tpi_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_tpi_jobobj_ready = 1; } memset(&g_tpi_c, 0, sizeof(g_tpi_c)); g_tpi_c.myhost = &g_tpi_c.myhost_info; g_tpi_c.myhost_info.name = (char *)"ctrl-test"; g_tpi_c.myhost_info.key = (char *)""; memset(&g_tpi_req_ep, 0, sizeof(g_tpi_req_ep)); memset(&g_tpi_req_priv, 0, sizeof(g_tpi_req_priv)); g_tpi_req_ep.fd = -1; g_tpi_req_ep.data = &g_tpi_req_priv; TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_tpi_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_tpi_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); g_tpi_c.qmgr = makeInstance(&xkvmgr_type, &str_params); g_tpi_c.workermgr = makeInstance(&xkvmgr_type, &str_params); g_tpi_c.resmgr = makeInstance(&xkvmgr_type, &str_params); TEST_ASSERT_NOT_NULL(g_tpi_c.jobmgr); TEST_ASSERT_NOT_NULL(g_tpi_c.qmgr); TEST_ASSERT_NOT_NULL(g_tpi_c.workermgr); TEST_ASSERT_NOT_NULL(g_tpi_c.resmgr); zarray_init(&g_tpi_c.sharedinst, 16, NULL); persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; cfg.sync_timeout_ms = 2000; snprintf(g_tpi_tmpdir, sizeof(g_tpi_tmpdir), "/tmp/xce_tpi63_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(g_tpi_tmpdir)); persist_set_workdir(&g_tpi_c.persist, g_tpi_tmpdir, "ctrl"); TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tpi_c.persist, &cfg, &g_tpi_c)); TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tpi_c.persist)); /* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until a * test explicitly delivers them via repl_on_ack. commit_set_pool is * deliberately never called -- ce->pool stays NULL, matching every * other fixture in this spec (see test_apply_pump.c's file header). */ TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_tpi_hce, &g_tpi_c.persist, 3, 8)); repl_set_commit_engine(&g_tpi_c.persist.repl, &g_tpi_hce); repl_broadcast_send_hook = tpi_drop_send_hook; g_tpi_pool_active = 0; if (pool_n_vcpus > 0) { TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_tpi_pool, pool_n_vcpus)); TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_tpi_pool)); commit_set_pool(&g_tpi_hce, &g_tpi_pool); g_tpi_pool_active = 1; } TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_tpi_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tpi_reaction_spy, &g_tpi_c)); tpi_spy_reset(); } static void tpi_teardown(void) { tpi_reply_obs_t drain = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &drain); /* free any leftover packets */ repl_broadcast_send_hook = NULL; commit_engine_fini(&g_tpi_hce); if (g_tpi_pool_active) { /* Mirrors test_apply_pump.c's tpr_teardown ordering exactly: * vtaskmgr_fini AFTER commit_engine_fini (so nothing can dispatch a * new vtask onto a pool mid-teardown) but BEFORE persist_stop. */ vtaskmgr_fini(&g_tpi_pool); g_tpi_pool_active = 0; } persist_stop(&g_tpi_c.persist); persist_fini(&g_tpi_c.persist); zarray_fini(&g_tpi_c.sharedinst); if (g_tpi_c.resmgr) dereference(g_tpi_c.resmgr); if (g_tpi_c.workermgr) dereference(g_tpi_c.workermgr); if (g_tpi_c.qmgr) dereference(g_tpi_c.qmgr); if (g_tpi_c.jobmgr) dereference(g_tpi_c.jobmgr); xce_hook_fini(&g_tpi_c.hooksys); } /* Creates a job directly in the fixture's jobmgr, bypassing WAL entirely — * for a pre-existing job that then receives a worker-reported fact. */ static job_submit_data_t g_tpi_seed_submit; static job_t *tpi_seed_job(uint64_t jobid, uint32_t state) { xce_job_t jdef; memset(&g_tpi_seed_submit, 0, sizeof(g_tpi_seed_submit)); g_tpi_seed_submit.queue = (char *)"q"; g_tpi_seed_submit.user = (char *)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.submit = &g_tpi_seed_submit; return makeJob(g_tpi_c.jobmgr, &jdef); } static job_t *tpi_find_job(uint64_t jobid) { job_t *j = NULL; lock(g_tpi_c.jobmgr, RTTI_LOCK_READ); int found = key_find(g_tpi_c.jobmgr, (void *)jobid, (void **)&j); unlock(g_tpi_c.jobmgr); return (found > 0) ? j : NULL; } /* A standby ACK only counts toward majority once this node's OWN copy is * durable (persist_durable_lsn) — poll briefly rather than assume the * flusher thread has already caught up, mirroring test_apply_pump.c's * tap_wait_durable. */ static void tpi_wait_durable(uint64_t lsn, uint32_t deadline_ms) { for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { if (atomic_load(&g_tpi_c.persist.durable_lsn) >= lsn) return; usleep(5000); } } /* Waits for the ordered apply pump to actually finish applying up through * `lsn` — needed even with ce->pool == NULL, since persist_start's * fsync_worker thread independently kicks the pump on every durable_lsn * advance, racing this test's own repl_on_ack-driven kick (see test_apply_ * pump.c's tap_wait_applied doc comment for the full mechanism and the * flake it fixes — this is that exact same lesson, reapplied here). */ static void tpi_wait_applied(uint64_t lsn, uint32_t deadline_ms) { for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { if (persist_last_applied_lsn(&g_tpi_c.persist) >= lsn) return; usleep(5000); } } /* task 6.4: waits until at least `want_n` replies have been observed on the * requester endpoint, accumulating across repeated tpi_scan_replies calls * (each call only drains packets newly enqueued since the last one). Needed * specifically for the real-pool burst test: tpi_wait_applied only proves * the WALK finished, not that every command's continuation hand-off has * fired -- a hand-off that self-heals late (persist_register_pending_apply, * bug-1415) runs on its OWN vtask thread, independently of and potentially * after the walk. */ static void tpi_wait_n_replies(tpi_reply_obs_t *obs, int want_n, uint32_t deadline_ms) { for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { tpi_scan_replies(&g_tpi_req_ep, obs); if (obs->n_replies >= want_n) return; usleep(5000); } tpi_scan_replies(&g_tpi_req_ep, obs); /* one last drain past the deadline */ } static void tpi_sleep_ms(uint32_t ms) { struct timespec ts; ts.tv_sec = ms / 1000u; ts.tv_nsec = (long)(ms % 1000u) * 1000000L; nanosleep(&ts, NULL); } /* ── Job submit request builder (mirrors test_commit_barrier_reply.c's * cbrp_arrange_jobsubmit — single-element, non-array submission). ── */ static job_startinfo_t g_tpi_jobsub_startinfo; static job_submit_data_t g_tpi_jobsub_data; static xce_job_submit_t g_tpi_jobsub_msg; static xce_request_t g_tpi_jobsub_req; static xce_peerinfo_t g_tpi_jobsub_from; static void tpi_arrange_jobsubmit(uint64_t jobid) { memset(&g_tpi_jobsub_startinfo, 0, sizeof(g_tpi_jobsub_startinfo)); g_tpi_jobsub_startinfo.cmdline = (char *)"/bin/true"; memset(&g_tpi_jobsub_data, 0, sizeof(g_tpi_jobsub_data)); g_tpi_jobsub_data.queue = (char *)"normal"; g_tpi_jobsub_data.user = (char *)"tester"; g_tpi_jobsub_data.start_info = &g_tpi_jobsub_startinfo; memset(&g_tpi_jobsub_msg, 0, sizeof(g_tpi_jobsub_msg)); g_tpi_jobsub_msg.jobid = jobid; g_tpi_jobsub_msg.array_size = 1; g_tpi_jobsub_msg.submit = &g_tpi_jobsub_data; g_tpi_jobsub_from.name = (char *)"requester"; g_tpi_jobsub_from.key = (char *)""; memset(&g_tpi_jobsub_req, 0, sizeof(g_tpi_jobsub_req)); g_tpi_jobsub_req.opcode = OPCode_jobsub; g_tpi_jobsub_req.seq = 21; g_tpi_jobsub_req.from = &g_tpi_jobsub_from; g_tpi_jobsub_req.body_which = Request_body_jobsubmsg; g_tpi_jobsub_req.body.jobsubmsg = &g_tpi_jobsub_msg; } /* ═══════════════════════════════════════════════════════════════════════ * Test 1 — command (job submit) and fact (job start) interleaved through * the SAME apply-pump pass and the SAME covering commit. * ═══════════════════════════════════════════════════════════════════════ * * Job X is submitted (command, task 3.2's continuation path); job Y * pre-exists and receives a real worker-reported START fact (task 4.1's * durable-append-only receipt path) — appended in that order, so X's * submit entry has the LOWER lsn. A single covering ACK commits both at * once; the pump must apply them in strict LSN order (X then Y) and gate * BOTH reactions on that same commit — proving the mixed-type ordering and * independent per-entry gating this task's boundary explicitly asks for, * not just two separately-proven single-type cases. */ static void test_pump_command_and_fact_interleave_through_same_commit(void) { const uint64_t jobid_x = 9500; /* submitted via handle_jobsubmit */ const uint64_t jobid_y = 9501; /* pre-existing, receives a START fact */ tpi_setup(0); job_t *jy = tpi_seed_job(jobid_y, JOB_STAT_PEND); int seeded_ok = (jy != NULL); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t lsn_before = g_tpi_c.persist.current_lsn; pthread_mutex_unlock(&g_tpi_c.persist.lock); tpi_arrange_jobsubmit(jobid_x); int submit_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); static unsigned int y_pids[1] = { 7001 }; job_pidinfo_t y_start = { .jobid = jobid_y, .npids = 1, .pids = &y_pids[0] }; int start_rc = on_job_start(&g_tpi_c, &y_start); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t watermark = g_tpi_c.persist.current_lsn - 1; pthread_mutex_unlock(&g_tpi_c.persist.lock); uint64_t x_lsn = lsn_before; /* first append after lsn_before */ uint64_t y_lsn = lsn_before + 1; /* second append */ /* ── Delayed-commit window: poll repeatedly across real time, proving * an absence, not just checking once. Neither reaction may fire, no * reply may be observable, job X must not exist, job Y's state must * stay PEND. ── */ enum { WINDOW_POLLS = 5, WINDOW_INTERVAL_MS = 20 }; bool window_job_x_absent[WINDOW_POLLS]; uint32_t window_jy_state[WINDOW_POLLS]; int window_spy_n[WINDOW_POLLS]; int window_replies[WINDOW_POLLS]; for (int i = 0; i < WINDOW_POLLS; i++) { tpi_sleep_ms(WINDOW_INTERVAL_MS); window_job_x_absent[i] = (tpi_find_job(jobid_x) == NULL); window_jy_state[i] = seeded_ok ? jy->job.state : 0; window_spy_n[i] = g_tpi_spy_n; tpi_reply_obs_t obs = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &obs); window_replies[i] = obs.n_replies; } /* ── Deliver ONE covering ACK past both entries, then wait for the * pump to actually finish applying (poll, per file header). ── */ tpi_wait_durable(watermark, 2000); repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); tpi_wait_applied(watermark, 2000); tpi_reply_obs_t final_obs = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &final_obs); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); job_t *jx_final = tpi_find_job(jobid_x); uint32_t jy_state_final = seeded_ok ? jy->job.state : 0; int spy_n_final = g_tpi_spy_n; tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); tpi_teardown(); /* Arrangement sanity */ TEST_ASSERT_TRUE(seeded_ok); TEST_ASSERT_EQUAL_INT(0, submit_rc); TEST_ASSERT_EQUAL_INT(0, start_rc); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, watermark, "exactly two entries (X's submit, Y's start) must append"); /* Delayed-commit window: zero side effects throughout (Req 5.3, 6.1) */ for (int i = 0; i < WINDOW_POLLS; i++) { TEST_ASSERT_TRUE_MESSAGE(window_job_x_absent[i], "job X must not exist in the job table before its submit " "entry commits, at every poll of the delayed-commit window " "(Req 2.1)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, window_jy_state[i], "job Y's fact must not mutate live state before its entry " "commits, at every poll of the delayed-commit window (Req 5.3)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, window_spy_n[i], "no reaction (worker slot release / scheduler dispatch " "trigger / client-apiserver notification) may fire for " "either entry before its commit, at every poll of the " "delayed-commit window (Req 5.3)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, window_replies[i], "no client reply may be observable before job X's submit " "entry commits (Req 2.2)"); } /* After the covering ACK: both entries committed and applied */ TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, "the covering ACK must advance committed_lsn over both entries"); TEST_ASSERT_NOT_NULL_MESSAGE(jx_final, "job X must be visible in the job table once its entry commits " "and applies"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jy_state_final, "job Y must reflect its applied START fact"); /* Strict LSN/commit order across the MIXED command+fact pair, through * the SAME pump pass — the actual new ground this test exists for. */ TEST_ASSERT_EQUAL_INT_MESSAGE(2, spy_n_final, "exactly two reactions must fire: one for X's submit, one for " "Y's fact"); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_NEW, spy_snap[0].detail, "the LOWER-lsn entry (X's submit) must apply and notify FIRST, " "even though it is a command and the second entry is a fact"); TEST_ASSERT_EQUAL_UINT64(jobid_x, spy_snap[0].jobid); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail, "the HIGHER-lsn entry (Y's fact) must apply and notify SECOND"); TEST_ASSERT_EQUAL_UINT64(jobid_y, spy_snap[1].jobid); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, spy_snap[1].state_at_fire, "Y's reaction must observe the ALREADY-applied RUN state — a " "consumer never sees the notification before the mutation"); /* No-watermark-recheck-needed proof (Req 4.3): at the exact instant * EACH reaction fired, committed_lsn already covered that entry. */ TEST_ASSERT_TRUE_MESSAGE( spy_snap[0].committed_lsn_at_fire >= x_lsn, "X's reaction must fire only once committed_lsn already covers " "X's own lsn -- no separate watermark check is ever needed"); TEST_ASSERT_TRUE_MESSAGE( spy_snap[1].committed_lsn_at_fire >= y_lsn, "Y's reaction must fire only once committed_lsn already covers " "Y's own lsn -- no separate watermark check is ever needed"); /* The client reply for X must never precede X's own reaction (the * mutation+notify), mirroring test_commit_barrier_reply.c's task 6.2 * ordering proof, now re-verified alongside an interleaved fact. */ TEST_ASSERT_FALSE_MESSAGE(spy_snap[0].reply_already_enqueued, "job X's client reply must be sent strictly AFTER X's own " "post-mutation notification, never before or concurrently (Req " "2.2)"); TEST_ASSERT_TRUE_MESSAGE(final_obs.n_replies > 0, "job X must eventually receive its success reply"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, final_obs.last_status, "job X's reply must report success"); TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, final_obs.last_data_which, "job X's success reply must carry its job id"); TEST_ASSERT_EQUAL_UINT64(jobid_x, final_obs.last_jobid); printf("test_pump_command_and_fact_interleave_through_same_commit: " "PASS\n"); } /* ═══════════════════════════════════════════════════════════════════════ * Test 2 — multiple fact types, genuine quorum-loss-then-restore window: * zero reactions throughout the outage, correct reactions after restore. * ═══════════════════════════════════════════════════════════════════════ * * Extends test_commit_barrier_reply.c's test_fact_state_unchanged_ * throughout_quorum_loss_outage (task 4.2), which proves no-mutation * during a genuine outage but never restores quorum to check what happens * next. This test drives the SAME kind of outage (repl_remove_standby, a * structurally-cannot-commit window) through START then FINISH for one * job, polls repeatedly for zero reactions during the outage, then * reconnects, delivers the covering ACK, and confirms both reactions fire * afterward, in order, each already-committed at fire time. */ static void test_pump_fact_reactions_withheld_during_outage_then_fire_after_restore( void) { const uint64_t jobid_z = 9600; tpi_setup(0); job_t *jz = tpi_seed_job(jobid_z, JOB_STAT_PEND); int seeded_ok = (jz != NULL); /* Genuine quorum loss: self alone is 1 of 3, never a majority, so * committed_lsn structurally cannot advance past this point (same * mechanism as test_fact_state_unchanged_throughout_quorum_loss_ * outage). */ repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a); repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t lsn_before = g_tpi_c.persist.current_lsn; pthread_mutex_unlock(&g_tpi_c.persist.lock); static unsigned int z_pids[1] = { 7101 }; job_pidinfo_t z_start = { .jobid = jobid_z, .npids = 1, .pids = &z_pids[0] }; int start_rc = on_job_start(&g_tpi_c, &z_start); xce_jobfinish_t z_finish = { .jobid = jobid_z, .exit_code = 0 }; int finish_rc = on_job_finish(&g_tpi_c, &z_finish); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t watermark = g_tpi_c.persist.current_lsn - 1; pthread_mutex_unlock(&g_tpi_c.persist.lock); uint64_t start_lsn = lsn_before; uint64_t finish_lsn = lsn_before + 1; /* ── Outage window: poll repeatedly across real time. Both facts are * durably appended and pending, but structurally cannot commit while * both standbys remain disconnected. ── */ enum { OUTAGE_POLLS = 5, OUTAGE_INTERVAL_MS = 20 }; uint32_t outage_state[OUTAGE_POLLS]; int outage_spy_n[OUTAGE_POLLS]; uint64_t outage_committed[OUTAGE_POLLS]; for (int i = 0; i < OUTAGE_POLLS; i++) { tpi_sleep_ms(OUTAGE_INTERVAL_MS); outage_state[i] = seeded_ok ? jz->job.state : 0; outage_spy_n[i] = g_tpi_spy_n; outage_committed[i] = atomic_load(&g_tpi_c.persist.committed_lsn); } /* ── Restore quorum and deliver the covering ACK. ── */ TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); tpi_wait_durable(watermark, 2000); repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); tpi_wait_applied(watermark, 2000); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); uint32_t state_final = seeded_ok ? jz->job.state : 0; int spy_n_final = g_tpi_spy_n; tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); tpi_teardown(); TEST_ASSERT_TRUE(seeded_ok); TEST_ASSERT_EQUAL_INT(0, start_rc); TEST_ASSERT_EQUAL_INT(0, finish_rc); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, watermark, "both facts (START, FINISH) must durably append during the " "outage"); for (int i = 0; i < OUTAGE_POLLS; i++) { TEST_ASSERT_TRUE_MESSAGE(outage_committed[i] < start_lsn, "the outage must still be genuinely in effect at this poll " "-- committed_lsn must not have advanced"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, outage_state[i], "job Z's live state must remain unchanged at every point " "during the outage (Req 6.1, 6.2) -- no local-apply escape " "hatch exists"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_spy_n[i], "no reaction may fire for either fact while quorum is " "unreachable, at every poll of the outage window (Req 5.3, " "6.1, 6.2)"); } TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, "a covering ACK after reconnect must advance committed_lsn over " "the retained fact suffix"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, state_final, "job Z must reflect both applied facts (START then FINISH) once " "quorum is restored and the entries commit"); TEST_ASSERT_EQUAL_INT_MESSAGE(2, spy_n_final, "exactly two reactions must fire once quorum is restored -- one " "per retained fact, none lost, none duplicated"); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[0].detail); TEST_ASSERT_EQUAL_UINT64(jobid_z, spy_snap[0].jobid); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, spy_snap[0].state_at_fire, "the START fact (lower lsn) must apply and notify FIRST, " "reflecting RUN -- retained entries commit and apply in their " "original log order (Req 6.3)"); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail); TEST_ASSERT_EQUAL_UINT64(jobid_z, spy_snap[1].jobid); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, spy_snap[1].state_at_fire, "the FINISH fact (higher lsn) must apply and notify SECOND, " "reflecting the now-terminal DONE state"); TEST_ASSERT_TRUE_MESSAGE( spy_snap[0].committed_lsn_at_fire >= start_lsn, "the START reaction must fire only once committed_lsn already " "covers its own lsn -- no separate watermark check is ever " "needed (Req 4.3)"); TEST_ASSERT_TRUE_MESSAGE( spy_snap[1].committed_lsn_at_fire >= finish_lsn, "the FINISH reaction must fire only once committed_lsn already " "covers its own lsn -- no separate watermark check is ever " "needed (Req 4.3)"); printf("test_pump_fact_reactions_withheld_during_outage_then_fire_" "after_restore: PASS\n"); } /* ═══════════════════════════════════════════════════════════════════════ * Test 3 — task 6.4: MIXED facts AND commands through a genuine quorum-loss- * then-restore window, proving strict original-log-order catch-up. * ═══════════════════════════════════════════════════════════════════════ * * Distinguishing scope vs this file's Test 2 above (task 6.3) and * test_commit_barrier_reply.c's test_fact_state_unchanged_throughout_ * quorum_loss_outage (task 4.2), both of which drive a genuine repl_remove_ * standby outage with FACTS ONLY: this test interleaves THREE job-submit * COMMANDS with TWO facts, five entries total, appended in COMMAND, FACT, * COMMAND, FACT, COMMAND order specifically so a bug that grouped all * commands before all facts (or vice versa) -- rather than genuinely * preserving append order -- would be caught by the strict per-slot * sequence assertion below, not merely "everything eventually applied, in * some order." Requirements 6.3 (retained entries commit/apply in original * log order) and 7.3 (never applies ahead of an earlier-committing entry) * are the direct target; 9.4 is checked negatively -- the outage must never * spuriously resolve a registered command continuation as a FAILURE (the * demotion/truncation path that requirement's other clause governs), only * ever as a genuine, later, COMMITTED success once quorum returns. * * Uses the standard NULL-pool fixture (tpi_setup(0)) deliberately, exactly * like this file's other two tests: with ce->pool == NULL, commit_resolve_ * pending()'s continuation dispatch and the apply-pump kick both fall back * to their synchronous inline path (dispatch_continuation, apply_pump_drain * -- see commit.c), so the ENTIRE catch-up (register, apply, notify, reply) * happens deterministically on this test's own thread inside the single * repl_on_ack() call below. That determinism is what makes the strict, * order-sensitive sequence assertion below meaningful rather than flaky -- * Test 4 below separately re-verifies the same ordering invariant survives * genuine concurrency, and additionally stress-tests task 6.2's bug-1415 * self-heal under a real multi-entry burst. */ static void test_pump_mixed_facts_and_commands_catchup_original_order(void) { const uint64_t jobid_a = 9700; /* command 1 */ const uint64_t jobid_w = 9701; /* pre-existing, receives both facts */ const uint64_t jobid_b = 9702; /* command 2 */ const uint64_t jobid_c = 9703; /* command 3 */ tpi_setup(0); job_t *jw = tpi_seed_job(jobid_w, JOB_STAT_PEND); int seeded_ok = (jw != NULL); /* Genuine quorum loss -- same mechanism as this file's Test 2 (task * 6.3) and test_commit_barrier_reply.c's task 4.2 test: both standbys * removed, self alone is 1 of 3, never a majority. */ repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a); repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t lsn_before = g_tpi_c.persist.current_lsn; pthread_mutex_unlock(&g_tpi_c.persist.lock); /* Append order: COMMAND, FACT, COMMAND, FACT, COMMAND -- deliberately * NOT grouped by type. */ tpi_arrange_jobsubmit(jobid_a); int submit_a_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); static unsigned int w_pids[1] = { 7201 }; job_pidinfo_t w_start = { .jobid = jobid_w, .npids = 1, .pids = &w_pids[0] }; int start_rc = on_job_start(&g_tpi_c, &w_start); tpi_arrange_jobsubmit(jobid_b); int submit_b_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); xce_jobfinish_t w_finish = { .jobid = jobid_w, .exit_code = 0 }; int finish_rc = on_job_finish(&g_tpi_c, &w_finish); tpi_arrange_jobsubmit(jobid_c); int submit_c_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t watermark = g_tpi_c.persist.current_lsn - 1; pthread_mutex_unlock(&g_tpi_c.persist.lock); /* ── Outage window: poll repeatedly across real time -- all five * entries are durably appended and retained, but structurally cannot * commit while both standbys remain disconnected (Req 6.1). ── */ enum { OUTAGE_POLLS = 5, OUTAGE_INTERVAL_MS = 20 }; bool outage_a_absent[OUTAGE_POLLS]; bool outage_b_absent[OUTAGE_POLLS]; bool outage_c_absent[OUTAGE_POLLS]; uint32_t outage_w_state[OUTAGE_POLLS]; int outage_spy_n[OUTAGE_POLLS]; uint64_t outage_committed[OUTAGE_POLLS]; int outage_replies[OUTAGE_POLLS]; for (int i = 0; i < OUTAGE_POLLS; i++) { tpi_sleep_ms(OUTAGE_INTERVAL_MS); outage_a_absent[i] = (tpi_find_job(jobid_a) == NULL); outage_b_absent[i] = (tpi_find_job(jobid_b) == NULL); outage_c_absent[i] = (tpi_find_job(jobid_c) == NULL); outage_w_state[i] = seeded_ok ? jw->job.state : 0; outage_spy_n[i] = g_tpi_spy_n; outage_committed[i] = atomic_load(&g_tpi_c.persist.committed_lsn); tpi_reply_obs_t obs = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &obs); outage_replies[i] = obs.n_replies; } /* ── Restore quorum and deliver ONE covering ACK past all five * entries. ── */ TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); tpi_wait_durable(watermark, 2000); repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); tpi_wait_applied(watermark, 2000); tpi_reply_obs_t final_obs = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &final_obs); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); job_t *ja_final = tpi_find_job(jobid_a); job_t *jb_final = tpi_find_job(jobid_b); job_t *jc_final = tpi_find_job(jobid_c); uint32_t jw_state_final = seeded_ok ? jw->job.state : 0; int spy_n_final = g_tpi_spy_n; tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); tpi_teardown(); /* Arrangement sanity */ TEST_ASSERT_TRUE(seeded_ok); TEST_ASSERT_EQUAL_INT(0, submit_a_rc); TEST_ASSERT_EQUAL_INT(0, start_rc); TEST_ASSERT_EQUAL_INT(0, submit_b_rc); TEST_ASSERT_EQUAL_INT(0, finish_rc); TEST_ASSERT_EQUAL_INT(0, submit_c_rc); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 4, watermark, "exactly five entries (3 commands, 2 facts) must durably append " "during the outage"); for (int i = 0; i < OUTAGE_POLLS; i++) { TEST_ASSERT_TRUE_MESSAGE(outage_committed[i] < lsn_before, "the outage must still be genuinely in effect at this poll -- " "committed_lsn must not have advanced past any retained " "entry"); TEST_ASSERT_TRUE_MESSAGE(outage_a_absent[i], "command A's job must not exist before its entry commits, at " "every poll of the outage window (Req 2.1) -- no local-apply " "escape hatch exists"); TEST_ASSERT_TRUE_MESSAGE(outage_b_absent[i], "command B's job must not exist before its entry commits, at " "every poll of the outage window (Req 2.1)"); TEST_ASSERT_TRUE_MESSAGE(outage_c_absent[i], "command C's job must not exist before its entry commits, at " "every poll of the outage window (Req 2.1)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, outage_w_state[i], "job W's facts must not mutate live state before their " "entries commit, at every poll of the outage window (Req 6.1, " "6.2)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_spy_n[i], "no reaction for any of the five retained entries may fire " "while quorum is unreachable, at every poll (Req 5.3, 6.1, " "6.2)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_replies[i], "no client reply for any of the three retained commands may " "be observable while quorum is unreachable -- confirms the " "outage never spuriously resolves a registered continuation " "as a failure either (Req 9.4)"); } /* After the covering ACK: all five entries committed and applied. */ TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, "a covering ACK after reconnect must advance committed_lsn over " "the entire retained suffix"); TEST_ASSERT_NOT_NULL_MESSAGE(ja_final, "command A's job must be visible once its entry commits and " "applies"); TEST_ASSERT_NOT_NULL_MESSAGE(jb_final, "command B's job must be visible once its entry commits and " "applies"); TEST_ASSERT_NOT_NULL_MESSAGE(jc_final, "command C's job must be visible once its entry commits and " "applies"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, jw_state_final, "job W must reflect both applied facts (START then FINISH) once " "quorum is restored and its entries commit"); /* The actual new ground this test exists for: strict, original-log- * order catch-up across a MIX of commands and facts -- not grouped by * type, not reordered by any priority. A bug that applied all commands * before all facts (or vice versa) would still pass a "some order" * check but would fail this exact per-slot sequence assertion. */ TEST_ASSERT_EQUAL_INT_MESSAGE(5, spy_n_final, "exactly five reactions must fire: one per retained entry, none " "lost, none duplicated"); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_NEW, spy_snap[0].detail, "slot 0 (lowest lsn) must be command A's submit -- original " "append order"); TEST_ASSERT_EQUAL_UINT64(jobid_a, spy_snap[0].jobid); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail, "slot 1 must be W's START fact, applying SECOND even though it is " "a fact sandwiched between two commands"); TEST_ASSERT_EQUAL_UINT64(jobid_w, spy_snap[1].jobid); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, spy_snap[1].state_at_fire); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_NEW, spy_snap[2].detail, "slot 2 must be command B's submit -- a bug that grouped both " "facts before both remaining commands would place W's FINISH here " "instead"); TEST_ASSERT_EQUAL_UINT64(jobid_b, spy_snap[2].jobid); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_UPDATE, spy_snap[3].detail, "slot 3 must be W's FINISH fact"); TEST_ASSERT_EQUAL_UINT64(jobid_w, spy_snap[3].jobid); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, spy_snap[3].state_at_fire); TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_NEW, spy_snap[4].detail, "slot 4 (highest lsn) must be command C's submit -- last in, last " "applied"); TEST_ASSERT_EQUAL_UINT64(jobid_c, spy_snap[4].jobid); for (int i = 0; i < 5; i++) { uint64_t entry_lsn = lsn_before + (uint64_t)i; TEST_ASSERT_TRUE_MESSAGE( spy_snap[i].committed_lsn_at_fire >= entry_lsn, "each reaction must fire only once committed_lsn already " "covers its own lsn -- no separate watermark check is ever " "needed (Req 4.3)"); } /* Every retained command's continuation resolves as a genuine, later * success -- never a spurious failure from the outage itself (Req 9.3, * 9.4). */ TEST_ASSERT_EQUAL_INT_MESSAGE(3, final_obs.n_replies, "exactly three replies must be observable: one per retained " "command, none lost, none duplicated"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_a), "command A must receive exactly one successful reply"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_b), "command B must receive exactly one successful reply"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_c), "command C must receive exactly one successful reply"); printf("test_pump_mixed_facts_and_commands_catchup_original_order: " "PASS\n"); } /* ═══════════════════════════════════════════════════════════════════════ * Test 4 — task 6.4: a REAL multi-entry catch-up burst on a genuine * multi-vcpu pool, stress-confirming task 6.2's bug-1415 self-healing fix * under realistic concurrency rather than one forced interleaving. * ═══════════════════════════════════════════════════════════════════════ * * test_apply_pump.c's own task 6.2 regression test (test_apply_pump_late_ * pending_apply_registration_self_heals) already proves the fix for ONE * specific, artificially-forced interleaving on a single lsn, deliberately * blocking the continuation until the walk has already passed it. That is * necessary but not sufficient evidence for THIS task's charter: a genuine * quorum-loss-then-restore catch-up commits a whole BATCH of retained * entries at once, which fires MULTIPLE independent continuation-dispatch * vtasks concurrently against the SAME single walk instance -- N * independent race instances in flight together, not one. This test drives * that real scenario: four job-submit commands and two facts, all durably * retained during a genuine repl_remove_standby outage, released by ONE * covering ACK on a REAL 4-vcpu vtaskmgr_t (unlike every other fixture in * this file and spec, which keep ce->pool == NULL and so structurally * cannot reproduce this race class at all -- see test_apply_pump.c's task * 6.2 test header for why). * * Two invariants must both hold under this real concurrency, not just when * pool == NULL: * 1. The apply-pump walk itself is single-threaded per persist_t (CAS- * coalesced kicks, `p->apply_pump_lock`) -- so the six reactions must * STILL fire in strict original append order, exactly as Test 3 above * proves for the NULL-pool case. Re-asserting the identical per-slot * sequence here proves that invariant survives four independent * continuation vtasks concurrently racing the walk, not just the * deterministic inline-fallback path. * 2. Every one of the four commands must receive EXACTLY one successful * reply -- zero (a bug-1415-class permanent hang) and more than one * (a double-fire) are both failures this test is built to catch, * whichever side of the per-lsn race actually occurs on a given run. */ static void test_pump_burst_catchup_real_pool_no_hang_no_double_fire(void) { const int N_VCPUS = 4; /* > 1 is the whole point -- see header */ const uint64_t jobid_d = 9800; /* command 1 */ const uint64_t jobid_w2 = 9810; /* pre-existing, receives both facts */ const uint64_t jobid_e = 9801; /* command 2 */ const uint64_t jobid_f = 9802; /* command 3 */ const uint64_t jobid_g = 9803; /* command 4 */ tpi_setup(N_VCPUS); job_t *jw2 = tpi_seed_job(jobid_w2, JOB_STAT_PEND); int seeded_ok = (jw2 != NULL); repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a); repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t lsn_before = g_tpi_c.persist.current_lsn; pthread_mutex_unlock(&g_tpi_c.persist.lock); /* Append order: D, W-START, E, F, W-FINISH, G -- two commands back to * back (E, F) to also stress adjacent-command dispatch concurrency. */ tpi_arrange_jobsubmit(jobid_d); int submit_d_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); static unsigned int w2_pids[1] = { 7301 }; job_pidinfo_t w2_start = { .jobid = jobid_w2, .npids = 1, .pids = &w2_pids[0] }; int start_rc = on_job_start(&g_tpi_c, &w2_start); tpi_arrange_jobsubmit(jobid_e); int submit_e_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); tpi_arrange_jobsubmit(jobid_f); int submit_f_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); xce_jobfinish_t w2_finish = { .jobid = jobid_w2, .exit_code = 0 }; int finish_rc = on_job_finish(&g_tpi_c, &w2_finish); tpi_arrange_jobsubmit(jobid_g); int submit_g_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); pthread_mutex_lock(&g_tpi_c.persist.lock); uint64_t watermark = g_tpi_c.persist.current_lsn - 1; pthread_mutex_unlock(&g_tpi_c.persist.lock); enum { OUTAGE_POLLS = 5, OUTAGE_INTERVAL_MS = 20 }; bool outage_d_absent[OUTAGE_POLLS]; bool outage_g_absent[OUTAGE_POLLS]; uint32_t outage_w2_state[OUTAGE_POLLS]; int outage_spy_n[OUTAGE_POLLS]; uint64_t outage_committed[OUTAGE_POLLS]; int outage_replies[OUTAGE_POLLS]; for (int i = 0; i < OUTAGE_POLLS; i++) { tpi_sleep_ms(OUTAGE_INTERVAL_MS); outage_d_absent[i] = (tpi_find_job(jobid_d) == NULL); outage_g_absent[i] = (tpi_find_job(jobid_g) == NULL); outage_w2_state[i] = seeded_ok ? jw2->job.state : 0; outage_spy_n[i] = g_tpi_spy_n; outage_committed[i] = atomic_load(&g_tpi_c.persist.committed_lsn); tpi_reply_obs_t obs = { 0 }; tpi_scan_replies(&g_tpi_req_ep, &obs); outage_replies[i] = obs.n_replies; } TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); TEST_ASSERT_TRUE( repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); tpi_wait_durable(watermark, 2000); repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); tpi_wait_applied(watermark, 3000); /* The walk finishing (tpi_wait_applied above) does not itself prove * every command's client reply has fired -- see file header: a LATE- * registering continuation's hand-off fires on its OWN vtask thread, * independently of and potentially after the walk. Poll separately, * bounded, for all four replies. */ tpi_reply_obs_t final_obs = { 0 }; tpi_wait_n_replies(&final_obs, 4, 3000); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); uint64_t applied_final = persist_last_applied_lsn(&g_tpi_c.persist); job_t *jd_final = tpi_find_job(jobid_d); job_t *je_final = tpi_find_job(jobid_e); job_t *jf_final = tpi_find_job(jobid_f); job_t *jg_final = tpi_find_job(jobid_g); uint32_t jw2_state_final = seeded_ok ? jw2->job.state : 0; int spy_n_final = g_tpi_spy_n; tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); int pending_d = persist_has_pending_apply(&g_tpi_c.persist, lsn_before); int pending_e = persist_has_pending_apply(&g_tpi_c.persist, lsn_before + 2); int pending_f = persist_has_pending_apply(&g_tpi_c.persist, lsn_before + 3); int pending_g = persist_has_pending_apply(&g_tpi_c.persist, lsn_before + 5); tpi_teardown(); TEST_ASSERT_TRUE(seeded_ok); TEST_ASSERT_EQUAL_INT(0, submit_d_rc); TEST_ASSERT_EQUAL_INT(0, start_rc); TEST_ASSERT_EQUAL_INT(0, submit_e_rc); TEST_ASSERT_EQUAL_INT(0, submit_f_rc); TEST_ASSERT_EQUAL_INT(0, finish_rc); TEST_ASSERT_EQUAL_INT(0, submit_g_rc); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 5, watermark, "exactly six entries (4 commands, 2 facts) must durably append " "during the outage"); for (int i = 0; i < OUTAGE_POLLS; i++) { TEST_ASSERT_TRUE_MESSAGE(outage_committed[i] < lsn_before, "the outage must still be genuinely in effect at this poll"); TEST_ASSERT_TRUE_MESSAGE(outage_d_absent[i], "command D's job must not exist before its entry commits, at " "every poll of the outage window"); TEST_ASSERT_TRUE_MESSAGE(outage_g_absent[i], "command G's job must not exist before its entry commits, at " "every poll of the outage window"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, outage_w2_state[i], "job W2's facts must not mutate live state before their " "entries commit, at every poll of the outage window"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_spy_n[i], "no reaction for any of the six retained entries may fire " "while quorum is unreachable, at every poll"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_replies[i], "no client reply for any of the four retained commands may be " "observable while quorum is unreachable"); } TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, "a covering ACK after reconnect must advance committed_lsn over " "the entire retained suffix"); TEST_ASSERT_TRUE_MESSAGE(applied_final >= watermark, "the apply pump must fully catch up through the retained suffix"); TEST_ASSERT_NOT_NULL(jd_final); TEST_ASSERT_NOT_NULL(je_final); TEST_ASSERT_NOT_NULL(jf_final); TEST_ASSERT_NOT_NULL(jg_final); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, jw2_state_final, "job W2 must reflect both applied facts once quorum is restored"); /* Invariant 1 (see file header): the walk is single-threaded per * persist_t regardless of pool -- strict original append order must * survive real concurrent continuation dispatch, not just the * deterministic NULL-pool case Test 3 above covers. */ TEST_ASSERT_EQUAL_INT_MESSAGE(6, spy_n_final, "exactly six reactions must fire: one per retained entry, none " "lost, none duplicated, even under real concurrent dispatch"); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_NEW, spy_snap[0].detail); TEST_ASSERT_EQUAL_UINT64(jobid_d, spy_snap[0].jobid); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail); TEST_ASSERT_EQUAL_UINT64(jobid_w2, spy_snap[1].jobid); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, spy_snap[1].state_at_fire); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_NEW, spy_snap[2].detail); TEST_ASSERT_EQUAL_UINT64(jobid_e, spy_snap[2].jobid); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_NEW, spy_snap[3].detail); TEST_ASSERT_EQUAL_UINT64(jobid_f, spy_snap[3].jobid); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[4].detail); TEST_ASSERT_EQUAL_UINT64(jobid_w2, spy_snap[4].jobid); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, spy_snap[4].state_at_fire); TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_NEW, spy_snap[5].detail); TEST_ASSERT_EQUAL_UINT64(jobid_g, spy_snap[5].jobid); /* Invariant 2 (see file header): task 6.2's bug-1415 self-heal must * resolve EVERY command's continuation exactly once, whichever side of * the walk-vs-registration race actually occurred for it on this run * -- neither a hang (fewer than 4 replies observed within the bounded * poll) nor a double-fire (more than 4, or any jobid appearing twice) * is acceptable. */ TEST_ASSERT_EQUAL_INT_MESSAGE(4, final_obs.n_replies, "exactly four replies must be observable: one per retained " "command -- fewer means a bug-1415-class hang, more means a " "double-fire"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_d), "command D must receive exactly one successful reply"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_e), "command E must receive exactly one successful reply"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_f), "command F must receive exactly one successful reply"); TEST_ASSERT_TRUE_MESSAGE( tpi_reply_has_exactly_one_success(&final_obs, jobid_g), "command G must receive exactly one successful reply"); /* No orphaned registration may remain linked on pending_applies for * any of the four commands once every reply has fired (mirrors * test_apply_pump.c's task 6.2 regression test's own final check). */ TEST_ASSERT_EQUAL_INT(0, pending_d); TEST_ASSERT_EQUAL_INT(0, pending_e); TEST_ASSERT_EQUAL_INT(0, pending_f); TEST_ASSERT_EQUAL_INT(0, pending_g); printf("test_pump_burst_catchup_real_pool_no_hang_no_double_fire: " "PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_apply_pump_integration(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_pump_command_and_fact_interleave_through_same_commit); RUN_TEST( test_pump_fact_reactions_withheld_during_outage_then_fire_after_restore); RUN_TEST(test_pump_mixed_facts_and_commands_catchup_original_order); RUN_TEST(test_pump_burst_catchup_real_pool_no_hang_no_double_fire); }