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

fix(job-array-single-wal-record): free jobArray dataset on scheduler receive path

Browse source 
sched_handle_dataset()'s Data_data_jobArray case never freed
dataset->rows (the decoded xce_job_submit_t*) after sched_on_job_array()
consumed it -- free_Dataset() is an empty no-op for every table type, so
nothing else owned this call. One bounded leak per delivered array push.

Free it via the existing free_JobSubmit_ptr() teardown on both the
success and error path, mirroring the sibling list-shaped cases'
ownership-transfer convention.

Rewrote test_job_array_expand.c's init_job_array() fixture to
heap-allocate + strdup its fields (matching the real wire-decode shape)
since freeing the old stack/string-literal fixture crashed. Redelivery
test now builds a fresh descriptor per delivery, matching how a real
redelivery is always a freshly-decoded wire message.

Closes task 7.3 (job-array-single-wal-record), the last of the three
post-GO residual gaps reopened 2026-07-16.
This commit is contained in:
Rongsong Shen 2026-07-17 07:13:12 +00:00
parent ea8d9d3c4f
commit 53bd686897
3 changed files with 55 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -187,7 +187,7 @@
- _Requirements: 8.1, 8.2, 8.4_
- _Boundary: Non-Regression & Testing_
- [ ] 7. Residual gap closure (post-GO follow-up, user-directed 2026-07-16)
- [x] 7. Residual gap closure (post-GO follow-up, user-directed 2026-07-16)
- [x] 7.1 Close the bulk-insert-phase partial-visibility gap (bug-1443)
- Add a new, purely-additive capacity-reservation primitive (`kvmgr_reserve(var kvmgr, int n)` in `kvmgr.c`, following the existing non-vtable `kvmgr_capture_keys()` template exactly — same "caller MUST hold the kvmgr write lock" contract, same bare-extern-function style rather than a new RTTI vtable slot) that pre-grows `c->jobmgr`'s backing `xzarray` for `n` additional entries via a new small non-static wrapper in `zarray.c` calling the existing, unmodified `xzarray_ensure_items(a, a->count + n)` — zero changes to any existing function in either file
- Call `kvmgr_reserve(c->jobmgr, array_size)` in `restore_job_array()` immediately before the bulk-insert pass begins; on reservation failure, treat exactly like today's existing pre-mutation failure paths (discard the scratch list, `c->jobmgr` never touched, return negative) — this eliminates the dominant, N-scaling failure trigger (the large `realloc()` growing the value store) structurally, before any `key_add()` call in the bulk-insert loop can run
@ -207,7 +207,7 @@
- _Boundary: Live Apply Resource Fence (`src/controller/persist.c`'s `replay_wal()`) — reuses existing task 4.1/4.2 primitives verbatim, no new fence mechanism_
- _Depends: 7.1 is not a prerequisite (independent root cause), but should land first since both touch `persist.c`_
- [ ] 7.3 Fix the bounded per-commit leak on the scheduler receive path (task 5.3b's self-flagged gap)
- [x] 7.3 Fix the bounded per-commit leak on the scheduler receive path (task 5.3b's self-flagged gap)
- `sched_handle_dataset()`'s `Data_data_jobArray` case never frees `dataset->rows` (the decoded `xce_job_submit_t*`) after `sched_on_job_array()` consumes it — `free_Dataset()` is an empty no-op for every table type, so this is the one call site responsible for freeing this specific variant's payload
- Free `dataset->rows` (and its embedded submit spec, following whatever the existing `xce_job_submit_t`/`JobSubmitData` teardown convention already is elsewhere in this file) after `sched_on_job_array()` returns, in both the success and any early-return/error path through that case
- Observable: a leak-detection test run (ASan) around a real array push shows zero growth attributable to this path, where it previously showed one bounded leak per delivered array push
@ -216,6 +216,8 @@
## Implementation Notes
- **Task 7.3**: `sched_handle_dataset()`'s `Data_data_jobArray` case now calls `free_JobSubmit_ptr((xce_job_submit_t**)&dataset->rows)` + `dataset->nrows = 0` after `sched_on_job_array()` returns (success and error paths alike), mirroring the sibling list-shaped cases' ownership-transfer convention. `free_JobSubmit_ptr()` is the existing capnp-generated teardown for this exact struct shape (frees `submit`'s 7 strings + `start_info`'s 10 strings + both nested structs + the top-level struct) — no new free logic was needed, just the missing call site. Fixing this exposed that `test_job_array_expand.c`'s white-box fixture (`init_job_array()`) built `dataset->rows` from stack-allocated structs with string-literal fields; freeing those crashes (SEGV in `free_JobSubmitData`). The fixture was rewritten to heap-allocate + `strdup` (matching the real `decode_JobSubmit_ptr()` wire-decode shape), and `test_job_array_redelivery_does_not_duplicate` now builds a fresh descriptor per delivery (the first is consumed+freed by the handler, so reusing one pointer across two calls is no longer valid — this also matches production, where a redelivery is always a freshly-decoded wire message, never a reused pointer). Verified via direct ASAN runs of all 4 scheduler job-array test binaries (leak detection is disabled by default in the meson test harness): zero leak growth traceable to the jobArray path in any of them (confirmed by grepping leak stack traces, not just byte counts, since fixture-init debt is a separate pre-existing, unrelated leak source in the same binaries).
- **Task 1.3**: `meson test --suite xce`'s `libxce test` target SIGABRTs under ASAN leak detection whenever any code path exercises a `job_init()` failure — root cause is the vendored `rtti` library's `makeInstance()` not freeing its allocation when a type's `.init()` returns negative (no `.c` source available, headers-only in the nix store). This is pre-existing and independently reproduced against an unmodified pre-task `jobj.c`, not introduced by this feature. It will fire again on any future task whose tests exercise a `create_job()`/`makeJob()` failure path (e.g. task 3.1/3.4's fault-injection tests for `restore_job_array()`) — expect the same SIGABRT-not-assertion-failure signature (`grep ":FAIL"` on the Unity log will be empty; the failure is LeakSanitizer-only) and do not treat it as a new regression without the same stash-and-reproduce check task 1.3's review used.
- **Task 1.3**: `job_obj_init()`/`job_obj_fini()` are now idempotent (atomic_flag-guarded) — safe to call more than once without an intervening `fini()`. Any future test construction can rely on this instead of coordinating a single call site.
- **Task 1.3**: `makeJobsub()`/`makeJobstartinfo()` now return `-ENODEV` (not a generic `-EINVAL`) when the interning table isn't initialized — if a later task's error handling branches on `job_init()`'s return value, `-ENODEV` is a distinct, checkable signal for "table not ready" vs. other construction failures.

8
src/scheduler/src/handler.c
View file

@ -365,8 +365,12 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset)
* into the pend-update queue -- every array element is its own
* independent clone (see make_array_element/clone_job_submit_data
* above) -- so the original (jobid, arraySize, submit) descriptor
* stays exactly as owned/freed by whatever handed it to us. Do not
* null it out here. */
* is never consumed downstream. free_Dataset()/free_Dataset_ptr()
* are a no-op for every table type, so this is the one call site
* responsible for releasing it, on both the success and error
* path. */
free_JobSubmit_ptr((xce_job_submit_t**)&dataset->rows);
dataset->nrows = 0;
break;
}

