From b0c7cfe548a25dceff5e26ec0b31ebd839268cb7 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 15 Jul 2026 02:30:06 +0000 Subject: [PATCH] feat(job-array-single-wal-record): wire array expansion into apply_wal_entry Adds case WAL_JOB_ARRAY_SUBMIT to apply_wal_entry() (standby catch-up dispatch), calling restore_job_array() hook-free, matching the existing WAL_JOB_SUBMIT case's shape exactly. Review found apply_wal_entry() has zero production call sites -- a 2026-07-10 refactor (ha_standby_advance_committed_and_apply) moved standby entity materialization onto the same apply_committed_entry() pump the primary uses; the code comment claiming otherwise predated that refactor and was never updated. Requirement 3.1/3.2/4.1 are fully delivered by task 3.2 alone. This task's wiring is kept as defensive parity with WAL_JOB_SUBMIT's existing, equally-dead case in the same switch, not as an independent delivery path -- design.md and tasks.md corrected accordingly. Task 3.3 of job-array-single-wal-record. --- .../job-array-single-wal-record/design.md | 4 +- .../job-array-single-wal-record/tasks.md | 5 +- src/controller/persist.c | 21 ++ .../testsuite/test_job_array_apply.c | 242 +++++++++++++++++- 4 files changed, 260 insertions(+), 12 deletions(-) diff --git a/.kiro/specs/job-array-single-wal-record/design.md b/.kiro/specs/job-array-single-wal-record/design.md index 3df3f63..be715c2 100644 --- a/.kiro/specs/job-array-single-wal-record/design.md +++ b/.kiro/specs/job-array-single-wal-record/design.md @@ -62,7 +62,7 @@ The brief's original framing (`job_submit()` looping `create_job()` per element) | 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 dispatch | **Correction (found during task 3.3 review, 2026-07-14):** `apply_wal_entry()` (persist.c:2574) has **zero production call sites** — a 2026-07-10 refactor (`ha_standby_advance_committed_and_apply`, ha.c:8024) moved standby entity materialization onto the SAME `apply_committed_entry()`/`persist_apply_pump_kick` pump the primary uses; the persist.c:3722 comment claiming `apply_wal_entry()` "remains the live standby catch-up dispatch" predates that refactor and was never updated. `apply_committed_entry()` (persist.c:2721) is the **only** live dispatcher — for primary live apply, boot replay, *and* standby catch-up. | Requirement 3/4.1 is fully delivered by task 3.2's `apply_committed_entry()` wiring alone. Task 3.3's parallel wiring into `apply_wal_entry()` is kept as defensive parity with `WAL_JOB_SUBMIT`'s existing (also-dead) case in the same switch, not as a requirement-delivery path. | | 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 | @@ -273,7 +273,7 @@ int persist_submit_job_array_async(persist_t* p, uint64_t base_jobid, - 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). +- 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). **Correction (2026-07-14, see the Existing Architecture Analysis table's "Live apply dispatch" row):** `apply_wal_entry()` has zero production call sites — `apply_committed_entry()` alone delivers Requirement 3.1/3.2/4.1 for live apply, boot replay, and standby catch-up. The `apply_wal_entry()` wiring is retained as defensive parity with `WAL_JOB_SUBMIT`'s existing (also-dead) case in the same switch, not as a requirement-delivery path. - 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. diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index ee5e61f..ce2c106 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -67,10 +67,10 @@ - _Requirements: 3.1, 3.4_ - _Boundary: Array Apply Expansion_ -- [ ] 3.3 Wire array expansion into standby catch-up +- [x] 3.3 Wire array expansion into standby catch-up (defensive parity — see note below) - Call 3.1's same primitive from the separate standby catch-up/replicated-apply dispatch path for the new WAL record type, without firing any notification (matching that path's existing hook-free behavior for the single-job case) - Observable: a test applies the same committed array record through both the live-apply dispatch and the standby catch-up dispatch and confirms byte-identical resulting job state either way - - _Requirements: 3.1, 3.2, 4.1_ + - _Requirements: 3.1, 3.2, 4.1 — delivered by task 3.2 alone (see Implementation Notes); this task is defensive parity, not an independent delivery path_ - _Boundary: Array Apply Expansion_ - [ ] 3.4 Unit tests for array expansion @@ -173,4 +173,5 @@ - **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.3**: `apply_wal_entry()` (persist.c:2574) is dead code in production — confirmed via exhaustive grep (zero call sites outside its own declaration/definition/comments) and via `ha.c:8024`'s `ha_standby_advance_committed_and_apply` (dated 2026-07-10), whose own doc comment explicitly says standby entity materialization now goes through the SAME `apply_committed_entry()`/`persist_apply_pump_kick` pump the primary uses, not `apply_wal_entry()`. The persist.c:3722 comment claiming otherwise predates that refactor and was never updated (design.md corrected accordingly). Requirement 3.1/3.2/4.1 are fully delivered by task 3.2 alone. Task 3.3's code was kept (not discarded) as defensive parity with `WAL_JOB_SUBMIT`'s existing, also-dead case in the same switch — it is tested and correct, just not currently reachable in production. If `apply_wal_entry()` is ever revived or removed entirely in a future spec, this task's case should move or be deleted with it. - **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/persist.c b/src/controller/persist.c index 1275c27..3fb0c58 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -2627,6 +2627,27 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, restore_job(p->cluster, 0, payload, payload_len, lsn); break; + case WAL_JOB_ARRAY_SUBMIT: + /* job-array-single-wal-record task 3.3 (design.md "Array Apply + * Expansion"): the array-record analog of WAL_JOB_SUBMIT's + * restore_job() call directly above, dispatched from this SEPARATE, + * still-live standby catch-up / replicated-apply seam so it can + * never diverge from apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT + * case (Req 3.1, 3.2, 4.1). Mirrors the single-job case's exact + * minimalism: no xlist_t, no hook fire — this dispatcher is + * deliberately hook-free (standby catch-up does not notify the + * local scheduler/API server; that is the primary's job, done via + * apply_committed_entry()). out_new_jobs is not needed here, so + * NULL is passed (restore_job_array()'s out_new_jobs parameter is + * optional — see test_rja_out_new_jobs_is_optional, + * test_job_array_apply.c). The return value is intentionally + * discarded, matching every other case in this switch today (task + * 4.1 — Live Apply Resource Fence — is what later makes this + * function's callers respect a negative rc; not this task's + * boundary). */ + restore_job_array(p->cluster, payload, payload_len, lsn, NULL); + break; + case WAL_JOB_START: case WAL_JOB_FINISH: case WAL_JOB_FAIL: diff --git a/src/controller/testsuite/test_job_array_apply.c b/src/controller/testsuite/test_job_array_apply.c index d1597fd..2ea94c3 100644 --- a/src/controller/testsuite/test_job_array_apply.c +++ b/src/controller/testsuite/test_job_array_apply.c @@ -1,12 +1,17 @@ -/* test_job_array_apply.c — job-array-single-wal-record task 3.1 +/* test_job_array_apply.c — job-array-single-wal-record tasks 3.1 / 3.3 * * 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). + * 2.3, 3.3, 4.1, 5.1, 5.2, 5.3, 9.3, 9.6), PLUS (task 3.3, added below the + * task-3.1 tests) dispatch-level tests proving apply_wal_entry()'s new + * WAL_JOB_ARRAY_SUBMIT case -- the standby catch-up / replicated-apply + * dispatcher, a SEPARATE, still-live function from apply_committed_entry() + * (design.md "Existing Architecture Analysis" table) -- expands a committed + * array record identically to apply_committed_entry() while staying + * hook-free, matching that dispatcher's existing single-job case (Req 3.1, + * 3.2, 4.1). * - * Covers: + * Covers (task 3.1, restore_job_array() called DIRECTLY): * - 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). @@ -17,19 +22,32 @@ * 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. + * Covers (task 3.3, driving apply_wal_entry()/apply_committed_entry()): + * - apply_wal_entry() constructs every array element and fires NO + * notification hook (Req 3.1, 3.2). + * - apply_committed_entry() and apply_wal_entry() applied to the SAME + * committed array record produce byte-for-byte identical resulting job + * state (Req 3.1, 3.2, 4.1). + * + * Fixture (task 3.1 tests): 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). + * + * Fixture (task 3.3 tests, tajd_* below): a richer cluster_t (jobmgr + + * hooksys) since these tests drive the dispatch functions -- not + * restore_job_array() directly -- and must be able to prove the ABSENCE of + * a hook fire. Mirrors test_apply_committed_entry.c's tace_setup shape. */ #include #include #include +#include #include #include #include @@ -39,7 +57,10 @@ #include "cluster.h" #include "controller.capnp.h" +#include "hook.h" #include "jobj.h" +#include "persist.h" +#include "xce/cluster_def.h" #include "xce/types.h" #include "xlist.h" #include "xstring.h" @@ -477,6 +498,209 @@ static void test_rja_out_new_jobs_is_optional(void) printf("test_rja_out_new_jobs_is_optional: PASS\n"); } +/* ── Task 3.3: apply_wal_entry() standby catch-up dispatch ─────────────── */ + +static cluster_t g_tajd_c; +static persist_t g_tajd_p; +static int g_tajd_hook_fires; + +static int tajd_spy_hook(int type, int situation, int detail, void* data, + va_list ap) +{ + (void)type; (void)detail; (void)data; (void)ap; + if (situation == HOOK_AFTER_EVENT) + g_tajd_hook_fires++; + return 0; +} + +static void tajd_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_tajd_c, 0, sizeof(g_tajd_c)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tajd_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + g_tajd_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_tajd_c.jobmgr); + + memset(&g_tajd_p, 0, sizeof(g_tajd_p)); + g_tajd_p.cluster = &g_tajd_c; + + g_tajd_hook_fires = 0; + +#ifdef XCE_TESTING + job_array_apply_test_clear_force_fail(); + job_array_apply_test_clear_bulk_insert_force_fail(); +#endif +} + +static void tajd_teardown(void) +{ +#ifdef XCE_TESTING + job_array_apply_test_clear_force_fail(); + job_array_apply_test_clear_bulk_insert_force_fail(); +#endif + xce_hook_fini(&g_tajd_c.hooksys); + if (g_tajd_c.jobmgr) + dereference(g_tajd_c.jobmgr); +} + +static job_t* tajd_find_job(cluster_t* c, uint64_t jobid) +{ + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + int found = key_find(c->jobmgr, (void*)jobid, (void**)&j); + unlock(c->jobmgr); + return (found > 0) ? j : NULL; +} + +/* apply_wal_entry()'s WAL_JOB_ARRAY_SUBMIT case constructs every element + * through restore_job_array() and fires NO notification -- mirroring the + * existing WAL_JOB_SUBMIT case's hook-free behavior at this same dispatch + * seam (standby catch-up does not notify the local scheduler/API server; + * that is apply_committed_entry()'s job, Req 3.1, 3.2). */ +static void test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook(void) +{ + tajd_setup(); + + const uint64_t base_jobid = 78000; + const uint64_t array_size = 3; + const uint64_t entry_lsn = 80; + 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)); + + TEST_ASSERT_EQUAL_INT(0, + xce_add_hook(&g_tajd_c.hooksys, XCE_CONTROLLER_HOOK_JOB, + tajd_spy_hook, NULL)); + + int rc = apply_wal_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn, + payload, (uint32_t)len); + TEST_ASSERT_EQUAL_INT(0, rc); + + for (uint64_t i = 0; i < array_size; i++) { + job_t* j = tajd_find_job(&g_tajd_c, base_jobid + i); + TEST_ASSERT_NOT_NULL_MESSAGE(j, + "every array index must be constructed via restore_job_array" + " through apply_wal_entry's standby catch-up dispatch"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(entry_lsn, j->submit_lsn, + "each constructed element must be tagged with the entry's lsn"); + } + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tajd_hook_fires, + "apply_wal_entry's standby catch-up dispatch must never fire a" + " notification hook, matching its existing single-job WAL_JOB_SUBMIT" + " case"); + + free(payload); + tajd_teardown(); + printf("test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook: " + "PASS\n"); +} + +/* The task 3.3 Observable: applying the SAME committed array record through + * apply_committed_entry() (live apply / boot replay) and through + * apply_wal_entry() (standby catch-up) must produce byte-for-byte identical + * resulting job state either way -- both dispatchers share the one + * restore_job_array() expansion primitive, so they can never diverge (Req + * 3.1, 3.2, 4.1). Uses two independent fixture instances (setup/apply/ + * snapshot/teardown, run twice) rather than two live cluster_t at once, to + * keep each dispatch call's state isolated and the comparison unambiguous. */ +typedef struct { + uint64_t jobid; + uint32_t state; + uint64_t submit_lsn; + char queue[32]; + char user[32]; +} tajd_snapshot_t; + +static void tajd_snapshot_array(cluster_t* c, uint64_t base_jobid, + uint64_t array_size, tajd_snapshot_t* out) +{ + for (uint64_t i = 0; i < array_size; i++) { + job_t* j = tajd_find_job(c, base_jobid + i); + TEST_ASSERT_NOT_NULL(j); + out[i].jobid = j->job.jobid; + out[i].state = j->job.state; + out[i].submit_lsn = j->submit_lsn; + out[i].queue[0] = '\0'; + out[i].user[0] = '\0'; + if (j->job.submit != NULL) { + if (j->job.submit->queue != NULL) + strncpy(out[i].queue, j->job.submit->queue, + sizeof(out[i].queue) - 1); + if (j->job.submit->user != NULL) + strncpy(out[i].user, j->job.submit->user, + sizeof(out[i].user) - 1); + } + } +} + +static void +test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_identical(void) +{ + const uint64_t base_jobid = 79000; + const uint64_t array_size = 4; + const uint64_t entry_lsn = 90; + 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)); + + tajd_snapshot_t via_committed[4]; + tajd_snapshot_t via_wal[4]; + + /* Path 1: live-apply / boot-replay dispatch. */ + tajd_setup(); + int rc1 = apply_committed_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn, + payload, (uint32_t)len); + TEST_ASSERT_EQUAL_INT((int)array_size, rc1); + tajd_snapshot_array(&g_tajd_c, base_jobid, array_size, via_committed); + tajd_teardown(); + + /* Path 2: standby catch-up / replicated-apply dispatch. */ + tajd_setup(); + int rc2 = apply_wal_entry(&g_tajd_p, WAL_JOB_ARRAY_SUBMIT, entry_lsn, + payload, (uint32_t)len); + TEST_ASSERT_EQUAL_INT(0, rc2); + tajd_snapshot_array(&g_tajd_c, base_jobid, array_size, via_wal); + tajd_teardown(); + + for (uint64_t i = 0; i < array_size; i++) { + TEST_ASSERT_EQUAL_UINT64_MESSAGE(via_committed[i].jobid, + via_wal[i].jobid, "jobid must match across both dispatchers"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(via_committed[i].state, + via_wal[i].state, "state must match across both dispatchers"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(via_committed[i].submit_lsn, + via_wal[i].submit_lsn, + "submit_lsn must match across both dispatchers"); + TEST_ASSERT_EQUAL_STRING_MESSAGE(via_committed[i].queue, + via_wal[i].queue, "queue must match across both dispatchers"); + TEST_ASSERT_EQUAL_STRING_MESSAGE(via_committed[i].user, + via_wal[i].user, "user must match across both dispatchers"); + } + + free(payload); + printf("test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_" + "identical: PASS\n"); +} + /* ── Test group entry point ───────────────────────────────────────────── */ void test_job_array_apply(void) @@ -491,5 +715,7 @@ void test_job_array_apply(void) #endif RUN_TEST(test_rja_null_or_empty_payload_returns_error); RUN_TEST(test_rja_out_new_jobs_is_optional); + RUN_TEST(test_tajd_apply_wal_entry_constructs_all_and_fires_no_hook); + RUN_TEST(test_tajd_apply_committed_entry_vs_apply_wal_entry_byte_identical); UNITY_END(); }