From 7c755dc4d7f1f1f6ebc2d5c81a52be5436287a4b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 14 Jul 2026 14:16:29 +0000 Subject: [PATCH] feat(job-array-single-wal-record): implement restore_job_array primitive 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. --- .../job-array-single-wal-record/design.md | 485 +++++++++++++++++ .../job-array-single-wal-record/tasks.md | 3 +- src/controller/include/cluster.h | 27 + src/controller/job.c | 334 ++++++++++++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_job_array_apply.c | 495 ++++++++++++++++++ 7 files changed, 1346 insertions(+), 1 deletion(-) create mode 100644 .kiro/specs/job-array-single-wal-record/design.md create mode 100644 src/controller/testsuite/test_job_array_apply.c diff --git a/.kiro/specs/job-array-single-wal-record/design.md b/.kiro/specs/job-array-single-wal-record/design.md new file mode 100644 index 0000000..3df3f63 --- /dev/null +++ b/.kiro/specs/job-array-single-wal-record/design.md @@ -0,0 +1,485 @@ +# Design Document: job-array-single-wal-record + +## Overview + +**Purpose**: This feature collapses a job-array submission's persistence, replication, and downstream-notification cost from O(array size) to O(1). Today, `handle_jobsubmit()` builds one WAL payload per array element and appends them through `persist_submit_job_payloads_async()`, registering one commit continuation per element (job.c:1055). A 10,000-element array therefore produces 10,000 WAL appends and 10,000 commit registrations before a single client reply is sent. This design replaces that per-element loop with one `WAL_JOB_ARRAY_SUBMIT` record — base job spec plus `(base_jobid, array_size)` — committed through exactly one majority-commit barrier, and expanded into individual `job_t` entries inside the existing `apply_committed_entry()`/`apply_wal_entry()` seam shared by live apply, restart replay, and standby catch-up. + +**Users**: Cluster operators submitting large job arrays (commit-engine load, replication bandwidth, and failover replay time no longer scale linearly with array size); maintainers of the controller's submit/apply/push code (one array-shaped record instead of N single-job records to reason about). + +**Impact**: Adds one new `wal_entry_type` (`WAL_JOB_ARRAY_SUBMIT`) alongside the unchanged single-job `WAL_JOB_SUBMIT`; adds one new shared expansion primitive (`restore_job_array()`) consumed by both live dispatchers of committed WAL entries; adds a compact push variant to the existing scheduler/API-server notification path; closes a live apply-pump gap (a resource-exhaustion apply failure is currently silently skipped rather than retried); hardens five specific, pre-existing gaps in `src/lib/jobj.c`'s job-construction primitives that this feature's N-way apply-time loop amplifies. + +### Goals +- One WAL record, one majority-commit barrier, one scheduler push, one API-server push per job-array submission — regardless of array size. +- Online apply, restart replay, and standby catch-up expand a committed array record identically, through one shared primitive. +- A committed array is applied deterministically: content-derived failures are impossible post-commit (caught pre-commit); resource-exhaustion failures fence the affected node without rejecting the array or diverging from peers, and self-heal automatically. +- Zero behavior change to non-array (single-job) submission: same WAL format, same commit-barrier count, same wire shape. + +### Non-Goals +- Per-element override semantics (array elements remain uniform clones of one base spec, distinguished only by jobid — no client-facing API change). +- Bulk query/cancel of a whole array as one operation (deferred; array elements can be in heterogeneous states at request time). +- Any change to the commit engine's or commit barrier's internal mechanics (consumed unchanged from `ha-commit-engine`/`ha-commit-barrier-rollout`). +- Scheduler dispatch/allocation policy for array elements (scheduled as ordinary single jobs once expanded). + +## Boundary Commitments + +### This Spec Owns +- The `WAL_JOB_ARRAY_SUBMIT` wire format (reusing the existing `JobSubmit` capnp struct) and its encode/append/commit path in `handle_jobsubmit()`. +- The shared array-expansion primitive (`restore_job_array()`) and its wiring into both live dispatchers of committed WAL entries (`apply_committed_entry()`, `apply_wal_entry()`). +- Atomicity of one array record's apply: all-or-nothing per attempt, idempotent re-apply, no partial/divergent element sets ever observable. +- The compact scheduler/API-server push variant for a committed array and the corresponding local-expansion logic on both receivers. +- The live apply-pump return-value fix (currently discarded) and the new live, self-healing, non-promotable fence for a resource-exhaustion apply failure. +- Hardening `src/lib/jobj.c`'s `job_init`/`makeJobsub`/`job_destroy`/`decode_Job`/`free_Job` for the five specific gaps requirements.md names, since the apply-time N-way loop amplifies each by array size. + +### Out of Boundary +- The apply-ordering/hook-decoupling refactor itself (delivered by `raft-apply-decoupling`, consumed unchanged) — `persist_job_hook_fun`'s `WAL_JOB_SUBMIT` branch is already a no-op and stays untouched. +- The non-array single-job submission WAL format, commit path, or reply plumbing — `array_size == 1` continues through today's exact code, unmodified. +- The commit engine's/commit barrier's internal mechanics — consumed via `commit_submit_async()` unchanged. +- Bulk array cancel/query as one client operation (deferred to a future spec — heterogeneous per-element state at request time is a materially different problem). +- Scheduler placement/dispatch policy for array elements — out of scope once elements are ordinary schedulable jobs. + +### Allowed Dependencies +- `raft-apply-decoupling` (complete): `apply_committed_entry()` and its dispatch contract, `persist_apply_pump_run()`'s LSN-ordered walk, `persist_register_pending_apply()`/`persist_take_pending_apply()` hand-off, `jobsubmit_async_ctx_t`'s batched-reply mechanism (consumed as-is, `pending` set to 1 for the array path instead of N). +- `ha-commit-engine`/`ha-commit-barrier-rollout` (complete): `commit_submit_async()`, per-LSN idempotency guards — consumed unchanged. +- bug-1419's in-flight jobid-range registry (`inflight_range_try_add`/`inflight_range_remove`/`jobid_range_exists`, job.c) — already range-shaped; consumed unchanged, called exactly once per array submission as it is today. +- `idgen()`'s existing `MAX_ARRAY_IDX` (32767) cap on `array_size`, enforced at jobid-minting time in the Go apiserver before a request ever reaches the controller — relied on as the structural bound satisfying Requirement 7, not re-implemented. +- `scheduler-controller-msg-gap` (complete): the controller→scheduler push protocol and its established pattern for adding a new `Data` union variant. + +### Revalidation Triggers +- Any change to `apply_committed_entry()`'s or `apply_wal_entry()`'s dispatch signature or contract (per `raft-apply-decoupling`'s own trigger: a new WAL entry type not added to both switches silently fails to apply). +- Any change to `struct JobSubmit`'s capnp shape (`controller.capnp:533`) — it is now load-bearing for three wire uses (client request, WAL payload, push payload), not one. +- Any change to `jobsub_table`/`makeJobsub()`'s interning contract in `src/lib/jobj.c` — every array element's `job_t` construction depends on it. +- Any change to `ha.c`'s promotion-gate eligibility conditions — the new Live Apply Resource Fence check must stay wired at both existing call sites (periodic release, election-time). +- Any change to `MAX_ARRAY_IDX`/`N_ARRAY_IDX_BITS` (`src/include/config.h`) or to `idgen()`'s `arraySize` validation — Requirement 7's "no chunking needed" conclusion depends on this cap continuing to hold upstream of the controller. + +## Architecture + +### Existing Architecture Analysis + +The brief's original framing (`job_submit()` looping `create_job()` per element) predates `raft-apply-decoupling` task 3.2 and is stale. Verified current state: + +| Aspect | Current state | Gap vs. this design | +|---|---|---| +| Array submission WAL cost | `handle_jobsubmit()` → `build_job_submit_payloads()` builds N plain `xce_job_t` payloads (job.c:221) → `persist_submit_job_payloads_async()` appends N `WAL_JOB_SUBMIT` entries, registers N commit continuations (persist.c:4846) | Requirement 1: N WAL appends / N commit barriers must become 1 | +| `create_job()` call site | Only inside the apply seam, via `restore_job()` (job.c:307) — never in `handle_jobsubmit` (removed by task 3.2) | None — this is already the correct location for this design's N-way expansion loop | +| Client reply batching | Already one reply per request regardless of `array_size`, via `jobsubmit_async_ctx_t` (job.c:754) and `pending` refcounting | None — reused as-is; the array path sets `pending = 1` instead of `plist.count` | +| Live apply dispatch | **Two** live dispatchers of `WAL_JOB_SUBMIT`, not one: `apply_committed_entry()` (persist.c:2721, primary live apply + boot replay) and `apply_wal_entry()` (persist.c:2574, standby catch-up/replicated-apply, explicitly not retired — persist.c:3722) | Requirement 3: both must expand an array record identically; today they share single-job construction only via `restore_job()` | +| Live apply-pump failure handling | `persist_apply_pump_run()` discards `apply_committed_entry()`'s return value (`(void)apply_committed_entry(...)`, persist.c:3281) and unconditionally advances `last_applied_lsn` | Requirement 9.4/9.5: a resource-exhaustion apply failure would silently be skipped forever, never retried, never fenced | +| Promotion-gate conditions | `ha.c` gates on `truncated`, `durable_floor` vs HWM, `apply_verdict == APPLY_DIVERGENT`, `self_restart_pending` — never on `last_applied_lsn` vs `committed_lsn` | Requirement 9.4: a node stuck on a resource-exhaustion apply failure looks fully promotable today | +| Client-facing array wire format | `struct JobSubmit` (`controller.capnp:533`) already carries `(jobid, arraySize, submit: JobSubmitData)` | None — this is exactly the compact shape Requirement 1/6 need; reused verbatim, not redesigned | +| Array-size bound | `idgen()` already rejects `arraySize > MAX_ARRAY_IDX + 1` (32768) at jobid-minting time, upstream of the controller (`clib.go:165`) | None — Requirement 7 is satisfied by this pre-existing cap plus the record's O(1) payload; no chunking policy needed | +| Scheduler/API-server push | `sched_job_new_fun`/`apiserver_job_new_fun` unconditionally call `xcesched_push_joblist`/`apiserver_push_joblist` with whatever `xlist_t` the hook fire carries — already N-capable by call count | Requirement 6: the wire payload itself still enumerates N full `Job`/`SchedJob` structs; a compact descriptor push is a new wire path, not a byproduct of the existing one | + +### Architecture Pattern & Boundary Map + +```mermaid +graph TB + Client[Client array submit] + Handler[handle_jobsubmit] + ArrayAppend[Single array WAL append] + CommitEngine[Commit engine one barrier] + ApplyPumpLive[Apply pump live] + ApplyPumpReplay[Boot replay] + StandbyCatchup[Standby catch up] + RestoreArray[restore_job_array shared helper] + Fence[Live apply resource fence] + HookFire[HOOK_DETAIL_JOB_NEW one fire N jobs] + SchedPush[Compact push to scheduler] + ApiPush[Compact push to apiserver] + SchedExpand[Scheduler local expansion] + ApiExpand[Apiserver local expansion] + + Client --> Handler + Handler --> ArrayAppend + ArrayAppend --> CommitEngine + CommitEngine -->|majority ack| ApplyPumpLive + ApplyPumpLive --> RestoreArray + ApplyPumpReplay --> RestoreArray + StandbyCatchup --> RestoreArray + RestoreArray -->|resource exhaustion| Fence + RestoreArray -->|success| HookFire + HookFire --> SchedPush + HookFire --> ApiPush + SchedPush --> SchedExpand + ApiPush --> ApiExpand +``` + +**Architecture Integration**: +- **Selected pattern**: extend the existing commit-then-apply pipeline (`raft-apply-decoupling`) with one new WAL entry type and one new shared expansion primitive, rather than introducing a parallel array-specific pipeline. The array record travels through the exact same append → majority-commit → apply-pump → hook-fire stages a single-job submission does; only the payload shape and the expansion fan-out differ. +- **Domain/feature boundaries**: `restore_job_array()` is the only place array-record bytes become N `job_t` entries — both live dispatchers call it, neither reimplements it, mirroring how `restore_job()` already anchors single-job construction for both dispatchers today. +- **Existing patterns preserved**: `jobsubmit_async_ctx_t`'s batched single-reply mechanism, `persist_register_pending_apply()`'s command hand-off, the in-flight jobid-range registry, the `t_apply_seam_notifying` thread-local guard, `wal_classify`'s WAIT/RETAIN table, the `Data` union's additive-variant extension pattern. +- **New components rationale**: `restore_job_array()` (no array-aware expansion primitive exists); the Live Apply Resource Fence (no live-path analog of the boot-only `recovery_unrecoverable`/`persist_recover_loud_fence_if_unrecoverable` fence exists); the compact push variant (existing push wire format is per-job, not descriptor-shaped). +- **Steering compliance**: preserves the synchronous-commit-as-policy invariant (`ha-async-io.md`) — the array append still runs off the iomgr thread through the unchanged commit-engine contract; the Live Apply Resource Fence follows the controller's Process Lifecycle invariant exactly (`src/controller/CLAUDE.md`: loud ERROR, non-promotable, stay alive, keep retrying, self-heal — never self-terminate). + +### Technology Stack + +| Layer | Choice / Version | Role in Feature | Notes | +|-------|------------------|------------------|-------| +| Controller core | C (gnu17), existing | All changes land in `src/controller/{job.c,persist.c,sched.c,apiserver.c,ha.c}` and `src/lib/jobj.c` | No new language/runtime | +| Serialization | Cap'n Proto (existing `controller.capnp`) | `WAL_JOB_ARRAY_SUBMIT` payload and the new compact push payload both reuse the existing `struct JobSubmit`; one new `Data` union variant added for the push path | No new struct; `encode_JobSubmit`/`decode_JobSubmit` already generated | +| Scheduler daemon | C (gnu17), existing | `src/scheduler/src/handler.c` gains a sibling case to `Data_data_schedjobs` for local array expansion | Follows `scheduler-controller-msg-gap`'s established extension pattern | +| API server | Go 1.23.4, existing | `src/frontend/apiserver/backend/` gains a sibling decode path to `cJobToModel` for local array expansion | CGO bridge, consistent with existing job-push consumption | +| Test harness | `controller_unit_test`, `xcelib_test`, `/xce-test-ha` | Unit coverage for the new apply path and fence; live soak for non-regression | Extends existing harnesses, no new framework | + +## File Structure Plan + +### Modified Files +- `src/include/wal_types.h` — add `WAL_JOB_ARRAY_SUBMIT` to `wal_entry_type` (a new value distinct from `WAL_JOB_SUBMIT`, keeping the single-job format untouched per Requirement 8.1). +- `src/controller/job.c` — `handle_jobsubmit()`: branch on `submit->array_size == 1` (today's exact code path, unmodified) vs. `> 1` (new single-append path: encode once via a new `capnp_encode_job_array()`-equivalent using `encode_JobSubmit`, append via a new `append_job_array_wal_entry_local()`, register one continuation via a new `persist_submit_job_array_async()`, set `actx->pending = 1`). `jobsubmit_commit_continuation`/`jobsubmit_apply_complete`/`jobsubmit_async_ctx_t` are reused unmodified — both are already parametric on `pending` and keyed by LSN, not by element count. New shared helper `restore_job_array()` (sibling to `restore_job()`, same file) implements atomic N-element construction: scratch-builds all elements first, bulk-inserts into `c->jobmgr` only once every element succeeds, and discards the scratch list directly (never touching `c->jobmgr`) on failure. +- `src/controller/persist.c` — `wal_classify()`: add `WAL_JOB_ARRAY_SUBMIT → WAL_MODE_WAIT` (same classification as `WAL_JOB_SUBMIT`). `apply_committed_entry()` and `apply_wal_entry()`: each gains one new `case WAL_JOB_ARRAY_SUBMIT` calling the shared `restore_job_array()`; `apply_committed_entry()`'s case additionally builds the N-element `xlist_t` and fires `HOOK_DETAIL_JOB_NEW` once (mirroring its existing single-job case); `apply_wal_entry()`'s case stays hook-free (mirroring its existing single-job case). `persist_apply_pump_run()`: capture `apply_committed_entry()`'s return value; on negative rc, stop the walk without advancing `last_applied_lsn`, log loud, engage the Live Apply Resource Fence (mirrors the existing `apply_pump_read_entry` failure branch immediately above it); on success following a prior failure at this LSN, clear the fence. New small primitives: `persist_apply_resource_fenced()`, `persist_mark_apply_resource_fenced()`, `persist_clear_apply_resource_fence()`. +- `src/controller/ha.c` — both promotion-gate eligibility call sites (periodic fence-release check, election-time check) gain one additional required condition: `!persist_apply_resource_fenced(&c->persist)`, threaded alongside the existing `apply_divergent`/`self_restart_pending` conditions. +- `src/controller/sched.c` — `sched_job_new_fun()`: branch on `l->count > 1` (the only way a `HOOK_DETAIL_JOB_NEW` fire ever carries more than one element, per Requirement 3.4) to call a new `xcesched_push_jobarray()` (encodes the compact `Data` union array variant) instead of `xcesched_push_joblist()`. +- `src/controller/apiserver.c` — `apiserver_job_new_fun()`: same `l->count > 1` branch, calling a new `apiserver_push_jobarray()`. +- `src/schema/controller.capnp` — `struct Data`'s union gains one new variant reusing `JobSubmit` (e.g. `jobArray @6: JobSubmit $C.mapname("job_array")`), following the existing additive-variant pattern (`schedjobs`/`jobs`/`workers`/`resources`/`resvals`). +- `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). + +### 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. +- `src/controller/testsuite/test_job_array_fence.c` — unit tests: `persist_apply_pump_run()` halts and does not advance `last_applied_lsn` on a negative `apply_committed_entry()` return; Live Apply Resource Fence engages, blocks promotion eligibility, and self-clears on a successful retry (Requirement 9.6's fault-injection scenario). +- `src/lib/testsuite/test_jobj_hardening.c` — unit tests for the five Requirement 10 gaps: `job_init` propagates a `makeJobsub()`/`makeString()` NULL; `jobsub_table` is initialized before first use under the previously-observed ordering; MOVE_PTR fields are confirmed NULL on a freshly-built array-element payload; `job_destroy` no longer leaks `accounts`; `decode_Job`/`free_Job` `ckptinfo` coverage is confirmed correct as-is (no wire field exists, so no leak). +- Extend `/xce-test-ha`'s soak scenario set with a large-array submission case (commit/push count assertions) and a chaotic-failover case submitting an array immediately before killing the primary (Requirement 8.2, 8.3). + +## System Flows + +### Array submission: append, commit, apply, push + +```mermaid +sequenceDiagram + participant C as Client + participant H as handle_jobsubmit + participant CE as Commit engine + participant AP as Apply pump + participant RA as restore_job_array + participant N as HOOK_DETAIL_JOB_NEW + participant S as Scheduler push + participant A as Apiserver push + + C->>H: array submit base spec plus array_size + H->>H: validate, jobid_range_exists, inflight_range_try_add (unchanged, range-shaped) + H->>CE: append ONE WAL_JOB_ARRAY_SUBMIT, register ONE continuation + CE-->>CE: majority ack (one barrier) + CE->>AP: resolve committed, hand off to apply pump + AP->>RA: restore_job_array (base_jobid, array_size, base spec, lsn) + alt all N constructed + RA->>N: fire once, N-element list + N->>S: compact push (base spec + range) + N->>A: compact push (base spec + range) + AP->>H: reply success (base jobid) + else resource exhaustion mid-loop + RA->>RA: discard scratch list (c->jobmgr never touched) + RA-->>AP: negative rc + Note over AP: last_applied_lsn NOT advanced; fence engaged; retried next kick + end +``` + +Key decision not obvious from the diagram: `restore_job_array()` builds every element in a scratch list and only bulk-inserts into `c->jobmgr` once the whole array has successfully constructed — so the array is never observably partial to a concurrent reader on the *success* path either, not only when a retry starts over from index 0 after a failure (Requirement 2.3, 9.4). + +### Live apply resource fence: engage and self-heal + +```mermaid +flowchart LR + Apply[apply_committed_entry fails resource exhaustion] --> Log[Loud ERROR logged] + Log --> Fence[Live Apply Resource Fence engaged] + Fence --> Gate[Promotion gate refuses this node] + Gate --> Retry[Apply pump retries same LSN on next kick] + Retry -->|still failing| Log + Retry -->|succeeds| Clear[Fence cleared by apply pump itself] + Clear --> Eligible[Node eligible for promotion again] +``` + +Key decision: the fence clears from inside the apply pump the moment the retried LSN succeeds — it does not depend on an election or on the unrelated WAL-durability catch-up check ever running, so a long-lived primary that never loses leadership still self-heals (Requirement 9.5). This is deliberately a *new*, narrowly-scoped fence rather than a reuse of `recovery_unrecoverable`, whose existing periodic clear is gated on WAL durability catch-up alone and would wrongly clear before the stuck apply actually resolves (see `research.md`, Design Decisions). + +## Requirements Traceability + +| Requirement | Summary | Components | +|---|---|---| +| 1.1-1.3 | Single WAL record, single commit barrier, no per-element append | WAL Array Submission | +| 2.1-2.3 | Atomic, all-or-nothing commit per record | WAL Array Submission (commit), Array Apply Expansion (no partial construction from an uncommitted record) | +| 3.1-3.4 | Shared expand-on-apply, online = replay = standby catch-up, idempotent re-apply, one hook fire | Array Apply Expansion | +| 4.1-4.3 | Standby reconstruction from one committed position | WAL Array Submission (replication carries one record, unchanged replication mechanics), Array Apply Expansion | +| 5.1-5.3 | Preserved per-element identity/addressing | Array Apply Expansion (unchanged `base_jobid + i` arithmetic — verification-only, no new mechanism) | +| 6.1-6.5 | Single compact push to scheduler and API server, local expansion | Compact Scheduler/API-Server Push | +| 7.1-7.2 | Bounded record size | WAL Array Submission (satisfied by construction — see Architecture table) | +| 8.1-8.4 | No regression to non-array submission or existing tests | Non-Regression & Testing (cross-cutting; `array_size == 1` path untouched) | +| 9.1-9.6 | Deterministic, total apply — pre-commit rejection, live resource fence | WAL Array Submission (9.1, 9.2 — pre-commit validation), Array Apply Expansion (9.3 — total apply post-commit), Live Apply Resource Fence (9.4, 9.5, 9.6) | +| 10.1-10.5 | jobj.c hardening | jobj.c Hardening | + +## Components and Interfaces + +| Component | Domain/Layer | Intent | Req Coverage | Key Dependencies (P0/P1) | Contracts | +|---|---|---|---|---|---| +| WAL Array Submission | job.c, persist.c, wal_types.h | Encode and commit one array record through one barrier | 1, 4.1-4.2, 7, 9.1, 9.2 | Commit Engine (P0), in-flight range registry (P0) | Service, Batch | +| Array Apply Expansion | persist.c, job.c (`restore_job_array`) | Expand a committed array record into N `job_t` entries, identically for live apply / replay / standby catch-up | 2, 3, 4.3, 5, 9.3 | `apply_committed_entry`/`apply_wal_entry` (P0), jobj.c primitives (P0) | Service, Batch | +| Live Apply Resource Fence | persist.c, ha.c | Fence a node non-promotable on a resource-exhaustion apply failure, self-heal on success | 9.4, 9.5, 9.6 | Apply pump (P0), promotion gate (P0) | State | +| Compact Scheduler/API-Server Push | sched.c, apiserver.c, controller.capnp, scheduler `handler.c`, apiserver `backend/` | One compact push per committed array, expanded locally by each receiver | 6 | `HOOK_DETAIL_JOB_NEW` (P0), `Data` union (P0) | Event, API | +| jobj.c Hardening | src/lib/jobj.c | Reliable, leak-free job construction under N-element reuse | 10 | RTTI object system (P0) | Service | + +### WAL Array Submission + +#### Single array record append and commit + +| Field | Detail | +|-------|--------| +| Intent | Encode the whole array as one `WAL_JOB_ARRAY_SUBMIT` record and commit it through one barrier | +| Requirements | 1.1, 1.2, 1.3, 4.1, 4.2, 7.1, 7.2, 9.1, 9.2 | + +**Responsibilities & Constraints** +- `handle_jobsubmit()` branches on `array_size`: `== 1` runs today's unmodified single-job path (Requirement 8.1); `> 1` runs the new single-append path. +- Pre-commit validation (`jobid_range_exists`, `inflight_range_try_add`) is unchanged and already range-shaped — no new range-claim logic is needed (Requirement 9.1). +- The record's encoded payload is `struct JobSubmit` (`jobid` = base, `arraySize`, one `submit: JobSubmitData`) — O(1) in `array_size` by construction, and `array_size` is already hard-capped at 32768 upstream by `idgen()` (Requirement 7 satisfied without a chunking policy). +- Rejection on failed pre-commit validation happens strictly before any WAL append (Requirement 9.2), identical ordering to today's single-job path. + +**Dependencies** +- Outbound: `commit_submit_async()` via a new `persist_submit_job_array_async()` (Criticality P0) +- Outbound: `inflight_range_try_add`/`jobid_range_exists` (Criticality P0, unchanged) +- Inbound: `handle_jobsubmit()` (Criticality P0) + +**Contracts**: Service [x] / Batch [x] + +##### Service Interface +```c +/* Encodes and appends ONE array-shaped WAL entry (WAL_JOB_ARRAY_SUBMIT, + * payload = encoded JobSubmit), registers ONE commit continuation. Mirrors + * persist_submit_job_payloads_async's per-entry contract but for a single + * array-shaped entry -- no loop over elements at this layer. */ +int persist_submit_job_array_async(persist_t* p, uint64_t base_jobid, + uint64_t array_size, + const xce_job_submit_t* base_spec, + uint64_t* out_lsn, + persist_job_commit_notify_fn fn, void* ctx); +``` +- Preconditions: `array_size > 1`; the `[base_jobid, base_jobid + array_size)` range has already been claimed via `inflight_range_try_add`. +- Postconditions: on success, exactly one WAL entry is durably appended locally and exactly one continuation is registered; `fn` is invoked exactly once with the outcome (mirrors `persist_submit_job_payloads_async`'s per-payload contract, specialized to one payload). +- Invariants: no `job_t` object exists in `c->jobmgr` for any array index until the commit succeeds and the apply seam runs (same non-negotiable ordering as the single-job path). + +##### Batch / Job Contract +- Trigger: `handle_jobsubmit()`'s array-size-branch, after pre-commit validation passes. +- Input / validation: `(base_jobid, array_size, base_spec)`, already range-claimed. +- Output / destination: one WAL entry; one registered commit continuation feeding `jobsubmit_commit_continuation`/`jobsubmit_apply_complete` unmodified (`actx->pending = 1`). +- Idempotency & recovery: a client retry after a commit timeout produces a second, independent `WAL_JOB_ARRAY_SUBMIT` entry for the same range — the array-scoped analog of bug-1419's existing single-job dedup story (see Testing Strategy, Requirement 8.3). + +### Array Apply Expansion + +#### restore_job_array — shared N-way expansion primitive + +| Field | Detail | +|-------|--------| +| Intent | Expand one committed array record into N `job_t` entries, atomically (scratch-build then bulk-insert — never partially visible), identically for every caller | +| Requirements | 2.1, 2.2, 2.3, 3.1, 3.2, 3.3, 3.4, 4.3, 5.1, 5.2, 5.3, 9.3 | + +**Responsibilities & Constraints** +- Decodes `struct JobSubmit` from the committed payload; for `i` in `[0, array_size)`, computes `jobid = base_jobid + i` (the exact arithmetic `build_job_submit_payloads` uses today, Requirement 5.1) and constructs a `job_t` via `create_job()`, sharing the one decoded `submit` pointer across elements (mirroring `restore_job()`'s existing `submit`/`ckptinfo` ownership discipline for a single job). +- Idempotent per index: if `jobid` already exists in `c->jobmgr`, that index is skipped, not treated as an error (handles a re-apply of an already-fully-applied record). If every index already exists, returns `-EEXIST` (mirrors `restore_job()`'s convention), and the caller fires no hook (Requirement 3.3). +- Scratch-construct-then-bulk-insert, not construct-then-rollback: all `array_size` elements are built via `create_job()`-equivalent construction into a local scratch list first, **never touching `c->jobmgr` while the construction loop is running**. Only after every element in the scratch list succeeds does one final pass insert all of them into `c->jobmgr` under a single write-lock acquisition (one atomic visibility step, not per-element lock/unlock cycles — this is what makes the array invisible-then-visible rather than visible-then-rolled-back on the success path). This means there is no window during *construction*, success or failure, where a concurrent reader of `c->jobmgr` (a status query, a cancel request) can observe some array indices present and others absent. If construction fails at index `k` (resource exhaustion only — content-derived failures are impossible here because Requirement 9.1's pre-commit validation already ran), the `0..k-1` scratch objects are destroyed directly (no `c->jobmgr` mutation ever occurred, so no `key_del`/rollback race is possible) and the function returns a negative rc. A retry re-enters the loop from index 0 against a fresh scratch list, not from a saved offset. + - **Residual gap in the bulk-insert step itself** (found during implementation, task 3.1 review round 1): the final insert-into-`c->jobmgr` pass is *not* actually failure-proof — `key_add()` can itself fail on allocation/growth exhaustion (`c->jobmgr`'s underlying table grows via `realloc()`, and no capacity-reservation primitive exists anywhere in this codebase's `kv.c`/`kvmgr.c`/`zarray.c` stack to pre-check this before mutating). Because the underlying table has no `key_del`, an already-inserted prefix `[0, inserted)` cannot be cleanly un-inserted if a later element in the same bulk pass fails. The implemented contract is therefore: `restore_job_array()` never *silently* claims success on a bulk-insert failure (it returns a negative errno, honestly), but a prefix of elements may be left genuinely, permanently visible in `c->jobmgr` while the overall call reports failure — and critically, that prefix is never included in `out_new_jobs`, so `HOOK_DETAIL_JOB_NEW` never fires for it, on this attempt or any later successful retry of the same LSN. Tasks 3.2/4.x must account for this: a bulk-insert-phase failure leaves elements that are individually addressable (status/cancel/completion work) but permanently un-pushed to the scheduler/API server unless something later re-derives and re-notifies them. This is a narrower, real exception to "atomically invisible-then-visible," not the general case — it fires only on late-stage resource exhaustion, not on ordinary construction failures (which remain fully atomic per the paragraph above). +- Called identically from `apply_committed_entry()` (which additionally builds the N-element list and fires `HOOK_DETAIL_JOB_NEW` once, Requirement 3.4) and from `apply_wal_entry()` (hook-free, matching its existing single-job case) — this is the one place expansion logic exists (Requirement 3.1, 3.2). +- Every constructed element is tagged with `submit_lsn = entry_lsn` (mirrors `restore_job()`, feeds the existing egress-watermark predicate unchanged). +- Exposes an `XCE_TESTING`-guarded fault-injection seam (a forceable construction failure at a caller-chosen element index, compiled out in production builds, matching this codebase's existing `#ifdef XCE_TESTING`-guarded test-only-symbol convention per `structure.md`) — the concrete unit `test_job_array_apply.c` drives directly to satisfy Requirement 9.6's fault-injection scenario. + +**Dependencies** +- Outbound: `create_job()` (job.c, unchanged), `jobj.c`'s hardened primitives (Criticality P0) +- Inbound: `apply_committed_entry()`, `apply_wal_entry()` (Criticality P0, both) + +**Contracts**: Service [x] / Batch [x] + +##### Service Interface +```c +/* Expands one committed WAL_JOB_ARRAY_SUBMIT payload into job_t entries. + * Builds every element into a local scratch list first -- c->jobmgr is + * never mutated until every element has successfully constructed, then one + * final pass inserts them all (a pass that cannot fail). Returns the count + * of NEWLY created elements (0..array_size) on success, -EEXIST if every + * index already existed (idempotent no-op, no hook should fire), or a + * negative errno on a genuine construction failure -- in which case the + * scratch list is discarded directly and c->jobmgr was never touched (Req + * 9.4: no partial set ever observable, on the failure path OR the success + * path). */ +int restore_job_array(cluster_t* c, const void* capnp_payload, size_t len, + uint64_t entry_lsn, xlist_t* out_new_jobs /* optional */); +``` +- Preconditions: `lsn` is committed; payload decodes to a valid `JobSubmit` (guaranteed by Requirement 9.1/9.3 — pre-commit validation makes a content-derived decode failure here impossible). +- Postconditions: either all `array_size` elements exist in `c->jobmgr` tagged with `entry_lsn`, or none of this attempt's elements do — and this holds at every instant, not only at return: `c->jobmgr` is untouched until the entire array is ready to insert as one bulk pass, so no concurrent reader can ever observe a partially-constructed array (Requirement 2.3). +- Invariants: applying the same `(WAL_JOB_ARRAY_SUBMIT, lsn, payload)` twice produces the same observable state as applying it once (Requirement 3.3). + +##### Batch / Job Contract +- Trigger: a new `case WAL_JOB_ARRAY_SUBMIT` in `apply_committed_entry()` (live apply pump, boot replay) and in `apply_wal_entry()` (standby catch-up). +- Input / validation: committed WAL payload, already CRC-verified by the point apply is reached (unchanged from the single-job path). +- Output / destination: N `job_t` entries in `c->jobmgr`; one `HOOK_DETAIL_JOB_NEW` fire (live-apply path only) carrying all N. +- Idempotency & recovery: guarded per-index existence check; `last_applied_lsn` is the resumable cursor for restart (unchanged mechanism, now also gated on this function's return value — see Live Apply Resource Fence). + +### Live Apply Resource Fence + +#### Apply-pump return-value gating and live promotion fence + +| Field | Detail | +|-------|--------| +| Intent | Never silently skip a resource-exhaustion apply failure; fence the node non-promotable while retrying; self-heal on success | +| Requirements | 9.4, 9.5, 9.6 | + +**Responsibilities & Constraints** +- `persist_apply_pump_run()` starts checking `apply_committed_entry()`'s return value (currently discarded) — on a negative result, the walk stops at that LSN without advancing `last_applied_lsn`, mirroring the existing `apply_pump_read_entry` failure branch immediately above it in the same function. This is a general fix (any WAL type's apply failure, not array-specific) that this feature's array-expansion failure is the first live trigger for. +- On engaging, logs a loud, operator-visible ERROR (LSN, reason) and sets a new, narrow fence flag — deliberately **not** `recovery_unrecoverable` (its existing periodic clear is gated on WAL-durability catch-up alone and would wrongly clear before the stuck apply actually resolves; see `research.md` Design Decisions). +- The fence clears itself, from inside the apply pump, the instant a retry of the same LSN succeeds — no dependency on an election or on any other periodic check running (Requirement 9.5). +- A fence on one node never blocks any other node's independent apply pump — each node's fence state is local (Requirement 9.5, satisfied structurally: nothing here is cross-node). +- Both existing `ha.c` promotion-gate call sites (periodic fence-release check, election-time eligibility check) gain one additional required condition alongside the existing `apply_divergent`/`self_restart_pending` checks. + +**Dependencies** +- Outbound: none new — a local atomic flag on `persist_t` +- Inbound: `persist_apply_pump_run()` (sets/clears), `ha.c`'s two promotion-gate call sites (reads) (Criticality P0, both) + +**Contracts**: State [x] + +##### State Management +- State model: one new field on `persist_t` (an atomic LSN or bool) recording "an LSN is currently stuck failing to apply." Orthogonal to `recovery_unrecoverable` and `apply_verdict` — no shared state, no risk of one clearing the other's condition. +- Persistence & consistency: in-memory only, like `last_applied_lsn` itself — recomputed by construction (the pump re-derives "stuck" the next time it fails to apply the same LSN after a restart; it is not itself a correctness-critical persisted fact). +- Concurrency strategy: set/cleared only from `persist_apply_pump_run()`'s single-threaded walk (the existing `apply_pump_lock` already serializes this); read from `ha.c`'s promotion-gate paths under their existing locking. + +## Data Models + +No new persistent entity type. `job_t`, `xce_job_t`, and `job_submit_data_t` are unchanged — array elements are ordinary jobs once expanded, addressed exactly as single-submission jobs are today. + +### Data Contracts & Integration + +**WAL payload** (`WAL_JOB_ARRAY_SUBMIT`, new `wal_entry_type` value): the existing `struct JobSubmit` capnp struct, unmodified — + +```capnp +struct JobSubmit $C.mapname("xce_job_submit_t") { + jobid @0: UInt64; # base jobid + arraySize @1: UInt64 $C.mapname("array_size"); + submit @2: JobSubmitData; # one shared spec for every element +} +``` + +**Push payload** (new `Data` union variant, additive): reuses the same struct — + +```capnp +struct Data $C.mapname("xce_data_t"){ + data :union $C.mapuniontag("table") { + schedjobs @0: SchedJobData $C.mapname("jobs"); + schedworkers @1: WorkerData $C.mapname("workers"); + jobs @2: JobData $C.mapname("jobs"); + workers @3: WorkerData $C.mapname("workers"); + resources @4: ResourceData $C.mapname("resources"); + resvals @5: ResourceVals $C.mapname("resvals"); + jobArray @6: JobSubmit $C.mapname("job_array"); # NEW + } +} +``` + +Both reuses are deliberate (Requirement 6's own wire-shape note: "the same compact shape as the WAL record") — one capnp struct, three uses (client request, WAL payload, push payload), zero new encoder/decoder pairs to write or maintain. + +### Compact Scheduler/API-Server Push + +#### One push per committed array, expanded locally + +| Field | Detail | +|-------|--------| +| Intent | Deliver a committed array to the scheduler and API server as one compact descriptor; each expands it locally into N entries | +| Requirements | 6.1, 6.2, 6.3, 6.4, 6.5 | + +**Responsibilities & Constraints** +- `sched_job_new_fun()`/`apiserver_job_new_fun()` (the existing `HOOK_DETAIL_JOB_NEW` subscribers) branch on `l->count > 1` — the only way `apply_committed_entry()` ever fires this hook with more than one element (Requirement 3.4) — to call a new `xcesched_push_jobarray()`/`apiserver_push_jobarray()` instead of the existing per-job `..._push_joblist()`. The `l->count == 1` branch (ordinary single-job submission) is untouched (Requirement 8.1). +- The new push functions derive `(base_jobid, array_size)` from the already-constructed `xlist_t` (first element's jobid, list count) and encode one `Data.jobArray` message — never N individually-encoded `Job`/`SchedJob` structs. +- The scheduler (`handler.c`, sibling case to `Data_data_schedjobs`) and the Go apiserver (`backend/`, sibling path to `cJobToModel`) each expand the descriptor locally into N entries using the identical `base_jobid + i` arithmetic the controller itself uses — the same algorithm at three layers, not three different implementations. +- Redelivery on reconnect/rejoin must not double-expand: both receivers dedupe expanded entries by `jobid` using the exact same key space and lookup an ordinary single-job push already uses downstream (no new dedup mechanism) — a scheduler or API-server reconnect that re-receives an already-expanded array's compact push is a no-op per index, mirroring `restore_job_array()`'s own idempotent-per-index contract at the controller layer. +- Each receiver logs one identifiable confirmation line (base jobid + element count) once local expansion completes, satisfying Requirement 6.4's log-based verification without a new query API. +- Elements are treated as ordinary single jobs for scheduling purposes once expanded (Requirement 6.3) — no new scheduling logic. + +**Dependencies** +- Inbound: `HOOK_DETAIL_JOB_NEW` fire from `apply_committed_entry()` (Criticality P0) +- Outbound: `xce_push_data_req()` (unchanged transport, Criticality P0) + +**Contracts**: Event [x] / API [x] + +##### Event Contract +- Published events: `HOOK_DETAIL_JOB_NEW` (unchanged detail, unchanged signature) — the branch on element count lives entirely inside the two existing subscribers, not in the hook system itself. +- Subscribed events: none new. +- Ordering/delivery guarantees: unchanged — one push per committed array, delivered after the array's own commit (same ordering guarantee single-job pushes already have). + +##### API Contract +| Direction | Message | Payload | Notes | +|---|---|---|---| +| Controller → Scheduler | `Data.jobArray` push | `JobSubmit` (base jobid, array_size, one spec) | New variant; existing `xce_push_data_req` transport | +| Controller → API server | `Data.jobArray` push | `JobSubmit` (base jobid, array_size, one spec) | Same variant, same transport, per connected API server (mirrors existing fan-out) | + +**Implementation Notes** +- Integration: no new opcode — `Data`'s existing union discriminator (`table`) already routes to the new variant on both the scheduler and the API-server receive paths. +- Validation: confirm the scheduler's local expansion treats a partially-received/decode-failed array push the same way a malformed single-job push is handled today (existing error path, not new). +- Risks: the scheduler and API server must never assume `array_size == l->count` implies the array is contiguous starting at any index other than the pushed `jobid` — enforced by construction here since `base_jobid + i` is the only arithmetic either side uses. + +### jobj.c Hardening + +#### Reliable, leak-free construction under N-element reuse + +| Field | Detail | +|-------|--------| +| Intent | Close five specific, pre-existing gaps in job/job-submit construction that this feature's N-way apply-time loop amplifies by array size | +| Requirements | 10.1, 10.2, 10.3, 10.4, 10.5 | + +**Responsibilities & Constraints** +- **10.1** — `job_init()` (jobj.c:532) currently stores `makeJobsub()`'s return value into `dst->job.submit` unconditionally (jobj.c:543), including a NULL on allocation/interning failure, and still returns success. Fix: check the return, propagate failure. Same check for `makeString(src->ckptinfo)`. +- **10.2** — the previously observed `makeJobsub()` NULL-with-EINVAL-for-a-valid-`src`, traced to a likely `jobsub_table` (jobj.c:16) initialization-ordering issue: `job_obj_init()` must complete before the first `makeJob`/`makeJobsub` call on every path that can reach one, including this feature's new N-way `restore_job_array()` loop. Root cause to be confirmed and fixed at the actual ordering dependency, not papered over by 10.1's error propagation alone. +- **10.3** — `job_init()`'s `MOVE_PTR` fields (`pidinfo, allocs, accounts, run_usage, exit_info, pend_reasons`) are confirmed always NULL for a freshly-submitted job: `build_job_submit_payloads()` (job.c:229) and `restore_job_array()`'s equivalent per-element construction both `calloc()` a fresh `xce_job_t` per element and never set these six fields — repeated `MOVE_PTR` from a shared template is benign (NULL→NULL) and, for the array case specifically, each element gets its own independently-zeroed struct in the first place. No code change required; documented as a verified invariant. +- **10.4** — `job_destroy()`'s `accounts` leak (jobj.c:584, `/* FIXME??? */`) is fixed by adding the missing free call, following the existing pattern already used for `allocs`/`pidinfo`/`run_usage` in the same function. Array submission multiplies this leak by element count, making it a correctness requirement here, not a nice-to-have. +- **10.5** — verified: `struct Job`'s capnp schema (`controller.capnp:287`) has no `ckptinfo` field at all, so `decode_Job` never populates it and `free_Job` correctly never frees it — not a gap. Documented as a verified non-issue, no code change. + +**Dependencies** +- Inbound: `create_job()`/`makeJob()` (job.c, jobj.c) — every caller, including this feature's `restore_job_array()` (Criticality P0) + +**Contracts**: Service [x] + +##### Service Interface +```c +/* job_init's contract changes: propagates a construction failure instead of + * silently storing NULL and reporting success. */ +static int job_init(meta_obj_t* obj, va_list ap); /* now fails on + makeJobsub()/makeString(ckptinfo) returning NULL, not only on src==NULL */ +``` +- Preconditions: unchanged (`src != NULL`, `src->submit != NULL`). +- Postconditions: on success, `dst->job.submit` and `dst->job.ckptinfo` are never NULL-with-success; `job_obj_init()` has run before this call on every reachable path. +- Invariants: a `job_t` never exists in a half-initialized state visible to any caller. + +**Implementation Notes** +- Integration: `restore_job_array()`'s all-or-nothing scratch-discard (Array Apply Expansion component) is the safety net if 10.1's now-propagated failure occurs mid-array — it is treated exactly like any other `create_job()` failure in that loop, discarded from the scratch list, never inserted into `c->jobmgr`. +- Validation: `src/lib/testsuite/test_jobj_hardening.c` (File Structure Plan) exercises all five items directly, independent of the array-expansion path, since these are general `jobj.c` fixes the single-job path benefits from too (Requirement 8.1's note explicitly permits this). +- Risks: 10.2's root cause is not yet confirmed at design time — flagged as a task-level investigation, not a pre-decided fix, consistent with Design vs. Implementation (state the contract: "`makeJobsub` must never observe an uninitialized `jobsub_table`," not the exact ordering fix). **Boundary note**: this item's File Structure Plan entry is `src/lib/jobj.c` only, on the assumption the fix is local to `jobj.c`'s own init/call-ordering. If root-causing 10.2 shows the actual gap is elsewhere (e.g. controller boot sequencing, a test-harness call path that skips `job_obj_init()`), that is a signal to split 10.2 into its own task with its own explicit file scope at task-generation time, not to silently widen this component's boundary. + +## Error Handling + +### Error Strategy +- **Pre-commit validation failure** (Requirement 9.2): `handle_jobsubmit()` rejects before any WAL append, identical to today's single-job path — no new error class. +- **Commit failure/timeout for the array record**: the single registered continuation resolves failed; `jobsubmit_reply_failure_once()` fires (unchanged); no job ever constructed (Requirement 2.2, 2.3 by construction — nothing to roll back). +- **Post-commit content-derived apply failure**: structurally impossible per Requirement 9.3 (9.1's pre-commit validation already ran) — if it somehow occurred, it is a programming-invariant violation, handled the same way the Live Apply Resource Fence handles any negative `apply_committed_entry()` return: loud ERROR, fence, retry forever. No `assert()`/abort (`src/controller/CLAUDE.md` Process Lifecycle invariant). +- **Post-commit resource-exhaustion apply failure**: `restore_job_array()` rolls back this attempt's partial construction; `persist_apply_pump_run()` halts the walk at this LSN; Live Apply Resource Fence engages; retried on the next apply-pump kick; self-heals on success (Requirement 9.4, 9.5). +- **`jobj.c` construction failure** (post-hardening): propagates as an ordinary `create_job()` failure into the same rollback path above — no new error class, just a previously-silent failure now made visible. + +### Monitoring +- Loud `ERROR()` on Live Apply Resource Fence engagement, naming the stuck LSN and reason (mirrors `persist_recover_loud_fence_if_unrecoverable`'s existing log shape). +- `INFO()` on fence self-heal (LSN, retry count if tracked). +- One `INFO()` confirmation per array from the scheduler and one from the API server on successful local expansion (base jobid + element count), the log-based verification Requirement 8.4/6.4 rely on. + +## Testing Strategy + +- **Unit Tests**: + - `persist_submit_job_array_async()` appends exactly one WAL entry and registers exactly one continuation for `array_size > 1` (Requirement 1.1-1.3). + - `restore_job_array()`: N-way expansion produces `base_jobid..base_jobid+array_size-1`; idempotent re-apply (all-existing) is a no-op returning `-EEXIST`; a partial-existing edge case skips existing indices without duplicating (Requirement 2, 3.3, 5.1). + - Simulated construction failure at element `k` of `N`, driven via `restore_job_array()`'s `XCE_TESTING`-guarded injection seam (forces a construction failure at a chosen index): elements `0..k-1` exist only in the discarded scratch list, never in `c->jobmgr` (nothing to tear down there); a subsequent successful retry builds a fresh scratch list from index 0 and bulk-inserts once, producing the full, non-duplicated set (Requirement 9.4, fault-injection scenario for 9.6). + - `apply_committed_entry()` vs. `apply_wal_entry()` applied to the same committed array record produce byte-for-byte identical `c->jobmgr` state (Requirement 3.1, 3.2). + - `persist_apply_pump_run()`: a negative `apply_committed_entry()` return halts the walk without advancing `last_applied_lsn`; a subsequent success advances it and clears the Live Apply Resource Fence (Requirement 9.4, 9.5). + - Promotion-gate refusal while the Live Apply Resource Fence is set, at both existing `ha.c` call sites (Requirement 9.4). + - `jobj.c`: all five Requirement 10 items, independent of array submission. +- **Integration Tests**: + - End-to-end array submit: commit success → all N elements visible and individually addressable (status/cancel/completion) via existing per-element APIs, unchanged (Requirement 5.2, 5.3). + - Single push received by both the scheduler and API server, each expanding N local entries and logging one confirmation (Requirement 6.1-6.5). + - Non-array (`array_size == 1`) submission: byte-identical WAL record, identical commit-barrier count, identical reply, before and after this feature (Requirement 8.1). +- **Regression / Live Soak**: + - `/xce-test-ha` extended with a large-array submission scenario asserting WAL/commit-barrier count (one record/one barrier, not N) and failover replay time for the committed array (Requirement 8.3, 4.3). + - A client retrying an identical array submission across a primary failover follows the same cross-primary dedup mechanism as single-job submission (bug-1419), extended to the array's full index range as one unit — asserted not to double-submit or partially duplicate (Requirement 8.3). + - Full sequential and chaotic `/xce-test-ha` soak, including array submission immediately before killing the primary, shows no new job loss, divergence, or liveness stall (Requirement 8.2). + - Non-array (`array_size == 1`) submission WAL format, commit-barrier count, and reply are byte-identical before and after this feature (Requirement 8.1). + +### Performance & Scalability +- WAL/replication cost for an array submission becomes O(1) in `array_size` (one record, one barrier, one replicated entry) versus today's O(N) — the target metric is commit-engine load and replication bandwidth for a large-array submission, measured via the `/xce-test-ha` large-array scenario. +- Apply-time cost (constructing N `job_t` entries inside one `apply_committed_entry()` call) is O(N) CPU inside a single, strictly-ordered apply-pump step — an accepted, bounded cost (N is hard-capped at 32767 by `idgen()`), not a new unbounded-latency risk; noted here as a deliberate trade-off, not re-litigated as a chunking requirement (see Requirement 7's Architecture-table resolution). +- Failover replay time for a committed array is governed by the size of the one array record (O(1)) rather than a per-element replay cost proportional to array size, since replay decodes and expands through the same `restore_job_array()` primitive as live apply (Requirement 4.3) — measured via the `/xce-test-ha` large-array scenario's failover-replay-time assertion. diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index 7dcc655..f5d3ae2 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -50,7 +50,7 @@ - _Boundary: WAL Array Submission_ - [ ] 3. Array Apply Expansion: shared N-way expansion and dispatch wiring -- [ ] 3.1 Implement the shared array expansion primitive +- [x] 3.1 Implement the shared array expansion primitive - Decode the committed array payload and, for each index in the array's range, build a job entry without yet making it visible to any reader - Skip building an entry for any index that already exists (idempotent re-apply); if every index already exists, treat the whole record as an already-applied no-op - Once every index has successfully built, make the whole set visible in one atomic step; if any index fails to build, discard everything built in this attempt without ever making any of it visible — a retry starts over from the first index @@ -173,3 +173,4 @@ - **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. +- **Task 3.1**: `restore_job_array()`'s atomicity has one narrow, documented exception (design.md updated to match): the final bulk-insert-into-`c->jobmgr` pass can itself fail on allocation/growth exhaustion (`c->jobmgr` grows via `realloc()`; no capacity-reservation primitive exists anywhere in `kv.c`/`kvmgr.c`/`zarray.c` to pre-check this before mutating). On this specific failure, the function never silently claims success — it returns a negative errno — but an already-inserted prefix `[0, inserted)` can be left genuinely, permanently visible in `c->jobmgr` (no `key_del` exists to undo it), and that prefix is NEVER included in `out_new_jobs`, so `HOOK_DETAIL_JOB_NEW` never fires for it, not on this attempt or any later successful retry of the same LSN. **Tasks 3.2/4.x must account for this**: those elements are individually addressable (status/cancel/completion work normally) but permanently un-pushed to the scheduler/API server unless something later re-derives and re-notifies them — this is a real gap in the current design, not yet solved, and task 4.x's Live Apply Resource Fence design should explicitly decide whether/how to handle it (e.g. detecting this residual state and re-notifying on self-heal) rather than only handling the construction-phase failure case. diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index ad82f09..61db06e 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -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*); diff --git a/src/controller/job.c b/src/controller/job.c index 1b48a5d..046ca36 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -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. */ diff --git a/src/controller/meson.build b/src/controller/meson.build index 1225c40..19b9eb6 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 0f19b6b..878cf76 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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(); } diff --git a/src/controller/testsuite/test_job_array_apply.c b/src/controller/testsuite/test_job_array_apply.c new file mode 100644 index 0000000..d1597fd --- /dev/null +++ b/src/controller/testsuite/test_job_array_apply.c @@ -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 + +#include +#include +#include +#include +#include +#include + +#include + +#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(); +}