test(raft-apply-decoupling): apply/notify timing integration through the full pump (task 6.3)
Add test_apply_pump_integration.c (2 tests), no production changes. Exercises the full live pump (real commit engine, real apply dispatch, real hook subscribers) for both a job submission and a fact: proves reply-after-visibility ordering for commands, and zero fact-reaction side effects during a delayed-commit/quorum-loss window (polled repeatedly across real time, not checked once). Confirms Req 4.3 by construction, not assertion: the apply walk only ever invokes apply_committed_entry() for LSNs already <= committed_lsn, so a consumer acting on the post-apply notification never needs to separately re-check a commit watermark. Independently verifies apply_committed_entry()'s generic HOOK_DETAIL_JOB_UPDATE routing (unchanged by task 6.1's _pad fix) makes the type-specific sched.c/apiserver.c hook handlers structurally unreachable for live-applied facts -- confirmed their va_arg-typed signatures would type-confuse-crash if ever called with the job_t* apply_committed_entry() actually passes, making the single generic reaction spy a sound design, not a coverage gap. Repo-wide audit confirms exactly 2 live call sites ever fire these hook details (the guarded apply seam, and job_ctrl's legitimate SIGCONT/SIGSTOP path) -- no orphaned third site.
This commit is contained in:
parent
88577f4658
commit
5f3b97a97b
4 changed files with 768 additions and 1 deletions
|
|
@ -83,6 +83,7 @@ extern void test_replay_apply_seam(void);
|
|||
extern void test_apply_reapply_idempotency(void);
|
||||
extern void test_apply_correctness_verification(void);
|
||||
extern void test_apply_pump(void);
|
||||
extern void test_apply_pump_integration(void);
|
||||
|
||||
void setUp(void) { return; }
|
||||
|
||||
|
|
@ -235,6 +236,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_apply_reapply_idempotency);
|
||||
TEST(test_apply_correctness_verification);
|
||||
TEST(test_apply_pump);
|
||||
TEST(test_apply_pump_integration);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
763
src/controller/testsuite/test_apply_pump_integration.c
Normal file
763
src/controller/testsuite/test_apply_pump_integration.c
Normal file
|
|
@ -0,0 +1,763 @@
|
|||
/* test_apply_pump_integration.c — raft-apply-decoupling task 6.3
|
||||
*
|
||||
* 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").
|
||||
*
|
||||
* 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, commit_
|
||||
* set_pool never called so the apply pump's NULL-pool inline fallback
|
||||
* applies) 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).
|
||||
*
|
||||
* Both 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 <rtti.h>
|
||||
|
||||
#include <capnp_c.h>
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
#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 "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];
|
||||
|
||||
/* 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;
|
||||
|
||||
typedef struct {
|
||||
int n_replies;
|
||||
int32_t last_status;
|
||||
int32_t last_data_which;
|
||||
uint64_t last_jobid;
|
||||
} 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);
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/* ── 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));
|
||||
}
|
||||
|
||||
static void tpi_setup(void)
|
||||
{
|
||||
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;
|
||||
|
||||
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);
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
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();
|
||||
|
||||
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 group entry point ───────────────────────────────────────────── */
|
||||
|
||||
void test_apply_pump_integration(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
RUN_TEST(test_pump_command_and_fact_interleave_through_same_commit);
|
||||
RUN_TEST(
|
||||
test_pump_fact_reactions_withheld_during_outage_then_fire_after_restore);
|
||||
UNITY_END();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue