From d38ca72e76f21cd593112c364faa09a7e1b37c40 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 14 Jul 2026 10:32:59 +0000 Subject: [PATCH] feat(job-array-single-wal-record): wire array submissions into handle_jobsubmit Branches handle_jobsubmit() on array_size: ==1 continues through today's exact unmodified single-job path; >1 routes through a new handle_jobsubmit_array() helper that reuses jobsubmit_async_ctx_t with pending=1 and calls task 2.1's persist_submit_job_array_async() instead of looping N single-element appends. Range validation (jobid_range_exists/inflight_range_try_add) unchanged, same order, strictly before any WAL append, for both branches. Retires two pre-existing tests whose premise (N independent per-element WAL entries with mixed outcomes) is now structurally impossible with one entry per array; restores array-scoped job-non-existence coverage for the failed-commit case that an earlier draft of this change had dropped (Req 2.2/9.2). Task 2.2 of job-array-single-wal-record. --- .../job-array-single-wal-record/tasks.md | 2 +- src/controller/job.c | 74 ++++++ .../testsuite/test_commit_barrier_reply.c | 217 ++++++++---------- 3 files changed, 174 insertions(+), 119 deletions(-) diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index ac75c2e0..b211dcaf 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -32,7 +32,7 @@ - _Requirements: 1.1, 1.2, 1.3, 4.2, 9.2_ - _Boundary: WAL Array Submission_ -- [ ] 2.2 Wire array submissions into the request handler +- [x] 2.2 Wire array submissions into the request handler - Branch the submission handler on array size: exactly 1 continues through today's existing single-job code path completely unmodified; greater than 1 routes to 2.1's single-append path - Preserve the existing pre-commit range validation (collision check plus in-flight claim) unchanged and in the same order, still rejecting strictly before any WAL append - Reuse the existing batched-reply mechanism, setting it to expect exactly one outcome instead of one per element diff --git a/src/controller/job.c b/src/controller/job.c index c718f26d..1b48a5db 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -923,6 +923,72 @@ static void jobsubmit_commit_continuation(void* ctx, uint64_t lsn, int rc) jobsubmit_apply_complete, actx); } +/* job-array-single-wal-record task 2.2 (design.md §WAL Array Submission): + * the array-shaped sibling of handle_jobsubmit()'s single-job append + + * continuation-registration block below, taken for submit->array_size > 1. + * Called strictly AFTER handle_jobsubmit() has already run the SAME + * pre-commit range validation (jobid_range_exists + inflight_range_try_add) + * the single-job path uses — this function never repeats or reorders that + * validation (Req 9.1, 9.2); it only takes over at the point the two paths + * diverge. + * + * jobsubmit_async_ctx_t / jobsubmit_commit_continuation / jobsubmit_apply_ + * complete / jobsubmit_reply_success_once / jobsubmit_reply_failure_once are + * reused completely unmodified — task 2.1's persist_submit_job_array_async() + * plugs into the exact same commit-continuation contract as persist_submit_ + * job_payloads_async() does for the single-job path (persist.h). The only + * difference from the single-job construction below is actx->pending = 1 + * instead of one entry per array element (Req 1.1-1.3): exactly one + * WAL_JOB_ARRAY_SUBMIT record describes the whole array, so exactly one + * outcome must resolve before the batch's single reply fires (Req 9.3). No + * job_t/xce_job_t payload is built here — array expansion into individual + * job entries happens later, in the shared apply seam (task 3.x), not on + * this call path. */ +static int handle_jobsubmit_array(cluster_t* c, endpoint_t* ep, + xce_request_t* request, uint64_t jobid, + xce_job_submit_t* submit) +{ + xce_reply_t reply; + uint64_t array_lsn = 0; + jobsubmit_async_ctx_t* actx; + + init_reply_by_req(&reply, request); + + actx = calloc(1, sizeof(*actx)); + if (actx == NULL) { + inflight_range_remove(jobid, submit->array_size); + + reply.status = -ENOMEM; + reply.data_which = RequestReply_data_nodata; + + xce_reply(ep, c->myhost, &reply); + return -ENOMEM; + } + actx->c = c; + actx->ep = ep; + actx->seq = request->seq; + actx->opcode = request->opcode; + actx->jobid = jobid; + actx->array_size = submit->array_size; + actx->pending = 1; /* one WAL_JOB_ARRAY_SUBMIT record, one outcome */ + actx->replied = 0; + pthread_mutex_init(&actx->lock, NULL); + endpoint_ref(ep); + + /* Mirrors handle_jobsubmit's own call below: the return value carries + * no information this function still needs to act on — fn (jobsubmit_ + * commit_continuation) has already been invoked, or is already + * guaranteed to be invoked later, exactly once, by the time this call + * returns (persist_submit_job_array_async's own documented contract, + * persist.h). */ + (void)persist_submit_job_array_async(&c->persist, jobid, + submit->array_size, submit, + &array_lsn, + jobsubmit_commit_continuation, actx); + + return 0; +} + int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) { uint64_t jobid; @@ -980,6 +1046,14 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) return -errno; } + /* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 8.1, 9.1, 9.2): + * array_size == 1 falls through to today's single-job path below, + * completely unmodified (Req 8.1) — array_size > 1 diverges here, + * strictly AFTER the same pre-commit range validation above, into the + * single-record append path task 2.1 built. */ + if (submit->array_size > 1) + return handle_jobsubmit_array(c, ep, request, jobid, submit); + /* Build only the WAL payload data for each array element — no job * object is constructed or inserted into c->jobmgr until the commit * below succeeds (raft-apply-decoupling task 3.1 / Req 2.1, 2.3): a diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index f78b6600..351e2a90 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -1463,49 +1463,45 @@ static void test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs(v " PASS\n"); } -/* Closes task 3.2's flagged gap ("no array_size > 1 mixed-outcome test - * exists yet"): element 1 commits (covering ACK) while elements 2 and 3 - * are left to time out FAILED. Confirms the failed elements produce zero - * phantom jobs and that the committed element's independent resolution - * is not blocked or skipped by its failing siblings -- task 3.2's - * per-element independence, proven by a real test rather than by code - * inspection alone. - * - * raft-apply-decoupling task 7.3 (Req 11.1) update: this test originally - * asserted (task 3.2, before task 6.1/6.2 existed) that the committed - * element had NO job yet, only a pending-apply hand-off -- catching that - * hand-off's transient window from a second thread is inherently racy - * against the ordered apply pump's own consumption of it (proven flaky - * empirically across repeated runs of this test: sometimes the window - * was caught, sometimes it was already consumed by the time this test - * polled). Now that the pump exists, the deterministic, settled effect is - * that element 1's mutation applies (job1 becomes visible in c->jobmgr) - * independently of its failing siblings, exactly like any other committed - * entry -- the per-element independence this test exists to prove now - * extends from "resolution" to "apply", per the design's own stated - * intent that a command's apply is per-entry and never gated on sibling - * array elements (collapsing that behavior is job-array-single-wal- - * record's job, not this feature's). The client-visible batch reply is - * unaffected: it is still the single failure reply from the timed-out - * siblings, because jobsubmit_reply_success_once() is gated on the same - * actx->replied flag as jobsubmit_reply_failure_once() (job.c), and the - * failure reply always wins that race here. */ -static void test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements(void) +/* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 2.1-2.3): supersedes + * the pre-2.2 test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_ + * elements (raft-apply-decoupling task 3.2, bug-1408 lineage), which forced + * a genuinely "mixed" per-element outcome by ACKing only the first of + * THREE separate WAL_JOB_SUBMIT entries and letting the other two time out + * independently. That scenario is now structurally IMPOSSIBLE: task 2.1 + * collapsed an array submission into exactly ONE WAL_JOB_ARRAY_SUBMIT + * record, so there is exactly one LSN, one continuation, and one outcome + * for the whole array -- Req 2.1's "no state in which some elements of + * that record are committed and others are not" now holds by construction, + * not just by convention. This test proves that atomic-unit contract at + * the handler-wiring boundary THIS task owns: with no covering ACK ever + * delivered, the array's single entry times out as ONE unit and the batch + * gets exactly ONE -ETIMEDOUT reply -- the array-shaped analog of + * test_jobsubmit_returns_before_commit_on_failure_path_too above. + * (Per-element job VISIBILITY after a fully-COMMITTED array -- i.e. that + * every element eventually appears in c->jobmgr -- is task 3.x's + * restore_job_array()/apply-expansion boundary, not this one's -- see + * test_job_array_apply.c, design.md File Structure Plan. But the inverse, + * that a FAILED/timed-out commit constructs no job at all for any element + * the array would have spanned, is this task's own commit-side contract + * (Req 2.2/9.2: "if commit fails, treat as not committed, construct + * nothing") and is pinned below.) */ +static void test_jobsubmit_array_timeout_appends_exactly_one_wal_entry_and_replies_once(void) { const uint64_t base_jobid = 9140; - uint64_t lsn_before, watermark, lsn1; + const uint64_t array_size = 3; + uint64_t lsn_before, watermark; int create_rc, join_rc; cbrp_reply_obs_t obs = { 0 }; - bool job1_exists, job2_exists, job3_exists; + bool job0_exists, job1_exists, job2_exists; - cbrp_setup(300); /* short: elements 2/3 must resolve via the timeout - * sweep within a bounded window */ - cbrp_arrange_jobsubmit(base_jobid, 3); + cbrp_setup(300); /* short: the array's one entry must resolve via the + * timeout sweep within a bounded window */ + cbrp_arrange_jobsubmit(base_jobid, array_size); pthread_mutex_lock(&g_hc.persist.lock); lsn_before = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); - lsn1 = lsn_before; g_op_arg.fn = handle_jobsubmit; g_op_arg.req = &g_jobsub_req; @@ -1513,54 +1509,49 @@ static void test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_element create_rc = pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg); join_rc = (create_rc == 0) ? pthread_join(g_op_tid, NULL) : -1; - /* All three array elements append back-to-back (task 3.2's - * non-blocking, no per-element ACK gate) before handle_jobsubmit - * returns -- by the time pthread_join above returns, every element's - * WAL entry already exists. */ - watermark = cbrp_poll_append(&g_hc.persist, lsn_before + 2, 2000); + /* Exactly ONE WAL entry for the whole array, regardless of array_size + * (Req 1.1-1.3) -- current_lsn advances by exactly one, landing the + * array's single record at lsn_before itself. */ + watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); - /* Cover ONLY the first element with a majority ACK; elements 2 and 3 - * are left pending -- they must resolve FAILED via the timeout sweep, - * independently of element 1's success. */ - repl_on_ack(&g_hc.persist.repl, &g_hep_a, lsn1); - - /* Poll for the batch's single settled reply (the failing siblings' - * timeout), then read the deterministic end state of every element's - * job existence -- not a transient hand-off window (see doc comment - * above). */ + /* No ACK is ever delivered: the array's one entry must resolve FAILED + * via the timeout sweep, exactly like a single-job submission. */ cbrp_poll_reply(&g_req_ep, 300 + 1500, &obs); - job1_exists = cbrp_job_exists(base_jobid); - job2_exists = cbrp_job_exists(base_jobid + 1); - job3_exists = cbrp_job_exists(base_jobid + 2); + /* Req 2.2/9.2: a failed/timed-out array commit must construct ZERO + * job elements -- checked for every constituent jobid the array would + * have spanned, not just the base id, since the collapsed single-entry + * record could in principle be partially expanded by a defect. */ + job0_exists = cbrp_job_exists(base_jobid); + job1_exists = cbrp_job_exists(base_jobid + 1); + job2_exists = cbrp_job_exists(base_jobid + 2); cbrp_teardown(); TEST_ASSERT_EQUAL_INT_MESSAGE(0, create_rc, "pthread_create must succeed"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, join_rc, "pthread_join must succeed"); - TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 2, watermark, - "all three array elements must append before this poll returns"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, watermark, + "an array submission must append exactly ONE WAL entry regardless " + "of array_size (Req 1.1-1.3), never one per element"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, - "a partially-failed array batch must produce exactly one failure " - "reply for the whole request (jobsubmit_reply_failure_once)"); + "a timed-out array submission must produce exactly one reply for " + "the whole request, never one per element"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status, - "the batch's single reply must carry the failing elements' " - "-ETIMEDOUT outcome"); - TEST_ASSERT_TRUE_MESSAGE(job1_exists, - "the committed array element's mutation must apply independently " - "of its failing siblings, exactly like any other committed entry, " - "now that the ordered apply pump exists (raft-apply-decoupling " - "task 6.1/6.2) -- the client's batch reply still reports failure " - "because a failing sibling wins jobsubmit_async_ctx_t's " - "single-reply race, but that reply does not gate this element's " - "independent apply"); + "the array's single entry times out as ONE atomic unit -- there is " + "no per-element outcome left to mix (Req 2.1)"); + TEST_ASSERT_FALSE_MESSAGE(job0_exists, + "a failed/timed-out array commit must construct no job for the " + "base element (Req 2.2/9.2: treat as not committed, construct " + "nothing)"); + TEST_ASSERT_FALSE_MESSAGE(job1_exists, + "a failed/timed-out array commit must construct no job for any " + "constituent element (Req 2.2/9.2)"); TEST_ASSERT_FALSE_MESSAGE(job2_exists, - "a failed array element must never have a phantom job"); - TEST_ASSERT_FALSE_MESSAGE(job3_exists, - "a failed array element must never have a phantom job"); + "a failed/timed-out array commit must construct no job for any " + "constituent element (Req 2.2/9.2)"); - printf("test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements:" - " PASS\n"); + printf("test_jobsubmit_array_timeout_appends_exactly_one_wal_entry_and_" + "replies_once: PASS\n"); } /* ── handle_jobsubmit success path via the ordered apply pump @@ -1707,27 +1698,37 @@ static void test_jobsubmit_success_reply_after_apply_carries_jobid(void) printf("test_jobsubmit_success_reply_after_apply_carries_jobid: PASS\n"); } -/* Array submission (3 elements, base jobid 9210): a single covering ACK - * commits all three entries at once (mirroring test_apply_pump.c's - * burst-order coverage), and the pump applies + hands off each one - * independently, in ascending LSN order. Confirms task 3.2's per-element - * independent registration is correctly drained by task 6.2's per-element - * consumption: every element's own post-mutation notification fires before - * the single batch-level reply is ever observable, and every element is - * visible in c->jobmgr by the time that reply arrives. */ -static void test_jobsubmit_array_success_all_elements_apply_before_single_reply( +/* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 8.1, 9.1, 9.2, 9.3): + * supersedes the pre-2.2 test_jobsubmit_array_success_all_elements_apply_ + * before_single_reply (raft-apply-decoupling task 6.2), which asserted + * THREE independent WAL entries, three independent HOOK_DETAIL_JOB_NEW + * fires, and all three elements visible in c->jobmgr -- all of that was + * per-element APPLY-EXPANSION behavior belonging to the old per-element + * persist_submit_job_payloads_async() path. Task 2.1 collapsed the + * append/commit side to one WAL_JOB_ARRAY_SUBMIT record; task 3.x + * (restore_job_array(), not yet implemented) owns expanding that ONE + * committed record into N job_t entries and firing HOOK_DETAIL_JOB_NEW + * once with all of them (Req 3.4) -- see test_job_array_apply.c, design.md + * File Structure Plan. Until 3.x lands, apply_committed_entry()'s default + * case treats an unrecognized WAL_JOB_ARRAY_SUBMIT entry as a harmless + * no-op (no job, no hook, `default:` branch, persist.c) -- a committed-but- + * not-yet-expanded posture, not a bug in this task. + * + * What THIS task (2.2) owns and this test proves: a covering ACK over the + * array's single entry resolves it COMMITTED, and jobsubmit_apply_complete + * -- reused completely unmodified from the single-job path, with actx-> + * pending seeded to 1 instead of array_size -- fires exactly once and sends + * exactly one success reply carrying the BASE job id, regardless of + * array_size (Req 1.1-1.3, 9.3). */ +static void test_jobsubmit_array_success_appends_one_entry_and_replies_once_with_base_jobid( void) { const uint64_t base_jobid = 9210; const uint64_t array_size = 3; uint64_t lsn_before, watermark; cbrp_reply_obs_t final = { 0 }; - bool job_exists[3]; cbrp_setup(2000); - TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, - cbrp_jobsubmit_order_spy, &g_hc)); - cbrp_jobsubmit_order_spy_reset(); cbrp_arrange_jobsubmit(base_jobid, array_size); pthread_mutex_lock(&g_hc.persist.lock); @@ -1741,25 +1742,21 @@ static void test_jobsubmit_array_success_all_elements_apply_before_single_reply( cbrp_op_thread_fn, &g_op_arg)); TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); - /* All three elements append back-to-back before handle_jobsubmit - * returns (task 3.2) -- one covering ACK past the last of them commits - * the whole batch in a single jump. */ - watermark = cbrp_poll_append(&g_hc.persist, lsn_before + (array_size - 1), - 2000); + /* Exactly ONE WAL entry for the whole array (Req 1.1-1.3) -- the + * covering ACK below only ever needs to reach this ONE lsn, regardless + * of array_size. */ + watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); cbrp_poll_reply(&g_req_ep, 2000, &final); - job_exists[0] = cbrp_job_exists(base_jobid); - job_exists[1] = cbrp_job_exists(base_jobid + 1); - job_exists[2] = cbrp_job_exists(base_jobid + 2); - cbrp_teardown(); - TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + (array_size - 1), watermark, - "all three array elements must append before this poll returns"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, watermark, + "an array submission must append exactly ONE WAL entry regardless " + "of array_size, never one per element (Req 1.1-1.3)"); TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0, - "the batch must eventually reply once every element applies"); + "the batch must eventually reply once its single entry commits"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, final.n_replies, "an all-succeeding array batch must produce exactly ONE reply for " "the whole request, not one per element (Req 9.3)"); @@ -1770,28 +1767,10 @@ static void test_jobsubmit_array_success_all_elements_apply_before_single_reply( "the batch success reply must carry a job id"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, final.last_jobid, "the batch success reply must echo the BASE job id, matching " - "pre-task-3.1 semantics (git history commit 6f1fa97), never a " - "per-element id"); + "pre-existing semantics, never a per-element id"); - TEST_ASSERT_TRUE_MESSAGE(job_exists[0], - "array element 0 must be visible in c->jobmgr by reply time"); - TEST_ASSERT_TRUE_MESSAGE(job_exists[1], - "array element 1 must be visible in c->jobmgr by reply time"); - TEST_ASSERT_TRUE_MESSAGE(job_exists[2], - "array element 2 must be visible in c->jobmgr by reply time"); - - TEST_ASSERT_EQUAL_INT_MESSAGE(3, g_jobsubmit_order_spy_fire_count, - "each of the three array elements must independently fire its own " - "post-mutation notification (task 3.2's per-element registration, " - "drained one at a time by task 6.2)"); - TEST_ASSERT_FALSE_MESSAGE(g_jobsubmit_order_spy_reply_seen_early, - "no array element's post-mutation notification may ever observe " - "the batch's single reply already enqueued -- the reply can only " - "follow the LAST element's own mutation, and every earlier " - "element's mutation must equally precede it"); - - printf("test_jobsubmit_array_success_all_elements_apply_before_single_" - "reply: PASS\n"); + printf("test_jobsubmit_array_success_appends_one_entry_and_replies_once_" + "with_base_jobid: PASS\n"); } /* REMOVED (raft-apply-decoupling task 3.2): the array-partial-construction- @@ -3348,11 +3327,13 @@ void test_commit_barrier_reply(void) RUN_TEST(test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry); RUN_TEST(test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job); RUN_TEST(test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs); - RUN_TEST(test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements); + /* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 2.1-2.3, 9.3): + * one WAL_JOB_ARRAY_SUBMIT record, one continuation, one reply */ + RUN_TEST(test_jobsubmit_array_timeout_appends_exactly_one_wal_entry_and_replies_once); /* handle_jobsubmit success path via the ordered apply pump * (raft-apply-decoupling task 6.2, Req 1.1, 2.1-2.3) */ RUN_TEST(test_jobsubmit_success_reply_after_apply_carries_jobid); - RUN_TEST(test_jobsubmit_array_success_all_elements_apply_before_single_reply); + RUN_TEST(test_jobsubmit_array_success_appends_one_entry_and_replies_once_with_base_jobid); /* Receipt-ACK verdict pins (ha-commit-barrier-rollout task 3.3, Req 2.3) * — green on landing */ RUN_TEST(test_jobdispatch_receipt_ack_no_wal_append);