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

feat(scheduler-dispatch-commit): controller resolves retryable dispatch failures with bounded escalation (task 3.3)

Browse source 
The four retryable-classified failure sites in dispatch_a_job_() (worker
lookup, worker null, worker not ready, RPC failure) now resolve instead
of just returning an error code: lock(j, RTTI_LOCK_WRITE) -> genuinely
classify via xce_classify_dispatch_fail() -> increment dispatch_fail_count
-> within XCE_DISPATCH_FAIL_RETRY_BOUND (3) resolves retryable (nothing
further under the lock), past it resolves terminal via mark_job_fail()
exactly like task 3.2's sites -> unlock(j) -> dereference(j) (closing
bug-1627's 4-site reference leak) -> outside the lock, notify_dispatch()
fires only on the retryable outcome. Each site's natural error code is
captured into a stable local before any intervening call, since errno
isn't guaranteed stable once lock()/ERROR()/free_launchinfo() sit between
the original failure and the eventual return.

handle_jobdispatch() is intentionally untouched -- it still unconditionally
calls notify_dispatch() itself for every outcome, so the retryable path
now temporarily double-notifies until task 4.1 removes that caller-side
call. Expected, not a defect.

New test_dispatch_fail_retry.c (4 tests, including a bounded-escalation
test driving 4 consecutive RPC failures on the same job: 3rd retries,
4th resolves EXIT) plus a fixed stale assertion in test_dispatch_commit.c
and additive, read-only notify-capture accessors in
test_commit_barrier_reply.c (the binary's sole notify() stub host).

Independently reviewed and approved, including explicit scrutiny of the
cross-file test accessor addition and the stale-assertion fix -- both
confirmed legitimate, not scope creep or regression-hiding.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Cxa1PmqAB5PyXfJaNvj8TN
This commit is contained in:
Rongsong Shen 2026-08-23 17:37:07 +00:00
parent f756c4b09f
commit 05c45ef3c8
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 880 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

251
src/controller/job.c
View file

@ -23,6 +23,19 @@ static int record_job_finish_fact(cluster_t*, job_t*, xce_jobfinish_t*);
static int record_job_fail_fact(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_update(cluster_t*, job_t*, int);
/* scheduler-dispatch-commit task 3.3: dispatch_a_job_()'s four retryable
* failure sites (job.c, below) need to notify the scheduler of a retryable
* failure -- notify_dispatch() is defined AFTER dispatch_a_job_() in this
* file with no forward declaration until now. Mirrors mark_job_fail's own
* forward declaration immediately above; body/signature unmodified. */
static void notify_dispatch(cluster_t*, endpoint_t*, uint64_t, int);
/* scheduler-dispatch-commit task 3.3 (Req 4.3): default bounded-retry limit
* for a job's consecutive retryable dispatch failures with no intervening
* successful dispatch -- design.md's own stated default (3 consecutive
* failures). The failure that pushes job.dispatch_fail_count past this
* bound resolves the job to terminal instead of retrying again. */
#define XCE_DISPATCH_FAIL_RETRY_BOUND 3
/* A job is terminal once it carries a DONE or EXIT bit. Terminal jobs reject
* backward (terminal -> non-terminal) transitions and treat re-applied
@ -1025,27 +1038,202 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
}
unlock(c->workermgr);
if (rc < 0) {
/* scheduler-dispatch-commit task 3.3 (Req 4.1, 4.3, 4.4, 4.5):
* resolves this retryable (transient/environmental) dispatch
* failure -- closes bug-1627's leak site #1 (no dereference(j)
* existed on this path before this task). rc already holds the
* stable natural error code from key_find() above; no further call
* touches it before it is returned below. Nothing was ever
* committed to j->job.allocs on this path (task 3.1 gates that
* commit on a successful xce_startjob_req(), never reached here),
* so Req 4.4/4.5's "release any tentative allocation data" is
* satisfied vacuously by construction. */
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
uint32_t failcount;
lock(j, RTTI_LOCK_WRITE);
dfclass = xce_classify_dispatch_fail(
XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP);
switch (dfclass) {
case XCE_DISPATCH_FAIL_RETRYABLE:
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
} else {
failinfo.code = rc;
failinfo.detail = "dispatch failed: worker lookup failed "
"repeatedly, retry bound exceeded";
mark_job_fail(c, j, &failinfo);
}
break;
case XCE_DISPATCH_FAIL_TERMINAL:
default:
/* Impossible per today's classification table (confirmed:
* xce_classify_dispatch_fail() returns RETRYABLE for this site
* unconditionally) -- fail loud and fall through to the same
* terminal resolution as a safe fallback, mirroring
* xce_classify_dispatch_fail()'s own default: convention. */
ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP "
"classified TERMINAL (%d) -- impossible per the current "
"classification table; resolving terminal as a safe "
"fallback", (int)dfclass);
#ifdef XCE_TESTING
assert(0
&& "XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP classified "
"terminal -- update the classification table or "
"this call site");
#endif
failinfo.code = rc;
failinfo.detail = "dispatch failed: worker lookup failed";
mark_job_fail(c, j, &failinfo);
break;
}
unlock(j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, rc);
}
return rc;
}
if (w == NULL) {
int dfrc;
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
uint32_t failcount;
/* scheduler-dispatch-commit task 3.3: capture errno FIRST, before
* any other call (free_launchinfo() included) -- errno is not
* guaranteed stable across subsequent calls. */
dfrc = -errno;
free_launchinfo(&launchinfo);
return -errno;
/* Req 4.1, 4.3, 4.4, 4.5: resolves this retryable (transient/
* environmental) dispatch failure -- closes bug-1627's leak site
* #2 (no dereference(j) existed on this path before this task). */
lock(j, RTTI_LOCK_WRITE);
dfclass = xce_classify_dispatch_fail(
XCE_DISPATCH_FAIL_SITE_WORKER_NULL);
switch (dfclass) {
case XCE_DISPATCH_FAIL_RETRYABLE:
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
} else {
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not found "
"repeatedly, retry bound exceeded";
mark_job_fail(c, j, &failinfo);
}
break;
case XCE_DISPATCH_FAIL_TERMINAL:
default:
ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_WORKER_NULL "
"classified TERMINAL (%d) -- impossible per the current "
"classification table; resolving terminal as a safe "
"fallback", (int)dfclass);
#ifdef XCE_TESTING
assert(0
&& "XCE_DISPATCH_FAIL_SITE_WORKER_NULL classified "
"terminal -- update the classification table or "
"this call site");
#endif
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not found";
mark_job_fail(c, j, &failinfo);
break;
}
unlock(j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, dfrc);
}
return dfrc;
}
lock(w, RTTI_LOCK_READ);
if ((w->ep == NULL) || (w->host.status != 0)) {
int dfrc;
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
uint32_t failcount;
ERROR("worker %s is not in correct state, status %x ep %p",
w->host.name, w->host.status, w->ep);
free_launchinfo(&launchinfo);
SET_ERROR(E_XCE_BAD_WORKER);
/* scheduler-dispatch-commit task 3.3: capture errno immediately
* after SET_ERROR(), before any further call -- errno is not
* guaranteed stable across unlock(w)/dereference(w)/lock(j)/etc. */
dfrc = -errno;
unlock(w);
dereference(w);
return -errno;
/* Req 4.1, 4.3, 4.4, 4.5: resolves this retryable (transient/
* environmental) dispatch failure -- closes bug-1627's leak site
* #3 (no dereference(j) existed on this path before this task). */
lock(j, RTTI_LOCK_WRITE);
dfclass = xce_classify_dispatch_fail(
XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY);
switch (dfclass) {
case XCE_DISPATCH_FAIL_RETRYABLE:
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
} else {
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not ready "
"repeatedly, retry bound exceeded";
mark_job_fail(c, j, &failinfo);
}
break;
case XCE_DISPATCH_FAIL_TERMINAL:
default:
ERROR("dispatch_a_job_: "
"XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY classified "
"TERMINAL (%d) -- impossible per the current "
"classification table; resolving terminal as a safe "
"fallback", (int)dfclass);
#ifdef XCE_TESTING
assert(0
&& "XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY classified "
"terminal -- update the classification table or "
"this call site");
#endif
failinfo.code = dfrc;
failinfo.detail = "dispatch failed: worker not ready";
mark_job_fail(c, j, &failinfo);
break;
}
unlock(j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, dfrc);
}
return dfrc;
}
/* FIXME: we may consider to optimize when a job has enter a queue? */
@ -1183,8 +1371,67 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
unlock(j);
} else {
xce_dispatch_fail_class_t dfclass;
job_failinfo_t failinfo = { .jobid = alloc->jobid };
bool resolved_retryable = false;
uint32_t failcount;
SET_ERROR(-rc);
PERROR("xce_startjob_req");
/* scheduler-dispatch-commit task 3.3 (Req 4.1, 4.3, 4.4, 4.5):
* resolves this retryable (transient/environmental) dispatch
* failure -- closes bug-1627's leak site #4 (no dereference(j)
* existed on this path before this task). rc already holds the
* stable, captured return of xce_startjob_req() -- no call between
* that assignment and here has touched it. */
lock(j, RTTI_LOCK_WRITE);
dfclass = xce_classify_dispatch_fail(
XCE_DISPATCH_FAIL_SITE_RPC_FAILURE);
switch (dfclass) {
case XCE_DISPATCH_FAIL_RETRYABLE:
failcount = ++j->dispatch_fail_count;
if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) {
resolved_retryable = true;
} else {
failinfo.code = rc;
failinfo.detail = "dispatch failed: xce_startjob_req "
"failed repeatedly, retry bound "
"exceeded";
mark_job_fail(c, j, &failinfo);
}
break;
case XCE_DISPATCH_FAIL_TERMINAL:
default:
/* Impossible per today's classification table (confirmed:
* xce_classify_dispatch_fail() returns RETRYABLE for this site
* unconditionally) -- fail loud and fall through to the same
* terminal resolution as a safe fallback, mirroring
* xce_classify_dispatch_fail()'s own default: convention. */
ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_RPC_FAILURE "
"classified TERMINAL (%d) -- impossible per the current "
"classification table; resolving terminal as a safe "
"fallback", (int)dfclass);
#ifdef XCE_TESTING
assert(0
&& "XCE_DISPATCH_FAIL_SITE_RPC_FAILURE classified "
"terminal -- update the classification table or "
"this call site");
#endif
failinfo.code = rc;
failinfo.detail = "dispatch failed: xce_startjob_req failed";
mark_job_fail(c, j, &failinfo);
break;
}
unlock(j);
dereference(j);
if (resolved_retryable) {
notify_dispatch(c, ep, alloc->jobid, rc);
}
}
return rc;

