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

feat(job-array-single-wal-record): wire array expansion into live apply

Browse source 
Adds case WAL_JOB_ARRAY_SUBMIT to apply_committed_entry() (the live
apply pump / boot replay dispatch seam): calls restore_job_array()
(task 3.1) instead of restore_job(), and on success fires
HOOK_DETAIL_JOB_NEW exactly once carrying every newly created
element in restore_job_array()'s own out_new_jobs list -- never once
per element. No hook fires on an idempotent no-op (-EEXIST) or any
genuine construction failure, honestly propagating task 3.1's
documented bulk-insert-phase residual gap rather than papering over
it. apply_wal_entry() (standby catch-up) is untouched -- task 3.3.

Task 3.2 of job-array-single-wal-record.
This commit is contained in:
Rongsong Shen 2026-07-14 14:46:03 +00:00
parent 7c755dc4d7
commit 0c4573b54a
3 changed files with 281 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/job-array-single-wal-record/tasks.md
View file

@ -60,7 +60,7 @@
- _Boundary: Array Apply Expansion_
- _Depends: 1.3_
- [ ] 3.2 Wire array expansion into live apply
- [x] 3.2 Wire array expansion into live apply
- Call 3.1's primitive from the live-apply/boot-replay dispatch path for the new WAL record type
- After a successful expansion, fire the existing job-created notification exactly once, carrying every newly created element in one list — never once per element
- Observable: applying one committed array record produces exactly one notification event carrying all N elements

44
src/controller/persist.c
View file

@ -2776,6 +2776,50 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
return 1;
}
case WAL_JOB_ARRAY_SUBMIT: {
/* job-array-single-wal-record task 3.2 (design.md "Array Apply
* Expansion"): the array-record analog of WAL_JOB_SUBMIT's
* restore_job() above, dispatched from this SAME shared seam so live
* apply, boot replay (both callers of this function), and (task
* 3.3) standby catch-up can never diverge in how they expand a
* committed array record (Req 3.1, 3.2).
*
* restore_job_array()'s return generalizes restore_job()'s
* three-way contract to N elements: a positive count = the number
* of genuinely NEW elements this call constructed (this is exactly,
* and only, what the hook fire below carries — Req 3.4); -EEXIST =
* every index in the record already existed (idempotent
* whole-record no-op, Req 3.3) — no hook; a negative errno = a
* genuine construction/resource-exhaustion failure (Req 9.4) — no
* hook, propagated so persist_apply_pump_run() (task 4.1) can halt
* and fence on it rather than silently skipping it. */
xlist_t out_new_jobs;
memset(&out_new_jobs, 0, sizeof(out_new_jobs));
int rc = restore_job_array(c, payload, len, lsn, &out_new_jobs);
if (rc == -EEXIST)
return 0; /* idempotent re-apply (replay / resync) — no hook */
if (rc < 0)
return rc;
/* Exactly ONE fire carrying every newly created element in one
* list — never once per element (Req 3.4). out_new_jobs is already
* the precise newly-created set restore_job_array() built; no
* re-derivation via key_find needed. */
t_apply_seam_notifying = true;
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_NEW, c, &out_new_jobs);
t_apply_seam_notifying = false;
/* Only the list's own backing array is released here — ownership of
* every job_t* it holds already transferred to c->jobmgr inside
* restore_job_array()'s bulk-insert pass, mirroring
* test_job_array_apply.c's own xlist_fini(&out) usage after a
* successful call. */
xlist_fini(&out_new_jobs);
return rc;
}
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_STOP:

236
src/controller/testsuite/test_apply_committed_entry.c
View file

