From d16847185ef19ceb63e2d28db8005cb18f5340c4 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 15 Jul 2026 10:10:14 +0000 Subject: [PATCH] feat(job-array-single-wal-record): scheduler-side local expansion of the compact push Adds sched_on_job_array() (handler.c), a sibling case to Data_data_schedjobs decoding the new Data.jobArray variant and expanding it locally into N ordinary schedulable tasks via base_jobid + i, routed through the existing unmodified per-job pipeline (sched_on_jobs()). Each element gets its own deep clone (clone_job_submit_data/clone_job_startinfo) of the shared submit spec, since per-element teardown (free_Job_ptr) frees it independently and a shared pointer would double-free. Dedup reuses sched_get_job() against the same kvtab sched_add_job() guards, rather than a new mechanism. A confirmation log line fires once per successful expansion (skipped on a fully-duplicate redelivery). Adds a new task (5.3b) discovered during this task's review: the shared wire dispatcher extract_data() (src/lib/request.c) has no case for Data_data_jobArray and silently drops every real array push before any receiver's handler runs -- outside every current task's boundary, confirmed to affect the future API-server receive path (task 5.4) identically. design.md's File Structure Plan corrected to record this. User confirmed sequencing: fix it now, before 5.4. Task 5.3 of job-array-single-wal-record (Requirements 6.3, 6.4, 6.5, 8.4). --- .../job-array-single-wal-record/design.md | 1 + .../job-array-single-wal-record/tasks.md | 13 +- src/scheduler/src/handler.c | 225 +++++++++++++++ src/scheduler/testsuite/meson.build | 14 + src/scheduler/testsuite/test_handlers.c | 18 +- .../testsuite/test_job_array_expand.c | 261 ++++++++++++++++++ 6 files changed, 526 insertions(+), 6 deletions(-) create mode 100644 src/scheduler/testsuite/test_job_array_expand.c diff --git a/.kiro/specs/job-array-single-wal-record/design.md b/.kiro/specs/job-array-single-wal-record/design.md index a4939fe..9dc3bd1 100644 --- a/.kiro/specs/job-array-single-wal-record/design.md +++ b/.kiro/specs/job-array-single-wal-record/design.md @@ -133,6 +133,7 @@ graph TB - `src/scheduler/src/handler.c` — sibling case to the existing `Data_data_schedjobs` (line 112) decoding the new array variant and expanding it locally into N schedulable tasks via the same `base_jobid + i` arithmetic used by the controller, logging one confirmation line per array (Requirement 6.4). - `src/frontend/apiserver/backend/` — sibling decode path to `cJobToModel` (`convert.go:31`) for the new array variant, expanding N `models.JobInfo` entries locally, logging one confirmation line per array (Requirement 6.4). - `src/lib/jobj.c` — `job_init()`: check `makeJobsub()`'s (and `makeString(src->ckptinfo)`'s) return value; propagate failure instead of storing NULL with a success return (Requirement 10.1). Root-cause and fix the `jobsub_table` initialization-ordering gap behind the previously observed NULL-with-EINVAL failure (Requirement 10.2). `job_destroy()`: fix the `accounts` leak (currently `/* FIXME??? */`, jobj.c:584) by adding the missing free call, following the existing pattern used for `allocs`/`pidinfo`/`run_usage` (Requirement 10.4). +- **`src/lib/request.c` (added during implementation, task 5.3b, discovered 2026-07-15):** `extract_data()` — the shared inbound-push reassembly dispatcher every receiver (scheduler, API server) passes through inside `vtask_entry()` (`src/lib/xcectx.c`) before its own handler ever runs. Missed by this component's original architecture analysis, which examined `sched.c`/`apiserver.c`/`handler.c` but not the transport dispatch layer beneath them. Its switch has no case for `Data_data_jobArray` and falls into `default: rc = -EINVAL`; because `vtask_entry()`'s `if (callnext)` treats any nonzero return (including a negative errno) as "reassembly complete, proceed," this silently frees the real payload and hands the receiver an empty `dataset_t`, with no log anywhere in the chain. Add a `case Data_data_jobArray:` passing the single, unchunked descriptor straight through — no pagination loop needed, unlike the existing LIST-shaped cases. ### New Test Files - `src/controller/testsuite/test_job_array_apply.c` — unit tests: single-record N-way expansion, idempotent re-apply (no duplicate elements), all-or-nothing scratch-discard on a simulated construction failure (never touches `c->jobmgr`), identical results from `apply_committed_entry()` vs. `apply_wal_entry()` for the same record. diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index e52fe60..2e0ef16 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -122,7 +122,7 @@ - _Requirements: 6.2, 6.4_ - _Boundary: Compact Scheduler/API-Server Push (controller-side apiserver push)_ -- [ ] 5.3 (P) Scheduler-side local expansion +- [x] 5.3 (P) Scheduler-side local expansion - Decode the compact push and expand it locally into N ordinary schedulable tasks using the same base-jobid-plus-index arithmetic the controller itself uses - Deduplicate by element identity so a redelivered push (e.g. after a scheduler reconnect) does not create duplicate tasks - Log one identifiable confirmation line (base jobid plus element count) once expansion completes @@ -131,6 +131,15 @@ - _Boundary: Compact Scheduler/API-Server Push (scheduler daemon receive path)_ - _Depends: 5.1_ +- [ ] 5.3b Fix the shared wire dispatcher to carry the compact array push + - `extract_data()` (`src/lib/request.c`) is the shared reassembly step every inbound push (scheduler and API-server alike) passes through inside `vtask_entry()` (`src/lib/xcectx.c`) before a receiver's handler ever runs. Its dispatch switch has no case for the `Data_data_jobArray` variant (added task 1.2) and falls into `default: rc = -EINVAL`. Because `vtask_entry()`'s `if (callnext)` treats ANY nonzero return — including a negative errno — as "reassembly complete, proceed," this doesn't surface as an error: the real payload is freed and the receiver is silently handed an empty, zeroed `dataset_t` (`rows=NULL, nrows=0`), with no log anywhere in the chain (`vtask_entry()` discards the handler's own return value unconditionally) + - Add a `case Data_data_jobArray:` to `extract_data()` that passes the single, unchunked `JobSubmit` descriptor straight through to the caller's `dataset_t` — no pagination/accumulation loop is needed (unlike the existing LIST-shaped cases), since this push is always exactly one descriptor per commit, matching the "wire shape note" both `xcesched_push_jobarray()` (task 5.1) and `apiserver_push_jobarray()` (task 5.2) already establish + - Confirms both the scheduler (task 5.3, already landed) and the future API-server Go receiver (task 5.4) share this exact fix — this task only needs to land once + - Observable: a synthetic `Data_data_jobArray` push driven through the REAL `vtask_entry()`/`extract_data()` path (not task 5.3's white-box `sched_handle_dataset()`-direct test) reaches the scheduler's `sched_on_job_array()` with the genuine payload intact, not an empty dataset + - _Requirements: 6.1, 6.2, 6.4, 8.4_ + - _Boundary: shared wire transport (`src/lib/request.c`, `src/lib/xcectx.c`) — out of the Compact Scheduler/API-Server Push component's original File Structure Plan; discovered during task 5.3's review (2026-07-15), design.md corrected to record it_ + - _Depends: 5.1, 5.2_ + - [ ] 5.4 (P) API-server-side local expansion - Decode the compact push and expand it locally into N per-element records using the same base-jobid-plus-index arithmetic - Deduplicate by element identity so a redelivered push does not create duplicate records @@ -180,3 +189,5 @@ - **Task 4.2**: New `_Atomic uint64_t apply_resource_fenced_lsn` field on `persist_t` (0 = clear, nonzero = the stuck LSN) plus `persist_apply_resource_fenced()`/`persist_mark_apply_resource_fenced()`/`persist_clear_apply_resource_fence()`, wired into `persist_apply_pump_run()`'s existing halt branch (engage) and unconditionally after every successful advance (clear). The unconditional clear is provably equivalent to "clear only on the fenced LSN's own successful retry" because `last_applied_lsn` has exactly 3 writers codebase-wide, the other two (`replay_wal`, `persist_recover`) are boot-only and temporally disjoint from the live pump, and a failed walk never advances past the stuck LSN — so the next success by construction IS that LSN's retry. **Known residual gap (out of this task's boundary, flagged by review, NOT yet fixed)**: `replay_wal()` (boot/restart replay, persist.c:3875-3888) still discards `apply_committed_entry()`'s failure return the same way the live pump did before task 4.1, and `persist_init()` unconditionally zeroes the fence field on every restart. A node that crashes while fenced and re-hits the identical resource-exhaustion condition during boot replay will silently pass it and look fully promotable while actually divergent — task 4.1/4.2's protection does not yet extend to the boot-replay path. This is a pre-existing gap (not introduced by this feature) but is now inconsistent with the live path's new guarantee; worth a dedicated follow-up task before anything downstream places full trust in this fence across a restart. - **Task 5.1**: New `xcesched_push_jobarray()` (`sched.c`) sends via direct `xce_request()`/`OPCode_push`/`Request_body_pushmsg` rather than the existing `xce_push_data_req()` helper — `xce_push_data_req()`'s dispatch switch hard-rejects (`-EINVAL`) anything outside jobs/workers/resources/resvals, so it cannot carry a single non-list `Data.jobArray` descriptor. design.md corrected to match (Dependencies + API Contract for the Compact Scheduler/API-Server Push component). Task 5.2's API-server push will hit the identical fork and should follow the same direct-`xce_request()` pattern rather than re-deriving it. One accepted, low-severity trade-off: the direct path does not go through `xce_push_data_req()`'s `checkf`-gated `check_controller`/`c->running` liveness check that every other push gets — a narrow shutdown-race behavioral difference from sibling pushes, not a new safety hazard; not fixed here since threading a liveness check into a single new call site duplicates rather than reuses the shared gate. `sched.c`/`dataex.c`/`egress.c` were never linked into `controller_unit_test` (two existing test files there ship their own fake `on_scheduler_join()`/`notify()`/`collect_confvars()` precisely because of this, and linking the real ones would collide) — task 5.1's tests live in a new standalone `sched_push_unit_test` binary instead; task 5.2 will likely face the same fork for `apiserver.c` and should check for an analogous existing-fakes collision before choosing a test-binary strategy. - **Task 5.2**: `apiserver_push_jobarray()` (`apiserver.c`) mirrors task 5.1's `xcesched_push_jobarray()` exactly — same field derivation, same direct-`xce_request()` transport, same `l->count > 1` discriminator — and additionally, correctly replicates `apiserver_push_joblist()`'s existing per-connected-server fan-out loop (`N_APISERVER_MAX` iteration, same lock discipline), independently verified against 2 simulated connected servers each receiving exactly one push. Confirmed the same `apiserver.c`/`dataex.c` link-collision task 5.1 hit for `sched.c` (a fake `notify()` already exists in `test_commit_barrier_reply.c`) — new standalone `apiserver_push_unit_test` binary added, correctly omitting `egress.c` since `apiserver_push_joblist()`, unlike `sched.c`'s push path, never references `job_submit_committed()`/`collect_jobs_committed()`. One low-severity, non-blocking nit (not fixed): the new test file's comment justifying its `apiserver_init()` idempotency guard is factually inaccurate — it claims a second `apiserver_init()` call would double-register hooks, but `xce_add_hook()` has deduped by `(fn,data)` since bug-845, so that specific claim is wrong; the guard itself is still correct/necessary for a different, real reason (avoids re-`mylock_init()`-ing an already-initialized mutex, UB). +- **Task 5.3**: `sched_on_job_array()` (`handler.c`) decodes `Data_data_jobArray`, derives each element's jobid via `base_jobid + i`, and routes constructed elements through the unmodified per-job pipeline (`sched_on_jobs()`). Each element gets its OWN deep clone (`clone_job_submit_data`/`clone_job_startinfo`) of the shared submit spec — required because `sched_pend_update_killer()`'s per-element teardown (`free_Job_ptr`) frees each element's `submit` independently and a shared pointer across N elements would double-free; independently verified. Dedup reuses `sched_get_job()` against the same `sched->jobs` kvtab `sched_add_job()`'s `key_add` authoritatively guards — sound at the data-integrity level, but review found a narrow, non-corrupting TOCTOU: the dedup pre-check runs synchronously at receive time while the actual insert only happens later when `sched_update()` drains `pend_updates[0]`, so a redelivery landing before that drain can re-construct/re-enqueue a duplicate (harmless) object and double the confirmation log line — the same pre-existing, task-5.3-unrelated pattern as `sched_update_jobs()`'s already-ignored `key_add` `-EEXIST` return on the ordinary single-job path. Not fixed (pre-existing class of gap, narrow window, non-corrupting). + **CRITICAL discovery during this task's review (2026-07-15), NOT a defect in task 5.3 itself**: the shared wire dispatcher `extract_data()` (`src/lib/request.c`) has no case for `Data_data_jobArray` and silently drops every real array push before ANY receiver's handler runs (see new task 5.3b, inserted immediately after this task per explicit user decision). Confirmed to affect task 5.4's future API-server receive path identically (both share the same `xcecli_ctx_init()` → `vtask_entry()` → `extract_data()` pipeline). Task 5.3's own tests are unaffected by this gap since they use the same established white-box convention as every sibling `handler.c` test (calling `sched_handle_dataset()` directly, bypassing the transport layer) — this is not a new testing shortcut, it just means task 5.3's passing tests do not, and were never intended to, prove real-wire delivery. diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 9f4205a..1394b64 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -1,6 +1,8 @@ #include #include #include +#include +#include #include #include "controller.capnp.h" @@ -37,6 +39,220 @@ static int sched_on_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs) return 0; } +/* job-array-single-wal-record task 5.3 (design.md "Compact Scheduler/ + * API-Server Push", File Structure Plan handler.c bullet; Req 6.3, 6.4, 6.5, + * 8.4): sibling case to Data_data_schedjobs (sched_handle_dataset, below) + * decoding the compact Data.jobArray descriptor -- task 5.1's + * xcesched_push_jobarray() wire shape, one JobSubmit{jobid, arraySize, + * submit} carrying the array's base jobid, element count and ONE shared job + * spec -- and expanding it locally into N ordinary schedulable tasks via the + * same base_jobid + i arithmetic the controller's own restore_job_array() + * (src/controller/job.c) uses to derive each element's identity. + * + * Each expanded element gets its OWN independent deep copy of the shared + * submit spec (clone_job_submit_data/clone_job_startinfo below), never a + * shared pointer: the existing per-job pipeline this feeds into + * (sched_on_jobs -> XCE_SCHED_OP_UPDATE_JOB -> sched_pend_update_killer) + * calls free_Job_ptr() on every element independently, which would + * double-free a submit pointer shared across more than one xce_job_t. */ +static job_startinfo_t* clone_job_startinfo(const job_startinfo_t* src) +{ + job_startinfo_t* d; + + if (src == NULL) { + return NULL; + } + + d = (job_startinfo_t*)calloc(1, sizeof(*d)); + if (d == NULL) { + return NULL; + } + +#define CLONE_STR_FIELD(f) \ + do { \ + if (src->f != NULL) { \ + d->f = strdup(src->f); \ + if (d->f == NULL) { \ + goto fail; \ + } \ + } \ + } while (0) + + CLONE_STR_FIELD(exechome); + CLONE_STR_FIELD(execcwd); + CLONE_STR_FIELD(execuser); + CLONE_STR_FIELD(cmdline); + CLONE_STR_FIELD(exdata); + CLONE_STR_FIELD(launch_type); + CLONE_STR_FIELD(q_pre_exec); + CLONE_STR_FIELD(q_post_exec); + CLONE_STR_FIELD(job_pre_exec); + CLONE_STR_FIELD(job_post_exec); + +#undef CLONE_STR_FIELD + + return d; + +fail: + free_JobStartInfo_ptr(&d); + return NULL; +} + +static job_submit_data_t* clone_job_submit_data(const job_submit_data_t* src) +{ + job_submit_data_t* d; + + if (src == NULL) { + return NULL; + } + + d = (job_submit_data_t*)calloc(1, sizeof(*d)); + if (d == NULL) { + return NULL; + } + + d->mintasks = src->mintasks; + d->maxtasks = src->maxtasks; + +#define CLONE_STR_FIELD(f) \ + do { \ + if (src->f != NULL) { \ + d->f = strdup(src->f); \ + if (d->f == NULL) { \ + goto fail; \ + } \ + } \ + } while (0) + + CLONE_STR_FIELD(queue); + CLONE_STR_FIELD(user); + CLONE_STR_FIELD(job_name); + CLONE_STR_FIELD(project_name); + CLONE_STR_FIELD(user_group); + CLONE_STR_FIELD(job_description); + CLONE_STR_FIELD(resreq); + +#undef CLONE_STR_FIELD + + if (src->start_info != NULL) { + d->start_info = clone_job_startinfo(src->start_info); + if (d->start_info == NULL) { + goto fail; + } + } + + return d; + +fail: + free_JobSubmitData_ptr(&d); + return NULL; +} + +/* One array element, freshly-submitted: JOB_STAT_PEND / receive-time + * sub_time, mirroring the state a genuinely-new single-job push arrives + * with. The compact wire descriptor (Req 1.1/6.1) carries only (jobid, + * array_size, submit) by design -- no per-element timestamps -- so there is + * no "real" sub_time to recover here. */ +static xce_job_t* make_array_element(const xce_job_submit_t* job_array, + uint64_t jobid) +{ + xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); + + if (j == NULL) { + return NULL; + } + + j->jobid = jobid; + j->state = JOB_STAT_PEND; + j->sub_time = (uint64_t)time(NULL); + + j->submit = clone_job_submit_data(job_array->submit); + if (j->submit == NULL) { + free(j); + return NULL; + } + + return j; +} + +static int sched_on_job_array(xce_scheduler_t* sched, + xce_job_submit_t* job_array) +{ + xce_job_t** jobs; + uint64_t base_jobid; + uint64_t array_size; + uint64_t i; + int n_new = 0; + int rc; + char jobidString[128]; + + if ((sched == NULL) || (job_array == NULL) || (job_array->submit == NULL) + || (job_array->array_size == 0)) { + return -EINVAL; + } + + base_jobid = job_array->jobid; + array_size = job_array->array_size; + + jobs = (xce_job_t**)calloc((size_t)array_size, sizeof(xce_job_t*)); + if (jobs == NULL) { + return -ENOMEM; + } + + for (i = 0; i < array_size; i++) { + uint64_t jobid = base_jobid + i; + + /* Req 6.5/8.4: deduplicate a redelivered push by element identity -- + * reuse the SAME jobid-keyed exists-check (sched_get_job against + * sched->jobs) the ordinary single-job path relies on downstream + * (sched_add_job -> key_add -> -EEXIST), applied here per-element so + * an already-materialized element is never re-queued. Not a new + * dedup mechanism. */ + if (sched_get_job(sched, jobid) != NULL) { + continue; + } + + jobs[n_new] = make_array_element(job_array, jobid); + if (jobs[n_new] == NULL) { + int j; + for (j = 0; j < n_new; j++) { + free_Job_ptr(&jobs[j]); + } + free(jobs); + return -ENOMEM; + } + n_new++; + } + + if (n_new == 0) { + /* Every index already exists: a fully-duplicate redelivery. Nothing + * new to schedule, and nothing to confirm -- Req 8.4 ties the + * confirmation log to a successful *expansion*; a redelivery that + * expands zero new elements is a no-op, not a fresh expansion + * (task 5.3 Observable note). */ + free(jobs); + return 0; + } + + fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid); + INFO("job array push received and expanded: base_jobid=%s " + "array_size=%llu new_tasks=%d", + jobidString, (unsigned long long)array_size, n_new); + + rc = sched_on_jobs(sched, n_new, jobs); + if (rc < 0) { + for (i = 0; i < (uint64_t)n_new; i++) { + free_Job_ptr(&jobs[i]); + } + free(jobs); + } + /* On success sched_on_jobs()/the pend-update killer now owns jobs[] and + * every element in it (mirrors the existing Data_data_schedjobs case + * below, which hands dataset->rows to sched_on_jobs the same way). */ + + return rc; +} + static int sched_on_workers(xce_scheduler_t* sched, int nworkers, xce_worker_t** workers) { @@ -143,6 +359,15 @@ static int sched_handle_dataset(xce_scheduler_t* sched, dataset_t* dataset) dataset->rows = NULL; } break; + case Data_data_jobArray: + rc = sched_on_job_array(sched, (xce_job_submit_t*)dataset->rows); + /* Unlike the list-shaped cases above, we never alias dataset->rows + * 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. */ + break; } return rc; diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index c438d63..25ccb19 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -30,6 +30,18 @@ sched_handlers_test = executable('sched_handlers_test', dependencies: [scheduler_test_deps], install: false) +# White-box test for the compact job-array push receive/expand path +# (job-array-single-wal-record task 5.3). It also #includes ../src/handler.c +# directly, but additionally needs a REAL sched->jobs kvmgr (sched_get_job/ +# sched_add_job) to exercise the per-element dedup check for real — kept as +# its own binary rather than folded into sched_handlers_test so that file's +# "self-contained, only sched_init_config stubbed" story stays accurate. +sched_job_array_test = executable('sched_job_array_test', + ['test_job_array_expand.c'], + include_directories:[incdir], + dependencies: [scheduler_test_deps], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -38,3 +50,5 @@ test('scheduler job-update test', sched_job_update_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) test('scheduler handlers test', sched_handlers_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler job-array expand test', sched_job_array_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_handlers.c b/src/scheduler/testsuite/test_handlers.c index b7cc423..2287bc1 100644 --- a/src/scheduler/testsuite/test_handlers.c +++ b/src/scheduler/testsuite/test_handlers.c @@ -19,16 +19,24 @@ /* Pull the real handlers under test into this TU. handler.c is self-contained * for the enqueue path: enq_pend_update and sched_job_state_change are both - * static within it; the dispatchers call lib helpers (linked). The only - * scheduler-internal extern handler.c references (in the config-init request - * path, not exercised by these enqueue tests) is sched_init_config — stubbed. */ -#include "../src/handler.c" - + * static within it; the dispatchers call lib helpers (linked). The + * scheduler-internal externs handler.c references but this file does not + * exercise are stubbed below: sched_init_config (config-init request path) + * and sched_get_job (job-array-single-wal-record task 5.3's per-element + * dedup check, Data_data_jobArray — not tested here, see + * test_job_array_expand.c for the real kvmgr-backed sched_get_job). */ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) { (void)s; (void)c; return 0; } +xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +{ + (void)sched; (void)jobid; return NULL; +} + +#include "../src/handler.c" + void setUp(void) { return; } void tearDown(void) { diff --git a/src/scheduler/testsuite/test_job_array_expand.c b/src/scheduler/testsuite/test_job_array_expand.c new file mode 100644 index 0000000..0ea8ba9 --- /dev/null +++ b/src/scheduler/testsuite/test_job_array_expand.c @@ -0,0 +1,261 @@ +/* + * test_job_array_expand.c - white-box tests for the scheduler's compact + * job-array push receive path (job-array-single-wal-record task 5.3, Req + * 6.3, 6.4, 6.5, 8.4). + * + * Strategy: pull the real (static) sched_handle_dataset / sched_on_job_array + * into this TU via #include "../src/handler.c" (same white-box pattern as + * test_handlers.c), driven against a REAL sched->jobs kvmgr so the + * per-element dedup check (sched_get_job) is exercised for real rather than + * faked. sched_add_job/sched_get_job below are the SAME kvmgr key_add/ + * key_find path production uses (mirrors test_job_update.c's precedent). + */ +#define UNIT_TEST 1 +#include +#include +#include +#include + +#include + +#include "kvmgr.h" +#include "scheduler.h" + +/* Pull the real handler-under-test into this TU. handler.c is otherwise + * self-contained for the dataset dispatch path (enq_pend_update is static + * within it); the only scheduler-internal extern it references that lives + * elsewhere is sched_init_config, stubbed below (same as test_handlers.c). */ +#include "../src/handler.c" + +int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c) +{ + (void)s; (void)c; return 0; +} + +/* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors + * test_job_update.c's precedent for exercising sched_get_job "for real"). */ +int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) +{ + int rc; + lock(sched->jobs, RTTI_LOCK_WRITE); + rc = key_add(sched->jobs, (void*)j->jobid, j); + unlock(sched->jobs); + return rc; +} + +xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +{ + int rc; + xce_sched_job_t* job = NULL; + lock(sched->jobs, RTTI_LOCK_READ); + rc = key_find(sched->jobs, (void*)jobid, (void**)&job); + unlock(sched->jobs); + if (rc <= 0) { + return NULL; + } + return job; +} + +void setUp(void) { return; } +void tearDown(void) +{ + printf("\n-----------------------------------------------------\n"); + return; +} + +static void fixture_init(xce_scheduler_t* sched) +{ + kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT, + .nslots = 59, + .delta = 16, + .datakiller = NULL }; + + memset(sched, 0, sizeof(*sched)); + mylock_init(&(sched->lock)); + sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->pend_updates[0], NULL); + sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); + dataq_init(sched->pend_updates[1], NULL); + sched->jobs = makeInstance(&xkvmgr_type, &jobs_params); +} + +static void seed_existing_job(xce_scheduler_t* sched, uint64_t jobid) +{ + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + job->jobid = 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) +{ + memset(submit, 0, sizeof(*submit)); + submit->queue = "normal"; + submit->user = "tester"; + submit->mintasks = 1; + submit->maxtasks = 1; + submit->resreq = "type == local"; + + memset(ja, 0, sizeof(*ja)); + ja->jobid = base_jobid; + ja->array_size = array_size; + ja->submit = submit; +} + +/* Pop one pending update request, asserting it is an UPDATE_JOB op; hands + * back njobs/jobs (caller now owns them -- free via free_Job_ptr + free() to + * mirror sched_pend_update_killer). Returns -1 if the queue is empty. */ +static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) +{ + data_item_t* it = dataq_deq(sched->pend_updates[0], NULL); + xce_sched_update_request_t* req; + int njobs; + + if (it == NULL) { + return -1; + } + + req = ITPTR_TO_UPDATE_REQ(it); + TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB, req->op); + + njobs = req->u.jobupdate.njobs; + if (jobs_out != NULL) { + *jobs_out = req->u.jobupdate.jobs; + } + free(req); + return njobs; +} + +/* ── Req 6.3/6.4: a compact Data.jobArray push expands locally into N + * ordinary schedulable tasks via base_jobid + i arithmetic ── */ + +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; + dataset_t dataset; + xce_job_t** jobs = NULL; + int njobs; + int i; + + fixture_init(&sched); + init_job_array(&ja, &submit, 5000, 3); + + memset(&dataset, 0, sizeof(dataset)); + dataset.table = Data_data_jobArray; + dataset.rows = (void*)&ja; + + TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0); + + njobs = pop_job_update(&sched, &jobs); + TEST_ASSERT_EQUAL_INT(3, njobs); + TEST_ASSERT_EQUAL_UINT64(5000, jobs[0]->jobid); + TEST_ASSERT_EQUAL_UINT64(5001, jobs[1]->jobid); + TEST_ASSERT_EQUAL_UINT64(5002, jobs[2]->jobid); + TEST_ASSERT_EQUAL_STRING("normal", jobs[0]->submit->queue); + TEST_ASSERT_EQUAL_STRING("normal", jobs[2]->submit->queue); + /* every element gets its OWN independent clone, not a shared pointer -- + * required so each is independently free_Job_ptr()-able downstream + * (sched_pend_update_killer) without a double-free. */ + TEST_ASSERT_TRUE(jobs[0]->submit != jobs[1]->submit); + TEST_ASSERT_TRUE(jobs[1]->submit != jobs[2]->submit); + + for (i = 0; i < njobs; i++) { + free_Job_ptr(&jobs[i]); + } + free(jobs); +} + +/* ── Req 6.5/8.4: a redelivered push (e.g. after a scheduler reconnect) + * must not create duplicate tasks -- reuse the jobid-keyed exists-check + * (sched_get_job) the ordinary single-job path relies on ── */ + +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; + + /* first delivery: 2 new tasks enqueued */ + TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0); + njobs = pop_job_update(&sched, &jobs); + TEST_ASSERT_EQUAL_INT(2, njobs); + for (i = 0; i < njobs; i++) { + /* simulate the async consumer (scheduler.c's sched_update_jobs -> + * sched_add_job, not linked into this white-box TU) materializing + * the element into sched->jobs, so the redelivery's dedup check + * sees them as already existing -- exactly what a real reconnect + * scenario looks like from the handler's point of view. */ + seed_existing_job(&sched, jobs[i]->jobid); + free_Job_ptr(&jobs[i]); + } + free(jobs); + + /* redelivery of the IDENTICAL push: every index already exists */ + TEST_ASSERT_EQUAL_INT(0, sched_handle_dataset(&sched, &dataset)); + TEST_ASSERT_EQUAL_INT(-1, pop_job_update(&sched, NULL)); /* nothing new */ +} + +/* ── Guard-clause coverage on the decode step itself ── */ + +static void test_job_array_null_submit_returns_einval(void) +{ + xce_scheduler_t sched; + xce_job_submit_t ja; + + fixture_init(&sched); + + memset(&ja, 0, sizeof(ja)); + ja.jobid = 1; + ja.array_size = 3; + ja.submit = NULL; + + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja)); +} + +static void test_job_array_zero_size_returns_einval(void) +{ + xce_scheduler_t sched; + job_submit_data_t submit; + xce_job_submit_t ja; + + fixture_init(&sched); + init_job_array(&ja, &submit, 42, 0); + + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja)); +} + +static void test_job_array_null_descriptor_returns_einval(void) +{ + xce_scheduler_t sched; + fixture_init(&sched); + + TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, NULL)); +} + +int main(int argc, char* argv[]) +{ + (void)argc; (void)argv; + UNITY_BEGIN(); + RUN_TEST(test_job_array_push_expands_into_n_tasks, 0); + RUN_TEST(test_job_array_redelivery_does_not_duplicate, 0); + RUN_TEST(test_job_array_null_submit_returns_einval, 0); + RUN_TEST(test_job_array_zero_size_returns_einval, 0); + RUN_TEST(test_job_array_null_descriptor_returns_einval, 0); + return UNITY_END(); +}