feat(job-array-one-based-index): guard scheduler array expansion (task 3.1)

The scheduler expands an array push only when the first ID carries the
array marking and the size fits in its block (size <= 32768 - idx, no
addition); otherwise it logs an ERROR naming the array identity and
expands nothing. Pre-upgrade arrays starting at index 0 still expand.
The INFO log names the array identity (array_jobid=...). This also
closes a crash: a malformed push with a huge size overflowed calloc.
Expansion tests move to tagged block|1 first IDs.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

.kiro/specs/job-array-one-based-index/tasks.md

@@ -50,7 +50,7 @@
   - _Requirements: 7.7, 7.10, 7.11_
 
 - [ ] 3. Scheduler and worker
-- [ ] 3.1 (P) Guard scheduler array expansion and log the array identity
+- [x] 3.1 (P) Guard scheduler array expansion and log the array identity
   - Expand an array push only when the first ID carries the array marking and its size fits within the block (checked without addition); otherwise log an ERROR naming the array identity and size and expand nothing
   - Log the received array by its array identity
   - Move the existing expansion tests (translation, array-expand and wire end-to-end tests) to tagged block-plus-one first IDs

src/scheduler/src/driver.c

@@ -466,6 +466,21 @@ static int translate_one_pending_update(xce_sched_core_t*           core,
         base_jobid = job_array->jobid;
         array_size = job_array->array_size;
 
+        /* SchedGuard (job-array-one-based-index, Req 5.9): expand only a
+         * tagged first ID whose elements all fit in its 32768-index block.
+         * The size check subtracts instead of adding, so a UInt64 size
+         * cannot wrap. Index 0 (pre-upgrade arrays) is accepted. The
+         * controller never sends anything else; defensive only. */
+        if (!is_array_tag(base_jobid)
+            || (array_size > (32768 - array_idx(base_jobid)))) {
+            fmt_jobid(jobidString, sizeof(jobidString) - 1,
+                      array_jobid(base_jobid));
+            ERROR("%s UPDATE_JOB_ARRAY: not a well-formed array, not "
+                  "expanded: array_jobid=%s array_size=%llu",
+                  __func__, jobidString, (unsigned long long)array_size);
+            return 0;
+        }
+
         jobs = (xce_job_t**)calloc((size_t)array_size, sizeof(xce_job_t*));
         if (jobs == NULL) {
             ERROR("%s UPDATE_JOB_ARRAY: out of memory allocating a "
@@ -506,8 +521,9 @@ static int translate_one_pending_update(xce_sched_core_t*           core,
             return 0;
         }
 
-        fmt_jobid(jobidString, sizeof(jobidString) - 1, base_jobid);
+        fmt_jobid(jobidString, sizeof(jobidString) - 1,
-        INFO("job array push received and expanded: base_jobid=%s "
+                  array_jobid(base_jobid));
+        INFO("job array push received and expanded: array_jobid=%s "
              "array_size=%llu new_tasks=%d",
              jobidString, (unsigned long long)array_size, n_new);
 

src/scheduler/testsuite/test_driver_translate.c

@@ -409,6 +409,13 @@ static xce_job_t* make_wire_job(uint64_t jobid, uint32_t state,
     return j;
 }
 
+/* job-array-one-based-index (SchedGuard, Req 5.9): a real array's first
+ * element ID carries TAG_JOB_ARRAY and sits in a 32768-aligned block of the
+ * sequence; element k of block `blk` is ARR_ID(blk, k). */
+#define ARR_ID(blk, k)                                                         \
+    MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY,                                      \
+               (((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | (uint64_t)(k))
+
 /* Compact job-array descriptor, same shape init_job_array() (test_job_
  * array_expand.c) builds -- heap-owned strings, matching what a real
  * XCE_SCHED_OP_UPDATE_JOB_ARRAY request now carries unexpanded (connection-
@@ -720,7 +727,7 @@ static void test_translate_job_rusage_null_rusages_skips(void)
 static void test_translate_job_array_expands_into_n_events(void)
 {
     xce_sched_core_t            core;
-    xce_job_submit_t*           ja  = make_wire_job_array(9000, 3);
+    xce_job_submit_t*           ja  = make_wire_job_array(ARR_ID(1, 1), 3);
     xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
     xce_core_event_t            ev;
     int                         rc;
@@ -734,9 +741,9 @@ static void test_translate_job_array_expands_into_n_events(void)
     TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type);
     TEST_ASSERT_EQUAL_UINT64(42, ev.ts);
     TEST_ASSERT_EQUAL_INT(3, ev.u.jobs_push.njobs);
-    TEST_ASSERT_EQUAL_UINT64(9000, ev.u.jobs_push.jobs[0]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 1), ev.u.jobs_push.jobs[0]->jobid);
-    TEST_ASSERT_EQUAL_UINT64(9001, ev.u.jobs_push.jobs[1]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 2), ev.u.jobs_push.jobs[1]->jobid);
-    TEST_ASSERT_EQUAL_UINT64(9002, ev.u.jobs_push.jobs[2]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(1, 3), ev.u.jobs_push.jobs[2]->jobid);
     TEST_ASSERT_EQUAL_STRING("normal", ev.u.jobs_push.jobs[0]->submit->queue);
     TEST_ASSERT_EQUAL_STRING("normal", ev.u.jobs_push.jobs[2]->submit->queue);
     /* every element gets its OWN independent clone, not a shared pointer --
@@ -763,7 +770,7 @@ static void test_translate_job_array_expands_into_n_events(void)
 static void test_translate_job_array_skips_already_existing_elements(void)
 {
     xce_sched_core_t            core;
-    xce_job_submit_t*           ja  = make_wire_job_array(9100, 3);
+    xce_job_submit_t*           ja  = make_wire_job_array(ARR_ID(2, 1), 3);
     xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
     xce_core_event_t            ev;
     xce_sched_job_t             existing0;
@@ -774,11 +781,11 @@ static void test_translate_job_array_skips_already_existing_elements(void)
     req->u.jobarrayupdate.job_array = ja;
 
     memset(&existing0, 0, sizeof(existing0));
-    existing0.jobid = 9100;
+    existing0.jobid = ARR_ID(2, 1);
     sched_add_job(&core, &existing0);
 
     memset(&existing1, 0, sizeof(existing1));
-    existing1.jobid = 9101;
+    existing1.jobid = ARR_ID(2, 2);
     sched_add_job(&core, &existing1);
 
     rc = translate_one_pending_update(&core, req, 1, &ev);
@@ -786,7 +793,7 @@ static void test_translate_job_array_skips_already_existing_elements(void)
     TEST_ASSERT_EQUAL_INT(1, rc);
     TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type);
     TEST_ASSERT_EQUAL_INT(1, ev.u.jobs_push.njobs);
-    TEST_ASSERT_EQUAL_UINT64(9102, ev.u.jobs_push.jobs[0]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(2, 3), ev.u.jobs_push.jobs[0]->jobid);
 }
 
 /* Every index already exists: a fully-duplicate redelivery -- nothing new
@@ -795,7 +802,7 @@ static void test_translate_job_array_skips_already_existing_elements(void)
 static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
 {
     xce_sched_core_t            core;
-    xce_job_submit_t*           ja  = make_wire_job_array(9200, 2);
+    xce_job_submit_t*           ja  = make_wire_job_array(ARR_ID(3, 1), 2);
     xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
     xce_core_event_t            ev;
     xce_sched_job_t             existing0;
@@ -806,11 +813,11 @@ static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
     req->u.jobarrayupdate.job_array = ja;
 
     memset(&existing0, 0, sizeof(existing0));
-    existing0.jobid = 9200;
+    existing0.jobid = ARR_ID(3, 1);
     sched_add_job(&core, &existing0);
 
     memset(&existing1, 0, sizeof(existing1));
-    existing1.jobid = 9201;
+    existing1.jobid = ARR_ID(3, 2);
     sched_add_job(&core, &existing1);
 
     rc = translate_one_pending_update(&core, req, 1, &ev);
@@ -820,6 +827,81 @@ static void test_translate_job_array_all_duplicates_returns_zero_noop(void)
     TEST_ASSERT_NULL(req->u.jobarrayupdate.expanded_jobs);
 }
 
+/* SchedGuard (Req 5.9): pushes that are not a well-formed array are logged
+ * at ERROR and expand nothing. expect_not_expanded() runs one through
+ * translate and asserts nothing was produced. */
+static void expect_not_expanded(uint64_t first, uint64_t array_size)
+{
+    xce_sched_core_t            core;
+    xce_job_submit_t*           ja  = make_wire_job_array(first, array_size);
+    xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
+    xce_core_event_t            ev;
+    int                         rc;
+
+    fixture_init(&core);
+    req->u.jobarrayupdate.job_array = ja;
+
+    rc = translate_one_pending_update(&core, req, 1, &ev);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(0, req->u.jobarrayupdate.expanded_njobs);
+    TEST_ASSERT_NULL(req->u.jobarrayupdate.expanded_jobs);
+    TEST_ASSERT_NULL(sched_get_job(&core, first));
+}
+
+static void test_translate_job_array_untagged_first_not_expanded(void)
+{
+    expect_not_expanded(9000, 3);
+}
+
+static void test_translate_job_array_overflowing_block_not_expanded(void)
+{
+    /* index 32766 + 3 elements would run past 32767 into the next block */
+    expect_not_expanded(ARR_ID(5, 32766), 3);
+    /* index 1 + 32768 elements: one past the block */
+    expect_not_expanded(ARR_ID(5, 1), 32768);
+}
+
+static void test_translate_job_array_max_u64_size_not_expanded(void)
+{
+    /* idx + size would wrap to 1 with an addition-based check */
+    expect_not_expanded(ARR_ID(6, 2), UINT64_MAX);
+}
+
+/* Boundary: the last two indices of the block fit exactly. */
+static void test_translate_job_array_block_end_expands(void)
+{
+    xce_sched_core_t            core;
+    xce_job_submit_t*           ja  = make_wire_job_array(ARR_ID(7, 32766), 2);
+    xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
+    xce_core_event_t            ev;
+
+    fixture_init(&core);
+    req->u.jobarrayupdate.job_array = ja;
+
+    TEST_ASSERT_EQUAL_INT(1, translate_one_pending_update(&core, req, 1, &ev));
+    TEST_ASSERT_EQUAL_INT(2, ev.u.jobs_push.njobs);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(7, 32766), ev.u.jobs_push.jobs[0]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(7, 32767), ev.u.jobs_push.jobs[1]->jobid);
+}
+
+/* A pre-upgrade array (tagged, numbered from 0) still expands to 0..n-1. */
+static void test_translate_job_array_pre_upgrade_index_zero_expands(void)
+{
+    xce_sched_core_t            core;
+    xce_job_submit_t*           ja  = make_wire_job_array(ARR_ID(8, 0), 3);
+    xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
+    xce_core_event_t            ev;
+
+    fixture_init(&core);
+    req->u.jobarrayupdate.job_array = ja;
+
+    TEST_ASSERT_EQUAL_INT(1, translate_one_pending_update(&core, req, 1, &ev));
+    TEST_ASSERT_EQUAL_INT(3, ev.u.jobs_push.njobs);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(8, 0), ev.u.jobs_push.jobs[0]->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_ID(8, 2), ev.u.jobs_push.jobs[2]->jobid);
+}
+
 /* A never-decoded/malformed descriptor (NULL job_array, the enqueue side
  * should never produce this, guarded defensively here anyway, same
  * discipline every other case in this function follows) skips cleanly. */
@@ -1078,18 +1160,19 @@ static void test_drain_translates_and_pushes_job_array_op(void)
 
     fixture_init(&core);
 
-    ja  = make_wire_job_array(9300, 3);
+    ja                              = make_wire_job_array(ARR_ID(4, 1), 3);
-    req = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
+    req                             = make_req(XCE_SCHED_OP_UPDATE_JOB_ARRAY);
     req->u.jobarrayupdate.job_array = ja;
     enqueue(&core, req);
 
     rc = xce_sched_driver_translate_pending_events(&core, NULL);
 
     TEST_ASSERT_EQUAL_INT(1, rc);
-    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9300));
+    TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 1)));
-    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9301));
+    TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 2)));
-    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 9302));
+    TEST_ASSERT_NOT_NULL(sched_get_job(&core, ARR_ID(4, 3)));
-    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, sched_get_job(&core, 9300)->state);
+    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND,
+                             sched_get_job(&core, ARR_ID(4, 1))->state);
 }
 
 /* ---------------------------------------------------------------------- */
