From 7c4afbbb8318c7ef560034183c65c65188f5182d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 14 Jul 2026 09:50:00 +0000 Subject: [PATCH] feat(job-array-single-wal-record): implement single-record array append/commit Adds persist_submit_job_array_async(): encodes one JobSubmit-shaped payload (base jobid, array size, one shared spec), appends exactly one WAL_JOB_ARRAY_SUBMIT entry, and registers exactly one commit continuation regardless of array size -- reuses the existing job_commit_adapter_ctx_t/job_commit_resolve_adapter machinery, no per-element loop. Not yet wired into handle_jobsubmit (task 2.2). Also fixes a test-helper bug (WAL offset back-computed from the post-append cursor, only valid for zero-payload entries) that corrupted state across tests when exercising a nonzero-payload array-submit entry. Task 2.1 of job-array-single-wal-record. --- .../job-array-single-wal-record/tasks.md | 2 +- src/controller/persist.c | 153 ++++++++ .../testsuite/test_commit_barrier_rollout.c | 363 ++++++++++++++++++ src/include/persist.h | 42 ++ 4 files changed, 559 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index 8c7aca0..ac75c2e 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -25,7 +25,7 @@ - _Boundary: jobj.c Hardening_ - [ ] 2. WAL Array Submission: encode, append, commit -- [ ] 2.1 Implement the single-record array append and commit path +- [x] 2.1 Implement the single-record array append and commit path - Encode one compact payload carrying (base jobid, array size, one shared job spec) — the same shape as the client's own submit request - Append it as exactly one `WAL_JOB_ARRAY_SUBMIT` WAL entry and register exactly one commit continuation for it, regardless of array size - Observable: submitting an array produces one WAL append call and one registered commit continuation in a controlled test, not one per element diff --git a/src/controller/persist.c b/src/controller/persist.c index dda83ac..18dbc05 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -4925,6 +4925,159 @@ int persist_submit_job_payloads_async(persist_t* p, xlist_t* payloads, return first_rc; } +/* job-array-single-wal-record task 2.1: serialize a job-array submission + * (base jobid + array size + one shared job spec) to a malloc'd Cap'n Proto + * buffer, reusing the SAME capnp struct (JobSubmit) the client's own submit + * request uses -- mirrors capnp_encode_job above, which mirrors encode_Job/ + * new_Job/write_Job for a single job, except here via encode_JobSubmit/ + * new_JobSubmit/write_JobSubmit. The encoded payload is O(1) in array_size: + * arraySize is a fixed-width uint64 field, not a per-element list -- so this + * function's output size never grows with array_size (Req 7.1). Caller must + * free(*out) on success. */ +static int capnp_encode_job_array(const xce_job_submit_t* s, uint8_t** out, + size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct JobSubmit cjs; + JobSubmit_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_JobSubmit(cs, &cjs, (xce_job_submit_t*)s); + jp = new_JobSubmit(cs); + write_JobSubmit(&cjs, jp); + capn_setp(capn_root(&c), 0, jp.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; +} + +/* job-array-single-wal-record task 2.1: append-only, single-record + * counterpart of append_job_wal_entry_local for a WHOLE job array. Encodes + * ONE JobSubmit-shaped payload (base_jobid, array_size, base_spec->submit) + * and appends it as exactly one WAL_JOB_ARRAY_SUBMIT entry via the SAME + * primary-append primitive append_job_wal_entry_local already uses + * (ha_primary_log_append / persist_wal_async) -- never one entry per array + * element, regardless of array_size (Req 1.1, 1.2, 1.3). base_jobid/ + * array_size are taken as explicit parameters rather than read from + * base_spec->jobid/base_spec->array_size, so the caller's own + * already-range-claimed values are always what gets committed, independent + * of whatever same-shaped fields the client's own wire request happened to + * carry. */ +static int append_job_array_wal_entry_local(persist_t* p, uint64_t base_jobid, + uint64_t array_size, + const xce_job_submit_t* base_spec, + uint64_t* out_lsn) +{ + xce_job_submit_t wire = { .jobid = base_jobid, + .array_size = array_size, + .submit = base_spec->submit }; + uint8_t* buf = NULL; + size_t sz = 0; + int rc; + uint64_t lsn; + + if (capnp_encode_job_array(&wire, &buf, &sz) != 0) + return -ENOMEM; + + /* Snapshot the LSN this entry will receive BEFORE the append (same + * convention as append_job_wal_entry_local / submit_job_wal_entry_and_ + * commit: persist_wal_async / ha_primary_log_append assigns + * hdr.lsn = current_lsn then increments). */ + pthread_mutex_lock(&p->lock); + lsn = p->current_lsn; + pthread_mutex_unlock(&p->lock); + + if (p->cluster != NULL) + rc = ha_primary_log_append(p->cluster, WAL_JOB_ARRAY_SUBMIT, buf, + (uint32_t)sz, NULL); + else + rc = persist_wal_async(p, WAL_JOB_ARRAY_SUBMIT, buf, (uint32_t)sz); + free(buf); + + if (rc == 0) + *out_lsn = lsn; + return rc; +} + +/* job-array-single-wal-record task 2.1 (design.md §WAL Array Submission, + * Service Interface): encodes and appends ONE array-shaped WAL entry + * (WAL_JOB_ARRAY_SUBMIT) and registers ONE commit continuation for it, + * regardless of array_size -- the single-record sibling of + * persist_submit_job_payloads_async's per-payload loop, specialized to + * exactly one payload. No job_t/xce_job_t object is constructed here or + * anywhere on this call path -- array expansion into individual job + * entries happens later, in the shared apply seam (task 3.x's + * restore_job_array), never here. + * + * Mirrors persist_submit_job_payloads_async's contract exactly: + * - a local encode/append failure resolves `fn` synchronously with that + * failure and returns the same negative rc; + * - the LSN-0 / engine-not-wired edge cases (documented on + * persist_submit_job_payloads_async above) resolve `fn` inline with + * rc=0; + * - otherwise `fn` is registered via commit_submit_async and invoked + * exactly once, asynchronously, once the entry's commit outcome is + * known. */ +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) +{ + uint64_t lsn = 0; + int rc; + + if (p == NULL || base_spec == NULL || base_spec->submit == NULL || + out_lsn == NULL || fn == NULL || array_size <= 1) + return -EINVAL; + + rc = append_job_array_wal_entry_local(p, base_jobid, array_size, + base_spec, &lsn); + if (rc != 0) { + fn(ctx, 0, rc); + return rc; + } + *out_lsn = lsn; + + /* LSN-0 / engine-not-wired edge (see persist_submit_job_payloads_async's + * doc comment above for the full rationale -- identical here, + * specialized to one entry instead of a loop). */ + if (lsn == 0 || p->repl.ce == NULL) { + fn(ctx, lsn, 0); + return 0; + } + + job_commit_adapter_ctx_t* a = calloc(1, sizeof(*a)); + if (a == NULL) { + fn(ctx, lsn, -ENOMEM); + return -ENOMEM; + } + a->fn = fn; + a->ctx = ctx; + + int reg_rc = commit_submit_async(p->repl.ce, lsn, + job_commit_resolve_adapter, a); + if (reg_rc != 0) { + free(a); + ERROR("persist_submit_job_array_async: lsn=%lu appended but " + "commit_submit_async registration failed (rc=%d) -- the " + "entry remains durable and will still be caught up " + "eventually via replay, but this caller will not be told " + "its outcome via this path", + (unsigned long)lsn, reg_rc); + fn(ctx, lsn, reg_rc); + return reg_rc; + } + + return 0; +} + void persist_register_pending_apply(persist_t* p, uint64_t lsn, pending_apply_notify_fn on_applied, void* ctx) diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index 34fff02..8d1281f 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -1724,6 +1724,364 @@ static void test_config_ack_wire_path_completes_barrier(void) printf("test_config_ack_wire_path_completes_barrier: PASS\n"); } +/* ════════════════════════════════════════════════════════════════════════ + * job-array-single-wal-record task 2.1 — persist_submit_job_array_async: + * the single-record array append and commit path (Req 1.1, 1.2, 1.3, 4.2, + * 9.2). Fixture mirrors test_fact_hook_append_is_retained_and_nonblocking + * above (standalone persist_t, no cluster_t -- p->cluster stays NULL so + * the append routes through the persist_wal_async fallback branch inside + * append_job_array_wal_entry_local; real tmpdir WAL; wired commit engine + * with 2 stub standbys; ACKs driven through repl_on_ack). commit_engine_init + * here never wires a vtaskmgr pool (ce->pool stays NULL), so + * commit_resolve_pending's continuation dispatch is the synchronous, + * same-thread inline fallback (cerebrum precedent: "the NO-POOL + * inline-dispatch fallback calls fn() synchronously ON THE CALLING + * THREAD") -- a covering repl_on_ack call resolves a registered + * continuation before it returns, so no polling is needed anywhere below. + * + * Deliberately calls persist_submit_job_array_async() DIRECTLY -- job.c's + * handle_jobsubmit is not wired to this function yet (that is task 2.2's + * boundary, not this one's); nothing in job.c/handle_jobsubmit is touched + * here. + * ════════════════════════════════════════════════════════════════════════ */ + +static persist_t g_jap; /* stub persist for array-submit tests */ +static commit_engine_t g_jace; +static int g_jaep_a, g_jaep_b; /* dummy standby endpoints */ +static char g_jap_tmpdir[256]; + +static void jap_setup(uint32_t sync_timeout_ms) +{ + persist_config_t cfg; + + memset(&g_jap, 0, sizeof(g_jap)); + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = sync_timeout_ms; + + snprintf(g_jap_tmpdir, sizeof(g_jap_tmpdir), "/tmp/xce_jap21_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_jap_tmpdir)); + persist_set_workdir(&g_jap, g_jap_tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_jap, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_jap)); + + /* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until + * a test explicitly delivers them via repl_on_ack. */ + TEST_ASSERT_TRUE( + repl_add_standby(&g_jap.repl, &g_jaep_a, "jasb_a", 7420) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_jap.repl, &g_jaep_b, "jasb_b", 7421) >= 0); + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_jace, &g_jap, 3, 8)); + repl_set_commit_engine(&g_jap.repl, &g_jace); + repl_broadcast_send_hook = cbr_drop_send_hook; + + /* Reserved-0 contract: a fresh store starts current_lsn=1 -- the first + * array record lands at LSN 1 with no priming append needed. */ + TEST_ASSERT_EQUAL_UINT64(1, g_jap.current_lsn); +} + +static void jap_teardown(void) +{ + repl_broadcast_send_hook = NULL; + commit_engine_fini(&g_jace); + persist_stop(&g_jap); + persist_fini(&g_jap); + /* tmpdir left on disk — /tmp is ephemeral */ +} + +/* Minimal job_submit_data_t + xce_job_submit_t base spec -- same shape as + * cbrp_arrange_jobsubmit (test_commit_barrier_reply.c) and + * test_data_job_array_variant_round_trip (test_e2e_capnp.c); start_info + * left NULL, which that round-trip test already proves + * encode_JobStartInfo_ptr handles safely. */ +static job_submit_data_t g_jap_submit; +static xce_job_submit_t g_jap_spec; + +static void jap_arrange_base_spec(void) +{ + memset(&g_jap_submit, 0, sizeof(g_jap_submit)); + g_jap_submit.queue = (char *)"normal"; + g_jap_submit.user = (char *)"tester"; + g_jap_submit.mintasks = 1; + g_jap_submit.maxtasks = 1; + g_jap_submit.start_info = NULL; + g_jap_submit.job_name = (char *)"array-job"; + + memset(&g_jap_spec, 0, sizeof(g_jap_spec)); + /* Deliberately NOT set to a test's real base_jobid/array_size: proves + * persist_submit_job_array_async uses its own explicit base_jobid/ + * array_size PARAMETERS, never these same-shaped struct fields. */ + g_jap_spec.jobid = 0; + g_jap_spec.array_size = 0; + g_jap_spec.submit = &g_jap_submit; +} + +/* Read the WAL entry header located at explicit segment offset `at` back + * from the active segment file. Unlike read_last_wal_hdr in + * test_ha_role_lifecycle.c (which back-computes `seg_offset - + * sizeof(hdr)` and is only ever used for zero-payload WAL_NOOP entries, + * where that arithmetic happens to land on the header), this helper takes + * the header's start offset EXPLICITLY -- the caller snapshots + * p->seg_offset immediately before the append it wants to read back. + * Back-computing from the post-append p->seg_offset is only valid when + * payload_len == 0; every WAL_JOB_ARRAY_SUBMIT entry here carries a + * nonzero capnp payload, so `seg_offset - sizeof(hdr)` would land inside + * the payload bytes, not the header (caught by task 2.1 RED: read garbage + * decoded as a bogus type). */ +static int jap_read_wal_hdr_at(persist_t *p, off_t at, wal_entry_hdr_t *out) +{ + if (at < 0 || p->seg_offset < (uint64_t)at + sizeof(wal_entry_hdr_t)) + return -ENOENT; + + int fd = open(p->seg_path, O_RDONLY); + if (fd < 0) + return -errno; + + ssize_t r = pread(fd, out, sizeof(*out), at); + close(fd); + + if (r != (ssize_t)sizeof(*out)) + return -EIO; + return 0; +} + +/* Read the header AND payload bytes of the WAL entry located at explicit + * segment offset `at` back from the active segment file. Caller must + * free(*out_payload) (NULL/0-len payload leaves *out_payload NULL). */ +static int jap_read_wal_entry_at(persist_t *p, off_t at, + wal_entry_hdr_t *out_hdr, + uint8_t **out_payload) +{ + int rc = jap_read_wal_hdr_at(p, at, out_hdr); + if (rc != 0) + return rc; + + if (out_hdr->payload_len == 0) { + *out_payload = NULL; + return 0; + } + + int fd = open(p->seg_path, O_RDONLY); + if (fd < 0) + return -errno; + + off_t payload_at = at + (off_t)sizeof(wal_entry_hdr_t); + uint8_t *buf = (uint8_t *)malloc(out_hdr->payload_len); + TEST_ASSERT_NOT_NULL(buf); + ssize_t r = pread(fd, buf, out_hdr->payload_len, payload_at); + close(fd); + + if (r != (ssize_t)out_hdr->payload_len) { + free(buf); + return -EIO; + } + *out_payload = buf; + return 0; +} + +/* Notify-callback capture (persist_job_commit_notify_fn). */ +typedef struct { + int call_count; + uint64_t last_lsn; + int last_rc; +} jap_notify_obs_t; + +static jap_notify_obs_t g_jap_obs; + +static void jap_notify_reset(void) +{ + memset(&g_jap_obs, 0, sizeof(g_jap_obs)); +} + +static void jap_notify_fn(void *ctx, uint64_t lsn, int rc) +{ + jap_notify_obs_t *obs = (jap_notify_obs_t *)ctx; + obs->call_count++; + obs->last_lsn = lsn; + obs->last_rc = rc; +} + +/* ── One WAL append, one commit continuation, decodable back to the same + * (base_jobid, array_size, submit spec) — Req 1.1, 1.2, 1.3, 4.2, 9.2 ── */ + +static void test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation(void) +{ + const uint64_t base_jobid = 9500; + const uint64_t array_size = 5; + uint64_t lsn_before, out_lsn = 0; + int rc; + off_t entry_off; + wal_entry_hdr_t hdr; + uint8_t *payload = NULL; + + /* Arrange */ + jap_setup(2000); + jap_arrange_base_spec(); + jap_notify_reset(); + lsn_before = g_jap.current_lsn; + entry_off = (off_t)g_jap.seg_offset; + + /* Act: append the array record; the continuation is registered but not + * yet resolved (no covering ACK delivered yet). */ + rc = persist_submit_job_array_async(&g_jap, base_jobid, array_size, + &g_jap_spec, &out_lsn, + jap_notify_fn, &g_jap_obs); + + /* Assert: exactly one WAL entry was appended -- current_lsn advanced by + * exactly 1, regardless of array_size=5 */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "a valid array submission must append and register successfully"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, out_lsn, + "the array record must land at the pre-append current_lsn"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn, + "exactly ONE WAL entry must be appended for the whole array, " + "not one per element"); + + /* ...and it is the ONE entry on disk, typed WAL_JOB_ARRAY_SUBMIT, + * decodable back to the same (base_jobid, array_size, submit spec) -- + * NOT g_jap_spec's own (deliberately mismatched) jobid/array_size + * fields, proving the function used its explicit parameters */ + TEST_ASSERT_EQUAL_INT(0, + jap_read_wal_entry_at(&g_jap, entry_off, &hdr, &payload)); + TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)WAL_JOB_ARRAY_SUBMIT, hdr.type, + "the array record's WAL type must be WAL_JOB_ARRAY_SUBMIT"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(out_lsn, hdr.lsn, + "the on-disk entry's lsn must match the returned out_lsn"); + + { + struct capn cctx; + TEST_ASSERT_EQUAL_INT(0, + capn_init_mem(&cctx, payload, hdr.payload_len, 0)); + + JobSubmit_ptr jp; + jp.p = capn_getp(capn_root(&cctx), 0, 1); + TEST_ASSERT_TRUE(jp.p.type != CAPN_NULL); + + 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); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, decoded.jobid, + "decoded base jobid must match the base_jobid PARAMETER"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(array_size, decoded.array_size, + "decoded array_size must match the array_size PARAMETER"); + TEST_ASSERT_NOT_NULL(decoded.submit); + TEST_ASSERT_EQUAL_STRING("normal", decoded.submit->queue); + TEST_ASSERT_EQUAL_STRING("tester", decoded.submit->user); + + free_JobSubmit(&decoded); + capn_free(&cctx); + } + free(payload); + + /* ...exactly one continuation is registered -- not yet resolved before + * any ACK arrives */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count, + "the notify callback must not fire before a covering commit"); + + /* ...and fires EXACTLY ONCE, with rc==0, once a covering majority ACK + * resolves the single registered continuation */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); + repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, + "the notify callback must fire EXACTLY ONCE for the whole array, " + "not once per element"); + TEST_ASSERT_EQUAL_UINT64(out_lsn, g_jap_obs.last_lsn); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, + "a majority-committed array record must resolve rc=0"); + + jap_teardown(); + printf("test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation: PASS\n"); +} + +/* ── Encoded payload size does not depend on array_size (Req 7.1, 7.2) ──── */ + +static void test_job_array_submit_payload_size_independent_of_array_size(void) +{ + wal_entry_hdr_t hdr_small, hdr_large; + uint64_t out_lsn = 0; + int rc; + off_t off_small, off_large; + + /* Arrange */ + jap_setup(2000); + jap_arrange_base_spec(); + jap_notify_reset(); + + /* Act: array_size=2, then array_size=10000, SAME base spec -- both + * entries encoded through the identical code path in the same fixture + * so their on-disk payload lengths are directly comparable. */ + off_small = (off_t)g_jap.seg_offset; + rc = persist_submit_job_array_async(&g_jap, 9600, 2, &g_jap_spec, + &out_lsn, jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_small, &hdr_small)); + + jap_notify_reset(); + off_large = (off_t)g_jap.seg_offset; + rc = persist_submit_job_array_async(&g_jap, 20000, 10000, &g_jap_spec, + &out_lsn, jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, jap_read_wal_hdr_at(&g_jap, off_large, &hdr_large)); + + /* Assert: the encoded payload's size does not depend on array_size -- + * arraySize is a fixed-width uint64 field, not a per-element list */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len, + hdr_large.payload_len, + "the WAL_JOB_ARRAY_SUBMIT payload size must be IDENTICAL for " + "array_size=2 and array_size=10000 -- only the fixed-width " + "arraySize field differs, never the encoded byte length"); + + /* Cleanup: resolve both pending continuations with one covering ACK + * before teardown (hygiene — avoids leaving unresolved slots for + * commit_engine_fini to silently drop). */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); + repl_on_ack(&g_jap.repl, &g_jaep_a, hdr_large.lsn); + + jap_teardown(); + printf("test_job_array_submit_payload_size_independent_of_array_size: PASS\n"); +} + +/* ── Invalid-argument rejection (defensive input validation) ────────────── */ + +static void test_job_array_submit_rejects_array_size_one_or_less(void) +{ + uint64_t out_lsn = 0; + int rc; + + /* Arrange */ + jap_setup(2000); + jap_arrange_base_spec(); + jap_notify_reset(); + + /* Act / Assert: array_size<=1 is a single-job submission's shape, not + * this function's precondition (design.md: "array_size > 1") -- reject + * before any WAL append, no continuation registered, no callback fired */ + rc = persist_submit_job_array_async(&g_jap, 9700, 1, &g_jap_spec, + &out_lsn, jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc, + "array_size == 1 must be rejected -- not this function's job"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn, + "a rejected submission must never append a WAL entry"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count, + "a rejected submission's callback contract is 'never called', not " + "'called with an error' — the caller never registered anything"); + + rc = persist_submit_job_array_async(&g_jap, 9700, 0, &g_jap_spec, + &out_lsn, jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc, + "array_size == 0 must be rejected"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_jap.current_lsn, + "a rejected submission must never append a WAL entry"); + + jap_teardown(); + printf("test_job_array_submit_rejects_array_size_one_or_less: PASS\n"); +} + /* ── Test group entry point ──────────────────────────────────────────────── */ void test_commit_barrier_rollout(void) @@ -1767,5 +2125,10 @@ void test_commit_barrier_rollout(void) /* Task 5.2 remediation r1 — production wire-path ACK ingestion * (completes the unlanded half of ha-config-bootstrap task 8.1) */ RUN_TEST(test_config_ack_wire_path_completes_barrier); + /* job-array-single-wal-record task 2.1 — persist_submit_job_array_async: + * single-record array append/commit path (Req 1.1, 1.2, 1.3, 4.2, 9.2) */ + RUN_TEST(test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation); + RUN_TEST(test_job_array_submit_payload_size_independent_of_array_size); + RUN_TEST(test_job_array_submit_rejects_array_size_one_or_less); UNITY_END(); } diff --git a/src/include/persist.h b/src/include/persist.h index 53f340a..cee80be 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -10,6 +10,13 @@ #include "wal_types.h" /* wal_entry_hdr_t, wal_entry_type — P2-4: shared with repl.h */ #include "repl.h" /* replicator_t — P2-4: embedded in persist_t */ #include "xlist.h" /* xlist_t — persist_submit_wal_and_commit job batch */ +#include "xce/types.h" /* xce_job_submit_t — job-array-single-wal-record task 2.1: + * persist_submit_job_array_async's base_spec parameter. + * Header-guarded (_XCE_TYPES_H_), no circular dependency + * on persist.h — safe to pull in this early; the later, + * pre-existing #include "xce/types.h" further down this + * file (GC retention predicates) is now a harmless + * no-op re-inclusion. */ /* Forward declaration */ typedef struct cluster_s cluster_t; @@ -1077,6 +1084,41 @@ int persist_submit_job_payloads_async(persist_t* p, xlist_t* payloads, persist_job_commit_notify_fn fn, void* ctx); +/* job-array-single-wal-record task 2.1 (design.md §WAL Array Submission, + * Service Interface): encodes and appends the WHOLE array as ONE + * WAL_JOB_ARRAY_SUBMIT entry (base_jobid, array_size, one shared + * base_spec->submit) and registers ONE commit continuation for it, + * regardless of array_size -- the single-record sibling of + * persist_submit_job_payloads_async's per-payload loop. No job_t/xce_job_t + * object is constructed here or anywhere on this call path (mirrors the + * "nothing exists until apply" invariant persist_submit_job_payloads_async + * already documents) -- array expansion into individual job entries happens + * later, in the shared apply seam (job-array-single-wal-record task 3.x). + * + * Preconditions: array_size > 1 (array_size <= 1 returns -EINVAL; a + * single-job submission stays on the unmodified persist_submit_wal_and_ + * commit / persist_submit_job_payloads_async path); base_spec != NULL and + * base_spec->submit != NULL; the caller has already range-claimed + * [base_jobid, base_jobid + array_size) (job.c's inflight_range_try_add) -- + * not re-validated here, same division of responsibility as the payload- + * loop sibling above. + * + * Postconditions: on success, exactly one WAL entry is durably appended + * locally (*out_lsn set to its LSN) and exactly one continuation is + * registered; `fn` is invoked exactly once with the outcome -- either + * synchronously, right here (a local encode/append failure, the once-per- + * cluster-lifetime LSN-0 edge, or an engine-not-wired context), or later, + * asynchronously, once the entry's commit outcome is known. Returns 0 once + * the entry is durably appended locally and its outcome resolution path is + * established (sync or async); a negative rc otherwise -- `fn` has already + * been invoked (or is already guaranteed to be invoked later) by the time + * this function returns, regardless of its return value. */ +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); + /* raft-apply-decoupling task 3.2: producer-side hand-off for a command * whose commit_submit_async() continuation resolved COMMITTED. Per task * 3.2's own contract, a COMMITTED outcome must NOT apply state or reply to