@ -66,6 +66,31 @@ static int tace_spy_hook(int type, int situation, int detail, void* data,
return 0;
}
/* ── Array-aware spy hook: like tace_spy_hook (counts every AFTER_EVENT
* firing into the same g_tace_hook_fires counter), additionally recording
* the xlist_t element count seen on the most recent HOOK_DETAIL_JOB_NEW
* firing -- this is what lets a test distinguish "fired once carrying N
* elements" from "fired N times carrying one element each" (task 3.2,
* Req 3.4). va_arg order mirrors test_apply_pump.c's tap_spy_hook: the
* WAL_JOB_SUBMIT/WAL_JOB_ARRAY_SUBMIT HOOK_DETAIL_JOB_NEW call shape is
* (cluster_t*, xlist_t*) -- consume the leading cluster_t* first. ────────── */
static int g_tace_last_new_list_count;
static int tace_array_spy_hook(int type, int situation, int detail,
void* data, va_list ap)
{
(void)type; (void)data;
if (situation != HOOK_AFTER_EVENT)
return 0;
g_tace_hook_fires++;
if (detail == HOOK_DETAIL_JOB_NEW) {
(void)va_arg(ap, cluster_t*);
xlist_t* jlist = va_arg(ap, xlist_t*);
g_tace_last_new_list_count = jlist ? jlist->count : -1;
}
return 0;
}
static void tace_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
@ -166,6 +191,44 @@ static int tace_encode_worker(const xce_worker_t* w, uint8_t** out,
return 0;
}
/* Encodes a (base_jobid, array_size, spec) JobSubmit payload -- the exact
* shape a committed WAL_JOB_ARRAY_SUBMIT record carries (mirrors
* test_job_array_apply.c's taj_encode_job_submit and this file's own
* tace_encode_job/tace_encode_worker pattern above). Caller frees *out. */
static int tace_encode_job_submit_array(uint64_t base_jobid,
uint64_t array_size,
job_submit_data_t* spec,
uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
struct JobSubmit cjs;
JobSubmit_ptr jsp;
xce_job_submit_t s;
memset(&s, 0, sizeof(s));
s.jobid = base_jobid;
s.array_size = array_size;
s.submit = spec;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_JobSubmit(cs, &cjs, &s);
jsp = new_JobSubmit(cs);
write_JobSubmit(&cjs, jsp);
capn_setp(capn_root(&c), 0, jsp.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* tace_find_job(uint64_t jobid)
{
@ -433,6 +496,174 @@ static void test_ace_worker_ctrl_join_restores_and_notifies(void)
printf("test_ace_worker_ctrl_join_restores_and_notifies: PASS\n");
}
/* ── WAL_JOB_ARRAY_SUBMIT (job-array-single-wal-record task 3.2) ────────── */
static job_submit_data_t tace_array_make_spec(void)
{
job_submit_data_t s;
memset(&s, 0, sizeof(s));
s.queue = (char*)"normal";
s.user = (char*)"tester";
s.mintasks = 1;
s.maxtasks = 1;
s.job_name = (char*)"array-job";
return s;
}
static void test_ace_job_array_submit_constructs_all_and_notifies_once(void)
{
tace_setup();
const uint64_t base_jobid = 5100;
const uint64_t array_size = 4;
job_submit_data_t spec = tace_array_make_spec();
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_job_submit_array(base_jobid,
array_size, &spec,
&payload, &plen));
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
tace_array_spy_hook, NULL));
int rc = apply_committed_entry(&g_tace_p, WAL_JOB_ARRAY_SUBMIT, 60,
payload, (uint32_t)plen);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, rc,
"a genuine full-array creation must return the count of newly "
"created elements (restore_job_array(), not restore_job(), was "
"invoked)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"applying one committed array record must fire exactly one "
"notification event, never one per element");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, g_tace_last_new_list_count,
"the single notification must carry all N newly created elements "
"in one list, not a 1-element list fired N times");
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = tace_find_job(base_jobid + i);
TEST_ASSERT_NOT_NULL_MESSAGE(j,
"every array index must be constructed via restore_job_array");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(60, j->submit_lsn,
"each constructed element must be tagged with the entry's lsn");
}
free(payload);
tace_teardown();
printf("test_ace_job_array_submit_constructs_all_and_notifies_once: "
"PASS\n");
}
static void test_ace_job_array_submit_idempotent_replay_no_extra_notify(void)
{
tace_setup();
const uint64_t base_jobid = 5200;
const uint64_t array_size = 3;
job_submit_data_t spec = tace_array_make_spec();
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_job_submit_array(base_jobid,
array_size, &spec,
&payload, &plen));
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
tace_array_spy_hook, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size,
apply_committed_entry(&g_tace_p, WAL_JOB_ARRAY_SUBMIT, 61, payload,
(uint32_t)plen),
"the first apply must genuinely construct every element");
TEST_ASSERT_EQUAL_INT(1, g_tace_hook_fires);
/* Re-apply the SAME committed record (replay / resync redelivery): every
* index already exists, so this is a whole-record idempotent no-op --
* restore_job_array() returns -EEXIST, apply_committed_entry() must
* translate that to 0 and fire NO further notification (Req 3.3). */
int rc2 = apply_committed_entry(&g_tace_p, WAL_JOB_ARRAY_SUBMIT, 61,
payload, (uint32_t)plen);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc2,
"an idempotent whole-record re-apply must return 0, not a genuine "
"creation count");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"an idempotent duplicate array re-apply must not fire a second "
"notification");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL(tace_find_job(base_jobid + i));
}
free(payload);
tace_teardown();
printf("test_ace_job_array_submit_idempotent_replay_no_extra_notify: "
"PASS\n");
}
#ifdef XCE_TESTING
static void test_ace_job_array_submit_forced_failure_fires_no_hook(void)
{
tace_setup();
const uint64_t base_jobid = 5300;
const uint64_t array_size = 5;
job_submit_data_t spec = tace_array_make_spec();
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_job_submit_array(base_jobid,
array_size, &spec,
&payload, &plen));
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
tace_array_spy_hook, NULL);
/* Force restore_job_array()'s scratch-build loop to fail constructing
* element index 2 -- simulates a resource-exhaustion apply failure
* (Req 9.4). apply_committed_entry() must propagate the negative errno
* and must NOT fire HOOK_DETAIL_JOB_NEW at all. */
job_array_apply_test_set_force_fail_at(2);
int rc = apply_committed_entry(&g_tace_p, WAL_JOB_ARRAY_SUBMIT, 62,
payload, (uint32_t)plen);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"a genuine construction failure must propagate as a negative errno");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tace_hook_fires,
"a failed array expansion must fire NO notification");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NULL_MESSAGE(tace_find_job(base_jobid + i),
"an all-or-nothing failed attempt must leave nothing visible");
}
/* Clear the seam and retry the identical committed record: this time
* construction succeeds fully, the hook fires exactly once with all N
* elements, and the earlier failed attempt did not poison this one. */
job_array_apply_test_clear_force_fail();
int rc2 = apply_committed_entry(&g_tace_p, WAL_JOB_ARRAY_SUBMIT, 62,
payload, (uint32_t)plen);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, rc2,
"a retry after the injected failure clears must construct every "
"element");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"the successful retry must fire exactly one notification");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, g_tace_last_new_list_count,
"the retry's single notification must carry all N elements");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL(tace_find_job(base_jobid + i));
}
free(payload);
tace_teardown();
printf("test_ace_job_array_submit_forced_failure_fires_no_hook: PASS\n");
}
#endif /* XCE_TESTING */
/* ── Defensive / boundary ─────────────────────────────────────────────── */
static void test_ace_null_persist_returns_error(void)
@ -453,6 +684,11 @@ void test_apply_committed_entry(void)
RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify);
RUN_TEST(test_ace_queue_ctrl_updates_and_notifies);
RUN_TEST(test_ace_worker_ctrl_join_restores_and_notifies);
RUN_TEST(test_ace_job_array_submit_constructs_all_and_notifies_once);
RUN_TEST(test_ace_job_array_submit_idempotent_replay_no_extra_notify);
#ifdef XCE_TESTING
RUN_TEST(test_ace_job_array_submit_forced_failure_fires_no_hook);
#endif
RUN_TEST(test_ace_null_persist_returns_error);
UNITY_END();
}