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

feat(scheduler-dispatch-commit): controller commits alloc+RUN transition together on successful dispatch (task 3.1)

Browse source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Cxa1PmqAB5PyXfJaNvj8TN
This commit is contained in:
Rongsong Shen 2026-08-23 15:24:55 +00:00
parent d87f4bb333
commit d9aedc4b34
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
5 changed files with 699 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

7
.kiro/specs/scheduler-dispatch-commit/tasks.md
View file

@ -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 <group>`) 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.

51
src/controller/job.c
View file

@ -1035,6 +1035,24 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
free_launchinfo(&launchinfo);
/* 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 (j->job.allocs != NULL) {
free_JobAlloc_ptr(&(j->job.allocs));
}
@ -1044,15 +1062,44 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
MOVE_PTR(j->job.allocs->allocs, allocs);
alloc->allocs = NULL;
if (rc < 0) {
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;

1
src/controller/meson.build
View file

@ -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',

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

@ -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();
}

633
src/controller/testsuite/test_dispatch_commit.c Normal file
View file

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