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

feat(job-array-single-wal-record): implement restore_job_array primitive

Browse source 
Adds the shared N-way array expansion primitive: decodes a committed
WAL_JOB_ARRAY_SUBMIT payload, scratch-builds every element via a
throwaway kv-table (never touching c->jobmgr during construction),
then makes the whole set visible in one bulk pass under a single
write-lock acquisition -- no window where a concurrent reader can
observe a partially-constructed array. Idempotent per index (skips
already-existing jobids; whole-record no-op if all exist). Includes
an XCE_TESTING-guarded fault-injection seam. Not yet wired into any
live dispatch path (tasks 3.2/3.3).

Two review rounds: round 1 found the bulk-insert pass silently
claimed success even when key_add() itself could fail on allocation
exhaustion (c->jobmgr grows via realloc, no reservation primitive
exists anywhere in kv.c/kvmgr.c/zarray.c to pre-check this). Fixed
by checking key_add()'s return and reporting failure honestly
instead of a false success. The underlying table has no key_del, so
an already-inserted prefix on this specific failure mode can remain
genuinely visible while the call reports failure -- documented as a
residual gap in design.md and tasks.md for tasks 3.2/4.x to account
for (that prefix is never pushed to scheduler/API server, even on a
later successful retry, since it's never included in out_new_jobs).

Task 3.1 of job-array-single-wal-record.
This commit is contained in:
Rongsong Shen 2026-07-14 14:16:29 +00:00
parent 78eb983a74
commit 7c755dc4d7
7 changed files with 1346 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

27
src/controller/include/cluster.h
View file

@ -45,6 +45,33 @@ extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*);
extern int load_jobs(void*);
extern int restore_job(cluster_t*, uint64_t, const void*, size_t, uint64_t);
/* job-array-single-wal-record task 3.1 (design.md Array Apply Expansion):
* shared N-way expansion primitive for a committed WAL_JOB_ARRAY_SUBMIT
* payload -- see the doc comment on the definition (job.c) for the full
* atomicity/idempotency contract. out_new_jobs is optional; when non-NULL
* and the call genuinely creates new elements, it is populated (ownership
* transferred) with every newly created job_t* in index order. */
extern int restore_job_array(cluster_t*, const void*, size_t, uint64_t,
xlist_t*);
#ifdef XCE_TESTING
/* job-array-single-wal-record task 3.1: forces restore_job_array()'s
* scratch-build loop to fail construction at the given 0-based element
* index (simulating resource exhaustion), so a test can prove the
* all-or-nothing atomicity contract. Pass a negative index, or call the
* clear helper, to disable. Compiled out of production. */
extern void job_array_apply_test_set_force_fail_at(int index);
extern void job_array_apply_test_clear_force_fail(void);
/* job-array-single-wal-record task 3.1 remediation: forces
* restore_job_array()'s BULK-INSERT pass (the key_add(c->jobmgr, ...) loop,
* distinct from the scratch-build loop above) to fail at the given 0-based
* index WITHIN THAT PASS, simulating c->jobmgr's backing store hitting a
* genuine realloc() OOM while growing under the write lock. Lets a test
* prove this failure mode is reported as a negative errno, never a false
* built.count success. Pass a negative index, or call the clear helper, to
* disable. Compiled out of production. */
extern void job_array_apply_test_set_bulk_insert_force_fail_at(int index);
extern void job_array_apply_test_clear_bulk_insert_force_fail(void);
#endif
extern int update_job(void*, uint64_t, xce_kv_list_t*);
extern int do_job_started(void*, uint64_t, int, int*, int, int*);

334
src/controller/job.c
View file

