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

feat(job-array-single-wal-record): implement single-record array append/commit

Browse source 
Adds persist_submit_job_array_async(): encodes one JobSubmit-shaped
payload (base jobid, array size, one shared spec), appends exactly
one WAL_JOB_ARRAY_SUBMIT entry, and registers exactly one commit
continuation regardless of array size -- reuses the existing
job_commit_adapter_ctx_t/job_commit_resolve_adapter machinery, no
per-element loop. Not yet wired into handle_jobsubmit (task 2.2).

Also fixes a test-helper bug (WAL offset back-computed from the
post-append cursor, only valid for zero-payload entries) that
corrupted state across tests when exercising a nonzero-payload
array-submit entry.

Task 2.1 of job-array-single-wal-record.
This commit is contained in:
Rongsong Shen 2026-07-14 09:50:00 +00:00
parent 76862bd9aa
commit 7c4afbbb83
4 changed files with 559 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

363
src/controller/testsuite/test_commit_barrier_rollout.c
View file

@ -1724,6 +1724,364 @@ static void test_config_ack_wire_path_completes_barrier(void)
printf("test_config_ack_wire_path_completes_barrier: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* job-array-single-wal-record task 2.1 — persist_submit_job_array_async:
* the single-record array append and commit path (Req 1.1, 1.2, 1.3, 4.2,
* 9.2). Fixture mirrors test_fact_hook_append_is_retained_and_nonblocking
* above (standalone persist_t, no cluster_t -- p->cluster stays NULL so
* the append routes through the persist_wal_async fallback branch inside
* append_job_array_wal_entry_local; real tmpdir WAL; wired commit engine
* with 2 stub standbys; ACKs driven through repl_on_ack). commit_engine_init
* here never wires a vtaskmgr pool (ce->pool stays NULL), so
* commit_resolve_pending's continuation dispatch is the synchronous,
* same-thread inline fallback (cerebrum precedent: "the NO-POOL
* inline-dispatch fallback calls fn() synchronously ON THE CALLING
* THREAD") -- a covering repl_on_ack call resolves a registered
* continuation before it returns, so no polling is needed anywhere below.
*
* Deliberately calls persist_submit_job_array_async() DIRECTLY -- job.c's
* handle_jobsubmit is not wired to this function yet (that is task 2.2's
* boundary, not this one's); nothing in job.c/handle_jobsubmit is touched
* here.
* ════════════════════════════════════════════════════════════════════════ */
static persist_t g_jap; /* stub persist for array-submit tests */
static commit_engine_t g_jace;
static int g_jaep_a, g_jaep_b; /* dummy standby endpoints */
static char g_jap_tmpdir[256];
static void jap_setup(uint32_t sync_timeout_ms)
{
persist_config_t cfg;
memset(&g_jap, 0, sizeof(g_jap));
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = sync_timeout_ms;
snprintf(g_jap_tmpdir, sizeof(g_jap_tmpdir), "/tmp/xce_jap21_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_jap_tmpdir));
persist_set_workdir(&g_jap, g_jap_tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_jap, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_jap));
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until
* a test explicitly delivers them via repl_on_ack. */
TEST_ASSERT_TRUE(
repl_add_standby(&g_jap.repl, &g_jaep_a, "jasb_a", 7420) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_jap.repl, &g_jaep_b, "jasb_b", 7421) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_jace, &g_jap, 3, 8));
repl_set_commit_engine(&g_jap.repl, &g_jace);
repl_broadcast_send_hook = cbr_drop_send_hook;
/* Reserved-0 contract: a fresh store starts current_lsn=1 -- the first
* array record lands at LSN 1 with no priming append needed. */
TEST_ASSERT_EQUAL_UINT64(1, g_jap.current_lsn);
}
static void jap_teardown(void)
{
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_jace);
persist_stop(&g_jap);
persist_fini(&g_jap);
/* tmpdir left on disk — /tmp is ephemeral */
}
/* Minimal job_submit_data_t + xce_job_submit_t base spec -- same shape as
* cbrp_arrange_jobsubmit (test_commit_barrier_reply.c) and
* test_data_job_array_variant_round_trip (test_e2e_capnp.c); start_info
* left NULL, which that round-trip test already proves
* encode_JobStartInfo_ptr handles safely. */
static job_submit_data_t g_jap_submit;
static xce_job_submit_t g_jap_spec;
static void jap_arrange_base_spec(void)
{
memset(&g_jap_submit, 0, sizeof(g_jap_submit));
g_jap_submit.queue = (char *)"normal";
g_jap_submit.user = (char *)"tester";
g_jap_submit.mintasks = 1;
g_jap_submit.maxtasks = 1;
g_jap_submit.start_info = NULL;
g_jap_submit.job_name = (char *)"array-job";
memset(&g_jap_spec, 0, sizeof(g_jap_spec));
/* Deliberately NOT set to a test's real base_jobid/array_size: proves
* persist_submit_job_array_async uses its own explicit base_jobid/
* array_size PARAMETERS, never these same-shaped struct fields. */
g_jap_spec.jobid = 0;
g_jap_spec.array_size = 0;
g_jap_spec.submit = &g_jap_submit;
}
/* Read the WAL entry header located at explicit segment offset `at` back
* from the active segment file. Unlike read_last_wal_hdr in
* test_ha_role_lifecycle.c (which back-computes `seg_offset -
* sizeof(hdr)` and is only ever used for zero-payload WAL_NOOP entries,
* where that arithmetic happens to land on the header), this helper takes
* the header's start offset EXPLICITLY -- the caller snapshots
* p->seg_offset immediately before the append it wants to read back.
* Back-computing from the post-append p->seg_offset is only valid when
* payload_len == 0; every WAL_JOB_ARRAY_SUBMIT entry here carries a
* nonzero capnp payload, so `seg_offset - sizeof(hdr)` would land inside
* the payload bytes, not the header (caught by task 2.1 RED: read garbage
* decoded as a bogus type). */
static int jap_read_wal_hdr_at(persist_t *p, off_t at, wal_entry_hdr_t *out)
{
if (at < 0 || p->seg_offset < (uint64_t)at + sizeof(wal_entry_hdr_t))
return -ENOENT;
int fd = open(p->seg_path, O_RDONLY);
if (fd < 0)
return -errno;
ssize_t r = pread(fd, out, sizeof(*out), at);
close(fd);
if (r != (ssize_t)sizeof(*out))
return -EIO;
return 0;
}
/* Read the header AND payload bytes of the WAL entry located at explicit
* segment offset `at` back from the active segment file. Caller must
* free(*out_payload) (NULL/0-len payload leaves *out_payload NULL). */
static int jap_read_wal_entry_at(persist_t *p, off_t at,
wal_entry_hdr_t *out_hdr,
uint8_t **out_payload)
{
int rc = jap_read_wal_hdr_at(p, at, out_hdr);
if (rc != 0)
return rc;
if (out_hdr->payload_len == 0) {
*out_payload = NULL;
return 0;
}
int fd = open(p->seg_path, O_RDONLY);
if (fd < 0)
return -errno;
off_t payload_at = at + (off_t)sizeof(wal_entry_hdr_t);
uint8_t *buf = (uint8_t *)malloc(out_hdr->payload_len);
TEST_ASSERT_NOT_NULL(buf);
ssize_t r = pread(fd, buf, out_hdr->payload_len, payload_at);
close(fd);
if (r != (ssize_t)out_hdr->payload_len) {
free(buf);
return -EIO;
}
*out_payload = buf;
return 0;
}
/* Notify-callback capture (persist_job_commit_notify_fn). */
typedef struct {
int call_count;
uint64_t last_lsn;
int last_rc;
} jap_notify_obs_t;
static jap_notify_obs_t g_jap_obs;
static void jap_notify_reset(void)
{
memset(&g_jap_obs, 0, sizeof(g_jap_obs));
}
static void jap_notify_fn(void *ctx, uint64_t lsn, int rc)
{
jap_notify_obs_t *obs = (jap_notify_obs_t *)ctx;
obs->call_count++;
obs->last_lsn = lsn;
obs->last_rc = rc;
}
/* ── One WAL append, one commit continuation, decodable back to the same
* (base_jobid, array_size, submit spec) — Req 1.1, 1.2, 1.3, 4.2, 9.2 ── */
static void test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation(void)
{
const uint64_t base_jobid = 9500;
const uint64_t array_size = 5;
uint64_t lsn_before, out_lsn = 0;
int rc;
off_t entry_off;
wal_entry_hdr_t hdr;
uint8_t *payload = NULL;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
lsn_before = g_jap.current_lsn;
entry_off = (off_t)g_jap.seg_offset;
/* Act: append the array record; the continuation is registered but not
* yet resolved (no covering ACK delivered yet). */
rc = persist_submit_job_array_async(&g_jap, base_jobid, array_size,
&g_jap_spec, &out_lsn,
jap_notify_fn, &g_jap_obs);
/* Assert: exactly one WAL entry was appended -- current_lsn advanced by
* exactly 1, regardless of array_size=5 */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a valid array submission must append and register successfully");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, out_lsn,
"the array record must land at the pre-append current_lsn");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn,
"exactly ONE WAL entry must be appended for the whole array, "
"not one per element");
/* ...and it is the ONE entry on disk, typed WAL_JOB_ARRAY_SUBMIT,
* decodable back to the same (base_jobid, array_size, submit spec) --
* NOT g_jap_spec's own (deliberately mismatched) jobid/array_size
* fields, proving the function used its explicit parameters */
TEST_ASSERT_EQUAL_INT(0,
jap_read_wal_entry_at(&g_jap, entry_off, &hdr, &payload));
TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)WAL_JOB_ARRAY_SUBMIT, hdr.type,
"the array record's WAL type must be WAL_JOB_ARRAY_SUBMIT");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(out_lsn, hdr.lsn,
"the on-disk entry's lsn must match the returned out_lsn");
{
struct capn cctx;
TEST_ASSERT_EQUAL_INT(0,
capn_init_mem(&cctx, payload, hdr.payload_len, 0));
JobSubmit_ptr jp;
jp.p = capn_getp(capn_root(&cctx), 0, 1);
TEST_ASSERT_TRUE(jp.p.type != CAPN_NULL);
struct JobSubmit wire;
memset(&wire, 0, sizeof(wire));
read_JobSubmit(&wire, jp);
xce_job_submit_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_JobSubmit(&decoded, &wire);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, decoded.jobid,
"decoded base jobid must match the base_jobid PARAMETER");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(array_size, decoded.array_size,
"decoded array_size must match the array_size PARAMETER");
TEST_ASSERT_NOT_NULL(decoded.submit);
TEST_ASSERT_EQUAL_STRING("normal", decoded.submit->queue);
TEST_ASSERT_EQUAL_STRING("tester", decoded.submit->user);
free_JobSubmit(&decoded);
capn_free(&cctx);
}
free(payload);
/* ...exactly one continuation is registered -- not yet resolved before
* any ACK arrives */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count,
"the notify callback must not fire before a covering commit");
/* ...and fires EXACTLY ONCE, with rc==0, once a covering majority ACK
* resolves the single registered continuation */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count,
"the notify callback must fire EXACTLY ONCE for the whole array, "
"not once per element");
TEST_ASSERT_EQUAL_UINT64(out_lsn, g_jap_obs.last_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc,
"a majority-committed array record must resolve rc=0");
jap_teardown();
printf("test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation: PASS\n");
}
/* ── Encoded payload size does not depend on array_size (Req 7.1, 7.2) ──── */
static void test_job_array_submit_payload_size_independent_of_array_size(void)
{
wal_entry_hdr_t hdr_small, hdr_large;
uint64_t out_lsn = 0;
int rc;
off_t off_small, off_large;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
/* Act: array_size=2, then array_size=10000, SAME base spec -- both
* entries encoded through the identical code path in the same fixture
* so their on-disk payload lengths are directly comparable. */
off_small = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 9600, 2, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_small, &hdr_small));
jap_notify_reset();
off_large = (off_t)g_jap.seg_offset;
rc = persist_submit_job_array_async(&g_jap, 20000, 10000, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_large, &hdr_large));
/* Assert: the encoded payload's size does not depend on array_size --
* arraySize is a fixed-width uint64 field, not a per-element list */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len,
hdr_large.payload_len,
"the WAL_JOB_ARRAY_SUBMIT payload size must be IDENTICAL for "
"array_size=2 and array_size=10000 -- only the fixed-width "
"arraySize field differs, never the encoded byte length");
/* Cleanup: resolve both pending continuations with one covering ACK
* before teardown (hygiene — avoids leaving unresolved slots for
* commit_engine_fini to silently drop). */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap));
repl_on_ack(&g_jap.repl, &g_jaep_a, hdr_large.lsn);
jap_teardown();
printf("test_job_array_submit_payload_size_independent_of_array_size: PASS\n");
}
/* ── Invalid-argument rejection (defensive input validation) ────────────── */
static void test_job_array_submit_rejects_array_size_one_or_less(void)
{
uint64_t out_lsn = 0;
int rc;
/* Arrange */
jap_setup(2000);
jap_arrange_base_spec();
jap_notify_reset();
/* Act / Assert: array_size<=1 is a single-job submission's shape, not
* this function's precondition (design.md: "array_size > 1") -- reject
* before any WAL append, no continuation registered, no callback fired */
rc = persist_submit_job_array_async(&g_jap, 9700, 1, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc,
"array_size == 1 must be rejected -- not this function's job");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn,
"a rejected submission must never append a WAL entry");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count,
"a rejected submission's callback contract is 'never called', not "
"'called with an error' — the caller never registered anything");
rc = persist_submit_job_array_async(&g_jap, 9700, 0, &g_jap_spec,
&out_lsn, jap_notify_fn, &g_jap_obs);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc,
"array_size == 0 must be rejected");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn,
"a rejected submission must never append a WAL entry");
jap_teardown();
printf("test_job_array_submit_rejects_array_size_one_or_less: PASS\n");
}
/* ── Test group entry point ──────────────────────────────────────────────── */
void test_commit_barrier_rollout(void)
@ -1767,5 +2125,10 @@ void test_commit_barrier_rollout(void)
/* Task 5.2 remediation r1 — production wire-path ACK ingestion
* (completes the unlanded half of ha-config-bootstrap task 8.1) */
RUN_TEST(test_config_ack_wire_path_completes_barrier);
/* job-array-single-wal-record task 2.1 — persist_submit_job_array_async:
* single-record array append/commit path (Req 1.1, 1.2, 1.3, 4.2, 9.2) */
RUN_TEST(test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation);
RUN_TEST(test_job_array_submit_payload_size_independent_of_array_size);
RUN_TEST(test_job_array_submit_rejects_array_size_one_or_less);
UNITY_END();
}