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 apply_wal_entry

Browse source 
Adds case WAL_JOB_ARRAY_SUBMIT to apply_wal_entry() (standby
catch-up dispatch), calling restore_job_array() hook-free, matching
the existing WAL_JOB_SUBMIT case's shape exactly.

Review found apply_wal_entry() has zero production call sites -- a
2026-07-10 refactor (ha_standby_advance_committed_and_apply) moved
standby entity materialization onto the same apply_committed_entry()
pump the primary uses; the code comment claiming otherwise predated
that refactor and was never updated. Requirement 3.1/3.2/4.1 are
fully delivered by task 3.2 alone. This task's wiring is kept as
defensive parity with WAL_JOB_SUBMIT's existing, equally-dead case
in the same switch, not as an independent delivery path -- design.md
and tasks.md corrected accordingly.

Task 3.3 of job-array-single-wal-record.
This commit is contained in:
Rongsong Shen 2026-07-15 02:30:06 +00:00
parent 0c4573b54a
commit b0c7cfe548
4 changed files with 260 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/controller/persist.c
View file

@ -2627,6 +2627,27 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
restore_job(p->cluster, 0, payload, payload_len, lsn);
break;
case WAL_JOB_ARRAY_SUBMIT:
/* job-array-single-wal-record task 3.3 (design.md "Array Apply
* Expansion"): the array-record analog of WAL_JOB_SUBMIT's
* restore_job() call directly above, dispatched from this SEPARATE,
* still-live standby catch-up / replicated-apply seam so it can
* never diverge from apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT
* case (Req 3.1, 3.2, 4.1). Mirrors the single-job case's exact
* minimalism: no xlist_t, no hook fire — this dispatcher is
* deliberately hook-free (standby catch-up does not notify the
* local scheduler/API server; that is the primary's job, done via
* apply_committed_entry()). out_new_jobs is not needed here, so
* NULL is passed (restore_job_array()'s out_new_jobs parameter is
* optional — see test_rja_out_new_jobs_is_optional,
* test_job_array_apply.c). The return value is intentionally
* discarded, matching every other case in this switch today (task
* 4.1 — Live Apply Resource Fence — is what later makes this
* function's callers respect a negative rc; not this task's
* boundary). */
restore_job_array(p->cluster, payload, payload_len, lsn, NULL);
break;
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:

242
src/controller/testsuite/test_job_array_apply.c
View file