1
src/controller/meson.build
View file

@ -181,6 +181,7 @@ controller_unit_test_src = [
'testsuite/test_job_dispatch.c',
'testsuite/test_dispatch_commit.c',
'testsuite/test_dispatch_fail_terminal.c',
'testsuite/test_dispatch_fail_retry.c',
'testsuite/test_upgrade_snapshot_restore.c',
'testsuite/test_upgrade_snapshot_apply.c',
'testsuite/test_bug857_regression.c',

2
src/controller/testsuite/controller_test.c
View file

@ -99,6 +99,7 @@ extern void test_sched_admin_dispatch(void);
extern void test_job_dispatch(void);
extern void test_dispatch_commit(void);
extern void test_dispatch_fail_terminal(void);
extern void test_dispatch_fail_retry(void);
void setUp(void) { return; }
@ -267,6 +268,7 @@ int main(int argc, char* argv[])
TEST(test_job_dispatch);
TEST(test_dispatch_commit);
TEST(test_dispatch_fail_terminal);
TEST(test_dispatch_fail_retry);
return UNITY_END();
}

37
src/controller/testsuite/test_commit_barrier_reply.c
View file

@ -133,6 +133,43 @@ static void cbrp_notify_capture_reset(void)
sizeof(g_notify_reply_already_enqueued));
}
/* scheduler-dispatch-commit task 3.3: minimal read-only accessors exposing
* this file's notify() capture (above) to other test files sharing this
* same controller_unit_test binary. This file owns the sole notify()
* definition linked into that binary (dataex.c's real one is excluded --
* see the "Link stubs" comment at the top of this file), so it is the only
* place any test in the binary can observe a notify_dispatch() (job.c)
* call. Read-only: does not alter this file's own capture/reset behavior
* or any of its own tests. Consumed by test_dispatch_fail_retry.c to
* confirm notify_dispatch() fires on the new retryable-resolution path. */
int cbrp_notify_capture_count_ext(void)
{
return g_notify_count;
}
int cbrp_notify_capture_event_ext(int idx)
{
return (idx >= 0 && idx < g_notify_count && idx < CBRP_MAX_NOTIFY)
? g_notify_events[idx] : -1;
}
uint64_t cbrp_notify_capture_jobid_ext(int idx)
{
return (idx >= 0 && idx < g_notify_count && idx < CBRP_MAX_NOTIFY)
? g_notify_jobids[idx] : 0;
}
int cbrp_notify_capture_code_ext(int idx)
{
return (idx >= 0 && idx < g_notify_count && idx < CBRP_MAX_NOTIFY)
? g_notify_codes[idx] : 0;
}
void cbrp_notify_capture_reset_ext(void)
{
cbrp_notify_capture_reset();
}
static worker_t *g_stub_ep_worker; /* NULL by default (3.1 behavior) */
/* clang-format off */

