lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(job-array-one-based-index): classify every job submission first (task 2.1)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-01 09:44:12 +00:00
parent ef728435d4
commit e6c484a0e7
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
11 changed files with 574 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

444
src/controller/testsuite/test_jobsubmit_validate.c Normal file
View file

@ -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 <rtti.h>
#include <errno.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <capnp_c.h>
#include <unity/unity.h>
#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);
}