@ -1,12 +1,17 @@
/* test_job_array_apply.c — job-array-single-wal-record task 3.1
/* test_job_array_apply.c — job-array-single-wal-record tasks 3.1 / 3.3
*
* Unit tests for restore_job_array() (job.c), the shared N-way array
* expansion primitive (design.md "Array Apply Expansion" / Req 2.1, 2.2,
* 2.3, 3.3, 4.1, 5.1, 5.2, 5.3, 9.3, 9.6). Calls restore_job_array()
* DIRECTLY -- it is not wired into apply_committed_entry()/apply_wal_entry()
* yet (that is task 3.2/3.3's boundary, not this one's).
* 2.3, 3.3, 4.1, 5.1, 5.2, 5.3, 9.3, 9.6), PLUS (task 3.3, added below the
* task-3.1 tests) dispatch-level tests proving apply_wal_entry()'s new
* WAL_JOB_ARRAY_SUBMIT case -- the standby catch-up / replicated-apply
* dispatcher, a SEPARATE, still-live function from apply_committed_entry()
* (design.md "Existing Architecture Analysis" table) -- expands a committed
* array record identically to apply_committed_entry() while staying
* hook-free, matching that dispatcher's existing single-job case (Req 3.1,
* 3.2, 4.1).
*
* Covers:
* Covers (task 3.1, restore_job_array() called DIRECTLY):
* - N-way expansion produces the correct contiguous jobid set, each
* element tagged with entry_lsn (Req 5.1, 5.2, 5.3).
* - Idempotent whole-record re-apply is a no-op (-EEXIST, nothing new).
@ -17,19 +22,32 @@
* the failed index), and a subsequent successful retry (seam cleared)
* produces the complete, non-duplicated set (Req 9.3, 9.4, 9.6).
*
* Fixture: a minimal cluster_t (jobmgr only -- the only field
* restore_job_array reads) mirrors test_apply_committed_entry.c's "bare
* fixture, no more than the function under test needs" philosophy.
* Covers (task 3.3, driving apply_wal_entry()/apply_committed_entry()):
* - apply_wal_entry() constructs every array element and fires NO
* notification hook (Req 3.1, 3.2).
* - apply_committed_entry() and apply_wal_entry() applied to the SAME
* committed array record produce byte-for-byte identical resulting job
* state (Req 3.1, 3.2, 4.1).
*
* Fixture (task 3.1 tests): a minimal cluster_t (jobmgr only -- the only
* field restore_job_array reads) mirrors test_apply_committed_entry.c's
* "bare fixture, no more than the function under test needs" philosophy.
* job_obj_init()/xstring_table_init() are process-global (shared across
* every test file in the controller_unit_test binary) -- guarded
* idempotently, never torn down here, matching every sibling fixture in
* this suite (test_apply_committed_entry.c, test_jobj_hardening.c).
*
* Fixture (task 3.3 tests, tajd_* below): a richer cluster_t (jobmgr +
* hooksys) since these tests drive the dispatch functions -- not
* restore_job_array() directly -- and must be able to prove the ABSENCE of
* a hook fire. Mirrors test_apply_committed_entry.c's tace_setup shape.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
@ -39,7 +57,10 @@
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xlist.h"
#include "xstring.h"
@ -477,6 +498,209 @@ static void test_rja_out_new_jobs_is_optional(void)
printf("test_rja_out_new_jobs_is_optional: PASS\n");
}
/* ── Task 3.3: apply_wal_entry() standby catch-up dispatch ─────────────── */
static cluster_t g_tajd_c;
static persist_t g_tajd_p;
static int g_tajd_hook_fires;
static int tajd_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_tajd_hook_fires++;
return 0;
}
static void tajd_setup(void)
{
kvmgr_params_t int_params
= { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 };
if (!g_taj_xstring_ready) {
TEST_ASSERT_TRUE(xstring_table_init(64) >= 0);
g_taj_xstring_ready = 1;
}
if (!g_taj_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_taj_jobobj_ready = 1;
}
memset(&g_tajd_c, 0, sizeof(g_tajd_c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_tajd_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_tajd_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_tajd_c.jobmgr);
memset(&g_tajd_p, 0, sizeof(g_tajd_p));
g_tajd_p.cluster = &g_tajd_c;
g_tajd_hook_fires = 0;
#ifdef XCE_TESTING
job_array_apply_test_clear_force_fail();
job_array_apply_test_clear_bulk_insert_force_fail();
#endif
}
static void tajd_teardown(void)
{
#ifdef XCE_TESTING
job_array_apply_test_clear_force_fail();
job_array_apply_test_clear_bulk_insert_force_fail();
#endif
xce_hook_fini(&g_tajd_c.hooksys);
if (g_tajd_c.jobmgr)
dereference(g_tajd_c.jobmgr);
}
static job_t* tajd_find_job(cluster_t* c, uint64_t jobid)
{
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
int found = key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
return (found > 0) ? j : NULL;
}
/* apply_wal_entry()'s WAL_JOB_ARRAY_SUBMIT case constructs every element
* through restore_job_array() and fires NO notification -- mirroring the
* existing WAL_JOB_SUBMIT case's hook-free behavior at this same dispatch
* seam (standby catch-up does not notify the local scheduler/API server;
* that is apply_committed_entry()'s job, Req 3.1, 3.2). */
static void test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook(void)
{
tajd_setup();
const uint64_t base_jobid = 78000;
const uint64_t array_size = 3;
const uint64_t entry_lsn = 80;
job_submit_data_t spec = taj_make_spec();
uint8_t* payload = NULL;
size_t len = 0;
TEST_ASSERT_EQUAL_INT(0,
taj_encode_job_submit(base_jobid, array_size, &spec, &payload, &len));
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_tajd_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
tajd_spy_hook, NULL));
int rc = apply_wal_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn,
payload, (uint32_t)len);
TEST_ASSERT_EQUAL_INT(0, rc);
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = tajd_find_job(&g_tajd_c, base_jobid + i);
TEST_ASSERT_NOT_NULL_MESSAGE(j,
"every array index must be constructed via restore_job_array"
" through apply_wal_entry's standby catch-up dispatch");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(entry_lsn, j->submit_lsn,
"each constructed element must be tagged with the entry's lsn");
}
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tajd_hook_fires,
"apply_wal_entry's standby catch-up dispatch must never fire a"
" notification hook, matching its existing single-job WAL_JOB_SUBMIT"
" case");
free(payload);
tajd_teardown();
printf("test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook: "
"PASS\n");
}
/* The task 3.3 Observable: applying the SAME committed array record through
* apply_committed_entry() (live apply / boot replay) and through
* apply_wal_entry() (standby catch-up) must produce byte-for-byte identical
* resulting job state either way -- both dispatchers share the one
* restore_job_array() expansion primitive, so they can never diverge (Req
* 3.1, 3.2, 4.1). Uses two independent fixture instances (setup/apply/
* snapshot/teardown, run twice) rather than two live cluster_t at once, to
* keep each dispatch call's state isolated and the comparison unambiguous. */
typedef struct {
uint64_t jobid;
uint32_t state;
uint64_t submit_lsn;
char queue[32];
char user[32];
} tajd_snapshot_t;
static void tajd_snapshot_array(cluster_t* c, uint64_t base_jobid,
uint64_t array_size, tajd_snapshot_t* out)
{
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = tajd_find_job(c, base_jobid + i);
TEST_ASSERT_NOT_NULL(j);
out[i].jobid = j->job.jobid;
out[i].state = j->job.state;
out[i].submit_lsn = j->submit_lsn;
out[i].queue[0] = '\0';
out[i].user[0] = '\0';
if (j->job.submit != NULL) {
if (j->job.submit->queue != NULL)
strncpy(out[i].queue, j->job.submit->queue,
sizeof(out[i].queue) - 1);
if (j->job.submit->user != NULL)
strncpy(out[i].user, j->job.submit->user,
sizeof(out[i].user) - 1);
}
}
}
static void
test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_identical(void)
{
const uint64_t base_jobid = 79000;
const uint64_t array_size = 4;
const uint64_t entry_lsn = 90;
job_submit_data_t spec = taj_make_spec();
uint8_t* payload = NULL;
size_t len = 0;
TEST_ASSERT_EQUAL_INT(0,
taj_encode_job_submit(base_jobid, array_size, &spec, &payload, &len));
tajd_snapshot_t via_committed[4];
tajd_snapshot_t via_wal[4];
/* Path 1: live-apply / boot-replay dispatch. */
tajd_setup();
int rc1 = apply_committed_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn,
payload, (uint32_t)len);
TEST_ASSERT_EQUAL_INT((int)array_size, rc1);
tajd_snapshot_array(&g_tajd_c, base_jobid, array_size, via_committed);
tajd_teardown();
/* Path 2: standby catch-up / replicated-apply dispatch. */
tajd_setup();
int rc2 = apply_wal_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn,
payload, (uint32_t)len);
TEST_ASSERT_EQUAL_INT(0, rc2);
tajd_snapshot_array(&g_tajd_c, base_jobid, array_size, via_wal);
tajd_teardown();
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_EQUAL_UINT64_MESSAGE(via_committed[i].jobid,
via_wal[i].jobid, "jobid must match across both dispatchers");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(via_committed[i].state,
via_wal[i].state, "state must match across both dispatchers");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(via_committed[i].submit_lsn,
via_wal[i].submit_lsn,
"submit_lsn must match across both dispatchers");
TEST_ASSERT_EQUAL_STRING_MESSAGE(via_committed[i].queue,
via_wal[i].queue, "queue must match across both dispatchers");
TEST_ASSERT_EQUAL_STRING_MESSAGE(via_committed[i].user,
via_wal[i].user, "user must match across both dispatchers");
}
free(payload);
printf("test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_"
"identical: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_job_array_apply(void)
@ -491,5 +715,7 @@ void test_job_array_apply(void)
#endif
RUN_TEST(test_rja_null_or_empty_payload_returns_error);
RUN_TEST(test_rja_out_new_jobs_is_optional);
RUN_TEST(test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook);
RUN_TEST(test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_identical);
UNITY_END();
}