@ -1,5 +1,6 @@
#include <errno.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>
@ -333,6 +334,339 @@ int restore_job(cluster_t* c, uint64_t jobid_unused,
return 0;
}
#ifdef XCE_TESTING
/* job-array-single-wal-record task 3.1: forceable, per-index construction
* failure for restore_job_array()'s scratch-build loop -- lets a test prove
* the all-or-nothing atomicity contract (a mid-array failure discards every
* element built THIS attempt, without ever touching c->jobmgr) without
* needing a real OOM. -1 = disabled (default); a value >= 0 forces the
* construction attempt for that 0-based element index within the CURRENT
* array to fail, simulating a resource-exhaustion apply failure (Req 9.4,
* 9.6). File-scope, not per-cluster_t: this task's own test only ever
* drives one cluster_t fixture at a time; a later task needing independent
* multi-node fault injection can revisit this. Compiled out of production
* (xce_controller is built without -DXCE_TESTING). */
static _Atomic int s_job_array_apply_force_fail_at = -1;
void job_array_apply_test_set_force_fail_at(int index)
{
atomic_store_explicit(&s_job_array_apply_force_fail_at, index,
memory_order_relaxed);
}
void job_array_apply_test_clear_force_fail(void)
{
atomic_store_explicit(&s_job_array_apply_force_fail_at, -1,
memory_order_relaxed);
}
/* job-array-single-wal-record task 3.1 remediation (review finding: the
* bulk-insert pass into c->jobmgr was previously unchecked and unfailable
* by test): forces the BULK-INSERT phase's key_add(c->jobmgr, ...) call --
* as opposed to the scratch-build phase's makeJob() above -- to fail at a
* given 0-based index WITHIN THE BULK-INSERT PASS, simulating the one real
* failure this pass can hit: c->jobmgr's backing xzarray hitting a genuine
* realloc() OOM while growing under the write lock (Req 9.4). -1 = disabled
* (default). Like a real allocation failure, the forced index is never
* actually inserted -- indices before it were already inserted for real and
* cannot be un-inserted (see the function doc comment's residual-gap
* paragraph). Compiled out of production. */
static _Atomic int s_job_array_bulk_insert_force_fail_at = -1;
void job_array_apply_test_set_bulk_insert_force_fail_at(int index)
{
atomic_store_explicit(&s_job_array_bulk_insert_force_fail_at, index,
memory_order_relaxed);
}
void job_array_apply_test_clear_bulk_insert_force_fail(void)
{
atomic_store_explicit(&s_job_array_bulk_insert_force_fail_at, -1,
memory_order_relaxed);
}
#endif /* XCE_TESTING */
/* job-array-single-wal-record task 3.1 (design.md Array Apply Expansion,
* Service Interface): expands one committed WAL_JOB_ARRAY_SUBMIT payload
* (base jobid, array_size, one shared job_submit_data_t) into `array_size`
* job_t entries, atomically -- shared by apply_committed_entry() (live
* apply / boot replay) and apply_wal_entry() (standby catch-up), task 3.2/
* 3.3, so the two dispatchers can never diverge in how they expand the same
* committed array record (Req 3.1, 3.2).
*
* Atomicity (Req 2.1, 2.2, 2.3, 9.3, 9.4 -- scratch-build-then-bulk-insert,
* NOT construct-then-rollback): every element is constructed via makeJob()
* -- the same construction create_job() wraps -- against a throwaway,
* function-local kv table (`scratch`, an xkvmgr_type instance identical in
* kind to c->jobmgr itself, so makeJob()'s internal locking/key_find/
* key_add calls behave exactly as they do in production). c->jobmgr is
* NEVER touched while this loop runs -- scratch is never visible to any
* other thread, so there is no window, success path or failure path, in
* which a concurrent reader of c->jobmgr (a status query, a cancel request)
* can observe some array indices present and others absent. Only after
* EVERY element has successfully constructed does one final pass insert
* them all into c->jobmgr under a single held write-lock acquisition. Every
* object here already exists, fully constructed, and no other writer can be
* racing these exact jobids, so key_add() cannot fail on EEXIST or
* content grounds -- but it CAN still fail on resource exhaustion: growing
* c->jobmgr's own backing store (a realloc() reachable through key_add ->
* xkvmgr_add -> vstore -> xzarray_store -> xzarray_ensure_items, kvmgr.c/
* zarray.c) happens WHILE this write lock is held, and c->jobmgr is sized
* far below MAX_ARRAY_IDX in production (jobmgr_nhash/jobmgr_delta,
* cluster.c), so a meaningfully large array is near-certain to trigger a
* growth cycle here. No capacity-reservation primitive exists anywhere in
* c->jobmgr's stack (xkvmgr_type/xzarray/kvtab -- kv.c, kvmgr.c, zarray.c
* all checked) to pre-size it before this loop starts mutating it, so this
* bulk-insert pass is a second, narrower, LATER possible failure point --
* see the residual-atomicity-gap paragraph below for how it differs from a
* scratch-build-phase failure. On a construction failure at index k during
* the SCRATCH-BUILD phase (resource exhaustion only -- Req 9.1's pre-commit
* validation already made a content-derived failure here impossible, Req
* 9.3), the 0..k-1 scratch-built elements are destroyed directly via
* job_dereference (job_t's teardown is wired through its reference
* interface, not the generic destroy() -- job_type.destroy is NULL, jobj.c)
* -- c->jobmgr was never mutated, so there is no key_del/rollback race to
* reason about. A retry re-enters this function from index 0 against a
* fresh scratch list, not from a saved offset.
*
* Idempotent per index (Req 3.3): if jobid already exists in c->jobmgr,
* that index is skipped (not an error) -- handles re-applying an
* already-fully-applied record (replay, or a resolved-but-retried commit).
* If every index already exists, returns -EEXIST (mirrors restore_job()'s
* own convention) and constructs nothing -- the caller must fire no hook.
*
* Every newly constructed element is tagged with submit_lsn = entry_lsn
* (mirrors restore_job(), feeds the existing egress-watermark predicate
* unchanged -- Req 5).
*
* out_new_jobs (optional): on a genuine (non-EEXIST) success, populated
* with every newly created job_t* in index order -- structurally identical
* to the ad hoc xlist_t apply_committed_entry() already builds for a single
* WAL_JOB_SUBMIT's HOOK_DETAIL_JOB_NEW fire, so task 3.2 can feed it
* directly to xce_run_hooks() as the array's one-shot, all-N-elements
* notification (Req 3.4). Left untouched (caller must not read it) on
* -EEXIST or a negative-errno return.
*
* Residual atomicity gap -- BULK-INSERT-phase failure only (narrower than
* the scratch-build-phase failure above, and NOT structurally preventable:
* no capacity-reservation primitive exists to pre-size c->jobmgr, see
* above): if key_add() itself fails partway through the bulk-insert pass
* (at index k of built.count), indices [0, k) are, by that point, REAL and
* PERMANENTLY visible in c->jobmgr -- xkvmgr_type exposes no key_del
* (kvmgr.c), so a partial bulk insert can never be rolled back after the
* fact. This function still reports the failure honestly (a negative
* errno, never a false built.count success) and releases the
* never-inserted [k, built.count) elements so they are not leaked, but it
* CANNOT restore the true "invisible-then-visible, nothing in between"
* guarantee for this specific failure point. Callers -- and task 4.x's Live
* Apply Resource Fence in particular -- must be aware that a negative
* return from this function may still have left a real, non-empty prefix
* of this attempt's elements live and findable in c->jobmgr, with no
* HOOK_DETAIL_JOB_NEW ever fired for them.
*
* Returns the count of newly created elements (1..array_size) on success,
* -EEXIST if every index already existed (idempotent whole-record no-op,
* no hook should fire), or a negative errno on a genuine failure -- in
* which case *out_new_jobs is left untouched and the caller must fire no
* hook. For a SCRATCH-BUILD-phase failure, the scratch list has already
* been discarded and c->jobmgr was never touched at all (Req 9.4: no
* partial set ever observable). For a BULK-INSERT-phase failure, see the
* residual atomicity gap paragraph above -- c->jobmgr may already hold a
* real, permanent prefix of this attempt's elements. */
int restore_job_array(cluster_t* c, const void* capnp_payload, size_t len,
uint64_t entry_lsn, xlist_t* out_new_jobs)
{
if (!c || !capnp_payload || len == 0)
return -EINVAL;
struct capn c_ctx;
capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0);
JobSubmit_ptr jp;
jp.p = capn_getp(capn_root(&c_ctx), 0, 1);
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);
capn_free(&c_ctx);
if (decoded.submit == NULL || decoded.array_size == 0) {
free_JobSubmit(&decoded);
return -EINVAL;
}
uint64_t base_jobid = decoded.jobid;
uint64_t array_size = decoded.array_size;
/* Throwaway construction target -- see the function doc comment above.
* datakiller is deliberately NULL: this scratch table's job_t values are
* always released explicitly (via `built`, below), never through the
* table's own teardown -- xkvmgr_type has no key_del, so relying on a
* datakiller here would make the success path (where ownership of every
* element moves to c->jobmgr) indistinguishable from the failure path
* (where every element must be released) purely from the table's own
* perspective. */
kvmgr_params_t scratch_params
= { .key_type = KVMGR_KEY_INT,
.nslots = (uint32_t)((array_size < 2) ? 2 : array_size),
.delta = (uint32_t)((array_size < 1) ? 1 : array_size),
.datakiller = NULL };
var scratch = makeInstance(&xkvmgr_type, &scratch_params);
if (scratch == NULL) {
int saved_rc = -errno;
free_JobSubmit(&decoded);
return saved_rc;
}
xlist_t built;
xlist_init(&built, (int)array_size);
time_t now = time(NULL);
int rc = 0;
for (uint64_t i = 0; i < array_size; i++) {
uint64_t jobid = base_jobid + i;
job_t* existing = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
int found = key_find(c->jobmgr, (void*)jobid, (void**)&existing);
unlock(c->jobmgr);
if (found > 0 && existing != NULL)
continue; /* idempotent per-index skip (Req 3.3) */
xce_job_t ej;
memset(&ej, 0, sizeof(ej));
ej.submit = decoded.submit; /* shared/borrowed across every
* element -- job_init's makeJobsub()
* interns/refcounts it, never takes
* ownership; freed once, below. */
ej.jobid = jobid;
ej.sub_time = now;
ej.queue_time = now;
ej.state = JOB_STAT_PEND;
#ifdef XCE_TESTING
int forced = atomic_load_explicit(&s_job_array_apply_force_fail_at,
memory_order_relaxed);
if (forced >= 0 && (uint64_t)forced == i) {
errno = ENOMEM; /* simulated resource-exhaustion failure */
rc = -ENOMEM;
break;
}
#endif
job_t* job = makeJob(scratch, &ej);
if (job == NULL) {
rc = -errno;
break;
}
job->submit_lsn = entry_lsn;
if (xlist_store(&built, job) < 0) {
dereference(job); /* never touched c->jobmgr -- safe to release
* directly (job_type.destroy is NULL; the
* reference interface is the only teardown
* path, jobj.c). */
rc = -ENOMEM;
break;
}
}
/* The one shared spec is always released here, exactly once, whether
* every element succeeded or not -- mirrors restore_job()'s
* unconditional free_Job() discipline. */
free_JobSubmit(&decoded);
if (rc < 0) {
/* All-or-nothing: discard everything built THIS attempt. c->jobmgr
* was never mutated, so there is nothing to roll back there -- a
* retry starts over from index 0 against a fresh scratch list. */
for (int k = 0; k < built.count; k++) {
job_t* j = (job_t*)xlist_get(&built, k);
if (j != NULL)
dereference(j);
}
xlist_fini(&built);
destroy(scratch);
return rc;
}
if (built.count == 0) {
/* Every index already existed -- idempotent whole-record no-op
* (Req 3.3). Nothing was constructed; *out_new_jobs is left
* untouched and no hook should fire. */
xlist_fini(&built);
destroy(scratch);
return -EEXIST;
}
/* One bulk pass, held under a single write-lock acquisition so no
* concurrent reader of c->jobmgr can ever observe a partial insert --
* this is what makes the array atomically invisible-then-visible on the
* SUCCESS path too, not only recoverable on the failure path (Req 2.3,
* 9.4). Every object already exists, fully constructed, and no other
* writer can be racing these exact jobids (only this apply path ever
* constructs them), so key_add() cannot fail on EEXIST or content
* grounds -- but it CAN still fail on resource exhaustion (c->jobmgr's
* backing store growing via realloc() under this write lock; see the
* function doc comment's "Residual atomicity gap" paragraph). Checked
* and reported below, never silently claimed as full success. */
lock(c->jobmgr, RTTI_LOCK_WRITE);
int inserted = 0;
int insert_rc = 0;
for (int k = 0; k < built.count; k++) {
job_t* j = (job_t*)xlist_get(&built, k);
int add_rc;
#ifdef XCE_TESTING
int forced_bulk = atomic_load_explicit(
&s_job_array_bulk_insert_force_fail_at, memory_order_relaxed);
if (forced_bulk >= 0 && forced_bulk == k) {
errno = ENOMEM; /* simulated resource-exhaustion failure */
add_rc = -ENOMEM;
} else
#endif
{
add_rc = key_add(c->jobmgr, (void*)j->job.jobid, (void*)j);
}
if (add_rc < 0) {
insert_rc = add_rc;
break;
}
inserted++;
}
unlock(c->jobmgr);
if (insert_rc < 0) {
/* See the function doc comment's "Residual atomicity gap" paragraph:
* indices [0, inserted) are already real and permanently visible in
* c->jobmgr (no key_del exists to undo them) -- only the
* never-inserted [inserted, built.count) tail is still ours to
* release here. */
for (int k = inserted; k < built.count; k++) {
job_t* j = (job_t*)xlist_get(&built, k);
if (j != NULL)
dereference(j);
}
xlist_fini(&built);
destroy(scratch);
return insert_rc;
}
int created = built.count;
if (out_new_jobs != NULL)
*out_new_jobs = built; /* transfer ownership of built's data array */
else
xlist_fini(&built);
destroy(scratch);
return created;
}
/* Apply a recorded job state transition during WAL recovery.
* Looks up the job (which must already exist from WAL_JOB_SUBMIT replay)
* and overwrites its state with the value stored in the WAL entry. */