31
src/controller/testsuite/test_dispatch_commit.c
View file

@ -277,12 +277,24 @@ static void test_dc_success_commits_alloc_state_and_resets_counter(void)
#define DC_JOBID_NOTFOUND 91002ULL
/* Regression safety net (untouched failure branch): a dispatch whose
* target worker cannot be found must leave the job exactly as it was
* before the attempt -- still PEND, allocs still uncommitted, the
* dispatch_fail_count untouched (that counter's own increment is task
* 3.3's job, not 3.1's) -- and must append no WAL entry. Confirms this
* task's restructuring did not disturb the five failure branches ahead of
* xce_startjob_req(). */
* target worker cannot be found must leave the job's ALLOCATION exactly
* as it was before the attempt -- still PEND, allocs still uncommitted --
* and must append no WAL entry. Confirms this task's restructuring did
* not disturb the five failure branches ahead of xce_startjob_req().
*
* scheduler-dispatch-commit task 3.3 update: dispatch_fail_count is no
* longer asserted unchanged here -- task 3.3 wired this exact failure
* branch (whichever of XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP/WORKER_NULL
* the runtime takes for an unregistered worker key) to increment the
* counter by exactly 1 as part of its retryable resolution (Req 4.3),
* which this test's own original comment explicitly anticipated
* ("incrementing it is task 3.3's job, not 3.1's"). The counter's own
* increment-by-1/bound behavior is covered in depth by
* test_dispatch_fail_retry.c; this test only re-confirms the seeded
* starting value (2) plus this task's own increment (+1) lands at 3,
* still within the bound (3), so the job stays PEND -- not a regression
* in this test's own subject (allocation-gating), just an update to an
* assertion that depended on timing task 3.3 had not landed yet. */
static void test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted(void)
{
job_t* j;
@ -346,9 +358,10 @@ static void test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted(void)
"commit, it does not change any failure branch's behavior");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"job.allocs must remain uncommitted on this failure path");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(2, j->dispatch_fail_count,
"this task does not touch dispatch_fail_count on any failure path "
"(incrementing it is task 3.3's job) -- it must be unchanged");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(3, j->dispatch_fail_count,
"task 3.3: this retryable failure path increments the counter by "
"exactly 1 from its seeded starting value (2 -> 3), still within "
"the bound (3)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a failed dispatch must append no WAL entry");