74
src/scheduler/testsuite/test_job_array_expand.c
View file

@ -86,23 +86,32 @@ static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid)
sched_add_job(sched, job);
}
/* Fills a caller-owned (jobArray, submit) pair with a small, valid, sharable
* job spec -- the same (jobid, arraySize, submit) shape task 5.1's
* xcesched_push_jobarray() sends. */
static void init_job_array(xce_job_submit_t* ja, job_submit_data_t* submit,
uint64_t base_jobid, uint64_t array_size)
/* Builds a heap-allocated, heap-owned-strings (jobArray, submit) pair -- the
* same shape decode_JobSubmit_ptr() produces on the real wire-decode path
* (controller.capnp.c), not a stack/string-literal stand-in. Task 7.3 made
* sched_handle_dataset()'s Data_data_jobArray case free_JobSubmit_ptr() its
* dataset->rows unconditionally, so any fixture routed through that handler
* must be genuinely heap-owned or the free segfaults on stack memory /
* string literals. Caller owns the returned pointer until it hands it to
* sched_handle_dataset() (which frees it) or frees it directly itself. */
static xce_job_submit_t* init_job_array(uint64_t base_jobid,
uint64_t array_size)
{
memset(submit, 0, sizeof(*submit));
submit->queue = "normal";
submit->user = "tester";
job_submit_data_t* submit
= (job_submit_data_t*)calloc(1, sizeof(*submit));
xce_job_submit_t* ja = (xce_job_submit_t*)calloc(1, sizeof(*ja));
submit->queue = strdup("normal");
submit->user = strdup("tester");
submit->mintasks = 1;
submit->maxtasks = 1;
submit->resreq = "type == local";
submit->resreq = strdup("type == local");
memset(ja, 0, sizeof(*ja));
ja->jobid = base_jobid;
ja->array_size = array_size;
ja->submit = submit;
return ja;
}
/* Pop one pending update request, asserting it is an UPDATE_JOB op; hands
@ -135,20 +144,21 @@ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out)
static void test_job_array_push_expands_into_n_tasks(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
xce_job_submit_t* ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
init_job_array(&ja, &submit, 5000, 3);
ja = init_job_array(5000, 3);
memset(&dataset, 0, sizeof(dataset));
dataset.table = Data_data_jobArray;
dataset.rows = (void*)&ja;
dataset.rows = (void*)ja;
/* sched_handle_dataset() owns and frees dataset.rows on return (task
* 7.3) -- do not touch ja/dataset.rows again after this call. */
TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);
njobs = pop_job_update(&sched, &jobs);
@ -177,21 +187,19 @@ static void test_job_array_push_expands_into_n_tasks(void)
static void test_job_array_redelivery_does_not_duplicate(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
dataset_t dataset;
xce_job_t** jobs = NULL;
int njobs;
int i;
fixture_init(&sched);
init_job_array(&ja, &submit, 6000, 2);
memset(&dataset, 0, sizeof(dataset));
dataset.table = Data_data_jobArray;
dataset.rows = (void*)&ja;
dataset.rows = (void*)init_job_array(6000, 2);
/* first delivery: 2 new tasks enqueued */
/* first delivery: 2 new tasks enqueued. sched_handle_dataset() owns and
* frees dataset.rows on return (task 7.3). */
TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);
njobs = pop_job_update(&sched, &jobs);
TEST_ASSERT_EQUAL_INT(2, njobs);
@ -206,7 +214,12 @@ static void test_job_array_redelivery_does_not_duplicate(void)
}
free(jobs);
/* redelivery of the IDENTICAL push: every index already exists */
/* redelivery of the IDENTICAL push: a genuinely new decode of the
* retransmitted wire message (same jobid/array_size/submit content, but
* its own freshly-allocated descriptor -- exactly what a real reconnect
* redelivers; the first descriptor is already freed by the call above).
* Every index already exists. */
dataset.rows = (void*)init_job_array(6000, 2);
TEST_ASSERT_EQUAL_INT(0, sched_handle_dataset(&sched, &dataset));
TEST_ASSERT_EQUAL_INT(-1, pop_job_update(&sched, NULL)); /* nothing new */
}
@ -231,13 +244,17 @@ static void test_job_array_null_submit_returns_einval(void)
static void test_job_array_zero_size_returns_einval(void)
{
xce_scheduler_t sched;
job_submit_data_t submit;
xce_job_submit_t ja;
xce_job_submit_t* ja;
fixture_init(&sched);
init_job_array(&ja, &submit, 42, 0);
ja = init_job_array(42, 0);
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja));
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, ja));
/* unlike sched_handle_dataset(), sched_on_job_array() never takes
* ownership of job_array -- this test built it directly, so it frees it
* directly. */
free_JobSubmit_ptr(&ja);
}
static void test_job_array_null_descriptor_returns_einval(void)
@ -335,13 +352,12 @@ static void test_wire_dispatch_chain_delivers_n_tasks(void)
/* Cleanup: extract_data() transferred ownership of the decoded
* xce_job_submit_t into dataset.rows (nulling the source pointer inside
* decoded.body.pushmsg); sched_handle_dataset()'s jobArray case
* deliberately does not null dataset.rows out (every element is its own
* clone -- see that function's own comment), so this test, as the
* descriptor's owner, frees it directly. decoded.body.pushmsg is now a
* decoded.body.pushmsg); sched_handle_dataset()'s jobArray case now
* frees dataset.rows itself and nulls it out (task 7.3), so there is
* nothing left for this test to free there. decoded.body.pushmsg is a
* hollow container, safe to free via the normal request teardown. */
TEST_ASSERT_NULL(dataset.rows);
xce_free_request(&decoded);
free_JobSubmit_ptr((xce_job_submit_t **)&dataset.rows);
}
int main(int argc, char* argv[])