1
src/controller/meson.build
View file

@ -165,6 +165,7 @@ controller_unit_test_src = [
'testsuite/test_apply_correctness_verification.c',
'testsuite/test_apply_pump.c',
'testsuite/test_apply_pump_integration.c',
'testsuite/test_job_array_apply.c',
'ctrlstatus.c',
'commit.c',
'job.c',

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

@ -85,6 +85,7 @@ extern void test_apply_reapply_idempotency(void);
extern void test_apply_correctness_verification(void);
extern void test_apply_pump(void);
extern void test_apply_pump_integration(void);
extern void test_job_array_apply(void);
void setUp(void) { return; }
@ -239,6 +240,7 @@ int main(int argc, char* argv[])
TEST(test_apply_correctness_verification);
TEST(test_apply_pump);
TEST(test_apply_pump_integration);
TEST(test_job_array_apply);
return UNITY_END();
}

495
src/controller/testsuite/test_job_array_apply.c Normal file
View file

@ -0,0 +1,495 @@
/* test_job_array_apply.c — job-array-single-wal-record task 3.1
*
* 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).
*
* Covers:
* - 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).
* - Idempotent PARTIAL re-apply skips only the indices that already
* exist (Req 3.3).
* - The forceable fault-injection seam: a forced failure partway through
* leaves ZERO elements of THIS attempt visible in c->jobmgr (not just
* 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.
* 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).
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "jobj.h"
#include "xce/types.h"
#include "xlist.h"
#include "xstring.h"
static int g_taj_jobobj_ready;
static int g_taj_xstring_ready;
static cluster_t g_taj_c;
static void taj_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_taj_c, 0, sizeof(g_taj_c));
g_taj_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_taj_c.jobmgr);
#ifdef XCE_TESTING
job_array_apply_test_clear_force_fail();
job_array_apply_test_clear_bulk_insert_force_fail();
#endif
}
static void taj_teardown(void)
{
#ifdef XCE_TESTING
job_array_apply_test_clear_force_fail();
job_array_apply_test_clear_bulk_insert_force_fail();
#endif
/* Mirrors test_apply_committed_entry.c's tace_teardown: the jobmgr's own
* datakiller is unset, so the job_t values it still holds are not
* individually released here -- accepted, matching that established
* fixture convention (the controller suite runs with
* ASAN_OPTIONS=detect_leaks=0). */
if (g_taj_c.jobmgr)
dereference(g_taj_c.jobmgr);
}
/* Look up a job_t wrapper by jobid (NULL if absent). */
static job_t* taj_find_job(uint64_t jobid)
{
job_t* j = NULL;
lock(g_taj_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_taj_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_taj_c.jobmgr);
return (found > 0) ? j : NULL;
}
/* Seeds a job directly into the fixture's jobmgr, bypassing
* restore_job_array entirely -- used to simulate "this index already
* exists" for the partial-idempotency test. */
static job_submit_data_t g_taj_seed_submit;
static void taj_seed_job(uint64_t jobid)
{
xce_job_t jdef;
memset(&g_taj_seed_submit, 0, sizeof(g_taj_seed_submit));
g_taj_seed_submit.queue = (char*)"seed-q";
g_taj_seed_submit.user = (char*)"seed-user";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &g_taj_seed_submit;
TEST_ASSERT_NOT_NULL(makeJob(g_taj_c.jobmgr, &jdef));
}
/* Encodes a (base_jobid, array_size, spec) JobSubmit payload -- the exact
* shape the committed WAL_JOB_ARRAY_SUBMIT record carries (mirrors
* persist.c's persist_submit_job_array_async encode side, and
* test_apply_committed_entry.c's tace_encode_job/tace_encode_worker
* pattern for the encode helper shape). Caller frees *out. */
static int taj_encode_job_submit(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;
}
static job_submit_data_t taj_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;
}
/* ── N-way expansion produces the correct contiguous set (Req 5.1-5.3) ── */
static void test_rja_expands_contiguous_set_and_tags_lsn(void)
{
taj_setup();
const uint64_t base_jobid = 71000;
const uint64_t array_size = 5;
const uint64_t entry_lsn = 777;
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));
xlist_t out;
memset(&out, 0, sizeof(out));
int rc = restore_job_array(&g_taj_c, payload, len, entry_lsn, &out);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, rc,
"a fresh array record must report array_size newly created elements");
TEST_ASSERT_EQUAL_INT((int)array_size, out.count);
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = taj_find_job(base_jobid + i);
TEST_ASSERT_NOT_NULL_MESSAGE(j, "every array index must be visible in c->jobmgr");
TEST_ASSERT_EQUAL_UINT64(base_jobid + i, j->job.jobid);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(entry_lsn, j->submit_lsn,
"every constructed element must be tagged with entry_lsn");
TEST_ASSERT_NOT_NULL(j->job.submit);
TEST_ASSERT_EQUAL_STRING("normal", j->job.submit->queue);
TEST_ASSERT_EQUAL_STRING("tester", j->job.submit->user);
/* out_new_jobs carries the same objects, in index order. */
TEST_ASSERT_EQUAL_PTR(j, xlist_get(&out, (int)i));
}
xlist_fini(&out);
free(payload);
taj_teardown();
printf("test_rja_expands_contiguous_set_and_tags_lsn: PASS\n");
}
/* ── Idempotent whole-record re-apply is a no-op (Req 3.3) ─────────────── */
static void test_rja_idempotent_whole_record_replay_is_noop(void)
{
taj_setup();
const uint64_t base_jobid = 72000;
const uint64_t array_size = 4;
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));
xlist_t out1;
memset(&out1, 0, sizeof(out1));
int rc1 = restore_job_array(&g_taj_c, payload, len, 10, &out1);
TEST_ASSERT_EQUAL_INT((int)array_size, rc1);
xlist_fini(&out1);
/* Re-apply the identical committed record (replay / retried commit). */
xlist_t out2;
memset(&out2, 0, sizeof(out2));
int rc2 = restore_job_array(&g_taj_c, payload, len, 10, &out2);
TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc2,
"re-applying an already-fully-applied record must be a whole-record"
" no-op (-EEXIST), constructing nothing new");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL(taj_find_job(base_jobid + i));
}
free(payload);
taj_teardown();
printf("test_rja_idempotent_whole_record_replay_is_noop: PASS\n");
}
/* ── Idempotent PARTIAL re-apply skips only existing indices (Req 3.3) ── */
static void test_rja_partial_reapply_skips_existing_indices(void)
{
taj_setup();
const uint64_t base_jobid = 73000;
const uint64_t array_size = 5;
job_submit_data_t spec = taj_make_spec();
uint8_t* payload = NULL;
size_t len = 0;
/* Index 2 already exists before the array apply ever runs. */
taj_seed_job(base_jobid + 2);
TEST_ASSERT_EQUAL_INT(0,
taj_encode_job_submit(base_jobid, array_size, &spec, &payload, &len));
xlist_t out;
memset(&out, 0, sizeof(out));
int rc = restore_job_array(&g_taj_c, payload, len, 20, &out);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size - 1, rc,
"one pre-existing index must be skipped, not treated as an error");
TEST_ASSERT_EQUAL_INT((int)array_size - 1, out.count);
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL_MESSAGE(taj_find_job(base_jobid + i),
"every index, pre-existing or newly created, must be present");
}
/* The pre-existing index must not appear among the NEWLY created set. */
for (int k = 0; k < out.count; k++) {
job_t* j = (job_t*)xlist_get(&out, k);
TEST_ASSERT_NOT_EQUAL_UINT64(base_jobid + 2, j->job.jobid);
}
xlist_fini(&out);
free(payload);
taj_teardown();
printf("test_rja_partial_reapply_skips_existing_indices: PASS\n");
}
#ifdef XCE_TESTING
/* ── Forced mid-array failure: zero elements visible; retry succeeds ────
* (Req 2.1, 2.2, 2.3, 9.3, 9.4, 9.6 -- the load-bearing atomicity contract:
* a mid-array failure must discard EVERY element built this attempt,
* including the ones "before" the forced failure index, and must never
* have touched c->jobmgr at all.) */
static void test_rja_forced_failure_leaves_nothing_visible_then_retry_succeeds(void)
{
taj_setup();
const uint64_t base_jobid = 74000;
const uint64_t array_size = 5;
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));
/* Force a construction failure at index 2 -- indices 0 and 1 will have
* "succeeded" in the scratch list before the forced failure fires. */
job_array_apply_test_set_force_fail_at(2);
xlist_t out1;
memset(&out1, 0, sizeof(out1));
int rc1 = restore_job_array(&g_taj_c, payload, len, 30, &out1);
TEST_ASSERT_TRUE_MESSAGE(rc1 < 0,
"a forced mid-array construction failure must return a negative rc");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NULL_MESSAGE(taj_find_job(base_jobid + i),
"ALL elements of a failed attempt must be invisible -- including"
" the ones that succeeded construction before the forced"
" failure index, not only the failed index itself");
}
/* Clear the seam and retry the IDENTICAL record from scratch. */
job_array_apply_test_clear_force_fail();
xlist_t out2;
memset(&out2, 0, sizeof(out2));
int rc2 = restore_job_array(&g_taj_c, payload, len, 30, &out2);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, rc2,
"a retry after the injected failure clears must produce the"
" complete, non-duplicated set");
TEST_ASSERT_EQUAL_INT((int)array_size, out2.count);
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = taj_find_job(base_jobid + i);
TEST_ASSERT_NOT_NULL(j);
TEST_ASSERT_EQUAL_UINT64(base_jobid + i, j->job.jobid);
}
xlist_fini(&out2);
free(payload);
taj_teardown();
printf("test_rja_forced_failure_leaves_nothing_visible_then_retry_succeeds: PASS\n");
}
/* ── Bulk-insert-phase failure: reported as an error, never a false
* built.count success (task 3.1 review remediation) ───────────────────────
* Unlike the scratch-build-phase seam above, this forces the SECOND phase
* -- the key_add(c->jobmgr, ...) bulk-insert pass itself -- to fail
* partway through, simulating c->jobmgr's backing store hitting a genuine
* realloc() OOM while growing under the write lock (Req 9.4). Because
* xkvmgr_type exposes no key_del (kvmgr.c), the indices already inserted
* for real before the forced failure index CANNOT be rolled back -- this
* test proves the function is honest about that: it reports a negative
* errno (never claims full built.count success), and the already-real
* prefix genuinely IS visible while the never-inserted tail genuinely is
* NOT -- matching the "Residual atomicity gap" paragraph documented on
* restore_job_array() (job.c). A subsequent retry (seam cleared) must not
* duplicate the already-real prefix -- it is picked up by the existing
* per-index idempotency skip (Req 3.3). */
static void test_rja_bulk_insert_failure_reports_error_not_false_success(void)
{
taj_setup();
const uint64_t base_jobid = 76000;
const uint64_t array_size = 5;
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));
/* Force the bulk-insert pass to fail inserting the 3rd element (index 2
* of built.count) -- indices 0 and 1 will already be genuinely,
* permanently inserted into c->jobmgr by the time it fires. */
job_array_apply_test_set_bulk_insert_force_fail_at(2);
xlist_t out;
memset(&out, 0, sizeof(out));
int rc = restore_job_array(&g_taj_c, payload, len, 50, &out);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"a bulk-insert-phase failure must be reported as a negative errno,"
" never as a false built.count success");
TEST_ASSERT_NOT_EQUAL_MESSAGE((int)array_size, rc,
"the failure must not be silently masked as full success");
/* The residual gap, proven directly: indices [0, 2) are already real
* (cannot be rolled back -- no key_del); indices [2, array_size) were
* never inserted anywhere. */
for (uint64_t i = 0; i < 2; i++) {
TEST_ASSERT_NOT_NULL_MESSAGE(taj_find_job(base_jobid + i),
"indices already key_add()'d for real before the forced failure"
" must remain visible -- they cannot be un-inserted");
}
for (uint64_t i = 2; i < array_size; i++) {
TEST_ASSERT_NULL_MESSAGE(taj_find_job(base_jobid + i),
"indices never reached by key_add() must not be visible");
}
xlist_fini(&out);
/* Clear the seam and retry the identical record: the two already-real
* indices are skipped by the existing per-index idempotency check
* (Req 3.3), and the remaining three are newly created. */
job_array_apply_test_clear_bulk_insert_force_fail();
xlist_t out2;
memset(&out2, 0, sizeof(out2));
int rc2 = restore_job_array(&g_taj_c, payload, len, 51, &out2);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size - 2, rc2,
"a retry must skip the two already-real indices and create only the"
" remaining three");
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL_MESSAGE(taj_find_job(base_jobid + i),
"every index must be present after the retry");
}
xlist_fini(&out2);
free(payload);
taj_teardown();
printf("test_rja_bulk_insert_failure_reports_error_not_false_success: PASS\n");
}
#endif /* XCE_TESTING */
/* ── Defensive / boundary ─────────────────────────────────────────────── */
static void test_rja_null_or_empty_payload_returns_error(void)
{
taj_setup();
TEST_ASSERT_TRUE(restore_job_array(&g_taj_c, NULL, 0, 1, NULL) < 0);
TEST_ASSERT_TRUE(restore_job_array(NULL, (const void*)"x", 1, 1, NULL) < 0);
taj_teardown();
printf("test_rja_null_or_empty_payload_returns_error: PASS\n");
}
static void test_rja_out_new_jobs_is_optional(void)
{
taj_setup();
const uint64_t base_jobid = 75000;
const uint64_t array_size = 3;
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));
int rc = restore_job_array(&g_taj_c, payload, len, 40, NULL);
TEST_ASSERT_EQUAL_INT((int)array_size, rc);
for (uint64_t i = 0; i < array_size; i++) {
TEST_ASSERT_NOT_NULL(taj_find_job(base_jobid + i));
}
free(payload);
taj_teardown();
printf("test_rja_out_new_jobs_is_optional: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_job_array_apply(void)
{
UNITY_BEGIN();
RUN_TEST(test_rja_expands_contiguous_set_and_tags_lsn);
RUN_TEST(test_rja_idempotent_whole_record_replay_is_noop);
RUN_TEST(test_rja_partial_reapply_skips_existing_indices);
#ifdef XCE_TESTING
RUN_TEST(test_rja_forced_failure_leaves_nothing_visible_then_retry_succeeds);
RUN_TEST(test_rja_bulk_insert_failure_reports_error_not_false_success);
#endif
RUN_TEST(test_rja_null_or_empty_payload_returns_error);
RUN_TEST(test_rja_out_new_jobs_is_optional);
UNITY_END();
}