@@ -1174,6 +1257,11 @@ int main(void)
     RUN_TEST(test_translate_job_array_skips_already_existing_elements);
     RUN_TEST(test_translate_job_array_all_duplicates_returns_zero_noop);
     RUN_TEST(test_translate_job_array_null_payload_returns_zero);
+    RUN_TEST(test_translate_job_array_untagged_first_not_expanded);
+    RUN_TEST(test_translate_job_array_overflowing_block_not_expanded);
+    RUN_TEST(test_translate_job_array_max_u64_size_not_expanded);
+    RUN_TEST(test_translate_job_array_block_end_expands);
+    RUN_TEST(test_translate_job_array_pre_upgrade_index_zero_expands);
     RUN_TEST(test_translate_unknown_op_skips);
     RUN_TEST(test_drain_translates_and_pushes_all_ten_op_types);
     RUN_TEST(test_drain_translates_and_pushes_job_array_op);

src/scheduler/testsuite/test_job_array_expand.c

@@ -37,6 +37,12 @@
  * elsewhere is sched_init_config, stubbed below (same as test_handlers.c). */
 #include "../src/handler.c"
 
+/* job-array-one-based-index (Req 5.9): array first IDs are tagged
+ * TAG_JOB_ARRAY, block-aligned, numbered from 1. */
+#define ARR_FIRST(blk)                                                         \
+    MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY,                                      \
+               (((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | 1)
+
 int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
 {
     (void)s;
@@ -179,7 +185,7 @@ static void test_job_array_push_enqueues_raw_descriptor_unexpanded(void)
     xce_job_submit_t* popped;
 
     fixture_init(&sched);
-    ja = init_job_array(5000, 3);
+    ja = init_job_array(ARR_FIRST(1), 3);
 
     memset(&dataset, 0, sizeof(dataset));
     dataset.table = XceData_data_jobArray;
@@ -193,7 +199,7 @@ static void test_job_array_push_enqueues_raw_descriptor_unexpanded(void)
 
     popped = pop_job_array_update(&sched);
     TEST_ASSERT_EQUAL_PTR(ja, popped); /* same descriptor, not a clone */
-    TEST_ASSERT_EQUAL_UINT64(5000, popped->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(1), popped->jobid);
     TEST_ASSERT_EQUAL_UINT64(3, popped->array_size);
     TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
 
@@ -210,7 +216,7 @@ static void test_job_array_null_submit_returns_einval(void)
     fixture_init(&sched);
 
     memset(&ja, 0, sizeof(ja));
-    ja.jobid      = 1;
+    ja.jobid      = ARR_FIRST(2);
     ja.array_size = 3;
     ja.submit     = NULL;
 
@@ -280,7 +286,7 @@ static void test_wire_dispatch_chain_delivers_job_array_descriptor(void)
     submit.resreq   = "type == local";
 
     memset(&job_array, 0, sizeof(job_array));
-    job_array.jobid      = 8000;
+    job_array.jobid      = ARR_FIRST(3);
     job_array.array_size = 4;
     job_array.submit     = &submit;
 
@@ -316,7 +322,7 @@ static void test_wire_dispatch_chain_delivers_job_array_descriptor(void)
      * correctly addressed via base_jobid + array_size. */
     popped = pop_job_array_update(&sched);
     TEST_ASSERT_NOT_NULL(popped);
-    TEST_ASSERT_EQUAL_UINT64(8000, popped->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(3), popped->jobid);
     TEST_ASSERT_EQUAL_UINT64(4, popped->array_size);
     TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
     free_JobSubmit_ptr(&popped);

src/scheduler/testsuite/test_job_array_wire_e2e.c

@@ -80,6 +80,12 @@
  * self-contained; sched_init_config is its only unresolved extern). */
 #include "../src/handler.c"
 
+/* job-array-one-based-index (Req 5.9): array first IDs are tagged
+ * TAG_JOB_ARRAY, block-aligned, numbered from 1. */
+#define ARR_FIRST(blk)                                                         \
+    MAKE_JOBID(1, 2, 1000, TAG_JOB_ARRAY,                                      \
+               (((uint64_t)(blk)) << N_ARRAY_IDX_BITS) | 1)
+
 int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
 {
     (void)s;
@@ -292,14 +298,14 @@ static void test_one_push_enqueues_job_array_descriptor(void)
 
     fixture_init(&sched);
 
-    build_encoded_push(&pkt, 42000, 6, &submit_storage);
+    build_encoded_push(&pkt, ARR_FIRST(1), 6, &submit_storage);
 
     TEST_ASSERT_TRUE(
         deliver_via_real_dispatch_chain(pkt, &sched, &ds) >= 0);
 
     popped = pop_job_array_update(&sched);
     TEST_ASSERT_NOT_NULL(popped);
-    TEST_ASSERT_EQUAL_UINT64(42000, popped->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(1), popped->jobid);
     TEST_ASSERT_EQUAL_UINT64(6, popped->array_size);
     TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
     free_JobSubmit_ptr(&popped);
@@ -335,14 +341,14 @@ static void test_redelivery_enqueues_independent_second_descriptor(void)
     /* Build the wire bytes twice (two independent xce_encode() calls with
      * identical inputs) so each delivery genuinely goes through its own
      * encode and decode -- not one decode fed to the handler twice. */
-    build_encoded_push(&pkt1, 43000, 4, &submit_storage);
+    build_encoded_push(&pkt1, ARR_FIRST(2), 4, &submit_storage);
-    build_encoded_push(&pkt2, 43000, 4, &submit_storage);
+    build_encoded_push(&pkt2, ARR_FIRST(2), 4, &submit_storage);
 
     /* First delivery. */
     TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt1, &sched, &ds1) >= 0);
     popped1 = pop_job_array_update(&sched);
     TEST_ASSERT_NOT_NULL(popped1);
-    TEST_ASSERT_EQUAL_UINT64(43000, popped1->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(2), popped1->jobid);
     TEST_ASSERT_EQUAL_UINT64(4, popped1->array_size);
     free_JobSubmit_ptr(&popped1);
     free_JobSubmit_ptr((xce_job_submit_t**)&ds1->rows);
@@ -356,7 +362,7 @@ static void test_redelivery_enqueues_independent_second_descriptor(void)
     TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt2, &sched, &ds2) >= 0);
     popped2 = pop_job_array_update(&sched);
     TEST_ASSERT_NOT_NULL(popped2);
-    TEST_ASSERT_EQUAL_UINT64(43000, popped2->jobid);
+    TEST_ASSERT_EQUAL_UINT64(ARR_FIRST(2), popped2->jobid);
     TEST_ASSERT_EQUAL_UINT64(4, popped2->array_size);
     free_JobSubmit_ptr(&popped2);