563
src/controller/testsuite/test_dispatch_fail_retry.c Normal file
View file

@ -0,0 +1,563 @@
/* test_dispatch_fail_retry.c -- scheduler-dispatch-commit task 3.3
*
* Proves dispatch_a_job_() (job.c) now RESOLVES a retryable (transient/
* environmental) dispatch failure -- incrementing the consecutive-failure
* counter (job_t.dispatch_fail_count, task 1.3) under the job's own write
* lock, reporting the failure via notify_dispatch() so the job can be
* retried when the incremented count is still within the bound, or
* resolving the job to permanently failed (mirroring task 3.2's terminal
* resolution exactly) once the bound is exceeded -- instead of leaking the
* job reference it holds, as it did before this task (bug-1627).
*
* Covers the four failure sites design.md's "Controller Dispatch-Failure
* Classification" component always classifies RETRYABLE today (confirmed
* by task 1.1's own xce_classify_dispatch_fail() switch, unmodified/reused
* here, not re-tested):
*
* - XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP (workermgr key_find() rc<0)
* - XCE_DISPATCH_FAIL_SITE_WORKER_NULL (worker not found, w==NULL)
* - XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY (E_XCE_BAD_WORKER)
* - XCE_DISPATCH_FAIL_SITE_RPC_FAILURE (xce_startjob_req() rc<0)
*
* test_worker_notfound_resolves_retryable (below) seeds an unregistered
* worker key, matching test_dispatch_commit.c's own established fixture
* for the "worker not found" scenario -- that file's own header notes this
* takes dispatch_a_job_()'s WORKER_NULL branch (w stays NULL, rc>=0 from
* key_find()) rather than WORKER_LOOKUP (a genuine key_find() rc<0, which
* this test suite has no black-box way to force without reaching into
* librtti's own internals, an external/opaque dependency). Both sites
* share the exact same classify/resolve procedure in job.c (this task's
* whole point is that the four retryable sites are structurally
* identical), so this one test's assertions equally prove either site's
* behavior regardless of which the runtime actually takes -- it is
* deliberately written not to depend on which.
*
* Req 4.1: a retryable dispatch failure below the bound reports the
* failure via notify_dispatch() so the job can be retried -- job.state
* stays PEND, dispatch_fail_count increments by exactly 1.
* Req 4.3: consecutive retryable failures with no intervening success are
* evaluated against a bounded limit (design.md's stated default: 3) --
* the 4th consecutive failure resolves the job to terminal instead of
* retrying again.
* Req 4.4: a job returned to pending following a retryable failure has no
* stale resource-allocation/commitment data -- job.allocs stays NULL
* (task 3.1 gates its only commit on a successful xce_startjob_req(),
* never reached on any of these four failure paths, so this is satisfied
* vacuously by construction -- asserted here explicitly anyway).
* Req 4.5: same vacuous-by-construction guarantee for the bound-exceeded
* terminal resolution.
* Also (bug-1627's actual fix): each of the four sites now
* dereference(j)s the reference taken at the top of dispatch_a_job_() --
* proven via the same manual "canary" reference technique task 3.2's
* test_dispatch_fail_terminal.c established (see that file's header for
* the full rationale; job_t's real refcount lives in jobj.c's file-local
* job_private_t with no accessor exposed to a test).
*
* notify_dispatch() observability: this suite's controller_unit_test
* binary excludes dataex.c (the real notify()'s home) -- the sole notify()
* definition linked into this binary is test_commit_barrier_reply.c's own
* capture stub. That file exposes a handful of minimal, read-only
* accessor functions (cbrp_notify_capture_*_ext, added by this task) so
* this file can observe a notify_dispatch() call without duplicating any
* mocking machinery.
*
* dispatch_a_job_() is static/file-local; reached here through
* dispatch_a_job_test(), task 3.1's existing XCE_TESTING-only wrapper --
* reused directly, not duplicated.
*
* Fixture mirrors test_dispatch_fail_terminal.c's dft_setup/dft_teardown
* pattern (real jobmgr/workermgr/qmgr/hooksys + a real disk-backed
* persist_t with cluster == NULL) trimmed to what these four failure sites
* need. No hook functions are ever registered against g_dfr.hooksys in
* this fixture -- mark_job_fail()'s two xce_run_hooks() calls therefore
* iterate zero registered handlers and are safe no-ops here.
*/
#include <rtti.h>
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "netio.h"
#include "persist.h"
#include "qobj.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"
/* dispatch_a_job_test: task 3.1's XCE_TESTING-only wrapper (job.c), thinly
* exposing the static dispatch_a_job_() for direct testing. */
extern int dispatch_a_job_test(cluster_t* c, endpoint_t* ep,
xce_job_alloc_t* alloc);
/* test_commit_barrier_reply.c's minimal notify()-capture accessors (added
* by this task) -- see this file's header. */
extern int cbrp_notify_capture_count_ext(void);
extern int cbrp_notify_capture_event_ext(int idx);
extern uint64_t cbrp_notify_capture_jobid_ext(int idx);
extern int cbrp_notify_capture_code_ext(int idx);
extern void cbrp_notify_capture_reset_ext(void);
/* jobj.c's job type table and the process-global xstring table are shared
* across every test file in the controller_unit_test binary -- guard
* idempotently, mirroring every sibling fixture in this suite. */
static int g_dfr_jobobj_ready;
static cluster_t g_dfr;
static char g_dfr_tmpdir[256];
static void dfr_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32, .delta = 16 };
persist_config_t cfg;
if (makeString((char*)"dfr-probe") == NULL)
xstring_table_init(64);
if (!g_dfr_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_dfr_jobobj_ready = 1;
}
memset(&g_dfr, 0, sizeof(g_dfr));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_dfr.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_dfr.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_dfr.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_dfr.qmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_dfr.jobmgr);
TEST_ASSERT_NOT_NULL(g_dfr.workermgr);
TEST_ASSERT_NOT_NULL(g_dfr.qmgr);
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
strcpy(g_dfr_tmpdir, "/tmp/xce_dfr33_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_dfr_tmpdir));
persist_set_workdir(&g_dfr.persist, g_dfr_tmpdir, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_dfr.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_dfr.persist));
}
static void dfr_teardown(void)
{
char cmd[512];
persist_stop(&g_dfr.persist);
persist_fini(&g_dfr.persist);
if (g_dfr.workermgr) dereference(g_dfr.workermgr);
if (g_dfr.qmgr) dereference(g_dfr.qmgr);
if (g_dfr.jobmgr) dereference(g_dfr.jobmgr);
xce_hook_fini(&g_dfr.hooksys);
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_dfr_tmpdir);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_dfr_tmpdir);
}
/* Seeds a queue and a PEND job with a real submit/start_info against
* g_dfr.jobmgr/qmgr (dispatch_a_job_() reads submit->queue and
* submit->start_info). Returns the seeded job_t* (borrowed -- g_dfr.jobmgr
* owns the creation reference). */
static job_t* dfr_seed_job(uint64_t jobid, const char* qname)
{
xce_queue_t qdef;
job_startinfo_t startinfo;
job_submit_data_t submit;
xce_job_t jdef;
job_t* j;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)qname;
TEST_ASSERT_NOT_NULL(makeQueue(g_dfr.qmgr, &qdef));
memset(&startinfo, 0, sizeof(startinfo));
startinfo.cmdline = (char*)"/bin/true";
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)qname;
submit.user = (char*)"tester";
submit.start_info = &startinfo;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
j = makeJob(g_dfr.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j);
return j;
}
/* Registers a worker in g_dfr.workermgr with ep pointed at the given
* endpoint (host.status defaults to 0/ready via calloc). */
static void dfr_seed_worker(const char* wname, endpoint_t* ep)
{
xce_worker_t wdef;
hostattrs_t* attrs;
worker_t* w = NULL;
attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)wname;
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_dfr.workermgr, &wdef));
lock(g_dfr.workermgr, RTTI_LOCK_READ);
key_find(g_dfr.workermgr, (void*)wname, (void**)&w);
unlock(g_dfr.workermgr);
TEST_ASSERT_NOT_NULL(w);
w->ep = ep;
}
/* Canary-reference technique (see test_dispatch_fail_terminal.c's header
* for the full rationale): arm BEFORE the call under test, release AFTER
* and re-touch j -- an over-release inside dispatch_a_job_test() would
* have freed j while the canary is still held, a heap-use-after-free ASAN
* catches on the post-call touch. */
static void dfr_arm_ref_canary(job_t* j)
{
reference(j);
}
static void dfr_assert_ref_not_over_released(job_t* j, uint64_t expect_jobid)
{
dereference(j);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(expect_jobid, j->job.jobid,
"j must still be safely readable after releasing the canary "
"reference -- an over-release inside dispatch_a_job_test() would "
"have freed j while the canary was still held, making this a "
"heap-use-after-free");
}
/* A "broken" endpoint: real socket fd, real iomgr, deliberately NOT
* registered in the iomgr's epoll -- EPOLL_CTL_MOD on it fails with ENOENT
* (the exact post-failed-reconnect condition endpoint_update_epoll()
* surfaces, bug-646), making xce_startjob_req() -> xce_request() ->
* endpoint_enq() fail deterministically and repeatably. This is the same
* technique test_netio_arm_failure.c/test_request_send_propagation.c
* (src/lib/testsuite/) already establish and prove live for exactly this
* failure mode; dataq_enq() itself still always succeeds first (the
* request bytes get queued), only the poll-arm step fails, so this is
* safe to call repeatedly against the same endpoint -- ep->flags never
* reaches the target value, so the same failure recurs every call. */
static iomgr_t g_dfr_broken_iomgr;
static endpoint_t* g_dfr_broken_ep;
static void dfr_setup_broken_ep(void)
{
memset(&g_dfr_broken_iomgr, 0, sizeof(g_dfr_broken_iomgr));
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&g_dfr_broken_iomgr));
g_dfr_broken_ep = endpoint_get();
TEST_ASSERT_NOT_NULL(g_dfr_broken_ep);
g_dfr_broken_ep->fd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(g_dfr_broken_ep->fd >= 0);
g_dfr_broken_ep->iomgr = &g_dfr_broken_iomgr;
g_dfr_broken_ep->ssl_handshake_done = 1;
g_dfr_broken_ep->flags = 0;
}
static void dfr_teardown_broken_ep(void)
{
endpoint_fini(g_dfr_broken_ep); /* closes fd, drains/frees queued pkts */
free(g_dfr_broken_ep);
g_dfr_broken_ep = NULL;
close(g_dfr_broken_iomgr.epoll_fd);
}
/* ── SITE 1/2: worker lookup finds nothing ─────────────────────────────
* See file header: covers whichever of XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP
* / XCE_DISPATCH_FAIL_SITE_WORKER_NULL the runtime actually takes -- both
* resolve via the identical classify/resolve procedure, so this test's
* assertions hold either way. */
#define DFR_JOBID_WORKER_NOTFOUND 91201ULL
static void test_worker_notfound_resolves_retryable(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
int before_count;
dfr_setup();
j = dfr_seed_job(DFR_JOBID_WORKER_NOTFOUND, "dfr-q1");
/* Deliberately no matching worker registered in g_dfr.workermgr. */
alloc = make_job_alloc(DFR_JOBID_WORKER_NOTFOUND, 1,
make_host_allocs(1,
make_host_alloc((char*)"no-such-worker-dfr", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
cbrp_notify_capture_reset_ext();
before_count = cbrp_notify_capture_count_ext();
dfr_arm_ref_canary(j);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"a dispatch to a nonexistent worker must still fail, unchanged "
"return contract from before this task");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"Req 4.1: a retryable dispatch failure below the bound must leave "
"the job PEND, eligible for another attempt");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"Req 4.4: no partial allocation data may be left behind");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(1, j->dispatch_fail_count,
"Req 4.3: the consecutive-failure counter must increment by "
"exactly 1");
dfr_assert_ref_not_over_released(j, DFR_JOBID_WORKER_NOTFOUND);
TEST_ASSERT_EQUAL_INT_MESSAGE(before_count + 1,
cbrp_notify_capture_count_ext(),
"Req 4.1: a below-bound retryable failure must call "
"notify_dispatch() exactly once");
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
cbrp_notify_capture_event_ext(before_count));
TEST_ASSERT_EQUAL_UINT64(DFR_JOBID_WORKER_NOTFOUND,
cbrp_notify_capture_jobid_ext(before_count));
TEST_ASSERT_EQUAL_INT(rc, cbrp_notify_capture_code_ext(before_count));
free_JobAlloc_ptr(&alloc);
dfr_teardown();
printf("test_worker_notfound_resolves_retryable: PASS\n");
}
/* ── SITE 3: XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY ───────────────────
* A registered worker whose ep is left NULL (calloc default) hits
* "(w->ep == NULL) || (w->host.status != 0)". */
#define DFR_JOBID_WORKER_NOTREADY 91202ULL
static void test_worker_notready_resolves_retryable(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
int before_count;
dfr_setup();
j = dfr_seed_job(DFR_JOBID_WORKER_NOTREADY, "dfr-q2");
dfr_seed_worker("dfr-wkr-notready", NULL); /* ep left NULL: not ready */
alloc = make_job_alloc(DFR_JOBID_WORKER_NOTREADY, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-notready", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
cbrp_notify_capture_reset_ext();
before_count = cbrp_notify_capture_count_ext();
dfr_arm_ref_canary(j);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"a dispatch to a not-ready worker must still fail, unchanged "
"return contract from before this task");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"Req 4.1: a retryable dispatch failure below the bound must leave "
"the job PEND, eligible for another attempt");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"Req 4.4: no partial allocation data may be left behind");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(1, j->dispatch_fail_count,
"Req 4.3: the consecutive-failure counter must increment by "
"exactly 1");
dfr_assert_ref_not_over_released(j, DFR_JOBID_WORKER_NOTREADY);
TEST_ASSERT_EQUAL_INT_MESSAGE(before_count + 1,
cbrp_notify_capture_count_ext(),
"Req 4.1: a below-bound retryable failure must call "
"notify_dispatch() exactly once");
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
cbrp_notify_capture_event_ext(before_count));
TEST_ASSERT_EQUAL_UINT64(DFR_JOBID_WORKER_NOTREADY,
cbrp_notify_capture_jobid_ext(before_count));
TEST_ASSERT_EQUAL_INT(rc, cbrp_notify_capture_code_ext(before_count));
free_JobAlloc_ptr(&alloc);
dfr_teardown();
printf("test_worker_notready_resolves_retryable: PASS\n");
}
/* ── SITE 4: XCE_DISPATCH_FAIL_SITE_RPC_FAILURE ────────────────────────
* A ready worker whose ep is deliberately "broken" (see dfr_setup_broken_ep
* above) makes xce_startjob_req() itself fail. */
#define DFR_JOBID_RPC_FAILURE 91203ULL
static void test_rpc_failure_resolves_retryable(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
int before_count;
dfr_setup();
dfr_setup_broken_ep();
j = dfr_seed_job(DFR_JOBID_RPC_FAILURE, "dfr-q3");
dfr_seed_worker("dfr-wkr-rpcfail", g_dfr_broken_ep);
alloc = make_job_alloc(DFR_JOBID_RPC_FAILURE, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-rpcfail", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
cbrp_notify_capture_reset_ext();
before_count = cbrp_notify_capture_count_ext();
dfr_arm_ref_canary(j);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"xce_startjob_req() against a broken (unarmed-poll) endpoint must "
"fail, unchanged return contract from before this task");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"Req 4.1: a retryable dispatch failure below the bound must leave "
"the job PEND, eligible for another attempt");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"Req 4.4: no partial allocation data may be left behind");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(1, j->dispatch_fail_count,
"Req 4.3: the consecutive-failure counter must increment by "
"exactly 1");
dfr_assert_ref_not_over_released(j, DFR_JOBID_RPC_FAILURE);
TEST_ASSERT_EQUAL_INT_MESSAGE(before_count + 1,
cbrp_notify_capture_count_ext(),
"Req 4.1: a below-bound retryable failure must call "
"notify_dispatch() exactly once");
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
cbrp_notify_capture_event_ext(before_count));
TEST_ASSERT_EQUAL_UINT64(DFR_JOBID_RPC_FAILURE,
cbrp_notify_capture_jobid_ext(before_count));
TEST_ASSERT_EQUAL_INT(rc, cbrp_notify_capture_code_ext(before_count));
free_JobAlloc_ptr(&alloc);
dfr_teardown_broken_ep();
dfr_teardown();
printf("test_rpc_failure_resolves_retryable: PASS\n");
}
/* ── Bounded escalation (Req 4.3) ───────────────────────────────────────
* Drives the SAME job through XCE_DISPATCH_FAIL_SITE_RPC_FAILURE
* repeatedly (easiest of the four to trigger deterministically and
* repeatably via the broken-endpoint technique). Confirms the 3rd
* consecutive failure (== design.md's stated default bound) still
* resolves retryable, and the 4th resolves TERMINAL instead. */
#define DFR_JOBID_BOUND 91204ULL
static void test_rpc_failure_bounded_escalation(void)
{
job_t* j;
int i;
dfr_setup();
dfr_setup_broken_ep();
j = dfr_seed_job(DFR_JOBID_BOUND, "dfr-q4");
dfr_seed_worker("dfr-wkr-bound", g_dfr_broken_ep);
/* Attempts 1-3: within the bound (XCE_DISPATCH_FAIL_RETRY_BOUND == 3,
* job.c) -- each must resolve retryable. */
for (i = 1; i <= 3; i++) {
xce_job_alloc_t* alloc;
int rc;
alloc = make_job_alloc(DFR_JOBID_BOUND, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-bound", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"an attempt within the bound must leave the job PEND, still "
"eligible for retry");
TEST_ASSERT_EQUAL_UINT32((uint32_t)i, j->dispatch_fail_count);
TEST_ASSERT_NULL(j->job.allocs);
free_JobAlloc_ptr(&alloc);
}
/* Attempt 4: exceeds the bound -- must resolve TERMINAL instead of
* retrying again. */
{
xce_job_alloc_t* alloc;
int rc;
alloc = make_job_alloc(DFR_JOBID_BOUND, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-bound", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT, j->job.state,
"Req 4.3: the failure that pushes dispatch_fail_count past the "
"bound must resolve the job to terminal instead of retrying "
"again");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(4, j->dispatch_fail_count,
"the counter itself keeps counting past the bound -- only the "
"resolution changes");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"Req 4.5: no partial allocation data may be left behind on the "
"terminal resolution either");
free_JobAlloc_ptr(&alloc);
}
dfr_teardown_broken_ep();
dfr_teardown();
printf("test_rpc_failure_bounded_escalation: PASS\n");
}
void test_dispatch_fail_retry(void)
{
UnitySetTestFile(__FILE__);
printf("\n>> dispatch_a_job_() retryable dispatch-failure resolution "
"tests (Req 4.1, 4.3, 4.4, 4.5)\n\n");
RUN_TEST(test_worker_notfound_resolves_retryable);
RUN_TEST(test_worker_notready_resolves_retryable);
RUN_TEST(test_rpc_failure_resolves_retryable);
RUN_TEST(test_rpc_failure_bounded_escalation);
}