diff --git a/src/controller/job.c b/src/controller/job.c index 2c4291a2..3d6fc543 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -8,6 +8,7 @@ #include #include "cluster.h" +#include "config_repl.h" /* XCE_LOST_AND_FOUND_NAME -- bug-1960 */ #include "controller.capnp.h" #include "jobj.h" #include "persist.h" @@ -2608,6 +2609,45 @@ jobsubmit_class_t jobsubmit_classify(uint64_t first_jobid, uint64_t size) return JOBSUBMIT_OK; } +/* bug-1960: queue admission for a NEW client submission. Returns 0 when + * `queue` takes new jobs; -E_XCE_BAD_QUEUE when the controller has no such + * queue or it is the reserved recovery queue (jobs reach lost_and_found only + * through WAL replay, never by submission; the apiserver rejects it too); + * -E_XCE_QUEUE_CLOSED when it is closed (LSF semantics: a closed queue takes + * no new submissions, its pending jobs still dispatch -- bug-1938). + * + * Reads in-memory queue state only, under the qmgr and queue read locks + * (the queue_ctrl()/get_queue_exec() lookup pattern), no blocking I/O. It + * belongs to handle_jobsubmit() alone: apply/replay of an already-committed + * WAL_JOB_SUBMIT fact (restore_job) must not consult it, so a job accepted + * before its queue was closed still applies on a standby or after restart. */ +static int jobsubmit_check_queue(cluster_t* c, const char* queue) +{ + queue_t* q = NULL; + uint32_t status; + + if (strcmp(queue, XCE_LOST_AND_FOUND_NAME) == 0) + return -E_XCE_BAD_QUEUE; + + lock(c->qmgr, RTTI_LOCK_READ); + key_find(c->qmgr, (void*)queue, (void**)&q); + if (q != NULL) { + reference(q); + } + unlock(c->qmgr); + + if (q == NULL) + return -E_XCE_BAD_QUEUE; + + lock(q, RTTI_LOCK_READ); + status = q->queue.status; + unlock(q); + + dereference(q); + + return (status & QUEUE_STAT_CLOSED) ? -E_XCE_QUEUE_CLOSED : 0; +} + int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) { uint64_t jobid; @@ -2664,6 +2704,29 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) return -errno; } + /* bug-1960: reject a submission to a closed, unknown or reserved queue + * before the in-flight registry claims its jobid range and before any + * WAL fact is appended -- nothing is stored, the client gets the + * queue's status alone. Covers the single-job and the array path + * (handle_jobsubmit_array diverges below). */ + rc = jobsubmit_check_queue(c, submit->submit->queue); + if (rc < 0) { + char idstr[64]; + + fmt_jobid(idstr, sizeof(idstr), submit->jobid); + ERROR("jobsubmit: rejected submission %s (%lu elements) to queue " + "'%s': %s", + idstr, (unsigned long)submit->array_size, submit->submit->queue, + (rc == -E_XCE_QUEUE_CLOSED) + ? "queue is closed" + : "no such queue or not a submission queue"); + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + + xce_reply(ep, c->myhost, &reply); + return rc; + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_NEW, c, submit); diff --git a/src/controller/testsuite/test_apply_committed_entry.c b/src/controller/testsuite/test_apply_committed_entry.c index 6208bed0..57c40fd3 100644 --- a/src/controller/testsuite/test_apply_committed_entry.c +++ b/src/controller/testsuite/test_apply_committed_entry.c @@ -327,6 +327,50 @@ static void test_ace_job_submit_constructs_and_notifies(void) printf("test_ace_job_submit_constructs_and_notifies: PASS\n"); } +/* bug-1960: the closed-queue check belongs to new client submissions only + * (handle_jobsubmit). A committed WAL_JOB_SUBMIT fact of a job accepted + * before its queue was closed must still apply on replay / standby catch-up, + * and a closed queue's pending jobs still dispatch (bug-1938). */ +static void test_ace_job_submit_closed_queue_still_applies(void) +{ + tace_setup(); + + xce_queue_t qdef; + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)"qclosed"; + qdef.status = QUEUE_STAT_CLOSED; + TEST_ASSERT_NOT_NULL(makeQueue(g_tace_c.qmgr, &qdef)); + + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"qclosed"; + submit.user = (char*)"tester"; + + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 503; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, 12, payload, + (uint32_t)plen), + "a committed submit fact for a now-closed queue must still apply"); + + job_t* j = tace_find_job(503); + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_EQUAL_STRING("qclosed", j->job.submit->queue); + + free(payload); + tace_teardown(); + printf("test_ace_job_submit_closed_queue_still_applies: PASS\n"); +} + static void test_ace_job_submit_duplicate_is_noop_no_extra_notify(void) { tace_setup(); @@ -1240,6 +1284,7 @@ void test_apply_committed_entry(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_ace_job_submit_constructs_and_notifies); + RUN_TEST(test_ace_job_submit_closed_queue_still_applies); RUN_TEST(test_ace_job_submit_duplicate_is_noop_no_extra_notify); RUN_TEST(test_ace_job_state_transition_applies_forward_and_notifies); RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify); diff --git a/src/controller/testsuite/test_apply_pump_integration.c b/src/controller/testsuite/test_apply_pump_integration.c index 64fbd0ca..fcd5ba78 100644 --- a/src/controller/testsuite/test_apply_pump_integration.c +++ b/src/controller/testsuite/test_apply_pump_integration.c @@ -127,6 +127,7 @@ #include "jobj.h" #include "netio.h" #include "persist.h" +#include "qobj.h" #include "repl.h" #include "request.h" #include "vtaskmgr.h" @@ -374,6 +375,15 @@ static void tpi_setup(int pool_n_vcpus) TEST_ASSERT_NOT_NULL(g_tpi_c.resmgr); zarray_init(&g_tpi_c.sharedinst, 16, NULL); + /* bug-1960: handle_jobsubmit admits only a known, open queue; the + * jobsub fixture submits to "normal". */ + { + xce_queue_t qdef; + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char *)"normal"; + TEST_ASSERT_NOT_NULL(makeQueue(g_tpi_c.qmgr, &qdef)); + } + persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; cfg.sync_timeout_ms = 2000; diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 3659f721..4876d463 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -413,6 +413,15 @@ static void cbrp_setup(uint32_t sync_timeout_ms) TEST_ASSERT_NOT_NULL(g_hc.resmgr); zarray_init(&g_hc.sharedinst, 64, NULL); + /* bug-1960: handle_jobsubmit admits only a known, open queue; the + * jobsub fixtures below submit to "normal". */ + { + xce_queue_t qdef; + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char *)"normal"; + TEST_ASSERT_NOT_NULL(makeQueue(g_hc.qmgr, &qdef)); + } + for (i = 0; i < CBRP_N_RES; i++) { xce_resource_t rdef; memset(&rdef, 0, sizeof(rdef)); diff --git a/src/controller/testsuite/test_jobsubmit_validate.c b/src/controller/testsuite/test_jobsubmit_validate.c index 3c6f084d..033eba12 100644 --- a/src/controller/testsuite/test_jobsubmit_validate.c +++ b/src/controller/testsuite/test_jobsubmit_validate.c @@ -66,8 +66,10 @@ #include "jobj.h" #include "kvmgr.h" #include "netio.h" +#include "qobj.h" #include "request.h" #include "utils.h" +#include "xstring.h" /* ── Job identities ──────────────────────────────────────────────────── */ @@ -173,10 +175,30 @@ typedef struct { uint64_t last_seq; } jsv_reply_obs_t; +/* bug-1960: the fixture's queue table -- "normal" open, JSV_CLOSED_QUEUE + * closed, and the reserved recovery queue that create_lost_and_found() + * builds at boot. */ +#define JSV_CLOSED_QUEUE "shut" + +static void jsv_add_queue(const char* name, uint32_t status) +{ + xce_queue_t qdef; + queue_t* q; + + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)name; + qdef.status = status; + q = makeQueue(g_jsv_c.qmgr, &qdef); + TEST_ASSERT_NOT_NULL_MESSAGE(q, "fixture: creating a queue"); + TEST_ASSERT_EQUAL_UINT32(status, q->queue.status); +} + static void jsv_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 64, .delta = 16 }; + kvmgr_params_t str_params + = { .key_type = KVMGR_KEY_STRING, .nslots = 16, .delta = 16 }; memset(&g_jsv_c, 0, sizeof(g_jsv_c)); memset(&g_jsv_ep, 0, sizeof(g_jsv_ep)); @@ -194,6 +216,14 @@ static void jsv_setup(void) g_jsv_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_jsv_c.jobmgr); + if (makeString((char*)"jsv-probe") == NULL) + xstring_table_init(64); + g_jsv_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_jsv_c.qmgr); + jsv_add_queue("normal", 0); + jsv_add_queue(JSV_CLOSED_QUEUE, QUEUE_STAT_CLOSED); + jsv_add_queue("lost_and_found", 0); + /* No WAL: an admitted submission's local append fails with -EIO (see * the file header). */ g_jsv_c.persist.wal_fd = -1; @@ -213,6 +243,8 @@ static void jsv_teardown(void) inflight_submits_reset_for_test(); if (g_jsv_c.jobmgr) dereference(g_jsv_c.jobmgr); + if (g_jsv_c.qmgr) + dereference(g_jsv_c.qmgr); xce_hook_fini(&g_jsv_c.hooksys); } @@ -863,6 +895,112 @@ static void test_jobsubmit_no_collision_admitted(void) } } +/* ── bug-1960: queue admission ───────────────────────────────────────── */ + +static int g_jsv_queue_hook_calls; + +static void jsv_queue_reply_hook(void) { g_jsv_queue_hook_calls += 1; } + +/* Submit (first, size) to `queue` and expect the rejection `status`: one + * reply, nothing stored, an ERROR line naming the queue, and -- the + * "before any job id is reserved or any WAL fact is written" half -- no + * answer through inflight_reply_and_remove(), i.e. the in-flight registry + * never held the range and the WAL append (whose failure in this fixture + * is answered -EIO through that same step) was never reached. */ +static void jsv_expect_queue_rejected(const char* queue, uint64_t first, + uint64_t size, int32_t status, + const char* what) +{ + char* out = NULL; + char msg[200]; + int rc; + + jsv_setup(); + g_jsv_queue_hook_calls = 0; + jobsubmit_test_reply_hook = jsv_queue_reply_hook; + jsv_arrange(first, size, false); + g_jsv_data.queue = (char*)queue; + + rc = jsv_submit_capture(&out); + jsv_expect_rejected(rc, status, first, size, what); + + snprintf(msg, sizeof(msg), + "%s: rejected before the in-flight registry and the WAL", what); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jsv_queue_hook_calls, msg); + snprintf(msg, sizeof(msg), "%s: an ERROR line is logged", what); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, "ERROR"), msg); + snprintf(msg, sizeof(msg), "%s: the log line names the queue", what); + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(out, queue), msg); + free(out); + jsv_teardown(); +} + +/* A closed queue takes no new submissions: single job, job array (the + * handle_jobsubmit_array path) and a one-element array. */ +static void test_jobsubmit_closed_queue_rejected(void) +{ + jsv_expect_queue_rejected(JSV_CLOSED_QUEUE, jsv_id(0, 9600), 1, + -E_XCE_QUEUE_CLOSED, "closed queue, single job"); + jsv_expect_queue_rejected(JSV_CLOSED_QUEUE, jsv_elem(1), 3, + -E_XCE_QUEUE_CLOSED, "closed queue, array of 3"); + jsv_expect_queue_rejected(JSV_CLOSED_QUEUE, jsv_elem(1), 1, + -E_XCE_QUEUE_CLOSED, + "closed queue, one-element array"); +} + +/* A queue the controller does not know is rejected the same way. */ +static void test_jobsubmit_unknown_queue_rejected(void) +{ + jsv_expect_queue_rejected("no-such-queue", jsv_id(0, 9610), 1, + -E_XCE_BAD_QUEUE, "unknown queue, single job"); + jsv_expect_queue_rejected("no-such-queue", jsv_elem(1), 3, + -E_XCE_BAD_QUEUE, "unknown queue, array of 3"); +} + +/* The reserved recovery queue exists but is not a submission target (the + * apiserver rejects it too; jobs only reach it through WAL replay). */ +static void test_jobsubmit_lost_and_found_rejected(void) +{ + jsv_expect_queue_rejected("lost_and_found", jsv_id(0, 9620), 1, + -E_XCE_BAD_QUEUE, "lost_and_found, single job"); + jsv_expect_queue_rejected("lost_and_found", jsv_elem(1), 3, + -E_XCE_BAD_QUEUE, "lost_and_found, array of 3"); +} + +/* An open queue is still admitted: the submission reaches the in-flight + * registry and the WAL (answered -EIO there by this fixture), for a + * single job and an array alike. */ +static void test_jobsubmit_open_queue_admitted(void) +{ + const struct { + const char* what; + uint64_t first; + uint64_t size; + } cases[] = { + { "open queue, single job", jsv_id(0, 9630), 1 }, + { "open queue, array of 3", jsv_elem(1), 3 }, + }; + size_t i; + + for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + jsv_reply_obs_t obs = { 0 }; + char* out = NULL; + + jsv_setup(); + g_jsv_queue_hook_calls = 0; + jobsubmit_test_reply_hook = jsv_queue_reply_hook; + 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); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jsv_queue_hook_calls, + 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) { @@ -915,4 +1053,8 @@ void test_jobsubmit_validate(void) RUN_TEST(test_jobsubmit_existing_ordinary_job_blocks); RUN_TEST(test_jobsubmit_no_collision_admitted); RUN_TEST(test_array_identity_exists_scan); + RUN_TEST(test_jobsubmit_closed_queue_rejected); + RUN_TEST(test_jobsubmit_unknown_queue_rejected); + RUN_TEST(test_jobsubmit_lost_and_found_rejected); + RUN_TEST(test_jobsubmit_open_queue_admitted); } diff --git a/src/include/xce/errnox.h b/src/include/xce/errnox.h index b6638701..392445f0 100644 --- a/src/include/xce/errnox.h +++ b/src/include/xce/errnox.h @@ -25,6 +25,7 @@ enum { 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_QUEUE_CLOSED, /* Submission to a closed queue (bug-1960) */ E_XCE_MAX }; diff --git a/src/lib/utils.c b/src/lib/utils.c index f14748f4..446ef3d7 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -440,8 +440,9 @@ static char* xce_err_str[] = { "invalid job", /* E_XCE_BAD_JOB */ "controller not ready", /* E_XCE_NOT_READY */ "controller is standby", /* E_XCE_STANDBY */ "array element numbers start at 1; " - "upgrade the submitting client" }; + "upgrade the submitting client", /* E_XCE_ARRAY_START0 */ + "queue is closed" }; /* E_XCE_QUEUE_CLOSED */ _Static_assert(sizeof(xce_err_str) / sizeof(xce_err_str[0]) == (size_t)(E_XCE_MAX - XCE_ERRNO_BASE),