diff --git a/.kiro/specs/scheduler-dispatch-commit/tasks.md b/.kiro/specs/scheduler-dispatch-commit/tasks.md index 74fa9c7a..6ad627c5 100644 --- a/.kiro/specs/scheduler-dispatch-commit/tasks.md +++ b/.kiro/specs/scheduler-dispatch-commit/tasks.md @@ -56,7 +56,7 @@ - _Requirements: 4.2, 4.5_ - _Depends: 1.1_ -- [ ] 3.3 Controller resolves a retryable (transient/environmental) dispatch failure, with bounded escalation +- [x] 3.3 Controller resolves a retryable (transient/environmental) dispatch failure, with bounded escalation - When a dispatch attempt fails for a reason the classification (Task 1.1) marks retryable, the controller increments the consecutive-failure counter (Task 1.3) under the same protected access; if the incremented count is still within a bounded limit, it reports the failure so the job can be retried, otherwise it resolves the job to a permanently failed state exactly as Task 3.2 does for a terminal failure - This covers the four failure sites that do not currently release the job reference they hold — fix each to release it as part of this same resolution step (Task 3.2's two sites already release theirs correctly and are unaffected) - Observable completion: a retryable dispatch failure below the bound leaves the job eligible for another attempt with its reference count restored to the pre-attempt value; a retryable failure that pushes the count past the bound resolves the job to permanently failed instead @@ -107,3 +107,8 @@ - Task 3.1: when validating a build against this suite's ASAN leak output, always replicate the project's own `ASAN_OPTIONS=detect_leaks=0` test env (`meson.build:229`) — a standalone binary run without it produces spurious leak reports (a red herring, not a regression) that can waste review time. - Task 3.2: `job_t`'s real reference count lives in `jobj.c`'s file-local `job_private_t`, with no accessor exposed anywhere in the codebase — a test proving no reference-count regression can't do a direct numeric comparison. The working pattern (used by `test_dispatch_fail_terminal.c`) is a "canary reference": take one extra manual `reference()` on the job **immediately before** the call under test (not after — arming it after only relies incidentally on other assertions crashing first, which doesn't independently prove anything), then `dereference()` it after the call and confirm the job is still safely readable — an over-release during the call frees the job while the canary is still held, so the post-call touch is a real ASAN-caught heap-use-after-free. - Task 3.2: reconfirmed bug-1429's hang again (still `test_commit_barrier_reply.c`, still unrelated to any change in this feature) — third+ reproduction, still not root-caused. Continues to require `controller_unit_test -t ` filtering for a clean isolated regression signal. +- Task 3.3: bug-1627 (the 4-site reference leak on `dispatch_a_job_()`'s retryable failure paths) is now FIXED — `dereference(j)` added at all four retryable sites, closing it as a direct consequence of this task's resolution restructuring. `.wolf/buglog.json` updated accordingly. +- Task 3.3: `notify_dispatch()` is `static` and defined AFTER `dispatch_a_job_()` in job.c with no forward declaration — needed one added near the top (mirroring `mark_job_fail`'s existing forward-decl precedent at job.c:24) so the new retryable-resolution code could call it. Any later task in this file needing to call a same-file function defined later in the file should check for/add a forward declaration the same way, rather than reordering definitions. +- Task 3.3: capturing a site's natural error code (`rc`/`errno`) into a stable local must happen BEFORE any intervening call once more logic is inserted before the eventual `return` — `errno` is not guaranteed stable across `lock()`/`ERROR()`/`free_launchinfo()` etc. This bit the worker-null and worker-not-ready sites specifically (both originally read `errno` directly at the `return` statement with nothing in between; task 3.3 had to capture it immediately after `SET_ERROR()`/at branch entry, before inserting the new lock/classify/resolve sequence). +- Task 3.3: `test_commit_barrier_reply.c` (a file otherwise owned by the `ha-commit-barrier-rollout` spec) is this binary's sole host of the `notify()` stub — any future task needing to observe `notify_dispatch()`'s calls in a controller unit test must add accessors there too (purely additive, read-only `_ext`-suffixed pass-throughs to its existing static capture globals — do not modify the stub body itself or its existing pinned tests). Confirmed as legitimate shared test infrastructure, not a boundary violation, by task 3.3's independent review. +- Task 3.3: the bound constant is `XCE_DISPATCH_FAIL_RETRY_BOUND = 3` (design.md's stated default), defined once near job.c's top and used at all four retryable sites — no magic numbers. 3 consecutive failures still retry; the 4th resolves terminal. diff --git a/src/controller/job.c b/src/controller/job.c index 195c2731..828196a6 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; diff --git a/src/controller/meson.build b/src/controller/meson.build index 1c48d534..0a45dd47 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index d34d818a..cc19fbec 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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(); } diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index c4916452..a78f0498 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -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 */ diff --git a/src/controller/testsuite/test_dispatch_commit.c b/src/controller/testsuite/test_dispatch_commit.c index de84be44..a96bfb48 100644 --- a/src/controller/testsuite/test_dispatch_commit.c +++ b/src/controller/testsuite/test_dispatch_commit.c @@ -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"); diff --git a/src/controller/testsuite/test_dispatch_fail_retry.c b/src/controller/testsuite/test_dispatch_fail_retry.c new file mode 100644 index 00000000..cdf91972 --- /dev/null +++ b/src/controller/testsuite/test_dispatch_fail_retry.c @@ -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 + +#include +#include +#include +#include +#include +#include +#include + +#include + +#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); +}