/* test_apply_reapply_idempotency.c — raft-apply-decoupling task 2.4 * * Extends task 2.2's coverage of apply_committed_entry() (persist.c) with the * idempotent/ordered re-application guarantees Req 3.4/7.1/7.2/7.3 require, * driven end-to-end through the real dispatch function (never by calling the * task-2.1 guard cores directly — test_apply_seam.c already covers those). * * Does NOT duplicate test_apply_committed_entry.c's existing coverage * (one duplicate-job-submit case, one directly-seeded-terminal backward- * rejected case). This file adds: * * 1. A command-family (WAL_JOB_SUBMIT) entry re-applied THREE times, * proving the mutation and notification stay pinned at "happened once" * across repeated redelivery, and that the job object's identity * (pointer) never changes — no reconstruction on re-apply. * 2. A fact-family (WAL_JOB_START) entry re-applied with the IDENTICAL * state twice — the JOB_TRANSITION_NOOP_DUPLICATE path inside * apply_job_update_transition() — which 2.2's tests never exercised * (2.2 covered forward-once and backward-from-a-directly-seeded- * terminal job, not a same-state duplicate). * 3. The task's named scenario: SUBMIT -> FINISH -> stale, previously- * unapplied START, all three driven through apply_committed_entry(), * confirming the job stays in its FINISH-terminal state after the * stale START and that the stale START fires NO notification. The * guard here is state-based (job's current lifecycle stage vs. the * requested next stage), not LSN-position-based — LSN ordering is the * apply pump's job (task 6.1, out of this task's boundary); this test * proves the state guard alone is sufficient to stop a regression even * if a stale entry ever reaches the dispatch seam out of order. */ #include #include #include #include #include #include #include #include "cluster.h" #include "controller.capnp.h" #include "hook.h" #include "jobj.h" #include "persist.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 test_apply_committed_entry.c/test_apply_seam.c. */ static int g_tari_jobobj_ready; static cluster_t g_tari_c; static persist_t g_tari_p; /* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ static int g_tari_hook_fires; static int tari_spy_hook(int type, int situation, int detail, void* data, va_list ap) { (void)type; (void)detail; (void)data; (void)ap; if (situation == HOOK_AFTER_EVENT) g_tari_hook_fires++; return 0; } static void tari_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, .nslots = 32, .delta = 16 }; if (makeString((char*)"tari-probe") == NULL) xstring_table_init(64); if (!g_tari_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_tari_jobobj_ready = 1; } memset(&g_tari_c, 0, sizeof(g_tari_c)); g_tari_c.myhost = &g_tari_c.myhost_info; g_tari_c.myhost_info.name = (char*)"ctrl-test"; g_tari_c.myhost_info.key = (char*)""; TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_tari_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_tari_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); g_tari_c.qmgr = makeInstance(&xkvmgr_type, &str_params); g_tari_c.workermgr = makeInstance(&xkvmgr_type, &str_params); g_tari_c.resmgr = makeInstance(&xkvmgr_type, &str_params); TEST_ASSERT_NOT_NULL(g_tari_c.jobmgr); TEST_ASSERT_NOT_NULL(g_tari_c.qmgr); TEST_ASSERT_NOT_NULL(g_tari_c.workermgr); TEST_ASSERT_NOT_NULL(g_tari_c.resmgr); zarray_init(&g_tari_c.sharedinst, 16, NULL); memset(&g_tari_p, 0, sizeof(g_tari_p)); g_tari_p.cluster = &g_tari_c; g_tari_hook_fires = 0; } static void tari_teardown(void) { zarray_fini(&g_tari_c.sharedinst); if (g_tari_c.resmgr) dereference(g_tari_c.resmgr); if (g_tari_c.workermgr) dereference(g_tari_c.workermgr); if (g_tari_c.qmgr) dereference(g_tari_c.qmgr); if (g_tari_c.jobmgr) dereference(g_tari_c.jobmgr); xce_hook_fini(&g_tari_c.hooksys); } /* Mirrors the file-static capnp_encode_job in persist.c (file-static, so * every test file that needs it keeps a local copy — same pattern as * test_apply_committed_entry.c's tace_encode_job / test_egress_filter.c's * ef_encode_job). */ static int tari_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) { struct capn c; struct capn_segment* cs; struct Job cj; Job_ptr jp; capn_init_malloc(&c); cs = capn_root(&c).seg; encode_Job(cs, &cj, (xce_job_t*)j); jp = new_Job(cs); write_Job(&cj, jp); capn_setp(capn_root(&c), 0, jp.p); *out_len = capn_size(&c); *out = malloc(*out_len); if (!*out) { capn_free(&c); return -ENOMEM; } capn_write_mem(&c, *out, *out_len, 0); capn_free(&c); return 0; } /* Look up a job_t wrapper by jobid (NULL if absent). */ static job_t* tari_find_job(uint64_t jobid) { job_t* j = NULL; lock(g_tari_c.jobmgr, RTTI_LOCK_READ); int found = key_find(g_tari_c.jobmgr, (void*)jobid, (void**)&j); unlock(g_tari_c.jobmgr); return (found > 0) ? j : NULL; } /* Creates a job directly in the fixture's jobmgr, bypassing * handle_jobsubmit/restore_job entirely — for seeding fact-transition * tests that don't need a real SUBMIT-through-dispatch step. */ static job_submit_data_t g_tari_seed_submit; static job_t* tari_seed_job(uint64_t jobid, uint32_t state) { xce_job_t jdef; memset(&g_tari_seed_submit, 0, sizeof(g_tari_seed_submit)); g_tari_seed_submit.queue = (char*)"q"; g_tari_seed_submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.submit = &g_tari_seed_submit; return makeJob(g_tari_c.jobmgr, &jdef); } /* ── 1. Command-family (WAL_JOB_SUBMIT) idempotent re-application ──────── */ static void test_tari_job_submit_reapplied_thrice_is_idempotent(void) { tari_setup(); job_submit_data_t submit; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; xce_job_t jdef; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = 701; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, tari_encode_job(&jdef, &payload, &plen)); TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_tari_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tari_spy_hook, NULL)); /* First application: real mutation + one notification. */ TEST_ASSERT_EQUAL_INT_MESSAGE(1, apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT, 100, payload, (uint32_t)plen), "a genuine new job creation must return 1 (bug-1419)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tari_hook_fires, "the first application must fire exactly one notification"); job_t* first = tari_find_job(701); TEST_ASSERT_NOT_NULL(first); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, first->job.state); /* Re-apply the SAME (wal_type, lsn, payload) twice more — simulating * replay/resync redelivering the identical committed entry. Req 3.4/7.1: * observable state must not change again, and the notification must not * fire again, on EITHER repeat. */ for (int i = 0; i < 2; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE(0, apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT, 100, payload, (uint32_t)plen), "an idempotent duplicate must return 0, not 1 (bug-1419)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tari_hook_fires, "re-applying an identical committed SUBMIT must never fire a " "second notification, no matter how many times it repeats"); job_t* again = tari_find_job(701); TEST_ASSERT_EQUAL_PTR_MESSAGE(first, again, "re-application must not reconstruct the job — the same object " "identity must persist across every redelivery"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, again->job.state); } free(payload); tari_teardown(); printf("test_tari_job_submit_reapplied_thrice_is_idempotent: PASS\n"); } /* ── 2. Fact-family (WAL_JOB_START) identical-duplicate re-application ─── */ static void test_tari_job_state_transition_reapplied_identical_is_idempotent( void) { tari_setup(); job_t* j = tari_seed_job(702, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); wal_jobstate_t js = { .jobid = 702, .state = JOB_STAT_RUN }; xce_add_hook(&g_tari_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tari_spy_hook, NULL); /* First application: forward PEND -> RUN, one notification. */ TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tari_p, WAL_JOB_START, 200, &js, sizeof(js))); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tari_hook_fires, "the first forward transition must fire exactly one notification"); /* Re-apply the IDENTICAL entry (same jobid, same requested state RUN, * same LSN) — this hits apply_job_update_transition()'s state==cur * NOOP_DUPLICATE guard specifically (Req 3.4/7.1), a path 2.2's own * tests never drove through apply_committed_entry (2.2 only covered a * forward-once apply and a backward-from-directly-seeded-terminal * apply, never a same-state duplicate). Do it twice to prove the no-op * is stable, not a one-shot fluke. */ for (int i = 0; i < 2; i++) { int rc = apply_committed_entry(&g_tari_p, WAL_JOB_START, 200, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "a duplicate identical fact re-application is a success no-op, " "not an apply failure"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, "a duplicate identical fact must leave the job's state " "unchanged"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tari_hook_fires, "a duplicate identical fact re-application must fire NO " "additional notification"); } tari_teardown(); printf("test_tari_job_state_transition_reapplied_identical_is_" "idempotent: PASS\n"); } /* ── 3. Named scenario: SUBMIT -> FINISH -> stale, previously-unapplied * START, entirely through apply_committed_entry() ────────────────── */ static void test_tari_finish_then_stale_start_preserves_terminal_state_via_dispatch(void) { tari_setup(); xce_add_hook(&g_tari_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tari_spy_hook, NULL); /* Step 1: SUBMIT the job for real, through the dispatch seam (not * seeded directly) — jobid 703 starts PEND. */ job_submit_data_t submit; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; xce_job_t jdef; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = 703; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, tari_encode_job(&jdef, &payload, &plen)); TEST_ASSERT_EQUAL_INT_MESSAGE(1, apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT, 300, payload, (uint32_t)plen), "a genuine new job creation must return 1 (bug-1419)"); free(payload); job_t* j = tari_find_job(703); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tari_hook_fires, "SUBMIT must fire exactly one notification before FINISH is ever " "applied"); /* Step 2: apply a FINISH entry (job-state-family wire shape, same as * START/UPDATE per 2.2's documented wal_jobstate_t routing deviation) — * drives the job to its terminal state. */ wal_jobstate_t finish = { .jobid = 703, .state = JOB_STAT_DONE }; TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tari_p, WAL_JOB_FINISH, 301, &finish, sizeof(finish))); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, "FINISH must drive the job to its terminal state"); TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_tari_hook_fires, "FINISH must fire exactly one notification (SUBMIT's + FINISH's = " "2 total so far)"); /* Step 3: apply a STALE, previously-unapplied START for the SAME job — * simulating a duplicate/out-of-order fact arriving after the job * already reached its terminal state (Req 7.1/7.2). The guard inside * apply_job_update_transition() is state-based (current lifecycle stage * vs. requested next stage) — it does not consult the LSN at all, so * this proves the state guard alone stops the regression regardless of * what LSN the stale entry carries (LSN-order enforcement is the apply * pump's job, task 6.1, out of this task's boundary). Use an LSN lower * than FINISH's to make the "out-of-order" framing literal. */ wal_jobstate_t stale_start = { .jobid = 703, .state = JOB_STAT_RUN }; int rc = apply_committed_entry(&g_tari_p, WAL_JOB_START, 250, &stale_start, sizeof(stale_start)); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "a guard-rejected stale START is a success no-op, not an apply " "failure"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, "the job must remain in its FINISH-terminal state after the stale " "START is applied"); TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_tari_hook_fires, "the stale, guard-rejected START must fire NO notification — the " "count must stay exactly where FINISH left it"); tari_teardown(); printf("test_tari_finish_then_stale_start_preserves_terminal_state_via_" "dispatch: PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_apply_reapply_idempotency(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_tari_job_submit_reapplied_thrice_is_idempotent); RUN_TEST(test_tari_job_state_transition_reapplied_identical_is_idempotent); RUN_TEST( test_tari_finish_then_stale_start_preserves_terminal_state_via_dispatch); }