diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index 628e40ef..63b7afdc 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -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. diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 1394b64c..a701b945 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -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; } diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c index 5c923781..888fca14 100644 --- a/src/scheduler/testsuite/test_job_array_expand.c +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -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[])