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) <noreply@anthropic.com>

.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

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

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
+ * a duplicate. Populated in handle_jobsubmit() before the job-table
- * passes (before any WAL work).
+ * 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
+ * inflight_range_try_add's doc comment) — handle_jobsubmit first calls
- * calls inflight_range_try_add() after this check passes, which atomically
+ * inflight_range_try_add(), which atomically closes the not-yet-applied
- * closes the remaining not-yet-applied window this function cannot see. */
+ * 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
+    /* job-array-one-based-index (Req 7.10, 7.11): the job-table checks run
-     * window jobid_range_exists' c->jobmgr check alone cannot see (see its
+     * AFTER registration, so a same-identity submission applied and
-     * doc comment). A false return means another submission (a genuine
+     * answered between the two steps cannot be missed by both (an applied
-     * concurrent collision, or -- the actual motivating case -- this exact
+     * submission is in c->jobmgr before it leaves the registry). An array
-     * request's own retried resend racing its still-in-flight original
+     * is checked by one scan of its whole block (element range and array
-     * attempt) already holds this range; reject exactly as the
+     * identity); an ordinary job keeps the element-range check. */
-     * jobid_range_exists branch above does. From here on every answer
+    if (is_array_tag(jobid)
-     * goes through inflight_reply_and_remove(), which sends the reply and
+            ? array_identity_exists(c, array_jobid(jobid))
-     * releases this registration as one step (job-array-one-based-index,
+            : jobid_range_exists(c, jobid, submit->array_size)) {
-     * Req 7.7, 7.10, 7.11) -- the early-outs below directly, the commit
+        reply.status     = -EEXIST;
-     * 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;
 
-        xce_reply(ep, c->myhost, &reply);
+        inflight_reply_and_remove(ep, c->myhost, &reply, jobid,
-        return -errno;
+                                  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):

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
+ * released together with the reply via inflight_reply_and_remove) -- not
- * still-unresolved window the test above proves. A live chaotic-soak run
+ * just DURING the still-unresolved window the test above proves. A live
- * (2026-07-08) showed a genuine WAL-level jobid duplicate (two committed
+ * 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

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);
 }