lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test(raft-apply-decoupling): idempotent and ordered re-application of apply seam (task 2.4)

Browse source 
Add test_apply_reapply_idempotency.c (3 tests, no production changes):
triple-reapplication of an identical job submit with hook-count and
pointer-identity assertions across every call, the NOOP_DUPLICATE
guard branch (untested by task 2.2), and the task's named scenario --
real SUBMIT then real FINISH then a stale out-of-order START, all
through the real apply_committed_entry() dispatch -- confirming the
job stays terminal with no spurious notification.

Closes out task-group 2 (shared apply seam for committed entries).
This commit is contained in:
Rongsong Shen 2026-07-04 00:44:55 +00:00
parent 50d016d905
commit 8febd663d5
4 changed files with 386 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/meson.build
View file

@ -160,6 +160,7 @@ controller_unit_test_src = [
'testsuite/test_apply_seam.c',
'testsuite/test_apply_committed_entry.c',
'testsuite/test_replay_apply_seam.c',
'testsuite/test_apply_reapply_idempotency.c',
'ctrlstatus.c',
'commit.c',
'job.c',

2
src/controller/testsuite/controller_test.c
View file

@ -80,6 +80,7 @@ extern void test_persist_snapshot_entities(void);
extern void test_apply_seam(void);
extern void test_apply_committed_entry(void);
extern void test_replay_apply_seam(void);
extern void test_apply_reapply_idempotency(void);
void setUp(void) { return; }
@ -229,6 +230,7 @@ int main(int argc, char* argv[])
TEST(test_apply_seam);
TEST(test_apply_committed_entry);
TEST(test_replay_apply_seam);
TEST(test_apply_reapply_idempotency);
return UNITY_END();
}

380
src/controller/testsuite/test_apply_reapply_idempotency.c Normal file
View file

@ -0,0 +1,380 @@
/* 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 <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <string.h>
#include <unity/unity.h>
#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(0, apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT,
100, payload,
(uint32_t)plen));
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(0,
apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT, 100, payload,
(uint32_t)plen));
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(0, apply_committed_entry(&g_tari_p, WAL_JOB_SUBMIT,
300, payload,
(uint32_t)plen));
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)
{
UNITY_BEGIN();
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);
UNITY_END();
}