From d9aedc4b34777ad118f94efad806d1e5b283ce56 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 23 Aug 2026 15:24:55 +0000 Subject: [PATCH] feat(scheduler-dispatch-commit): controller commits alloc+RUN transition together on successful dispatch (task 3.1) dispatch_a_job_()'s previously-unconditional job.allocs commit is now gated on xce_startjob_req() actually succeeding (rc >= 0). In that same successful branch, under a newly-acquired lock(j, RTTI_LOCK_WRITE) (a real pre-existing locking gap this call site had -- it took no lock on the job object itself before this), the job transitions to JOB_STAT_RUN, dispatch_fail_count resets to 0, and a WAL_JOB_DISPATCH fact is appended for durability. The failure path is otherwise unchanged -- classification and resolution of the six failure sites is tasks 3.2/3.3's job, not this one's. Adds an XCE_TESTING-only dispatch_a_job_test() wrapper (dispatch_a_job_() is static) and test_dispatch_commit.c, proving: state->RUN, allocs committed correctly, counter resets, and durability survives a real persist_recover() replay with no WAL_JOB_START present. Independently reviewed and approved. Two pre-existing, out-of-boundary findings surfaced and logged for later tasks: test_commit_barrier_reply.c's test_jobdispatch_receipt_ack_no_wal_append will need updating by task 5.4 (Req 5.3 anticipates this), and bug-1429's hang was reconfirmed a 3rd time (unrelated to this change). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Cxa1PmqAB5PyXfJaNvj8TN --- .../specs/scheduler-dispatch-commit/tasks.md | 7 +- src/controller/job.c | 67 +- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_dispatch_commit.c | 633 ++++++++++++++++++ 5 files changed, 699 insertions(+), 11 deletions(-) create mode 100644 src/controller/testsuite/test_dispatch_commit.c diff --git a/.kiro/specs/scheduler-dispatch-commit/tasks.md b/.kiro/specs/scheduler-dispatch-commit/tasks.md index 7eed5b5d..85607452 100644 --- a/.kiro/specs/scheduler-dispatch-commit/tasks.md +++ b/.kiro/specs/scheduler-dispatch-commit/tasks.md @@ -40,7 +40,7 @@ - [ ] 3. Core: controller transitions a job to running state only once dispatch to the worker succeeds, and resolves failures - Sub-tasks 3.1-3.3 all restructure the same function and share its per-job lock scope, so none of them are marked parallel to each other, regardless of how disjoint their individual concerns look. -- [ ] 3.1 Controller commits a job's resource allocation and running-state transition together, only on a successful dispatch +- [x] 3.1 Controller commits a job's resource allocation and running-state transition together, only on a successful dispatch - The controller's existing (currently unconditional) commit of a job's resource-allocation data becomes conditional on the dispatch request to the worker actually succeeding; the job's state transition to running happens in that same successful outcome, protected against concurrent access to the job record - Record the new durable dispatch-transition fact (Task 1.2) as part of this same successful outcome - Reset the consecutive-retryable-failure counter (Task 1.3) to zero as part of this same successful outcome, so a job that later fails again starts counting from zero rather than retaining count from before its last success @@ -100,3 +100,8 @@ - Tasks 3.2/3.3 (retryable/terminal resolution, both call `xce_classify_dispatch_fail()`) should follow the same fail-loud `default:` convention if they add their own exhaustive switches (e.g. over `xce_dispatch_fail_class_t`'s two values). - Operational note (not spec-relevant): implementer/reviewer subagents in this run consistently ended their turn mid-build (after starting `xce-build.sh controller`, before it finished) rather than blocking on it, requiring one resume-and-nudge each time. Future task dispatches in this run should explicitly instruct agents to poll/block synchronously until the build genuinely completes before composing their status/verdict block, to save round-trips. - Task 1.2: `apply_wal_entry()` (persist.c) is confirmed to have ZERO production call sites today — `persist_recover()`'s actual boot-time replay path (`replay_wal()`) and the standby catch-up path (`ha_standby_advance_committed_and_apply()`, post the 2026-07-10 `ha_apply_single_entry` refactor) both go through `apply_committed_entry()` exclusively. design.md's "WAL Dispatch Fact" component (its "Preconditions"-adjacent prose) still characterizes `apply_wal_entry()` as "the recovery/replay-time path reading WAL segments from disk after restart" — that's stale. Adding `WAL_JOB_DISPATCH` to both functions' fall-through groups was still correct (matches every sibling `WAL_JOB_*` type's identical dual-registration, harmless dead-code parity) and is NOT something to undo — but design.md's wording should be corrected before any later task cites `apply_wal_entry()`'s role from that doc without re-verifying. +- Task 3.1: `xce_request()`/every `xce_*_req()` wrapper (`src/lib/request.c`) is genuinely non-blocking fire-and-forget (capnp-encode + `endpoint_enq()` onto `ep->outq`, returns immediately — never waits for a reply). `endpoint_enq()` succeeds unconditionally against a zeroed `endpoint_t` with `ep->iomgr == NULL`. Net: testing a controller-side `dispatch_a_job_()` success path needs no live peer/socketpair/round-trip thread — a zeroed static `endpoint_t` (`.fd = -1`, no iomgr) is sufficient, matching `test_commit_barrier_reply.c`'s own established `g_wkr_disp_ep` fixture precedent. Reused directly for `test_dispatch_commit.c`; any later task exercising `dispatch_a_job_()`/`xce_startjob_req()` in a test should use the same pattern rather than building real network mocking. +- Task 3.1: **`test_commit_barrier_reply.c`'s `test_jobdispatch_receipt_ack_no_wal_append`** (a different, `ha-commit-barrier-rollout`-owned pinned test) asserts a successful dispatch cycle appends ZERO WAL entries — now false, since a successful dispatch appends exactly one `WAL_JOB_DISPATCH` fact (Req 2.4). Confirmed by both task 3.1's implementer and its independent reviewer; this is explicitly the kind of stale assertion Requirement 5.3 / task 5.4 exists to update, deliberately left unfixed here (out of 3.1's boundary, and that file's tests were unreachable in a full-suite run anyway due to the bug-1429 hang below, registered earlier in `controller_test.c`'s `main()`). **Task 5.4 must update this assertion** before claiming full-suite regression-clean. +- Task 3.1: reconfirmed **bug-1429** (`controller_unit_test` TIMEOUT/hang, pre-existing, unrooted) a 3rd time, now hanging inside `test_commit_barrier_reply.c`'s `test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs` — a 3rd distinct hang location across 3 sessions, confirmed unrelated to `dispatch_a_job_()` (that test never calls it) and independently reproduced twice more (standalone binary run past the meson harness's 300s cap, and a fully separate rebuild). Because this hang sits earlier in `controller_test.c`'s registration order than most of this feature's own new test files, a full un-filtered `meson test` run will NOT reach them — **task 5.4's full-suite validation must run with the hanging group filtered out (`controller_unit_test -t `) or budget for the 300s timeout**, not treat a bare `meson test` TIMEOUT as this feature's own failure. +- Task 3.1: confirmed pre-existing (not introduced by this task, not in this task's boundary to fix) — `dispatch_a_job_()`'s success path never calls `dereference(j)` on the reference taken at lookup; identical in the diff's before/after, and the sole caller `handle_jobdispatch()` never dereferences either. A candidate future bug-log entry, out of scope for this spec. +- 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. diff --git a/src/controller/job.c b/src/controller/job.c index e6d24c67..d1bfd056 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1035,24 +1035,71 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) free_launchinfo(&launchinfo); - if (j->job.allocs != NULL) { - free_JobAlloc_ptr(&(j->job.allocs)); - } - j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t)); - j->job.allocs->jobid = alloc->jobid; - j->job.allocs->nhosts = nc; - MOVE_PTR(j->job.allocs->allocs, allocs); - alloc->allocs = NULL; + /* scheduler-dispatch-commit task 3.1 (Req 2.1-2.4, 5.1): the allocation + * commit -- previously unconditional regardless of xce_startjob_req()'s + * outcome -- is now gated on the dispatch to the worker actually + * succeeding, and the job's transition to a running state happens in + * that same successful outcome. Protected under this component's own + * lock(j, RTTI_LOCK_WRITE): dispatch_a_job_() previously took no lock on + * `j` itself here (only on c->jobmgr/c->workermgr/w), a real + * pre-existing gap this restructuring corrects for this call site + * (design.md's Concurrency Strategy note -- not fixed anywhere else in + * this file). On failure, state is left untouched (still PEND) and + * nothing is committed, so there is nothing to release (Req 4.4/4.5 + * satisfied vacuously by construction for this path). The worker's own + * later confirmation that the job actually started + * (record_job_start_fact()/WAL_JOB_START) continues to fire exactly as + * it does today, unaffected by this change. */ + if (rc >= 0) { + lock(j, RTTI_LOCK_WRITE); - if (rc < 0) { + if (j->job.allocs != NULL) { + free_JobAlloc_ptr(&(j->job.allocs)); + } + j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t)); + j->job.allocs->jobid = alloc->jobid; + j->job.allocs->nhosts = nc; + MOVE_PTR(j->job.allocs->allocs, allocs); + alloc->allocs = NULL; + + j->job.state = JOB_STAT_RUN; + + /* Reset the consecutive-retryable-failure counter (task 1.3): a + * fresh successful dispatch clears any prior failure history, so a + * job that later fails again is evaluated against the bound as if + * it had only failed once (Req 4.3; the bound itself is exercised + * end-to-end by task 5.2, not here). */ + j->dispatch_fail_count = 0; + + /* Durable record of the dispatch-time RUN transition (task 1.2), + * mirroring record_job_start_fact()'s persist_submit_job_fact() + * call shape exactly -- _pad stays 0 (the "legacy: state is + * absolute value" convention, matching every other job-state fact + * committed synchronously here rather than via deferred apply). */ + (void)persist_submit_job_fact(&c->persist, WAL_JOB_DISPATCH, + alloc->jobid, JOB_STAT_RUN); + + unlock(j); + } else { SET_ERROR(-rc); PERROR("xce_startjob_req"); - return rc; } return rc; } +#ifdef XCE_TESTING +/* Test-only accessor: dispatch_a_job_() is static/file-local (its only + * caller, handle_jobdispatch(), is unmodified by this task) -- mirroring + * xce_classify_dispatch_fail_test()'s existing XCE_TESTING-only-export + * pattern earlier in this file (xce_controller itself is built without + * -DXCE_TESTING, so this never ships). */ +int dispatch_a_job_test(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) +{ + return dispatch_a_job_(c, ep, alloc); +} +#endif /* XCE_TESTING */ + int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) { uint64_t jobid = ctrinfo->jobid; diff --git a/src/controller/meson.build b/src/controller/meson.build index 0464c5dc..2efb11e3 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -179,6 +179,7 @@ controller_unit_test_src = [ 'testsuite/test_job_array_fence.c', 'testsuite/test_job_array_replay_fence.c', 'testsuite/test_job_dispatch.c', + 'testsuite/test_dispatch_commit.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 cf07de74..ebf796b1 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -97,6 +97,7 @@ extern void test_upgrade_snapshot_apply(void); extern void test_bug857_regression(void); extern void test_sched_admin_dispatch(void); extern void test_job_dispatch(void); +extern void test_dispatch_commit(void); void setUp(void) { return; } @@ -263,6 +264,7 @@ int main(int argc, char* argv[]) TEST(test_bug857_regression); TEST(test_sched_admin_dispatch); TEST(test_job_dispatch); + TEST(test_dispatch_commit); return UNITY_END(); } diff --git a/src/controller/testsuite/test_dispatch_commit.c b/src/controller/testsuite/test_dispatch_commit.c new file mode 100644 index 00000000..de84be44 --- /dev/null +++ b/src/controller/testsuite/test_dispatch_commit.c @@ -0,0 +1,633 @@ +/* test_dispatch_commit.c — scheduler-dispatch-commit task 3.1 + * + * Proves dispatch_a_job_()'s (job.c) restructured commit: the job's + * resource allocation and its transition to a running state are committed + * together, ONLY once the dispatch request to the worker actually succeeds + * (Req 2.1-2.4, 5.1) -- gating a commit that used to run unconditionally + * regardless of xce_startjob_req()'s outcome. Also proves the consecutive + * retryable-failure counter (job_t.dispatch_fail_count, task 1.3) resets to + * zero on that same successful outcome (Req 4.3), and that the new durable + * WAL_JOB_DISPATCH fact (task 1.2) is appended as part of it (Req 2.4). + * + * dispatch_a_job_() is static/file-local (its only caller, handle_ + * jobdispatch(), is untouched by this task); reached here through + * dispatch_a_job_test(), an XCE_TESTING-only wrapper this task adds -- + * mirroring xce_classify_dispatch_fail_test()'s existing precedent in the + * same file (test_job_dispatch.c, task 1.1). + * + * No live peer / synthetic reply thread is needed to make a dispatch + * "succeed": xce_startjob_req() (request.c) is a genuinely non-blocking, + * fire-and-forget send -- xce_request() only capnp-encodes the request and + * endpoint_enq()s it onto the endpoint's outq, it never waits for a reply. + * endpoint_enq() succeeds unconditionally against a zeroed endpoint_t with + * no iomgr registered: dataq_enq() (dataq.c) has no failure path, and + * endpoint_update_epoll() (netio.c) is a pure no-op whenever ep->iomgr == + * NULL. This exact "zeroed, no-iomgr endpoint" pattern is already + * established and proven live by test_commit_barrier_reply.c's + * cbrp_arrange_jobdispatch()/g_wkr_disp_ep fixture (see that file's own + * header comment on its reply-observation harness) -- this file's fixtures + * mirror that queue/job/worker/alloc construction, trimmed to what + * dispatch_a_job_() itself reads (no HA commit-engine/standby-registry + * machinery, which is only needed there for that file's own, unrelated + * reply-ordering assertions). + * + * Two fixture tiers, mirroring test_wal_job_dispatch_fact.c's own split: + * - dc_setup/dc_teardown: cluster_t (jobmgr/workermgr/qmgr/hooksys) + a + * real disk-backed persist_t initialized with cluster == NULL, so + * wal_hook_append_classified() (persist.c) takes the direct + * persist_wal_async() path -- no raft/commit-engine machinery needed + * for a plain RETAIN-mode fact append. + * - the durability test builds its own two short-lived cluster_t/ + * persist_t pairs directly, mirroring test_wal_job_dispatch_fact.c's + * wjd_make_fresh_persist/restart-replay pattern exactly. + */ + +#include + +#include +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "dataq.h" +#include "hook.h" +#include "jobj.h" +#include "kvmgr.h" +#include "netio.h" +#include "persist.h" +#include "qobj.h" +#include "wal_types.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 -- see this + * file's header and job.c's own comment beside the wrapper. */ +extern int dispatch_a_job_test(cluster_t* c, endpoint_t* ep, + xce_job_alloc_t* alloc); + +/* 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 + * (test_wal_job_dispatch_fact.c's g_wjd_jobobj_ready). */ +static int g_dc_jobobj_ready; + +static cluster_t g_dc; +static char g_dc_tmpdir[256]; + +/* Worker endpoint: zeroed, no iomgr -- xce_startjob_req()'s enqueue always + * succeeds against it (see file header). File-static, not stack-local, so + * it stays valid for the whole fixture's lifetime regardless of which + * function last touched worker_t.ep (mirrors test_commit_barrier_reply.c's + * g_wkr_disp_ep). */ +static endpoint_t g_dc_worker_ep; + +static void dc_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*)"dc-probe") == NULL) + xstring_table_init(64); + + if (!g_dc_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_dc_jobobj_ready = 1; + } + + memset(&g_dc, 0, sizeof(g_dc)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_dc.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_dc.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_dc.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_dc.qmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_dc.jobmgr); + TEST_ASSERT_NOT_NULL(g_dc.workermgr); + TEST_ASSERT_NOT_NULL(g_dc.qmgr); + + /* Real disk-backed persist, cluster == NULL: wal_hook_append_ + * classified() takes the direct persist_wal_async() path -- no raft/ + * commit-engine machinery needed for a plain fact append (this exact + * routing is what lets a bare cluster_t exercise dispatch_a_job_()'s + * real persist_submit_job_fact() call without a full HA harness). */ + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + strcpy(g_dc_tmpdir, "/tmp/xce_dc31_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_dc_tmpdir)); + persist_set_workdir(&g_dc.persist, g_dc_tmpdir, "ctrl-test"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_dc.persist, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_dc.persist)); + + memset(&g_dc_worker_ep, 0, sizeof(g_dc_worker_ep)); + g_dc_worker_ep.fd = -1; +} + +static void dc_teardown(void) +{ + char cmd[512]; + + persist_stop(&g_dc.persist); + persist_fini(&g_dc.persist); + + if (g_dc.workermgr) dereference(g_dc.workermgr); + if (g_dc.qmgr) dereference(g_dc.qmgr); + if (g_dc.jobmgr) dereference(g_dc.jobmgr); + xce_hook_fini(&g_dc.hooksys); + + snprintf(cmd, sizeof(cmd), "rm -rf %s", g_dc_tmpdir); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_dc_tmpdir); +} + +/* Frees every packet left queued on an endpoint's outq (the startjob + * request xce_startjob_req() enqueued) -- mirrors test_commit_barrier_ + * reply.c's cbrp_scan_replies, trimmed: this file never needs to decode the + * packets, only reclaim them. */ +static void dc_drain_ep(endpoint_t* ep) +{ + data_item_t* it; + + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t* pkt = ITPTR_TO_PACKET(it); + free_packet(pkt); + free(pkt); + } +} + +/* Seeds a queue, a PEND job with a real submit/start_info (deep-copied by + * makeJob via makeJobsub -- dispatch_a_job_() reads submit->queue and + * submit->start_info), and a ready worker whose ep is g_dc_worker_ep. + * Mirrors test_commit_barrier_reply.c's cbrp_arrange_jobdispatch fixture, + * built directly against g_dc rather than through handle_jobdispatch(). + * Returns the seeded job_t* (borrowed -- g_dc.jobmgr owns it). */ +static job_t* dc_seed_ready(uint64_t jobid, uint32_t start_fail_count, + const char* qname, const char* wname) +{ + xce_queue_t qdef; + xce_worker_t wdef; + hostattrs_t* attrs; + job_startinfo_t startinfo; + job_submit_data_t submit; + xce_job_t jdef; + job_t* j = NULL; + worker_t* w = NULL; + + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)qname; + TEST_ASSERT_NOT_NULL(makeQueue(g_dc.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_dc.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(j); + j->dispatch_fail_count = start_fail_count; + + 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_dc.workermgr, &wdef)); + + lock(g_dc.workermgr, RTTI_LOCK_READ); + key_find(g_dc.workermgr, (void*)wname, (void**)&w); + unlock(g_dc.workermgr); + TEST_ASSERT_NOT_NULL(w); + w->ep = &g_dc_worker_ep; /* host.status defaults to 0 (ready) via calloc */ + + return j; +} + +#define DC_JOBID_SUCCESS 91001ULL + +/* THE core proof (Req 2.1, 2.3, 4.3): a successful dispatch commits + * job.allocs and transitions job.state to RUN together, and resets + * dispatch_fail_count to 0 from a nonzero starting value -- all in one + * operation, with no dependence on any later worker confirmation. */ +static void test_dc_success_commits_alloc_state_and_resets_counter(void) +{ + job_t* j; + xce_job_alloc_t* alloc; + int rc; + + dc_setup(); + + j = dc_seed_ready(DC_JOBID_SUCCESS, 3 /* nonzero starting fail count */, + "dq", "wkr-dc"); + + alloc = make_job_alloc(DC_JOBID_SUCCESS, 1, + make_host_allocs(1, + make_host_alloc((char*)"wkr-dc", 1, + make_allocs(1, + make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); + TEST_ASSERT_NOT_NULL(alloc); + + rc = dispatch_a_job_test(&g_dc, NULL, alloc); + TEST_ASSERT_TRUE_MESSAGE(rc >= 0, + "a dispatch to a live, no-iomgr worker endpoint must succeed -- " + "xce_startjob_req() is a fire-and-forget enqueue, not a blocking " + "round trip"); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, + "the job must transition to RUN in the same operation that commits " + "the allocation, unaffected by whether the worker has separately " + "confirmed job start yet (Req 2.1, 2.3)"); + TEST_ASSERT_NOT_NULL(j->job.allocs); + TEST_ASSERT_EQUAL_UINT64(DC_JOBID_SUCCESS, j->job.allocs->jobid); + TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts); + TEST_ASSERT_NOT_NULL(j->job.allocs->allocs); + TEST_ASSERT_EQUAL_STRING("wkr-dc", j->job.allocs->allocs[0]->host); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, j->dispatch_fail_count, + "a successful dispatch must reset the consecutive retryable-" + "failure counter to zero (Req 4.3)"); + + free_JobAlloc_ptr(&alloc); /* allocs already MOVE_PTR'd out -- top-level + * struct only */ + dc_drain_ep(&g_dc_worker_ep); + dc_teardown(); + printf("test_dc_success_commits_alloc_state_and_resets_counter: PASS\n"); +} + +#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(). */ +static void test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted(void) +{ + job_t* j; + xce_job_alloc_t* alloc; + int rc; + uint64_t lsn_before, lsn_after; + + dc_setup(); + + /* Seed only the job/queue -- deliberately no matching worker in + * g_dc.workermgr, so key_find() finds nothing (w stays NULL) and + * dispatch_a_job_() takes its WORKER_NULL failure branch. */ + { + job_startinfo_t startinfo; + job_submit_data_t submit; + xce_job_t jdef; + xce_queue_t qdef; + + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)"dq2"; + TEST_ASSERT_NOT_NULL(makeQueue(g_dc.qmgr, &qdef)); + + memset(&startinfo, 0, sizeof(startinfo)); + startinfo.cmdline = (char*)"/bin/true"; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"dq2"; + submit.user = (char*)"tester"; + submit.start_info = &startinfo; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = DC_JOBID_NOTFOUND; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + j = makeJob(g_dc.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(j); + j->dispatch_fail_count = 2; + } + + alloc = make_job_alloc(DC_JOBID_NOTFOUND, 1, + make_host_allocs(1, + make_host_alloc((char*)"no-such-worker", 1, + make_allocs(1, + make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); + TEST_ASSERT_NOT_NULL(alloc); + + pthread_mutex_lock(&g_dc.persist.lock); + lsn_before = g_dc.persist.current_lsn; + pthread_mutex_unlock(&g_dc.persist.lock); + + rc = dispatch_a_job_test(&g_dc, NULL, alloc); + TEST_ASSERT_TRUE_MESSAGE(rc < 0, "dispatch to a nonexistent worker must " + "fail"); + + pthread_mutex_lock(&g_dc.persist.lock); + lsn_after = g_dc.persist.current_lsn; + pthread_mutex_unlock(&g_dc.persist.lock); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state, + "a dispatch failure ahead of xce_startjob_req() must leave the job " + "state untouched (still PEND) -- this task only gates the SUCCESS " + "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_UINT64_MESSAGE(lsn_before, lsn_after, + "a failed dispatch must append no WAL entry"); + + free_JobAlloc_ptr(&alloc); + dc_teardown(); + printf("test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted: " + "PASS\n"); +} + +/* ── Durability: WAL_JOB_DISPATCH survives a restart replay ───────────── + * + * Mirrors test_wal_job_dispatch_fact.c's own restart-replay test exactly, + * except the WAL_JOB_DISPATCH fact is produced by a REAL successful + * dispatch_a_job_() call (this task's actual new code), not a hand-built + * wal_jobstate_t -- proving the fact this task's restructuring appends is + * durable and independently sufficient to recover the job as RUNNING, with + * no WAL_JOB_START anywhere in the log. */ + +static persist_t* dc_make_fresh_persist(const char* tmpdir) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + persist_config_t cfg; + + if (!p) return NULL; + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + + persist_set_workdir(p, tmpdir, "ctrl-test"); + + if (persist_init(p, &cfg, NULL) != 0) { + free(p); + return NULL; + } + if (persist_start(p) != 0) { + persist_fini(p); + free(p); + return NULL; + } + return p; +} + +static void dc_teardown_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Encode helper mirroring test_wal_job_dispatch_fact.c's wjd_encode_job + * (capnp_encode_job is file-static in persist.c, so a direct WAL_JOB_SUBMIT + * seed needs its own encoder built from the exported encode_Job/new_Job/ + * write_Job codec chain). */ +static int dc_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Job cj; + Job_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Job(cs, &cj, (xce_job_t*)j); + jp = new_Job(cs); + write_Job(&cj, jp); + capn_setp(capn_root(&c), 0, jp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +#define DC_JOBID_REPLAY 91003ULL + +static void test_dc_success_appends_wal_job_dispatch_durably(void) +{ + char tmp[256]; + + strcpy(tmp, "/tmp/xce_dc31_replay_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: real cluster + real disk persist. Seed WAL_JOB_SUBMIT + * directly (so replay can reconstruct the job), then drive an actual + * successful dispatch_a_job_() call and let IT append WAL_JOB_DISPATCH + * -- this is the production code path under test, not a hand-built + * WAL entry. */ + { + cluster_t c; + 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; + endpoint_t wep; + job_t* j; + worker_t* w = NULL; + xce_job_alloc_t* alloc; + int rc; + + if (makeString((char*)"dc-probe") == NULL) + xstring_table_init(64); + if (!g_dc_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_dc_jobobj_ready = 1; + } + + memset(&c, 0, sizeof(c)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + c.workermgr = makeInstance(&xkvmgr_type, &str_params); + c.qmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(c.jobmgr); + TEST_ASSERT_NOT_NULL(c.workermgr); + TEST_ASSERT_NOT_NULL(c.qmgr); + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + persist_set_workdir(&c.persist, tmp, "ctrl-test"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); + + /* Durable WAL_JOB_SUBMIT seed -- no WAL_JOB_START anywhere in this + * log; the recovered RUN state must come from WAL_JOB_DISPATCH + * alone. */ + { + job_submit_data_t wal_submit; + xce_job_t wal_jdef; + uint8_t* buf = NULL; + size_t sz = 0; + + memset(&wal_submit, 0, sizeof(wal_submit)); + wal_submit.queue = (char*)"dq"; + wal_submit.user = (char*)"tester"; + + memset(&wal_jdef, 0, sizeof(wal_jdef)); + wal_jdef.jobid = DC_JOBID_REPLAY; + wal_jdef.state = JOB_STAT_PEND; + wal_jdef.submit = &wal_submit; + + TEST_ASSERT_EQUAL_INT(0, dc_encode_job(&wal_jdef, &buf, &sz)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(&c.persist, WAL_JOB_SUBMIT, buf, + (uint32_t)sz)); + free(buf); + } + + /* dc_seed_ready() operates against the module-global g_dc fixture, + * not usable here (this test's cluster_t is a local `c`, and + * dc_setup()/dc_teardown() are deliberately not called in this + * test) -- so the queue/job/worker are seeded directly below. */ + { + xce_queue_t qdef; + job_startinfo_t startinfo; + job_submit_data_t submit; + xce_job_t jdef; + hostattrs_t* attrs; + xce_worker_t wdef; + + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)"dq"; + TEST_ASSERT_NOT_NULL(makeQueue(c.qmgr, &qdef)); + + memset(&startinfo, 0, sizeof(startinfo)); + startinfo.cmdline = (char*)"/bin/true"; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"dq"; + submit.user = (char*)"tester"; + submit.start_info = &startinfo; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = DC_JOBID_REPLAY; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + j = makeJob(c.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(j); + + 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*)"wkr-replay"; + wdef.attrs = attrs; + TEST_ASSERT_NOT_NULL(makeWorker(c.workermgr, &wdef)); + } + + memset(&wep, 0, sizeof(wep)); + wep.fd = -1; + lock(c.workermgr, RTTI_LOCK_READ); + key_find(c.workermgr, (void*)"wkr-replay", (void**)&w); + unlock(c.workermgr); + TEST_ASSERT_NOT_NULL(w); + w->ep = &wep; + + alloc = make_job_alloc(DC_JOBID_REPLAY, 1, + make_host_allocs(1, + make_host_alloc((char*)"wkr-replay", 1, + make_allocs(1, + make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); + TEST_ASSERT_NOT_NULL(alloc); + + rc = dispatch_a_job_test(&c, NULL, alloc); + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); + + free_JobAlloc_ptr(&alloc); + dc_drain_ep(&wep); + + persist_stop(&c.persist); + persist_fini(&c.persist); + dereference(c.workermgr); + dereference(c.qmgr); + dereference(c.jobmgr); + xce_hook_fini(&c.hooksys); + } + + /* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only -- + * apply_committed_entry()'s WAL_JOB_SUBMIT/WAL_JOB_DISPATCH cases need + * nothing else), recover from disk -- simulates a controller restart. */ + { + persist_t* p2 = dc_make_fresh_persist(tmp); + cluster_t c2; + job_t* j2 = NULL; + int rc; + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, .delta = 16 }; + + TEST_ASSERT_NOT_NULL(p2); + + memset(&c2, 0, sizeof(c2)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX)); + c2.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(c2.jobmgr); + + p2->cluster = &c2; + rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(c2.jobmgr, RTTI_LOCK_READ); + key_find(c2.jobmgr, (void*)DC_JOBID_REPLAY, (void**)&j2); + unlock(c2.jobmgr); + TEST_ASSERT_NOT_NULL_MESSAGE(j2, + "restart replay must reconstruct the job via WAL_JOB_SUBMIT"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2->job.state, + "restart replay must recover the job as RUNNING from the " + "WAL_JOB_DISPATCH fact appended by dispatch_a_job_()'s real " + "success path -- no WAL_JOB_START anywhere in this log"); + + xce_hook_fini(&c2.hooksys); + if (c2.jobmgr) dereference(c2.jobmgr); + dc_teardown_persist(p2); + } + + { + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp); + } + + printf("test_dc_success_appends_wal_job_dispatch_durably: PASS\n"); +} + +void test_dispatch_commit(void) +{ + UnitySetTestFile(__FILE__); + printf("\n>> dispatch_a_job_() commit-gating tests (Req 2.1-2.4, 4.3, " + "5.1)\n\n"); + + RUN_TEST(test_dc_success_commits_alloc_state_and_resets_counter); + RUN_TEST(test_dc_worker_not_found_leaves_job_pend_and_alloc_uncommitted); + RUN_TEST(test_dc_success_appends_wal_job_dispatch_durably); +}