From fe5ddc98549627b06c5e228208b65e8cbdc66c12 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 1 Oct 2026 10:14:47 +0000 Subject: [PATCH] feat(job-array-one-based-index): collision checks after registration (task 2.3) handle_jobsubmit now registers the submission first and checks the job table second; a hit is answered -EEXIST and released through the atomic reply-and-release step. For array submissions, array_identity_exists() does one read-locked scan of the 32768 tagged element keys of the array's block (element 0 included for pre-upgrade arrays), covering both the element range and the array identity; unmarked submissions keep jobid_range_exists. An existing X|3 now blocks a new X|5. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../specs/job-array-one-based-index/tasks.md | 2 +- src/controller/include/cluster.h | 7 + src/controller/job.c | 88 ++++++-- .../testsuite/test_commit_barrier_reply.c | 7 +- .../testsuite/test_jobsubmit_validate.c | 201 ++++++++++++++++++ 5 files changed, 278 insertions(+), 27 deletions(-) diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index db6b2218..ceebe64b 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -43,7 +43,7 @@ - Done when the controller suite shows, from inside the hook, that the registry lock is held, the reply is already queued and the entry is still present; that the entry is gone afterwards; and that a retry of the same submission registered after the reply survives the original's context release - _Requirements: 7.7, 7.10, 7.11_ -- [ ] 2.3 Check element and array-identity collisions after registration +- [x] 2.3 Check element and array-identity collisions after registration - Move the job-table collision checks after registration; on a hit, reply duplicate-submission and release through the atomic step - For array submissions, one scan under a single job-table read lock over the 32768 element keys of the array's block answers both the element-range and the array-identity check; unmarked submissions keep the existing range check - Done when the controller suite shows: an existing element `X|3` blocks a new `X|5` with duplicate-submission; two concurrent submissions with the same array identity, the second gets duplicate-submission while the first is unanswered; once the first is answered with a failure, a new one with the same identity is accepted diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 72ea4a83..9ca999e2 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -121,6 +121,13 @@ typedef enum { } jobsubmit_class_t; extern jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, uint64_t size); + +/* job-array-one-based-index (Req 7.10): 1 if any of the 32768 element keys + * of array_id's block (TAG_JOB_ARRAY, elements 0..MAX_ARRAY_IDX, element 0 + * included for pre-upgrade arrays) is in c->jobmgr, 0 otherwise. One scan + * under a single read-lock hold; bounded CPU work, no I/O. array_id is the + * array identity (tag and element number cleared). */ +extern int array_identity_exists(cluster_t* c, uint64_t array_id); #ifdef XCE_TESTING /* bug-1419: clears job.c's process-lifetime in-flight-jobid registry. * Tests that deliberately leave a submission's commit continuation diff --git a/src/controller/job.c b/src/controller/job.c index 697c94f7..9753d47c 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -101,8 +101,8 @@ static job_t* create_job(cluster_t* c, xce_job_t* j) * process-local registry, and treat a hit there identically to a * c->jobmgr hit — both mean "this exact write already happened or is * already in flight," which is precisely what a retried write must see as - * a duplicate. Populated in handle_jobsubmit() right after this check - * passes (before any WAL work). + * a duplicate. Populated in handle_jobsubmit() before the job-table + * collision check (and before any WAL work). * * job-array-one-based-index (Req 7.7, 7.10, 7.11): an entry also carries * whether it is an array and its array identity (tag cleared, low 15 bits @@ -283,9 +283,13 @@ int jobsubmit_inflight_try_add_for_test(uint64_t base, uint64_t size) * happened". * * bug-1419: this check alone is no longer sufficient (see - * inflight_range_try_add's doc comment) — handle_jobsubmit additionally - * calls inflight_range_try_add() after this check passes, which atomically - * closes the remaining not-yet-applied window this function cannot see. */ + * inflight_range_try_add's doc comment) — handle_jobsubmit first calls + * inflight_range_try_add(), which atomically closes the not-yet-applied + * window this function cannot see, and runs this check only after that + * registration succeeds (job-array-one-based-index, Req 7.10, 7.11); + * a hit is answered and released through inflight_reply_and_remove(). + * Used for unmarked (tag-0) submissions only; an array submission is + * checked by array_identity_exists() instead. */ static bool jobid_range_exists(cluster_t* c, uint64_t base_jobid, uint64_t array_size) { @@ -304,6 +308,34 @@ static bool jobid_range_exists(cluster_t* c, uint64_t base_jobid, return false; } +/* job-array-one-based-index (Req 7.10): true if any element key of + * array_id's block -- MAKE_JOBID(.., TAG_JOB_ARRAY, A|k) for k = 0.. + * MAX_ARRAY_IDX, element 0 included for pre-upgrade arrays -- is in + * c->jobmgr. One scan under a single read-lock hold (key_find takes no lock + * of its own) answers both the element-range check for a classified array + * submission (its range lies inside the block) and the array-identity + * check, so it replaces jobid_range_exists() for tag-1 submissions. Bounded + * CPU work (32768 hash lookups), no I/O, no blocking (ha-async-io.md). */ +int array_identity_exists(cluster_t* c, uint64_t array_id) +{ + const uint64_t first = MAKE_JOBID( + EXTRACT_ID_CLUSTER(array_id), EXTRACT_ID_APISERVER(array_id), + EXTRACT_ID_TIMESTAMP(array_id), TAG_JOB_ARRAY, + EXTRACT_ARRAY_JOB(array_id)); + int found = 0; + + lock(c->jobmgr, RTTI_LOCK_READ); + for (uint64_t k = 0; k <= MAX_ARRAY_IDX && !found; k++) { + job_t* existing = NULL; + + if (key_find(c->jobmgr, (void*)(first | k), (void**)&existing) > 0 + && existing != NULL) + found = 1; + } + unlock(c->jobmgr); + return found; +} + /* Build one xce_job_t submission payload per array element, WITHOUT * constructing or registering any job object (raft-apply-decoupling task * 3.1 / Req 2.1, 2.3) — these are plain heap structs carrying only the data @@ -2452,7 +2484,7 @@ static void jobsubmit_commit_continuation(void* ctx, uint64_t lsn, int rc) * 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) + * pre-commit validation (inflight_range_try_add, then array_identity_exists) * 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. @@ -2599,7 +2631,19 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) jobid = submit->jobid; - if (jobid_range_exists(c, jobid, submit->array_size)) { + /* bug-1419: atomically claim this jobid range (and, for an array, its + * array identity) as in-flight, closing the window the c->jobmgr check + * alone cannot see (see jobid_range_exists' doc comment). A false + * return means another submission (a genuine concurrent collision, or + * -- the actual motivating case -- this exact request's own retried + * resend racing its still-in-flight original attempt) already holds an + * overlapping range or the same array identity; reject it + * duplicate-submission. From here on every answer goes through + * inflight_reply_and_remove(), which sends the reply and releases this + * registration as one step (job-array-one-based-index, Req 7.7, 7.10, + * 7.11) -- the early-outs below directly, the commit outcome via + * jobsubmit_reply_*_once. */ + if (!inflight_range_try_add(jobid, submit->array_size)) { errno = EEXIST; reply.status = -errno; reply.data_which = RequestReply_data_nodata; @@ -2608,24 +2652,22 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) return -errno; } - /* bug-1419: atomically claim this jobid range as in-flight, closing the - * window jobid_range_exists' c->jobmgr check alone cannot see (see its - * doc comment). A false return means another submission (a genuine - * concurrent collision, or -- the actual motivating case -- this exact - * request's own retried resend racing its still-in-flight original - * attempt) already holds this range; reject exactly as the - * jobid_range_exists branch above does. From here on every answer - * goes through inflight_reply_and_remove(), which sends the reply and - * releases this registration as one step (job-array-one-based-index, - * Req 7.7, 7.10, 7.11) -- the early-outs below directly, the commit - * outcome via jobsubmit_reply_*_once. */ - if (!inflight_range_try_add(jobid, submit->array_size)) { - errno = EEXIST; - reply.status = -errno; + /* job-array-one-based-index (Req 7.10, 7.11): the job-table checks run + * AFTER registration, so a same-identity submission applied and + * answered between the two steps cannot be missed by both (an applied + * submission is in c->jobmgr before it leaves the registry). An array + * is checked by one scan of its whole block (element range and array + * identity); an ordinary job keeps the element-range check. */ + if (is_array_tag(jobid) + ? array_identity_exists(c, array_jobid(jobid)) + : jobid_range_exists(c, jobid, submit->array_size)) { + reply.status = -EEXIST; reply.data_which = RequestReply_data_nodata; - xce_reply(ep, c->myhost, &reply); - return -errno; + inflight_reply_and_remove(ep, c->myhost, &reply, jobid, + submit->array_size); + errno = EEXIST; + return -EEXIST; } /* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 8.1, 9.1, 9.2): diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index dbe2cc25..29c96e14 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -1455,9 +1455,10 @@ static void test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry( /* bug-1419 residual: does the SAME dedup guard still hold AFTER the first * attempt has fully resolved (committed, applied, replied, in-flight claim - * released via jobsubmit_async_ctx_release) -- not just DURING the - * still-unresolved window the test above proves. A live chaotic-soak run - * (2026-07-08) showed a genuine WAL-level jobid duplicate (two committed + * released together with the reply via inflight_reply_and_remove) -- not + * just DURING the still-unresolved window the test above proves. A live + * chaotic-soak run (2026-07-08) showed a genuine WAL-level jobid duplicate + * (two committed * WAL_JOB_SUBMIT entries, same term, same primary) that neither known guard * should have let through if the retry arrived strictly before OR strictly * after resolution -- this test pins down whether the "after" window is as diff --git a/src/controller/testsuite/test_jobsubmit_validate.c b/src/controller/testsuite/test_jobsubmit_validate.c index 32c7ca5a..3c6f084d 100644 --- a/src/controller/testsuite/test_jobsubmit_validate.c +++ b/src/controller/testsuite/test_jobsubmit_validate.c @@ -33,6 +33,12 @@ * observes the state between the reply and the unlink; * jobsubmit_test_release_hook registers a retry between the reply and the * context release. + * + * Task 2.3 (Requirements 7.10, 7.11): the job-table checks run after + * registration. An existing element of the same array block (X|3 blocks a + * new X|5; element 0 and element 32767 included) or of the element range + * is answered -EEXIST through the atomic reply-and-release step, observed + * with the same reply hook; array_identity_exists() is also driven directly. */ /* rtti.h must come first -- it defines 'var' and 'typeinfo_t' used by kvmgr.h @@ -699,6 +705,196 @@ static void test_jobsubmit_same_array_identity_while_registered(void) jsv_teardown(); } +/* ── Task 2.3: job-table collisions after registration ───────────────── */ + +static job_submit_data_t g_jsv_seed_submit; + +/* Store an existing job `jobid` in the fixture's job table. */ +static void jsv_seed_job(uint64_t jobid) +{ + xce_job_t jdef; + + memset(&g_jsv_seed_submit, 0, sizeof(g_jsv_seed_submit)); + g_jsv_seed_submit.queue = (char*)"normal"; + g_jsv_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_PEND; + jdef.submit = &g_jsv_seed_submit; + + TEST_ASSERT_NOT_NULL_MESSAGE(makeJob(g_jsv_c.jobmgr, &jdef), + "fixture: seeding an existing job"); + TEST_ASSERT_TRUE(jsv_job_exists(jobid)); +} + +/* Submit (first, size) against a table holding `existing` and expect the + * duplicate-submission reply: answered through inflight_reply_and_remove + * while the submission is registered (so the table check ran after + * registration), its entry released afterwards, and nothing new stored. */ +static void jsv_expect_table_collision(uint64_t existing, uint64_t first, + uint64_t size, const char* what) +{ + char* out = NULL; + char msg[200]; + int is_array = is_array_tag(first) ? 1 : 0; + uint64_t array_id = is_array ? array_jobid(first) : 0; + uint64_t i; + int rc; + + jsv_setup(); + jsv_seed_job(existing); + + memset(&g_jsv_hook, 0, sizeof(g_jsv_hook)); + g_jsv_hook.base = first; + g_jsv_hook.size = size; + g_jsv_hook.is_array = is_array; + g_jsv_hook.array_id = array_id; + jobsubmit_test_reply_hook = jsv_reply_hook; + + jsv_arrange(first, size, false); + rc = jsv_submit_capture(&out); + free(out); + + { + jsv_reply_obs_t obs = { 0 }; + + jsv_drain(&obs); + snprintf(msg, sizeof(msg), "%s: exactly one reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, msg); + snprintf(msg, sizeof(msg), "%s: duplicate-submission reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, obs.last_status, msg); + snprintf(msg, sizeof(msg), "%s: reply echoes the request seq", what); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(21, obs.last_seq, msg); + snprintf(msg, sizeof(msg), "%s: handler return value", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc, msg); + } + snprintf(msg, sizeof(msg), + "%s: the table hit is answered through the atomic " + "reply-and-release step (registered first)", + what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_hook.calls, msg); + snprintf(msg, sizeof(msg), "%s: registry lock held across the reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(EBUSY, g_jsv_hook.trylock_rc, msg); + snprintf(msg, sizeof(msg), "%s: entry registered while answered", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_hook.entry_present, msg); + snprintf(msg, sizeof(msg), "%s: entry released after the reply", what); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, + jobsubmit_inflight_overlaps_for_test(first, size, is_array, array_id), + msg); + snprintf(msg, sizeof(msg), "%s: the existing job is unchanged", what); + TEST_ASSERT_TRUE_MESSAGE(jsv_job_exists(existing), msg); + for (i = 0; i < size && i < 8; i++) { + if (first + i == existing) + continue; + snprintf(msg, sizeof(msg), "%s: no job stored at offset %llu", what, + (unsigned long long)i); + TEST_ASSERT_FALSE_MESSAGE(jsv_job_exists(first + i), msg); + } + jsv_teardown(); +} + +/* 7.10: an existing element X|3 blocks a new X|5 (same array identity, + * disjoint elements) -- one-element and multi-element submissions. */ +static void test_jobsubmit_existing_element_blocks_same_identity(void) +{ + jsv_expect_table_collision(jsv_elem(3), jsv_elem(5), 1, + "existing X|3, new X|5 (1 element)"); + jsv_expect_table_collision(jsv_elem(3), jsv_elem(5), 4, + "existing X|3, new X|5..X|8"); + /* the identity scan covers the whole block, both ends */ + jsv_expect_table_collision(jsv_elem(MAX_ARRAY_IDX), jsv_elem(1), 2, + "existing X|32767, new X|1..X|2"); + /* a pre-upgrade element 0 holds the identity too */ + jsv_expect_table_collision(jsv_elem(0), jsv_elem(1), 3, + "existing pre-upgrade X|0, new X|1..X|3"); +} + +/* 7.10: the element-range check -- an existing element inside the range. */ +static void test_jobsubmit_existing_element_in_range_blocks(void) +{ + jsv_expect_table_collision(jsv_elem(2), jsv_elem(1), 3, + "existing X|2, new X|1..X|3"); + jsv_expect_table_collision(jsv_elem(1), jsv_elem(1), 1, + "existing X|1, new X|1"); +} + +/* Unmarked submissions keep the range check, now also after registration. */ +static void test_jobsubmit_existing_ordinary_job_blocks(void) +{ + jsv_expect_table_collision(jsv_id(0, 9500), jsv_id(0, 9500), 1, + "existing ordinary job, same ID"); +} + +/* No false collisions: another block's element, or an ordinary job whose + * sequence lies in X's block, leaves X's identity free -- the submission is + * admitted (it reaches the WAL and fails there with -EIO, not -EEXIST). */ +static void test_jobsubmit_no_collision_admitted(void) +{ + const uint64_t other_elem + = jsv_id(TAG_JOB_ARRAY, ((uint64_t)9 << N_ARRAY_IDX_BITS) | 3); + const uint64_t ordinary_in_block = jsv_id(0, JSV_BLOCK | 3); + const struct { + const char* what; + uint64_t existing; + uint64_t first; + uint64_t size; + } cases[] = { + { "other block's element", other_elem, jsv_elem(5), 2 }, + { "ordinary job in X's block", ordinary_in_block, jsv_elem(5), 1 }, + { "array element, new ordinary job", jsv_elem(3), jsv_id(0, 9510), 1 }, + }; + size_t i; + + for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + jsv_reply_obs_t obs = { 0 }; + char* out = NULL; + + jsv_setup(); + jsv_seed_job(cases[i].existing); + jsv_arrange(cases[i].first, cases[i].size, false); + (void)jsv_submit_capture(&out); + free(out); + jsv_drain(&obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, cases[i].what); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EIO, obs.last_status, cases[i].what); + jsv_teardown(); + } +} + +/* array_identity_exists: one scan over the block's 32768 element keys. */ +static void test_array_identity_exists_scan(void) +{ + const uint64_t other_block = (uint64_t)9 << N_ARRAY_IDX_BITS; + + jsv_setup(); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + array_identity_exists(&g_jsv_c, JSV_ARRAY_ID), + "empty table: identity free"); + jsv_seed_job(jsv_id(0, JSV_BLOCK | 3)); + jsv_seed_job(jsv_id(TAG_JOB_ARRAY, other_block | 1)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + array_identity_exists(&g_jsv_c, JSV_ARRAY_ID), + "an ordinary job or another block's " + "element does not hold X's identity"); + jsv_seed_job(jsv_elem(MAX_ARRAY_IDX)); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + array_identity_exists(&g_jsv_c, JSV_ARRAY_ID), + "element 32767 holds X's identity"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, array_identity_exists(&g_jsv_c, jsv_id(0, other_block)), + "the other block's identity is held too"); + jsv_teardown(); + + jsv_setup(); + jsv_seed_job(jsv_elem(0)); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + array_identity_exists(&g_jsv_c, JSV_ARRAY_ID), + "a pre-upgrade element 0 holds the identity"); + jsv_teardown(); +} + void test_jobsubmit_validate(void) { UnitySetTestFile(__FILE__); @@ -714,4 +910,9 @@ void test_jobsubmit_validate(void) RUN_TEST(test_jobsubmit_retry_survives_release_array); RUN_TEST(test_jobsubmit_inflight_overlap_rule); RUN_TEST(test_jobsubmit_same_array_identity_while_registered); + RUN_TEST(test_jobsubmit_existing_element_blocks_same_identity); + RUN_TEST(test_jobsubmit_existing_element_in_range_blocks); + RUN_TEST(test_jobsubmit_existing_ordinary_job_blocks); + RUN_TEST(test_jobsubmit_no_collision_admitted); + RUN_TEST(test_array_identity_exists_scan); }