diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index f5d3ae2d..ee5e61ff 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -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 diff --git a/src/controller/persist.c b/src/controller/persist.c index 18dbc058..1275c279 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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: diff --git a/src/controller/testsuite/test_apply_committed_entry.c b/src/controller/testsuite/test_apply_committed_entry.c index 6cba207d..864f1c6f 100644 --- a/src/controller/testsuite/test_apply_committed_entry.c +++ b/src/controller/testsuite/test_apply_committed_entry.c @@ -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(); }