From e6c484a0e765abd976b190b238f284b6615d4106 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 1 Oct 2026 09:44:12 +0000 Subject: [PATCH] feat(job-array-one-based-index): classify every job submission first (task 2.1) handle_jobsubmit classifies (first_jobid, array_size) before the field checks, the before-event hook and any registry access: a tagged array starting at index 0 is answered with the new E_XCE_ARRAY_START0 and an ERROR log asking for the client to be upgraded (7.1, wins per 7.9); size 0, tags 2/3, untagged multi-element submissions and arrays whose last element exceeds 32767 (checked without addition) get -EINVAL. Nothing is stored for a rejected submission. This also closes a crash: a size-0 submission used to reach payload building and overflow calloc. Existing array tests move to tagged block|1 first IDs. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../specs/job-array-one-based-index/tasks.md | 2 +- src/controller/include/cluster.h | 14 + src/controller/job.c | 57 +++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_commit_barrier_reply.c | 11 +- .../testsuite/test_commit_barrier_rollout.c | 39 +- .../testsuite/test_jobsubmit_validate.c | 444 ++++++++++++++++++ src/include/xce/errnox.h | 2 + src/lib/testsuite/test_error.c | 16 + src/lib/utils.c | 5 +- 11 files changed, 574 insertions(+), 19 deletions(-) create mode 100644 src/controller/testsuite/test_jobsubmit_validate.c diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index 264cbed9..dda5242d 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -25,7 +25,7 @@ - _Boundary: IdgenArray_ - [ ] 2. Controller: array admission and collision detection -- [ ] 2.1 Classify every submission before any other check +- [x] 2.1 Classify every submission before any other check - Add the distinct "array numbering starts at 1" error status and its message text to the shared error table - Classify a submission from its first ID and 64-bit size only: start-at-0 array first; then size 0, an unknown marking, an unmarked multi-element submission, and a last element above 32767 (computed without addition so no size can wrap) - Run the classification as the first step after the reply is initialised, before the existing field checks and the before-event hook; answer start-at-0 with the new status plus an ERROR log saying numbering starts at 1 and the client must be upgraded, and the other classes with invalid-argument; store nothing diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index ff3f7067..0085db66 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -107,6 +107,20 @@ extern int handle_join(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_reply(endpoint_t*, struct Request*, cluster_t*); extern int handle_jobdispatch(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_jobsubmit(cluster_t*, endpoint_t*, xce_request_t*); + +/* job-array-one-based-index (Req 7.1-7.5, 7.9): classification of a job + * submission from its first ID and its wire (UInt64) element count only. + * Pure, no locks. INDEX0: an array (tag TAG_JOB_ARRAY) whose first element + * number is 0 (7.1, checked first). INVALID: size 0 (7.4), a tag other than + * 0 or TAG_JOB_ARRAY (7.5), an unmarked multi-element submission (7.3), or + * an array whose last element number would exceed MAX_ARRAY_IDX (7.2). */ +typedef enum { + JOBSUBMIT_OK, + JOBSUBMIT_INDEX0, + JOBSUBMIT_INVALID +} jobsubmit_class_t; +extern jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, + uint64_t size); #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 716c47d7..0c2e6971 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -2431,6 +2431,31 @@ static int handle_jobsubmit_array(cluster_t* c, endpoint_t* ep, return 0; } +/* job-array-one-based-index (Req 7.1-7.5, 7.9; design "ControllerValidate"): + * classify a submission from its first ID and its wire element count only. + * 7.1 is checked first, so an index-0 array is answered as such whatever + * else is wrong with it (7.9); then 7.4, 7.5, 7.3 and 7.2, all INVALID. + * 7.2 is evaluated as size > 32768 - idx (idx >= 1 here), never as + * idx + size - 1 > 32767, so no 64-bit size can wrap past it. */ +jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, uint64_t size) +{ + uint64_t tag = EXTRACT_ID_TAG(first_jobid); + int is_array = is_array_tag(first_jobid); + uint64_t idx = array_idx(first_jobid); + + if (is_array && idx == 0) /* 7.1 */ + return JOBSUBMIT_INDEX0; + if (size == 0) /* 7.4 */ + return JOBSUBMIT_INVALID; + if (tag != 0 && !is_array) /* 7.5 */ + return JOBSUBMIT_INVALID; + if (!is_array && size > 1) /* 7.3 */ + return JOBSUBMIT_INVALID; + if (is_array && size > (MAX_ARRAY_IDX + 1) - idx) /* 7.2 */ + return JOBSUBMIT_INVALID; + return JOBSUBMIT_OK; +} + int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) { uint64_t jobid; @@ -2444,6 +2469,38 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) init_reply_by_req(&reply, request); + /* job-array-one-based-index (Req 7.1-7.6, 7.9): classify before the + * field checks, the BEFORE_EVENT hook and any registry/table access, so + * a rejected submission is answered with its class's status alone and + * nothing is stored. */ + switch (jobsubmit_classify(submit->jobid, submit->array_size)) { + case JOBSUBMIT_INDEX0: { + char idstr[64]; + + fmt_jobid(idstr, sizeof(idstr), submit->jobid); + ERROR("jobsubmit: rejected array submission %s (%lu elements) from " + "%s: array element numbers start at 1, this submission starts " + "at 0; the submitting client must be upgraded", + idstr, (unsigned long)submit->array_size, + (request->from != NULL && request->from->name != NULL) + ? request->from->name + : "unknown"); + reply.status = -E_XCE_ARRAY_START0; + reply.data_which = RequestReply_data_nodata; + + xce_reply(ep, c->myhost, &reply); + return -E_XCE_ARRAY_START0; + } + case JOBSUBMIT_INVALID: + reply.status = -EINVAL; + reply.data_which = RequestReply_data_nodata; + + xce_reply(ep, c->myhost, &reply); + return -EINVAL; + case JOBSUBMIT_OK: + break; + } + if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue) || (submit->submit->start_info == NULL) || NULLSTR(submit->submit->start_info->cmdline)) { diff --git a/src/controller/meson.build b/src/controller/meson.build index 6f1c92f4..bf8f9e41 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -110,6 +110,7 @@ controller_unit_test_src = [ 'testsuite/test_commit_engine.c', 'testsuite/test_commit_barrier_rollout.c', 'testsuite/test_commit_barrier_reply.c', + 'testsuite/test_jobsubmit_validate.c', 'testsuite/test_egress_filter.c', 'testsuite/test_demotion_convergence.c', 'testsuite/test_convergence_verification.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 45d17fc8..908e8f5f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -113,6 +113,7 @@ extern void test_dispatch_fail_retry(void); extern void test_jobdispatch_no_double_notify(void); extern void test_worker_resync(void); extern void test_standby_state_peer_idx(void); +extern void test_jobsubmit_validate(void); void setUp(void) { return; } @@ -303,6 +304,7 @@ int main(int argc, char* argv[]) TEST(test_jobdispatch_no_double_notify); TEST(test_worker_resync); TEST(test_standby_state_peer_idx); + TEST(test_jobsubmit_validate); return UNITY_END(); } diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 144b6d56..dbe2cc25 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -1636,7 +1636,12 @@ static void test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs(v * 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; + /* job-array-one-based-index task 2.1 (Req 7.1-7.3): an array + * submission must carry the array tag and start at element 1 of an + * aligned block; untagged multi-element or index-0 IDs are rejected by + * jobsubmit_classify before reaching the commit path under test. */ + const uint64_t base_jobid = MAKE_JOBID( + 0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)1 << N_ARRAY_IDX_BITS) | 1); const uint64_t array_size = 3; uint64_t lsn_before, watermark; int create_rc, join_rc; @@ -1871,7 +1876,9 @@ static void test_jobsubmit_success_reply_after_apply_carries_jobid(void) static void test_jobsubmit_array_success_appends_one_entry_and_replies_once_with_base_jobid( void) { - const uint64_t base_jobid = 9210; + /* tagged block|1 first ID (job-array-one-based-index task 2.1) */ + const uint64_t base_jobid = MAKE_JOBID( + 0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)2 << N_ARRAY_IDX_BITS) | 1); const uint64_t array_size = 3; uint64_t lsn_before, watermark; cbrp_reply_obs_t final = { 0 }; diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index c3cd0a0f..7af7468a 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -2324,16 +2324,24 @@ static void test_job_array_submit_one_entry_one_continuation_across_sizes(void) * test_job_array_submit_payload_size_independent_of_array_size above * compares an arbitrary "large" sample (10000) against a small one. This * test additionally exercises the REAL structural ceiling design.md relies - * on to satisfy Requirement 7 without a chunking policy: idgen() rejects - * arraySize > MAX_ARRAY_IDX + 1, so MAX_ARRAY_IDX + 1 (32768, config.h) is - * the largest array_size that can ever legally reach this function in - * production. Confirms the encoded payload at that exact ceiling is still - * identical in size to a tiny array's payload AND stays well under a sane - * absolute byte bound -- Requirement 7.1's "no single record exceeds a - * bounded size" is an absolute claim, not only a relative one. */ + * on to satisfy Requirement 7 without a chunking policy: since + * job-array-one-based-index task 2.1, handle_jobsubmit's jobsubmit_classify + * admits only tagged arrays numbered from 1 whose last element is at most + * MAX_ARRAY_IDX, so MAX_ARRAY_IDX (32767, config.h) elements starting at a + * tagged block|1 first ID is the largest array that can ever legally reach + * this function in production. Confirms the encoded payload at that exact + * ceiling is still identical in size to a tiny array's payload AND stays + * well under a sane absolute byte bound -- Requirement 7.1's "no single + * record exceeds a bounded size" is an absolute claim, not only a relative + * one. */ static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void) { - const uint64_t ceiling_array_size = MAX_ARRAY_IDX + 1; /* 32768 */ + const uint64_t ceiling_array_size = MAX_ARRAY_IDX; /* 32767 */ + /* tagged block|1 first IDs (job-array-one-based-index task 2.1) */ + const uint64_t small_first = MAKE_JOBID( + 0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)3 << N_ARRAY_IDX_BITS) | 1); + const uint64_t ceiling_first = MAKE_JOBID( + 0, 0, 0, TAG_JOB_ARRAY, ((uint64_t)4 << N_ARRAY_IDX_BITS) | 1); wal_entry_hdr_t hdr_small, hdr_ceiling; uint64_t out_lsn = 0; int rc; @@ -2344,7 +2352,7 @@ static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void) jap_notify_reset(); off_small = (off_t)g_jap.seg_offset; - rc = persist_submit_job_array_async(&g_jap, 40000, 2, &g_jap_spec, + rc = persist_submit_job_array_async(&g_jap, small_first, 2, &g_jap_spec, &out_lsn, jap_notify_fn, &g_jap_obs); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(0, @@ -2352,17 +2360,18 @@ static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void) jap_notify_reset(); off_ceiling = (off_t)g_jap.seg_offset; - rc = persist_submit_job_array_async(&g_jap, 50000, ceiling_array_size, - &g_jap_spec, &out_lsn, - jap_notify_fn, &g_jap_obs); + + rc = persist_submit_job_array_async(&g_jap, ceiling_first, + ceiling_array_size, &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_ceiling, &hdr_ceiling)); - TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len, - hdr_ceiling.payload_len, + TEST_ASSERT_EQUAL_UINT32_MESSAGE( + hdr_small.payload_len, hdr_ceiling.payload_len, "the payload size at the actual upstream array_size ceiling " - "(MAX_ARRAY_IDX + 1 = 32768) must be identical to a tiny array's " + "(MAX_ARRAY_IDX = 32767) must be identical to a tiny array's " "payload size -- proving Requirement 7's bound holds at the real " "cap, not just at an arbitrary large sample"); TEST_ASSERT_TRUE_MESSAGE(hdr_ceiling.payload_len < 4096, diff --git a/src/controller/testsuite/test_jobsubmit_validate.c b/src/controller/testsuite/test_jobsubmit_validate.c new file mode 100644 index 00000000..b4178050 --- /dev/null +++ b/src/controller/testsuite/test_jobsubmit_validate.c @@ -0,0 +1,444 @@ +/* + * test_jobsubmit_validate.c -- controller array admission + * (job-array-one-based-index task 2.1, Requirements 7.1-7.6, 7.9). + * + * Two layers: + * - jobsubmit_classify(first_jobid, size) over every 7.1-7.5 class, the + * 7.9 precedence (an index-0 array wins over every other break), and + * the no-wrap 7.2 bound (sizes of 2^32+2 and UINT64_MAX with index 2, + * neither truncated nor wrapped). + * - handle_jobsubmit end to end for rejected submissions: the reply + * status, the 7.1 controller ERROR line, and that no job is stored + * (7.6). Classification runs before the field checks, the hook and any + * registry/table/persist access, so a rejected submission needs only a + * reply endpoint, a job table to prove nothing was stored, and an + * initialised hook system (never reached once classification exists). + * + * Reply observation: as in test_commit_barrier_reply.c, the requester + * endpoint is a zeroed endpoint_t with no iomgr, so xce_reply() leaves the + * encoded packet on ep->outq; it is drained with endpoint_deq() and parsed + * with the raw capnp reader. + * + * Log observation: this test binary never calls tlog_init(), so ERROR() + * falls through tlog's early-print path straight to STDOUT_FILENO (the + * established precedent of test_worker_join.c's capture_stdout_during()). + */ + +/* rtti.h must come first -- it defines 'var' and 'typeinfo_t' used by kvmgr.h + */ +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "controller.capnp.h" +#include "xce/cluster_def.h" +#include "xce/errnox.h" +#include "xce/types.h" + +#include "cluster.h" +#include "hook.h" +#include "jobj.h" +#include "kvmgr.h" +#include "netio.h" +#include "request.h" +#include "utils.h" + +/* ── Job identities ──────────────────────────────────────────────────── */ + +/* An array block (sequence aligned to 32768) at a fixed cluster/apiserver/ + * timestamp; jsv_elem(k) is element k of it. */ +#define JSV_CLUSTER 3 +#define JSV_APISERVER 2 +#define JSV_TS 1700000000ULL +#define JSV_BLOCK ((uint64_t)5 << N_ARRAY_IDX_BITS) + +static uint64_t jsv_id(uint32_t tag, uint64_t seq) +{ + return MAKE_JOBID(JSV_CLUSTER, JSV_APISERVER, JSV_TS, tag, seq); +} + +static uint64_t jsv_elem(uint64_t k) +{ + return jsv_id(TAG_JOB_ARRAY, JSV_BLOCK | k); +} + +/* ── jobsubmit_classify ──────────────────────────────────────────────── */ + +typedef struct { + const char* what; + uint64_t first; + uint64_t size; + jobsubmit_class_t expect; +} jsv_case_t; + +static void test_jobsubmit_classify_table(void) +{ + const jsv_case_t cases[] = { + /* 7.1: tagged array starting at element 0 */ + { "7.1 index-0 array of 3", jsv_elem(0), 3, JOBSUBMIT_INDEX0 }, + { "7.1 index-0 array of 1", jsv_elem(0), 1, JOBSUBMIT_INDEX0 }, + { "7.1 index-0 array of 32768", jsv_elem(0), 32768, JOBSUBMIT_INDEX0 }, + /* 7.9: 7.1 wins over every 7.2-7.5 break */ + { "7.9 index-0 + size 0 (7.4)", jsv_elem(0), 0, JOBSUBMIT_INDEX0 }, + { "7.9 index-0 + last > 32767 (7.2)", jsv_elem(0), 40000, + JOBSUBMIT_INDEX0 }, + { "7.9 index-0 + UINT64_MAX size", jsv_elem(0), UINT64_MAX, + JOBSUBMIT_INDEX0 }, + /* 7.4: size 0, any tag */ + { "7.4 ordinary size 0", jsv_id(0, 77), 0, JOBSUBMIT_INVALID }, + { "7.4 array size 0", jsv_elem(1), 0, JOBSUBMIT_INVALID }, + /* 7.5: tags 2 and 3 (index 0 under them is not an array, so not + * 7.1) */ + { "7.5 tag 2", jsv_id(2, JSV_BLOCK | 1), 1, JOBSUBMIT_INVALID }, + { "7.5 tag 3", jsv_id(3, JSV_BLOCK | 1), 1, JOBSUBMIT_INVALID }, + { "7.5 tag 2 index 0", jsv_id(2, JSV_BLOCK), 3, JOBSUBMIT_INVALID }, + { "7.5 tag 3 index 0 size 0", jsv_id(3, JSV_BLOCK), 0, + JOBSUBMIT_INVALID }, + /* 7.3: unmarked multi-element */ + { "7.3 ordinary size 2", jsv_id(0, 77), 2, JOBSUBMIT_INVALID }, + { "7.3 ordinary size 3", jsv_id(0, 9140), 3, JOBSUBMIT_INVALID }, + { "7.3 ordinary UINT64_MAX", jsv_id(0, 77), UINT64_MAX, + JOBSUBMIT_INVALID }, + /* 7.2: last element above 32767, computed without addition */ + { "7.2 idx 1 size 32768", jsv_elem(1), 32768, JOBSUBMIT_INVALID }, + { "7.2 idx 2 size 32767", jsv_elem(2), 32767, JOBSUBMIT_INVALID }, + { "7.2 idx 32767 size 2", jsv_elem(32767), 2, JOBSUBMIT_INVALID }, + { "7.2 idx 2 size 2^32+2", jsv_elem(2), (1ULL << 32) + 2, + JOBSUBMIT_INVALID }, + { "7.2 idx 1 size 2^32+1", jsv_elem(1), (1ULL << 32) + 1, + JOBSUBMIT_INVALID }, + { "7.2 idx 2 size UINT64_MAX", jsv_elem(2), UINT64_MAX, + JOBSUBMIT_INVALID }, + /* accepted */ + { "ok ordinary single", jsv_id(0, 77), 1, JOBSUBMIT_OK }, + { "ok ordinary single seq 0", jsv_id(0, 0), 1, JOBSUBMIT_OK }, + { "ok one-element array", jsv_elem(1), 1, JOBSUBMIT_OK }, + { "ok array 1..3", jsv_elem(1), 3, JOBSUBMIT_OK }, + { "ok array 1..32767", jsv_elem(1), 32767, JOBSUBMIT_OK }, + { "ok array 2..32767", jsv_elem(2), 32766, JOBSUBMIT_OK }, + { "ok array 32767..32767", jsv_elem(32767), 1, JOBSUBMIT_OK }, + }; + size_t i; + char msg[160]; + + for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + snprintf(msg, sizeof(msg), "case %zu: %s", i, cases[i].what); + TEST_ASSERT_EQUAL_INT_MESSAGE( + cases[i].expect, jobsubmit_classify(cases[i].first, cases[i].size), + msg); + } +} + +/* ── handle_jobsubmit fixture ────────────────────────────────────────── */ + +static cluster_t g_jsv_c; +static endpoint_t g_jsv_ep; +static ep_private_t g_jsv_priv; +static job_startinfo_t g_jsv_startinfo; +static job_submit_data_t g_jsv_data; +static xce_job_submit_t g_jsv_msg; +static xce_request_t g_jsv_req; +static xce_peerinfo_t g_jsv_from; + +typedef struct { + int n_replies; + int32_t last_status; + int32_t last_data_which; + uint64_t last_seq; +} jsv_reply_obs_t; + +static void jsv_setup(void) +{ + kvmgr_params_t int_params + = { .key_type = KVMGR_KEY_INT, .nslots = 64, .delta = 16 }; + + memset(&g_jsv_c, 0, sizeof(g_jsv_c)); + memset(&g_jsv_ep, 0, sizeof(g_jsv_ep)); + memset(&g_jsv_priv, 0, sizeof(g_jsv_priv)); + + g_jsv_ep.fd = -1; + g_jsv_ep.data = &g_jsv_priv; + + g_jsv_c.myhost = &g_jsv_c.myhost_info; + g_jsv_c.myhost_info.name = (char*)"ctrl-test"; + g_jsv_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT( + 0, xce_hook_init(&g_jsv_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + g_jsv_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_jsv_c.jobmgr); +} + +static void jsv_drain(jsv_reply_obs_t* obs); + +static void jsv_teardown(void) +{ + jsv_reply_obs_t drain = { 0 }; + + jsv_drain(&drain); /* free any leftover packets */ + inflight_submits_reset_for_test(); + if (g_jsv_c.jobmgr) + dereference(g_jsv_c.jobmgr); + xce_hook_fini(&g_jsv_c.hooksys); +} + +/* A submission with every field the existing field checks require, unless + * empty_queue is set (then the queue is "" and the field check would reject + * it with -EINVAL). */ +static void jsv_arrange(uint64_t first, uint64_t size, bool empty_queue) +{ + memset(&g_jsv_startinfo, 0, sizeof(g_jsv_startinfo)); + g_jsv_startinfo.cmdline = (char*)"/bin/true"; + + memset(&g_jsv_data, 0, sizeof(g_jsv_data)); + g_jsv_data.queue = (char*)(empty_queue ? "" : "normal"); + g_jsv_data.user = (char*)"tester"; + g_jsv_data.start_info = &g_jsv_startinfo; + + memset(&g_jsv_msg, 0, sizeof(g_jsv_msg)); + g_jsv_msg.jobid = first; + g_jsv_msg.array_size = size; + g_jsv_msg.submit = &g_jsv_data; + + g_jsv_from.name = (char*)"requester"; + g_jsv_from.key = (char*)""; + + memset(&g_jsv_req, 0, sizeof(g_jsv_req)); + g_jsv_req.opcode = OPCode_jobsub; + g_jsv_req.seq = 21; + g_jsv_req.from = &g_jsv_from; + g_jsv_req.body_which = Request_body_jobsubmsg; + g_jsv_req.body.jobsubmsg = &g_jsv_msg; +} + +static void jsv_drain(jsv_reply_obs_t* obs) +{ + data_item_t* it; + + while ((it = endpoint_deq(&g_jsv_ep)) != NULL) { + io_packet_t* pkt = ITPTR_TO_PACKET(it); + struct capn c; + struct Request rq; + Request_ptr rp; + + capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0); + rp.p = capn_getp(capn_root(&c), 0, 1); + read_Request(&rq, rp); + if (rq.body_which == Request_body_replymsg) { + struct RequestReply rr; + read_RequestReply(&rr, rq.body.replymsg); + obs->n_replies += 1; + obs->last_status = rr.status; + obs->last_data_which = rr.data_which; + obs->last_seq = rr.seq; + } + capn_free(&c); + free_packet(pkt); + free(pkt); + } +} + +static bool jsv_job_exists(uint64_t jobid) +{ + job_t* j = NULL; + int rc; + + lock(g_jsv_c.jobmgr, RTTI_LOCK_READ); + rc = key_find(g_jsv_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_jsv_c.jobmgr); + return rc > 0 && j != NULL; +} + +/* Run handle_jobsubmit with STDOUT_FILENO redirected to a temp file; returns + * the handler's rc and the captured output (caller frees). */ +static int jsv_submit_capture(char** out) +{ + char tmp[] = "/tmp/xce_jsv_logcap_XXXXXX"; + int fd = mkstemp(tmp); + int saved, rc; + off_t len; + char* buf; + + TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "fixture: mkstemp must succeed"); + fflush(stdout); + saved = dup(STDOUT_FILENO); + TEST_ASSERT_TRUE_MESSAGE(saved >= 0, "fixture: dup(stdout) must succeed"); + TEST_ASSERT_TRUE_MESSAGE(dup2(fd, STDOUT_FILENO) >= 0, + "fixture: dup2 redirect must succeed"); + + rc = handle_jobsubmit(&g_jsv_c, &g_jsv_ep, &g_jsv_req); + + fflush(stdout); + dup2(saved, STDOUT_FILENO); + close(saved); + + len = lseek(fd, 0, SEEK_CUR); + TEST_ASSERT_TRUE(len >= 0); + buf = (char*)calloc(1, (size_t)len + 1); + TEST_ASSERT_NOT_NULL(buf); + if (len > 0) { + TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0); + TEST_ASSERT_TRUE(read(fd, buf, (size_t)len) == len); + } + close(fd); + unlink(tmp); + + *out = buf; + return rc; +} + +/* Assert one reply with `status`, the handler returned it, and none of the + * first min(size, 8) element identities is in the job table (7.6). */ +static void jsv_expect_rejected(int rc, int32_t status, uint64_t first, + uint64_t size, const char* what) +{ + jsv_reply_obs_t obs = { 0 }; + char msg[200]; + uint64_t i, n = size < 8 ? size : 8; + + 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: reply status", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(status, obs.last_status, msg); + snprintf(msg, sizeof(msg), "%s: reply carries no data", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_nodata, obs.last_data_which, + 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(status, rc, msg); + + if (n == 0) + n = 1; + for (i = 0; i < n; i++) { + snprintf(msg, sizeof(msg), "%s: no job stored at element %llu", what, + (unsigned long long)i); + TEST_ASSERT_FALSE_MESSAGE(jsv_job_exists(first + i), msg); + } +} + +/* 7.1 + 7.9: an index-0 array with an empty queue is answered with the + * distinct status (not the field check's -EINVAL) and the ERROR line. */ +static void test_jobsubmit_index0_empty_queue_start0_and_log(void) +{ + const uint64_t first = jsv_elem(0); + char* out = NULL; + char idstr[64]; + int rc; + + jsv_setup(); + jsv_arrange(first, 3, true); + + rc = jsv_submit_capture(&out); + jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 3, + "index-0 array, empty queue"); + + fmt_jobid(idstr, sizeof(idstr), first); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "ERROR"), + "7.1: an ERROR line must be logged"); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, idstr), + "7.1: the log line must name the job"); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "start at 1"), + "7.1: the log line must say array element " + "numbers start at 1"); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "must be upgraded"), + "7.1: the log line must say the submitting " + "client must be upgraded"); + free(out); + jsv_teardown(); +} + +/* 7.1 with every field valid: the same status and log, nothing stored. */ +static void test_jobsubmit_index0_valid_fields_start0(void) +{ + const uint64_t first = jsv_elem(0); + char* out = NULL; + int rc; + + jsv_setup(); + jsv_arrange(first, 3, false); + + rc = jsv_submit_capture(&out); + jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 3, + "index-0 array, valid fields"); + TEST_ASSERT_NOT_NULL(strstr(out, "must be upgraded")); + free(out); + jsv_teardown(); +} + +/* 7.9: index 0 with size 0 -- 7.1 alone answers. */ +static void test_jobsubmit_index0_size0_start0(void) +{ + const uint64_t first = jsv_elem(0); + char* out = NULL; + int rc; + + jsv_setup(); + jsv_arrange(first, 0, false); + + rc = jsv_submit_capture(&out); + jsv_expect_rejected(rc, -E_XCE_ARRAY_START0, first, 0, + "index-0 array, size 0"); + free(out); + jsv_teardown(); +} + +/* 7.2-7.5: answered -EINVAL before any field check, hook or registry; no + * 7.1 log line; nothing stored; the controller keeps serving (the next case + * runs against the same handler). */ +static void test_jobsubmit_invalid_classes_einval(void) +{ + const struct { + const char* what; + uint64_t first; + uint64_t size; + bool empty_queue; + } cases[] = { + { "7.4 size 0", jsv_id(0, 9300), 0, false }, + { "7.4 size 0, empty queue", jsv_id(0, 9301), 0, true }, + { "7.5 tag 2", jsv_id(2, JSV_BLOCK | 1), 1, false }, + { "7.5 tag 3", jsv_id(3, JSV_BLOCK | 1), 1, false }, + { "7.3 unmarked size 3", jsv_id(0, 9310), 3, false }, + { "7.2 idx 1 size 32768", jsv_elem(1), 32768, false }, + { "7.2 idx 2 size 2^32+2", jsv_elem(2), (1ULL << 32) + 2, false }, + { "7.2 idx 2 size UINT64_MAX", jsv_elem(2), UINT64_MAX, false }, + }; + size_t i; + + jsv_setup(); + for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + char* out = NULL; + int rc; + + jsv_arrange(cases[i].first, cases[i].size, cases[i].empty_queue); + rc = jsv_submit_capture(&out); + jsv_expect_rejected(rc, -EINVAL, cases[i].first, cases[i].size, + cases[i].what); + TEST_ASSERT_NULL_MESSAGE(strstr(out, "must be upgraded"), + "7.2-7.5 rejections must not log the 7.1 " + "line"); + free(out); + } + jsv_teardown(); +} + +void test_jobsubmit_validate(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_jobsubmit_classify_table); + RUN_TEST(test_jobsubmit_index0_empty_queue_start0_and_log); + RUN_TEST(test_jobsubmit_index0_valid_fields_start0); + RUN_TEST(test_jobsubmit_index0_size0_start0); + RUN_TEST(test_jobsubmit_invalid_classes_einval); +} diff --git a/src/include/xce/errnox.h b/src/include/xce/errnox.h index 2442d95a..b6638701 100644 --- a/src/include/xce/errnox.h +++ b/src/include/xce/errnox.h @@ -23,6 +23,8 @@ enum { E_XCE_REDIRECT, /* Standby is redirecting client to primary */ E_XCE_NOT_READY, /* Controller in CTRL_INIT or CTRL_ELECTING state */ E_XCE_STANDBY, /* Controller in CTRL_STANDBY; op is primary-only */ + E_XCE_ARRAY_START0, /* Array submission numbered from 0; numbering starts + * at 1, the submitting client must be upgraded */ E_XCE_MAX }; diff --git a/src/lib/testsuite/test_error.c b/src/lib/testsuite/test_error.c index fd048b3b..7d1af241 100644 --- a/src/lib/testsuite/test_error.c +++ b/src/lib/testsuite/test_error.c @@ -1,5 +1,6 @@ #include #include +#include #include #include "utils.h" @@ -76,6 +77,20 @@ static void test_strerror_out_of_range(void) TEST_ASSERT_TRUE(strcmp(buf, "unknown error") == 0); } +/* job-array-one-based-index (Req 7.1): the distinct array-numbering status + * sits right after E_XCE_STANDBY and its text says numbering starts at 1 + * and the client must be upgraded. */ +static void test_strerror_array_start0(void) +{ + char buf[128]; + + TEST_ASSERT_EQUAL_INT(E_XCE_STANDBY + 1, E_XCE_ARRAY_START0); + TEST_ASSERT_EQUAL_INT(0, + xce_strerror_r(E_XCE_ARRAY_START0, buf, sizeof(buf))); + TEST_ASSERT_NOT_NULL(strstr(buf, "start at 1")); + TEST_ASSERT_NOT_NULL(strstr(buf, "upgrade")); +} + void test_error(void) { printf("\n>> error test\n\n"); @@ -83,6 +98,7 @@ void test_error(void) RUN_TEST(test_error_setget); RUN_TEST(test_strerror_all_xce_codes); RUN_TEST(test_strerror_out_of_range); + RUN_TEST(test_strerror_array_start0); #if 0 RUN_TEST(test_strerror_r); #endif diff --git a/src/lib/utils.c b/src/lib/utils.c index 7f5c2d4e..f14748f4 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -438,7 +438,10 @@ static char* xce_err_str[] = { "invalid job", /* E_XCE_BAD_JOB */ "no known primary", /* E_XCE_NOMASTER */ "redirect to primary", /* E_XCE_REDIRECT */ "controller not ready", /* E_XCE_NOT_READY */ - "controller is standby" };/* E_XCE_STANDBY */ + "controller is standby", /* E_XCE_STANDBY */ + "array element numbers start at 1; " + "upgrade the submitting client" }; + /* E_XCE_ARRAY_START0 */ _Static_assert(sizeof(xce_err_str) / sizeof(xce_err_str[0]) == (size_t)(E_XCE_MAX - XCE_ERRNO_BASE),