/* test_worker_resync.c — worker-rejoin-resync task 2.1 * * Proves record_job_reconcile_pending_fact() (job.c) — the Requirement 4.6 * new fact-recording function that returns a worker-unrecognized job to * PEND, releasing any resource allocation it was carrying (bug-1631's own * deliberately-deferred trigger condition, closed here — see design.md's * Error Handling section) and durably appending a plain WAL_JOB_UPDATE fact * -- following dispatch_a_job_()'s eager-mutation-plus-explicit-notify * pattern (design.md "Controller: job.c — record_job_reconcile_pending_ * fact, record_job_reconcile_assign_fact"), NOT record_job_finish_fact's * deferred one. * * This task implements ONLY record_job_reconcile_pending_fact() itself -- * no live caller exists yet (dispatch_a_job_'s disconnected-worker split is * task 2.3; resync.c's report-apply glue is a later task), so every test * here directly exercises the exact sequence a real caller will perform: * lock -> call record_job_reconcile_pending_fact() -> check the returned * job_transition_result_t -> unlock -> (only when APPLIED) call * persist_notify_job_updated_already_persisted() itself -- proving the full * intended sequence end-to-end even though production wiring of that * sequence is a later task's job (design.md's Service Interface * Postconditions/Invariants: the caller, not this function, owns that * post-unlock notify call). * * Fixture mirrors test_dispatch_commit.c's dc_setup/dc_teardown exactly: a * bare cluster_t (jobmgr + hooksys) plus a real disk-backed persist_t * initialized with cluster == NULL, so wal_hook_append_classified() * (persist.c) takes the direct persist_wal_async() path for * persist_submit_job_fact()'s plain RETAIN-mode WAL_JOB_UPDATE append -- no * raft/commit-engine machinery needed. p->current_lsn before/after each * call proves (or disproves) a durable append happened, mirroring * test_dispatch_commit.c's test_dc_worker_not_found_leaves_job_pend_and_ * alloc_uncommitted's own LSN-delta idiom. */ #include #include #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 "persist.h" #include "resync.h" /* worker-rejoin-resync task 3.1 */ #include "wal_types.h" #include "wobj.h" /* worker-rejoin-resync task 3.1: makeWorker, for a real * RTTI-managed worker_t the timeout test can genuinely * lock() */ #include "xce/cluster_def.h" #include "xce/types.h" #include "xcealloc.h" #include "xstring.h" /* 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_twr_jobobj_ready; static cluster_t g_twr; static char g_twr_tmpdir[256]; /* worker-rejoin-resync task 3.1: no-op kvmgr datakiller for g_twr.workermgr, * mirroring production's own worker_obj_killer (cluster.c) -- a worker_t, * once created, is never explicitly torn down for the life of the process * (no key_del(c->workermgr, ...) call site exists anywhere in this * codebase), so there is nothing for this to actually free. */ static void twr_worker_obj_killer(void* arg) { (void)arg; } static void twr_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; persist_config_t cfg; if (makeString((char*)"twr-probe") == NULL) xstring_table_init(64); if (!g_twr_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_twr_jobobj_ready = 1; } memset(&g_twr, 0, sizeof(g_twr)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_twr.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_twr.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_twr.jobmgr); /* worker-rejoin-resync task 3.1: a real workermgr so the round-lifecycle * tests below can create a genuine RTTI-managed worker_t (via * makeWorker) where that matters (the timeout test, whose internal * resync_on_timeout callback genuinely lock()s `w`) -- key type mirrors * production's own workermgr_params (cluster.c, KVMGR_KEY_STRING, * keyed by hostname). */ { kvmgr_params_t workermgr_params = { .key_type = KVMGR_KEY_STRING, .nslots = 32, .delta = 16, .datakiller = twr_worker_obj_killer }; g_twr.workermgr = makeInstance(&xkvmgr_type, &workermgr_params); TEST_ASSERT_NOT_NULL(g_twr.workermgr); } /* Real disk-backed persist, cluster == NULL: 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 (mirrors test_dispatch_commit.c's dc_setup exactly, which * exercises dispatch_a_job_'s own real persist_submit_job_fact()-family * call the same way). */ cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; strcpy(g_twr_tmpdir, "/tmp/xce_twr21_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(g_twr_tmpdir)); persist_set_workdir(&g_twr.persist, g_twr_tmpdir, "ctrl-test"); TEST_ASSERT_EQUAL_INT(0, persist_init(&g_twr.persist, &cfg, NULL)); TEST_ASSERT_EQUAL_INT(0, persist_start(&g_twr.persist)); } static void twr_teardown(void) { char cmd[512]; persist_stop(&g_twr.persist); persist_fini(&g_twr.persist); if (g_twr.jobmgr) dereference(g_twr.jobmgr); if (g_twr.workermgr) dereference(g_twr.workermgr); xce_hook_fini(&g_twr.hooksys); snprintf(cmd, sizeof(cmd), "rm -rf %s", g_twr_tmpdir); if (system(cmd) != 0) fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_twr_tmpdir); } /* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay * entirely -- for seeding fact-transition tests, mirroring * test_wal_job_dispatch_fact.c's wjd_seed_job / test_dispatch_commit.c's * dc_seed_ready. */ static job_submit_data_t g_twr_seed_submit; static job_t* twr_seed_job(uint64_t jobid, uint32_t state) { xce_job_t jdef; memset(&g_twr_seed_submit, 0, sizeof(g_twr_seed_submit)); g_twr_seed_submit.queue = (char*)"q"; g_twr_seed_submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.submit = &g_twr_seed_submit; return makeJob(g_twr.jobmgr, &jdef); } /* One host ("twr-worker"), one resource ("slots", value 1.0) -- the same * minimal shape test_wal_job_dispatch_fact.c's/test_dispatch_commit.c's own * dispatch-alloc tests use. Caller owns the result (release via * free_JobAlloc_ptr, or hand ownership to job->job.allocs). */ static xce_job_alloc_t* twr_make_test_alloc(uint64_t jobid) { return make_job_alloc(jobid, 1, make_host_allocs(1, make_host_alloc((char*)"twr-worker", 1, make_allocs(1, make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); } static uint64_t twr_current_lsn(void) { uint64_t lsn; pthread_mutex_lock(&g_twr.persist.lock); lsn = g_twr.persist.current_lsn; pthread_mutex_unlock(&g_twr.persist.lock); return lsn; } /* Spy hook: counts HOOK_AFTER_EVENT/HOOK_DETAIL_JOB_UPDATE fires -- the * exact hook sched_job_update_fun/apiserver_job_update_fun subscribe to in * production, mirroring test_dispatch_commit.c's dc_job_update_spy_hook. */ static int g_twr_update_hook_fires; static int twr_job_update_spy_hook(int type, int situation, int detail, void* data, va_list ap) { (void)type; (void)data; (void)ap; if (situation == HOOK_AFTER_EVENT && detail == HOOK_DETAIL_JOB_UPDATE) g_twr_update_hook_fires++; return 0; } #define TWR_JOBID_APPLIED 81001ULL #define TWR_JOBID_NOOP 81002ULL #define TWR_JOBID_TERMINAL 81003ULL /* THE core proof (Req 4.6, 7.1-7.3; bug-1631 regression): a job that was * dispatched (real allocs committed, state RUN) and reconciles to * unrecognized ends the call -- synchronously, BEFORE any commit round-trip, * matching dispatch_a_job_'s own eager pattern, NOT record_job_finish_ * fact's deferred one -- with PEND state and no allocation, a durably * appended fact, and (once the caller unlocks and notifies, exactly as a * real caller must) the scheduler's notify path fired exactly once. */ static void test_reconcile_pending_applies_releases_allocs_and_notifies(void) { job_t* j; xce_job_alloc_t* alloc; int rc; uint64_t lsn_before, lsn_after; twr_setup(); g_twr_update_hook_fires = 0; TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB, twr_job_update_spy_hook, NULL)); j = twr_seed_job(TWR_JOBID_APPLIED, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); alloc = twr_make_test_alloc(TWR_JOBID_APPLIED); TEST_ASSERT_NOT_NULL(alloc); j->job.allocs = alloc; /* simulate dispatch_a_job_'s own committed alloc, * ownership transferred to job->job.allocs. */ lsn_before = twr_current_lsn(); lock(j, RTTI_LOCK_WRITE); rc = record_job_reconcile_pending_fact(&g_twr, j); /* Observable completion (design.md): PEND + no allocation, asserted * IMMEDIATELY after the call returns, still under the write lock, * before any commit round-trip -- the mutation is eager, not * deferred. */ TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, "an unrecognized RUN job must apply the PEND transition"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state, "job must be PEND synchronously, before any WAL commit/apply " "round trip"); TEST_ASSERT_NULL_MESSAGE(j->job.allocs, "bug-1631: a job reconciled from RUN back to PEND must release its " "resource allocation immediately, so a later redispatch never " "inherits a stale allocation from the prior dispatch"); unlock(j); lsn_after = twr_current_lsn(); TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before, "record_job_reconcile_pending_fact must durably append a " "WAL_JOB_UPDATE fact for an applied transition"); /* Caller's responsibility, AFTER unlocking (design.md Postconditions/ * Invariants): notify the scheduler through the existing notify path, * exactly once, for this actual mutation (Req 7.1-7.3). */ persist_notify_job_updated_already_persisted(&g_twr, j); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_twr_update_hook_fires, "the scheduler notify path must fire exactly once for this " "reconciled pending transition (Req 7.2) when the caller notifies " "after unlocking -- omitting it would reproduce bug-1679 at this " "new call site"); xce_disable_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB, twr_job_update_spy_hook); twr_teardown(); printf("test_reconcile_pending_applies_releases_allocs_and_notifies: " "PASS\n"); } /* A job with no allocation at all (e.g. never dispatched, already PEND): * reconciling it again must be a no-op per apply_job_update_transition's * own duplicate-state guard (JOB_TRANSITION_NOOP_DUPLICATE) -- no mutation, * no allocs touch (there is none to release), no durable fact, no crash * (Req 4.3). */ static void test_reconcile_pending_noop_on_already_pending_job(void) { job_t* j; int rc; uint64_t lsn_before, lsn_after; twr_setup(); j = twr_seed_job(TWR_JOBID_NOOP, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL(j->job.allocs); lsn_before = twr_current_lsn(); lock(j, RTTI_LOCK_WRITE); rc = record_job_reconcile_pending_fact(&g_twr, j); TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_NOOP_DUPLICATE, rc, "reconciling an already-PEND job must be treated as a duplicate " "no-op, matching apply_job_update_transition's own guard"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); TEST_ASSERT_NULL_MESSAGE(j->job.allocs, "a job that never held an allocation must still have none after a " "no-op reconcile"); unlock(j); lsn_after = twr_current_lsn(); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after, "a no-op reconcile must append no duplicate WAL fact (Req 4.3)"); twr_teardown(); printf("test_reconcile_pending_noop_on_already_pending_job: PASS\n"); } /* A job the controller already knows is terminal (DONE) must never be * clobbered back to PEND by a reconciliation report (Req 4.4) -- the * existing job_state_is_backward guard inside apply_job_update_transition, * reused unmodified, rejects it. */ static void test_reconcile_pending_rejects_terminal_job(void) { job_t* j; int rc; uint64_t lsn_before, lsn_after; twr_setup(); j = twr_seed_job(TWR_JOBID_TERMINAL, JOB_STAT_DONE); TEST_ASSERT_NOT_NULL(j); lsn_before = twr_current_lsn(); lock(j, RTTI_LOCK_WRITE); rc = record_job_reconcile_pending_fact(&g_twr, j); TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_REJECTED_BACKWARD, rc, "a terminal job must reject the backward PEND transition, never " "clobbering more-final state (Req 4.4)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, "the job's terminal state must be preserved exactly"); unlock(j); lsn_after = twr_current_lsn(); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after, "a rejected reconcile must append no WAL fact"); twr_teardown(); printf("test_reconcile_pending_rejects_terminal_job: PASS\n"); } /* ── record_job_reconcile_assign_fact() tests (task 2.2) ───────────────── * * Proves record_job_reconcile_assign_fact() (job.c) -- the Requirement 4.8 * new fact-recording function that records a job the worker reports running * that the controller has no existing non-terminal record of ("unqueried * extra", Req 3.6) as assigned to the reporting worker, through the same * dispatch_a_job_()-style eager-mutation-plus-explicit-notify pattern as * record_job_reconcile_pending_fact above, reusing bug-1678's NULL-alloc * persist_submit_job_dispatch_fact() path (design.md "Controller: job.c -- * record_job_reconcile_pending_fact, record_job_reconcile_assign_fact"). * * As with task 2.1, no live caller exists yet (resync.c's report-apply glue * is a later task), so every test here directly exercises the exact * sequence a real caller will perform: lock -> call * record_job_reconcile_assign_fact() -> check the result -> unlock -> call * persist_notify_job_updated_already_persisted() itself. */ #define TWR_JOBID_ASSIGN 81004ULL #define TWR_JOBID_ASSIGN_REPLAY 81005ULL /* THE core proof (Req 4.1, 4.3, 4.4, 4.7, 4.8, 7.1-7.3): a job the * controller never previously tracked, reported RUNNING by a worker, ends * the call -- synchronously, before any commit round-trip -- recorded as * running with no resource allocation (the controller genuinely does not * know what the scheduler originally computed, design.md's "Documented * scope decision"), a durably appended fact, and (once the caller unlocks * and notifies, exactly as a real caller must) the scheduler's notify path * fired exactly once. */ static void test_reconcile_assign_applies_running_no_allocs_and_notifies(void) { job_t* j; worker_t w; xce_job_state_entry_t report; int rc; uint64_t lsn_before, lsn_after; twr_setup(); g_twr_update_hook_fires = 0; TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB, twr_job_update_spy_hook, NULL)); /* Seeded with no allocation, mirroring a job the caller (task 3.2, not * yet implemented) constructed fresh for a jobid the controller had * never seen before -- there is nothing to release here, only * something new to record. */ j = twr_seed_job(TWR_JOBID_ASSIGN, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL(j->job.allocs); memset(&w, 0, sizeof(w)); w.host.name = (char*)"twr-worker"; report = (xce_job_state_entry_t){ .jobid = TWR_JOBID_ASSIGN, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 0 }; lsn_before = twr_current_lsn(); lock(j, RTTI_LOCK_WRITE); rc = record_job_reconcile_assign_fact(&g_twr, j, &w, &report); /* Observable completion (design.md / this task): RUN + no allocation, * asserted IMMEDIATELY after the call returns, still under the write * lock, before any commit round-trip -- the mutation is eager, not * deferred. */ TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, "an unqueried running job must always apply (no reject/no-op case)"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, "job must be RUN synchronously, before any WAL commit/apply round " "trip"); TEST_ASSERT_NULL_MESSAGE(j->job.allocs, "Req 4.8: a reconciliation-assigned job must never carry a " "fabricated resource allocation -- the controller genuinely does " "not know what the scheduler originally computed for it"); unlock(j); lsn_after = twr_current_lsn(); TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before, "record_job_reconcile_assign_fact must durably append a " "WAL_JOB_DISPATCH fact"); /* Caller's responsibility, AFTER unlocking (design.md Postconditions/ * Invariants): notify the scheduler through the existing notify path, * exactly once, for this actual mutation (Req 7.1-7.3). */ persist_notify_job_updated_already_persisted(&g_twr, j); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_twr_update_hook_fires, "the scheduler notify path must fire exactly once for this " "reconciled assignment (Req 7.2) when the caller notifies after " "unlocking -- omitting it would reproduce bug-1679 at this new " "call site"); xce_disable_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB, twr_job_update_spy_hook); twr_teardown(); printf("test_reconcile_assign_applies_running_no_allocs_and_notifies: " "PASS\n"); } /* Encode helper mirroring test_dispatch_commit.c's dc_encode_job / * 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 twr_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; } /* Fresh disk-backed persist_t pointed at an existing tmpdir -- mirrors * test_dispatch_commit.c's dc_make_fresh_persist/dc_teardown_persist * exactly, for this test's own Phase 2 (simulated restart). */ static persist_t* twr_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 twr_teardown_persist(persist_t* p) { persist_stop(p); persist_fini(p); free(p); } /* Restart-replay round trip -- this task's own explicit completion * criterion ("durably survives a restart replay with the same result"), * mirroring test_dispatch_commit.c's test_dc_success_appends_wal_job_ * dispatch_durably's Phase 1/Phase 2 shape exactly, but driving Phase 1's * WAL_JOB_DISPATCH fact via record_job_reconcile_assign_fact() instead of * dispatch_a_job_() -- proving bug-1678's NULL-alloc decode branch (already * tested on both apply seams by test_wal_job_dispatch_fact.c) round-trips * this new call site's real fact correctly too. Uses its own local * cluster_t/persist_t (not the shared g_twr fixture) so Phase 1's disk * state survives past that phase's own teardown, exactly like dc's * original. */ static void test_reconcile_assign_survives_restart_replay(void) { char tmp[256]; uint64_t captured_start_time = 0; strcpy(tmp, "/tmp/xce_twr22_replay_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); /* Phase 1: real cluster + real disk persist. Seed a durable * WAL_JOB_SUBMIT fact directly (so replay can reconstruct the job -- * this job was never dispatched through any controller, so there is no * other WAL_JOB_SUBMIT/DISPATCH in this log), seed the in-memory job_t * the same way twr_seed_job() does (bypassing dispatch/replay -- * record_job_reconcile_assign_fact() operates on a job_t* the caller * has already located/constructed, not this function's concern), then * drive the real function under test. */ { cluster_t c; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; persist_config_t cfg; job_t* j; worker_t w; xce_job_state_entry_t report; int rc; if (makeString((char*)"twr22-probe") == NULL) xstring_table_init(64); if (!g_twr_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_twr_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); TEST_ASSERT_NOT_NULL(c.jobmgr); 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)); { 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*)"q"; wal_submit.user = (char*)"tester"; memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = TWR_JOBID_ASSIGN_REPLAY; wal_jdef.state = JOB_STAT_PEND; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, twr_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); } { job_submit_data_t submit; xce_job_t jdef; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = TWR_JOBID_ASSIGN_REPLAY; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; j = makeJob(c.jobmgr, &jdef); TEST_ASSERT_NOT_NULL(j); } memset(&w, 0, sizeof(w)); w.host.name = (char*)"twr-worker-replay"; report = (xce_job_state_entry_t){ .jobid = TWR_JOBID_ASSIGN_REPLAY, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 0 }; lock(j, RTTI_LOCK_WRITE); rc = record_job_reconcile_assign_fact(&c, j, &w, &report); TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); TEST_ASSERT_NULL(j->job.allocs); captured_start_time = (uint64_t)j->job.start_time; TEST_ASSERT_NOT_EQUAL(0, captured_start_time); unlock(j); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); persist_stop(&c.persist); persist_fini(&c.persist); dereference(c.jobmgr); xce_hook_fini(&c.hooksys); } /* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only), * recover from disk -- simulates a controller restart. */ { persist_t* p2 = twr_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*)TWR_JOBID_ASSIGN_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_UINT64_MESSAGE(TWR_JOBID_ASSIGN_REPLAY, j2->job.jobid, "recovered job must be the same jobid"); 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 record_job_reconcile_" "assign_fact()'s real call (Req 4.8) -- no WAL_JOB_START " "anywhere in this log"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(captured_start_time, (uint64_t)j2->job.start_time, "restart replay must recover the exact dispatch-time " "start_time stamped by the real call in Phase 1"); TEST_ASSERT_NULL_MESSAGE(j2->job.allocs, "Req 4.8 / design.md's documented scope decision: replay must " "NOT fabricate a resource allocation for a reconciliation-" "assigned job -- allocs stays NULL exactly as it was set " "eagerly in Phase 1, exercising bug-1678's NULL-alloc decode " "branch"); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); twr_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_reconcile_assign_survives_restart_replay: PASS\n"); } /* ── resync.c round lifecycle tests (task 3.1) ─────────────────────────── * * Proves resync_arm_round()/resync_resolve_round()/resync_round_is_current() * (resync.c) -- the per-worker job-state reconciliation round lifecycle: * query-set construction (Req 1.1, 2.1, 2.2), a fresh monotonic round token * per arm distinguishing this round from any prior one for the same worker * (Req 6.1, 6.3), the XCE_WORKER_RECONCILING status bit (Req 5.1, 5.2), and * the Req 5.3 bounded timeout resolving the round without any blocking * call. Reuses the g_twr/twr_setup/twr_teardown fixture and twr_seed_job() * from task 2.1/2.2's tests above -- resync_arm_round only needs a * cluster_t with a real c->jobmgr (which g_twr already provides). * * No live caller exists yet (worker.c's rejoin_worker/apply_join_state * wiring is task 6.1; the report-apply glue that would call * resync_round_is_current for real is task 3.2) -- every test here calls * resync_arm_round/resync_resolve_round/resync_round_is_current directly, * exactly as a real caller (task 6.1) eventually will. * * resync_arm_round/resync_resolve_round/resync_round_is_current never * lock/unlock `w` themselves (their documented precondition is that the * CALLER already holds it) -- so a bare stack worker_t is sufficient and * safe for those tests, exactly mirroring how task 2.2's own * record_job_reconcile_assign_fact tests above already use a bare stack * `worker_t w` with no locking. The ONE exception is the timeout test: * resync_on_timeout (the internal delaymgr callback resync_test_fire_ * timeout drives directly) runs with no caller-held lock -- because a real * delaymgr fire has none -- so it genuinely calls lock()/unlock() on `w` * itself, which requires a real RTTI-managed worker_t (via makeWorker * against g_twr.workermgr), not a bare stack struct. */ #define TWR_JOBID_RESYNC_A 81101ULL #define TWR_JOBID_RESYNC_B 81102ULL #define TWR_JOBID_RESYNC_OTHER 81103ULL #define TWR_JOBID_RESYNC_DONE 81104ULL #define TWR_JOBID_RESYNC_NOALLOC 81105ULL #define TWR_JOBID_RESYNC_SOLO 81106ULL #define TWR_JOBID_RESYNC_TIMEOUT 81107ULL /* Same builder chain as twr_make_test_alloc() above, parameterized by host * -- twr_make_test_alloc() itself hardcodes "twr-worker", which does not * fit these tests' need for several DISTINCT hosts within one job table. */ static xce_job_alloc_t* twr_make_test_alloc_for_host(uint64_t jobid, const char* host) { return make_job_alloc(jobid, 1, make_host_allocs(1, make_host_alloc((char*)host, 1, make_allocs(1, make_alloc((char*)"slots", NULL, 0, 1.0f, 0))))); } /* True iff `jobid` appears anywhere in ids[0..n) -- used below to prove the * query set contains exactly the expected members, neither missing a real * match nor carrying an extra one. */ static int twr_ids_contains(const uint64_t* ids, int n, uint64_t jobid) { int i; for (i = 0; i < n; i++) { if (ids[i] == jobid) return 1; } return 0; } /* A real RTTI-managed worker_t (via makeWorker against g_twr.workermgr), * for the one test below that needs genuine lock()/unlock() to succeed on * `w` -- mirrors worker.c's own init_worker() construction * (`tmp.name = (char*)host; tmp.attrs = attrs; w = makeWorker(c->workermgr, * &tmp);`). */ static worker_t* twr_make_test_worker(const char* hostname) { hostattrs_t attrs; xce_worker_t tmp; memset(&attrs, 0, sizeof(attrs)); memset(&tmp, 0, sizeof(tmp)); tmp.name = (char*)hostname; tmp.attrs = &attrs; return makeWorker(g_twr.workermgr, &tmp); } /* Req 1.1, 2.1, 2.2: the query set matches EXACTLY the non-terminal jobs * assigned to the given host -- not fewer (a genuine match omitted), not * more (a different host's job, a terminal job on the SAME host, or a * never-dispatched job with no host assignment at all, leaking in). */ static void test_resync_arm_round_query_set_matches_exactly(void) { job_t* ja; job_t* jb; job_t* jother; job_t* jdone; job_t* jnoalloc; worker_t w; int rc; twr_setup(); ja = twr_seed_job(TWR_JOBID_RESYNC_A, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(ja); ja->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_A, "resync-host-A"); jb = twr_seed_job(TWR_JOBID_RESYNC_B, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(jb); jb->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_B, "resync-host-A"); /* Same non-terminal state, DIFFERENT host -- must NOT appear in * resync-host-A's query set. */ jother = twr_seed_job(TWR_JOBID_RESYNC_OTHER, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(jother); jother->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_OTHER, "resync-host-B"); /* SAME host, but terminal (DONE) -- must NOT appear (Req 1.1's * "non-terminal" qualifier). */ jdone = twr_seed_job(TWR_JOBID_RESYNC_DONE, JOB_STAT_DONE); TEST_ASSERT_NOT_NULL(jdone); jdone->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_DONE, "resync-host-A"); /* Never dispatched (no allocation at all) -- not assigned to ANY host, * so must NOT appear regardless of state. */ jnoalloc = twr_seed_job(TWR_JOBID_RESYNC_NOALLOC, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(jnoalloc); TEST_ASSERT_NULL(jnoalloc->job.allocs); memset(&w, 0, sizeof(w)); w.host.name = (char*)"resync-host-A"; rc = resync_arm_round(&g_twr, &w); TEST_ASSERT_EQUAL_INT_MESSAGE(2, rc, "exactly the 2 non-terminal resync-host-A jobs must be armed into " "the round"); TEST_ASSERT_NOT_NULL(w.resync); TEST_ASSERT_EQUAL_INT(2, w.resync->n_queried_jobids); TEST_ASSERT_TRUE_MESSAGE( twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, TWR_JOBID_RESYNC_A), "job A (non-terminal, resync-host-A) must be in the query set"); TEST_ASSERT_TRUE_MESSAGE( twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, TWR_JOBID_RESYNC_B), "job B (non-terminal, resync-host-A) must be in the query set"); TEST_ASSERT_FALSE_MESSAGE( twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, TWR_JOBID_RESYNC_OTHER), "a job assigned to a DIFFERENT host must not appear"); TEST_ASSERT_FALSE_MESSAGE( twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, TWR_JOBID_RESYNC_DONE), "a TERMINAL job on the same host must not appear (Req 1.1)"); TEST_ASSERT_FALSE_MESSAGE( twr_ids_contains(w.resync->queried_jobids, w.resync->n_queried_jobids, TWR_JOBID_RESYNC_NOALLOC), "a never-dispatched job (no host assignment at all) must not " "appear"); TEST_ASSERT_TRUE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), "XCE_WORKER_RECONCILING must be set once a round is armed"); resync_resolve_round(&g_twr, &w); TEST_ASSERT_NULL(w.resync); TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w.host.status)); twr_teardown(); printf("test_resync_arm_round_query_set_matches_exactly: PASS\n"); } /* Req 1.2: a worker with zero non-terminal jobs assigned is a documented * no-op -- no round armed, no status bit set. */ static void test_resync_arm_round_no_jobs_is_noop(void) { worker_t w; int rc; twr_setup(); memset(&w, 0, sizeof(w)); w.host.name = (char*)"resync-host-empty"; rc = resync_arm_round(&g_twr, &w); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "a worker with no non-terminal jobs must arm zero jobs (Req 1.2)"); TEST_ASSERT_NULL_MESSAGE(w.resync, "no resync_state_t must be attached for the Req 1.2 no-op case"); TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), "XCE_WORKER_RECONCILING must not be set for the Req 1.2 no-op case"); twr_teardown(); printf("test_resync_arm_round_no_jobs_is_noop: PASS\n"); } /* Req 6.1, 6.3: arming a round for a worker that already carries one * (simulating a disconnect interrupting an outstanding round, then a * rejoin) discards the prior round and produces a FRESH, DIFFERENT token -- * and the original token is provably no longer current once superseded. */ static void test_resync_arm_round_fresh_token_supersedes_prior(void) { job_t* j; worker_t w; int rc; uint64_t token1; uint64_t token2; twr_setup(); j = twr_seed_job(TWR_JOBID_RESYNC_SOLO, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_SOLO, "resync-host-solo"); memset(&w, 0, sizeof(w)); w.host.name = (char*)"resync-host-solo"; rc = resync_arm_round(&g_twr, &w); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_NOT_NULL(w.resync); token1 = w.resync->round_token; TEST_ASSERT_TRUE_MESSAGE(resync_round_is_current(&w, token1), "the round just armed must report its own token as current"); /* Simulated disconnect-then-rejoin: the round never resolved, but the * worker rejoins and is armed again. */ rc = resync_arm_round(&g_twr, &w); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_NOT_NULL(w.resync); token2 = w.resync->round_token; TEST_ASSERT_NOT_EQUAL_MESSAGE(token1, token2, "re-arming must produce a token distinct from the prior round's " "(Req 6.1, 6.3)"); TEST_ASSERT_FALSE_MESSAGE(resync_round_is_current(&w, token1), "the ORIGINAL token must now report stale/superseded (Req 6.3) -- " "the mechanism a later report-apply caller uses to discard a " "report carrying it"); TEST_ASSERT_TRUE_MESSAGE(resync_round_is_current(&w, token2), "the NEW token must report current"); resync_resolve_round(&g_twr, &w); twr_teardown(); printf("test_resync_arm_round_fresh_token_supersedes_prior: PASS\n"); } /* Req 5.1, 5.2: XCE_WORKER_RECONCILING is set once a round is armed and * cleared once resync_resolve_round runs -- resync_round_is_current also * correctly reports the resolved round as no longer current. */ static void test_resync_reconciling_flag_set_after_arm_cleared_after_resolve( void) { job_t* j; worker_t w; int rc; uint64_t token; twr_setup(); j = twr_seed_job(TWR_JOBID_RESYNC_SOLO, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_SOLO, "resync-host-flag"); memset(&w, 0, sizeof(w)); w.host.name = (char*)"resync-host-flag"; TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w.host.status)); rc = resync_arm_round(&g_twr, &w); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_TRUE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), "XCE_WORKER_RECONCILING must be set once armed (Req 5.1)"); token = w.resync->round_token; resync_resolve_round(&g_twr, &w); TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w.host.status), "XCE_WORKER_RECONCILING must clear once resolved (Req 5.2)"); TEST_ASSERT_NULL_MESSAGE(w.resync, "resync_resolve_round must detach/free w->resync"); TEST_ASSERT_FALSE_MESSAGE(resync_round_is_current(&w, token), "a resolved round's token must no longer report current"); /* Idempotent: resolving an already-resolved (NULL) round is a safe * no-op, never a double-free. */ resync_resolve_round(&g_twr, &w); TEST_ASSERT_NULL(w.resync); twr_teardown(); printf("test_resync_reconciling_flag_set_after_arm_cleared_after_" "resolve: PASS\n"); } /* Req 5.3: the timeout resolves an outstanding round -- and does so via a * plain, synchronous function call with no sleep/wait of any kind, proving * the resolution itself never blocks (ha-async-io.md's delaymgr * thread-role contract). g_twr never starts a real delaymgr (matching * every other bare-cluster_t fixture in this file), so resync_arm_round's * own `!c->delaymgr.running` guard already took the no-real-timer branch * when the round above was armed; resync_test_fire_timeout below drives * the EXACT SAME internal resolution logic * (resync_on_timeout) a real timer firing would have. */ static void test_resync_timeout_resolves_round_without_blocking(void) { job_t* j; worker_t* w; int rc; uint64_t token; twr_setup(); j = twr_seed_job(TWR_JOBID_RESYNC_TIMEOUT, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_RESYNC_TIMEOUT, "resync-host-timeout"); w = twr_make_test_worker("resync-host-timeout"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_NOT_NULL(w->resync); TEST_ASSERT_TRUE(IS_WORKER_RECONCILING(w->host.status)); token = w->resync->round_token; /* resync_on_timeout (invoked via resync_test_fire_timeout) locks/ * unlocks `w` itself -- this call must NOT be wrapped in an external * lock(w, ...), or it would deadlock against `w`'s own non-recursive * write lock. */ resync_test_fire_timeout(&g_twr, w, token); TEST_ASSERT_NULL_MESSAGE(w->resync, "the timeout must resolve (free/detach) the round"); TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), "the timeout must clear XCE_WORKER_RECONCILING so the worker is " "not withheld from dispatch indefinitely (Req 5.1, 5.2, 5.3)"); /* A second, late fire for the SAME (now-stale) token must be a safe * no-op -- never a double-free/double-clear against an already- * resolved round. This is exactly the round-token staleness check * (Req 6.3) protecting against a real delaymgr task that could not be * dequeued/cancelled (delay.c exposes no cancel primitive) firing late * against a resync_state_t that is already gone or has already been * replaced by a newer round. */ resync_test_fire_timeout(&g_twr, w, token); TEST_ASSERT_NULL(w->resync); TEST_ASSERT_FALSE(IS_WORKER_RECONCILING(w->host.status)); twr_teardown(); printf("test_resync_timeout_resolves_round_without_blocking: PASS\n"); } /* ── resync_apply_report() tests (task 3.2) ────────────────────────────── * * Proves resync_apply_report() (resync.c) -- the report-apply glue wiring * together task 2.1's record_job_reconcile_pending_fact, task 2.2's * record_job_reconcile_assign_fact, and task 3.1's resync_arm_round/ * resync_resolve_round/resync_round_is_current. Every test here uses a real * RTTI-managed worker_t (via twr_make_test_worker), since resync_apply_ * report locks/unlocks `w` itself (unlike resync_arm_round/resync_resolve_ * round/resync_round_is_current above, which require the caller to * pre-hold `w`'s lock and so tolerate a bare stack struct). */ #define TWR_JOBID_REPORT_FINOK 81108ULL #define TWR_JOBID_REPORT_FINFAIL 81109ULL #define TWR_JOBID_REPORT_RUNNING 81110ULL #define TWR_JOBID_REPORT_UNREC 81111ULL #define TWR_JOBID_REPORT_EXTRA 81112ULL #define TWR_JOBID_REPORT_DUP_A 81113ULL #define TWR_JOBID_REPORT_DUP_B 81114ULL #define TWR_JOBID_REPORT_STALE 81115ULL #define TWR_JOBID_REPORT_PARTIAL_A 81116ULL #define TWR_JOBID_REPORT_PARTIAL_B 81117ULL #define TWR_JOBID_REPORT_EXTRANOOP_Q 81118ULL #define TWR_JOBID_REPORT_EXTRANOOP_E 81119ULL #define TWR_JOBID_REPORT_FINISHTIME_REPLAY 81120ULL #define TWR_JOBID_REPORT_LOSTDISPATCH_Q 81121ULL #define TWR_JOBID_REPORT_LOSTDISPATCH 81122ULL /* THE core proof (Req 4.1-4.5, 6.1): a single report covering all five * distinct outcome kinds in one pass -- finished-success, finished-failure, * still-running (queried), unrecognized (queried), and an unqueried extra -- * leaves the job table in the correct state for each, routed through the * correct existing/new fact function, and resolves the round (outstanding * set empties -> XCE_WORKER_RECONCILING clears) once every QUERIED entry * has been answered. * * The unqueried-extra entry here is seeded TERMINAL (DONE), exercising * resync_apply_running_entry()'s branch (a) -- ALWAYS a no-op, regardless * of host (Req 4.4: a terminal job is never assign-eligible). This is * deliberate, not an oversight: round-2 review caught that this test * ORIGINALLY seeded j_extra TERMINAL and then asserted it became RUN -- * i.e. it baked in, and certified as PASSING, the exact Req 4.4 regression * a since-corrected two-way guard introduced (calling record_job_ * reconcile_assign_fact() on an already-terminal job, unconditionally * clobbering DONE/EXIT back to RUN). Branch (c) -- non-terminal AND * genuinely unassigned, the real "gets assigned" case -- is proven * separately, with full restart-replay rigor, by test_apply_report_ * unqueried_extra_unassigned_pend_job_gets_assigned below; branch (b) -- * non-terminal but already assigned (to this host or elsewhere) -- by * test_apply_report_unqueried_extra_already_tracked_is_noop. */ static void test_apply_report_all_five_outcomes_updates_job_table(void) { job_t* j_finok; job_t* j_finfail; job_t* j_running; job_t* j_unrec; job_t* j_extra; worker_t* w; int rc; uint64_t token; uint64_t lsn_before, lsn_after; xce_job_state_entry_t e_finok, e_finfail, e_running, e_unrec, e_extra; xce_job_state_entry_t* entries[5]; xce_job_state_report_t report; twr_setup(); j_finok = twr_seed_job(TWR_JOBID_REPORT_FINOK, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_finok); j_finok->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_FINOK, "resync-report-host"); j_finfail = twr_seed_job(TWR_JOBID_REPORT_FINFAIL, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_finfail); j_finfail->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_FINFAIL, "resync-report-host"); j_running = twr_seed_job(TWR_JOBID_REPORT_RUNNING, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_running); j_running->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_RUNNING, "resync-report-host"); j_unrec = twr_seed_job(TWR_JOBID_REPORT_UNREC, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_unrec); j_unrec->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_UNREC, "resync-report-host"); /* Seeded TERMINAL (DONE), so it is NOT part of the round's query set * (arm_round only collects non-terminal jobs), AND -- per Req 4.4 -- * must be left COMPLETELY untouched when later reported running as an * unqueried extra: a terminal record is never assign-eligible, full * stop, regardless of host (resync_apply_running_entry()'s branch (a), * resync.c). */ j_extra = twr_seed_job(TWR_JOBID_REPORT_EXTRA, JOB_STAT_DONE); TEST_ASSERT_NOT_NULL(j_extra); TEST_ASSERT_NULL(j_extra->job.allocs); w = twr_make_test_worker("resync-report-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT_MESSAGE(4, rc, "exactly the 4 non-terminal jobs on this host must be armed"); TEST_ASSERT_NOT_NULL(w->resync); token = w->resync->round_token; e_finok = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_FINOK, .status = JobStateStatus_finishedOk, .exit_code = 0, .finish_time_ms = 1000, .queried = 1 }; e_finfail = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_FINFAIL, .status = JobStateStatus_finishedFail, .exit_code = 7, .finish_time_ms = 2000, .queried = 1 }; e_running = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_RUNNING, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; e_unrec = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_UNREC, .status = JobStateStatus_unrecognized, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; e_extra = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_EXTRA, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 0 }; entries[0] = &e_finok; entries[1] = &e_finfail; entries[2] = &e_running; entries[3] = &e_unrec; entries[4] = &e_extra; report = (xce_job_state_report_t){ .n_total = 5, .pos = 0, .count = 5, .entries = entries }; lsn_before = twr_current_lsn(); rc = resync_apply_report(&g_twr, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); lsn_after = twr_current_lsn(); /* record_job_finish_fact_at() (bug-1747 fix)/record_job_fail_fact * (called via on_job_fail, unchanged -- Req 4.1) follow the SAME * durable-append-only, DEFERRED pattern as any ordinary worker * finish/fail report: the in-memory state mutation happens later, when * apply_committed_entry() replays the committed fact -- NOT * synchronously here (unlike record_job_reconcile_pending_fact/assign_ * fact's own EAGER pattern below, asserted synchronously exactly * because task 2.1/2.2 built those two functions to mutate eagerly). * This exact deferred behavior is already proven end-to-end (including * via a real restart replay) by test_wal_job_finish_fact.c's own * test_wjf_real_on_job_finish_survives_restart -- this task's own job * is only to prove routing (the correct fact TYPE, with the correct * worker-reported finish_time_ms for finishedOk, was durably appended * through the correct call), not to re-prove that already-tested apply * seam; test_apply_report_finished_success_uses_worker_reported_ * finish_time below proves the finish_time_ms plumbing specifically, * via a real restart replay. */ TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_finok->job.state, "finished-success routes through the DEFERRED record_job_finish_" "fact pattern -- state mutation happens later via apply_committed_" "entry, not synchronously here"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_finfail->job.state, "finished-failure routes through the DEFERRED record_job_fail_fact " "pattern -- state mutation happens later via apply_committed_entry, " "not synchronously here"); TEST_ASSERT_TRUE_MESSAGE(lsn_after >= lsn_before + 3, "at least 3 durable facts must be appended for this report: " "finished-success (WAL_JOB_FINISH), finished-failure (WAL_JOB_" "FAIL), and unrecognized (WAL_JOB_UPDATE) -- the still-running " "entry and the TERMINAL unqueried extra (Req 4.4 no-op) record " "none"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_running->job.state, "a queried still-running job must be left exactly as-is -- no fact " "recorded"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j_unrec->job.state, "unrecognized must apply through record_job_reconcile_pending_fact " "(-> PEND)"); TEST_ASSERT_NULL_MESSAGE(j_unrec->job.allocs, "bug-1631: reconciled-to-pending must release the allocation"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j_extra->job.state, "Req 4.4: a TERMINAL job reported running as an unqueried extra " "must be left COMPLETELY untouched -- never routed through " "record_job_reconcile_assign_fact, regardless of host"); TEST_ASSERT_NULL_MESSAGE(j_extra->job.allocs, "the terminal extra's allocation (already NULL) must stay " "untouched too"); TEST_ASSERT_NULL_MESSAGE(w->resync, "once every queried job (4 of them) has a reconciled outcome, the " "round must resolve"); TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), "XCE_WORKER_RECONCILING must clear once the round resolves"); twr_teardown(); printf("test_apply_report_all_five_outcomes_updates_job_table: PASS\n"); } /* Req 6.2: applying the identical report a second time (same round token) * must produce the same result as applying it once -- no further job-table * change, no additional WAL append. Since the first application already * fully resolves the round (both entries here are queried), the round * token is no longer current by the time the second, identical call * arrives -- resync_apply_report's own Req 6.1/6.3 stale-round discard is * exactly the mechanism that makes the duplicate a pure no-op. */ static void test_apply_report_duplicate_reapplication_is_noop(void) { job_t* ja; job_t* jb; worker_t* w; int rc; uint64_t token; uint64_t lsn_before_first, lsn_after_first, lsn_after_second; xce_job_state_entry_t ea, eb; xce_job_state_entry_t* entries[2]; xce_job_state_report_t report; twr_setup(); ja = twr_seed_job(TWR_JOBID_REPORT_DUP_A, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(ja); ja->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_REPORT_DUP_A, "resync-dup-host"); jb = twr_seed_job(TWR_JOBID_REPORT_DUP_B, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(jb); jb->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_REPORT_DUP_B, "resync-dup-host"); w = twr_make_test_worker("resync-dup-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT(2, rc); token = w->resync->round_token; ea = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_DUP_A, .status = JobStateStatus_finishedOk, .exit_code = 0, .finish_time_ms = 500, .queried = 1 }; eb = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_DUP_B, .status = JobStateStatus_unrecognized, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; entries[0] = &ea; entries[1] = &eb; report = (xce_job_state_report_t){ .n_total = 2, .pos = 0, .count = 2, .entries = entries }; lsn_before_first = twr_current_lsn(); rc = resync_apply_report(&g_twr, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); /* ea (finishedOk) routes through the DEFERRED record_job_finish_fact_at * pattern (bug-1747 fix) -- state mutation happens later via * apply_committed_entry, not synchronously here (see the "all five * outcomes" test's own comment for the full rationale); ja's state is * asserted unchanged (still RUN) below instead. */ TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, ja->job.state); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, jb->job.state); TEST_ASSERT_NULL(jb->job.allocs); TEST_ASSERT_NULL_MESSAGE(w->resync, "the round must resolve after the first, complete application"); lsn_after_first = twr_current_lsn(); TEST_ASSERT_TRUE_MESSAGE(lsn_after_first >= lsn_before_first + 2, "the first application must durably append a fact for each of the " "2 queried entries"); /* Second, IDENTICAL application -- same token, same report contents. */ rc = resync_apply_report(&g_twr, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); lsn_after_second = twr_current_lsn(); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, ja->job.state, "a duplicate re-application must change nothing further"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, jb->job.state, "a duplicate re-application must change nothing further"); TEST_ASSERT_NULL(jb->job.allocs); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_after_first, lsn_after_second, "a duplicate re-application for an already-resolved round must " "append no further WAL fact (Req 6.2)"); TEST_ASSERT_NULL(w->resync); twr_teardown(); printf("test_apply_report_duplicate_reapplication_is_noop: PASS\n"); } /* Req 6.1, 6.3: a report carrying a round token already superseded by a * newer round for the SAME worker is dropped in its entirety -- the job * table must be completely untouched, and the (unrelated, current) newer * round must be left exactly as it was. */ static void test_apply_report_stale_round_token_is_discarded(void) { job_t* j; worker_t* w; int rc; uint64_t token1, token2; uint64_t lsn_before, lsn_after; xce_job_state_entry_t e; xce_job_state_entry_t* entries[1]; xce_job_state_report_t report; twr_setup(); j = twr_seed_job(TWR_JOBID_REPORT_STALE, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); j->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_REPORT_STALE, "resync-stale-host"); w = twr_make_test_worker("resync-stale-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT(1, rc); token1 = w->resync->round_token; /* Simulated disconnect-then-rejoin: a fresh round supersedes the first * before its report ever arrives. */ rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT(1, rc); token2 = w->resync->round_token; TEST_ASSERT_NOT_EQUAL(token1, token2); e = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_STALE, .status = JobStateStatus_finishedOk, .exit_code = 0, .finish_time_ms = 999, .queried = 1 }; entries[0] = &e; report = (xce_job_state_report_t){ .n_total = 1, .pos = 0, .count = 1, .entries = entries }; lsn_before = twr_current_lsn(); /* A report carrying the now-stale token1 must be dropped entirely. */ rc = resync_apply_report(&g_twr, w, token1, &report); TEST_ASSERT_EQUAL_INT(0, rc); lsn_after = twr_current_lsn(); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, "a stale-round report must be dropped without touching the job " "table"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after, "a stale-round report must append no WAL fact"); TEST_ASSERT_NOT_NULL_MESSAGE(w->resync, "the CURRENT (newer) round must be left outstanding, untouched by " "the stale report"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(token2, w->resync->round_token, "the current round's own token must be unaffected"); TEST_ASSERT_EQUAL_INT(1, w->resync->n_queried_jobids); resync_resolve_round(&g_twr, w); twr_teardown(); printf("test_apply_report_stale_round_token_is_discarded: PASS\n"); } /* Req 5.1, 5.2: a partial report (not all queried jobs answered yet) must * NOT resolve the round -- the outstanding set stays non-empty and * XCE_WORKER_RECONCILING stays set -- until a LATER call with the * remaining entries completes it, proving resolution is genuinely * incremental against the outstanding set, not "resolve unconditionally * after any report." */ static void test_apply_report_partial_report_does_not_resolve_round(void) { job_t* ja; job_t* jb; worker_t* w; int rc; uint64_t token; uint64_t lsn_before, lsn_after; xce_job_state_entry_t ea, eb; xce_job_state_entry_t* entries_a[1]; xce_job_state_entry_t* entries_b[1]; xce_job_state_report_t report_a, report_b; twr_setup(); ja = twr_seed_job(TWR_JOBID_REPORT_PARTIAL_A, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(ja); ja->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_REPORT_PARTIAL_A, "resync-partial-host"); jb = twr_seed_job(TWR_JOBID_REPORT_PARTIAL_B, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(jb); jb->job.allocs = twr_make_test_alloc_for_host(TWR_JOBID_REPORT_PARTIAL_B, "resync-partial-host"); w = twr_make_test_worker("resync-partial-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT(2, rc); token = w->resync->round_token; ea = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_PARTIAL_A, .status = JobStateStatus_finishedOk, .exit_code = 0, .finish_time_ms = 111, .queried = 1 }; entries_a[0] = &ea; report_a = (xce_job_state_report_t){ .n_total = 1, .pos = 0, .count = 1, .entries = entries_a }; lsn_before = twr_current_lsn(); rc = resync_apply_report(&g_twr, w, token, &report_a); TEST_ASSERT_EQUAL_INT(0, rc); lsn_after = twr_current_lsn(); /* ea (finishedOk) routes through the DEFERRED record_job_finish_fact * pattern (via on_job_finish, unchanged) -- state mutation happens * later via apply_committed_entry, not synchronously here (see the * "all five outcomes" test's own comment for the full rationale); a * durable append is asserted instead. */ TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, ja->job.state); TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before, "job A's finished-success entry must durably append a fact"); TEST_ASSERT_NOT_NULL_MESSAGE(w->resync, "job B has not been answered yet -- the round must stay " "outstanding"); TEST_ASSERT_EQUAL_INT(1, w->resync->n_queried_jobids); TEST_ASSERT_TRUE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), "XCE_WORKER_RECONCILING must remain set while the round is still " "partially outstanding"); eb = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_PARTIAL_B, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; entries_b[0] = &eb; report_b = (xce_job_state_report_t){ .n_total = 1, .pos = 0, .count = 1, .entries = entries_b }; rc = resync_apply_report(&g_twr, w, token, &report_b); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, jb->job.state); TEST_ASSERT_NULL_MESSAGE(w->resync, "the round must resolve once the LAST outstanding queried job " "(B) is answered"); TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), "XCE_WORKER_RECONCILING must clear once the round fully resolves"); twr_teardown(); printf("test_apply_report_partial_report_does_not_resolve_round: " "PASS\n"); } /* Req 3.6/4.7/4.4 (the crux branching this task's own brief flagged as * needing careful resolution against job.c's record_job_reconcile_assign_ * fact doc comment; round-2 review strengthened this test after finding it * could not actually distinguish "correctly no-op'd" from "incorrectly * reassigned" -- see below): an unqueried extra reported RUNNING that the * controller ALREADY has a non-terminal record of (here: assigned to a * DIFFERENT host -- resync_apply_running_entry()'s branch (b), non-terminal * AND job->job.allocs != NULL) must NOT be routed through record_job_ * reconcile_assign_fact -- it is already correctly represented by simply * existing and being reported running. * * Proven three ways, precisely enough to actually catch a regression back * to either of the two review-caught defects (round 1's Req 4.8-axis bug; * round 2's Req 4.4-axis bug): * 1. allocs survives UNTOUCHED -- record_job_reconcile_assign_fact always * sets job->job.allocs to NULL (Req 4.8), so a surviving allocation * proves the no-op branch fired instead of the assign branch. * 2. start_time survives UNTOUCHED at its seeded sentinel value -- * record_job_reconcile_assign_fact ALSO unconditionally stamps * job->job.start_time = time(NULL) (bug-1697), even though the net * STATE effect (RUN -> RUN) looks harmless and is invisible to (1) or * to a bare state-equality assertion; this is exactly what round 1's * version of this test could not catch, since it never checked * start_time and the job was already RUN before the call either way. * 3. no additional WAL fact is appended (LSN unchanged) -- record_job_ * reconcile_assign_fact durably appends a fresh WAL_JOB_DISPATCH fact * unconditionally; a no-op must append nothing (the paired queried * "running" entry records nothing either, so the WHOLE report's LSN * delta must be exactly zero). */ static void test_apply_report_unqueried_extra_already_tracked_is_noop(void) { job_t* j_queried; job_t* j_extra; worker_t* w; int rc; uint64_t token; uint64_t lsn_before, lsn_after; time_t seeded_start_time; xce_job_state_entry_t e_queried, e_extra; xce_job_state_entry_t* entries[2]; xce_job_state_report_t report; twr_setup(); /* Keeps the round non-empty (arm_round needs >=1 non-terminal job on * w's own host to arm at all) -- unrelated to the extra under test. */ j_queried = twr_seed_job(TWR_JOBID_REPORT_EXTRANOOP_Q, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_queried); j_queried->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_EXTRANOOP_Q, "resync-extranoop-host"); /* Non-terminal (RUN), already assigned -- to a DIFFERENT host, so * arm_round's own query-set construction (non-terminal jobs assigned * to w's specific host) never includes it; it arrives only as an * unqueried extra in the report below. */ j_extra = twr_seed_job(TWR_JOBID_REPORT_EXTRANOOP_E, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j_extra); j_extra->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_EXTRANOOP_E, "resync-extranoop-OTHER-host"); /* A deliberately-distinguishable sentinel, far from "now" -- so a * wrongful record_job_reconcile_assign_fact() call (which stamps * time(NULL)) is caught precisely, not masked by coincidence. */ seeded_start_time = (time_t)424242; j_extra->job.start_time = seeded_start_time; w = twr_make_test_worker("resync-extranoop-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&g_twr, w); TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, "only j_queried is assigned to w's own host"); token = w->resync->round_token; e_queried = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_EXTRANOOP_Q, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; e_extra = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_EXTRANOOP_E, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 0 }; entries[0] = &e_queried; entries[1] = &e_extra; report = (xce_job_state_report_t){ .n_total = 2, .pos = 0, .count = 2, .entries = entries }; lsn_before = twr_current_lsn(); rc = resync_apply_report(&g_twr, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); lsn_after = twr_current_lsn(); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_extra->job.state, "an already-non-terminal extra must be left exactly as-is"); TEST_ASSERT_NOT_NULL_MESSAGE(j_extra->job.allocs, "record_job_reconcile_assign_fact must NOT have been called -- its " "own contract always sets allocs to NULL, so a surviving " "allocation proves the no-op branch fired instead"); TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)seeded_start_time, (uint64_t)j_extra->job.start_time, "record_job_reconcile_assign_fact must NOT have been called -- it " "unconditionally stamps start_time = time(NULL) (bug-1697), so a " "start_time still equal to the seeded sentinel (not \"now\") " "proves the no-op branch fired, catching what a bare state/allocs " "check alone cannot (round-2 review finding)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after, "neither entry in this report may append a WAL fact: the queried " "running entry never does, and the already-tracked unqueried " "extra must not durably re-append a WAL_JOB_DISPATCH fact for a " "job that was never actually lost/unassigned"); TEST_ASSERT_NULL_MESSAGE(w->resync, "the round's only queried job (j_queried) was answered -- it must " "resolve regardless of the unqueried extra"); twr_teardown(); printf("test_apply_report_unqueried_extra_already_tracked_is_noop: " "PASS\n"); } /* bug-1747 fix (Req 4.2): a finished-success reconciliation entry's * WORKER-reported finish_time_ms must end up as the job's recorded * finish_time after a real restart replay -- NOT reconciliation-processing * wall-clock time. Mirrors test_reconcile_assign_survives_restart_replay's * own Phase 1 (real cluster + disk persist, drive the real function under * test, flush) / Phase 2 (fresh persist_t + persist_recover(), assert on * the recovered job_t) shape, and test_wal_job_finish_fact.c's own * test_wjf_real_on_job_finish_survives_restart precedent for proving a * finish_time round-trip through a real WAL append + replay, not a * hand-built payload. Uses its own local cluster_t/persist_t/tmpdir (Phase * 1's disk state must survive past Phase 1's own teardown) -- only * g_twr.workermgr (via twr_make_test_worker) is shared with the rest of * this file's fixture, exactly as test_reconcile_assign_survives_restart_ * replay already establishes is safe (resync_apply_report's own locking * only ever touches `w` and the `c` passed to it, never workermgr). */ static void test_apply_report_finished_success_uses_worker_reported_finish_time(void) { char tmp[256]; uint64_t worker_reported_finish_time_s; strcpy(tmp, "/tmp/xce_twr24_finishtime_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); twr_setup(); /* A finish time deliberately, clearly DIFFERENT from "now" -- one full * day in the past -- so a reconciliation-processing-time bug * (time(NULL)) cannot accidentally produce a matching value. */ worker_reported_finish_time_s = (uint64_t)time(NULL) - 86400; /* Phase 1: real cluster + real disk persist. Seed a durable * WAL_JOB_SUBMIT fact directly (so replay can reconstruct the job -- * this job was never dispatched through any controller in this log) * and the in-memory job_t the same way (bypassing dispatch/replay), * already RUN and assigned to the reconciling worker's host, then * drive a real resync_apply_report() call carrying the worker-reported * finish_time_ms. */ { cluster_t c; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; persist_config_t cfg; job_t* j; worker_t* w; int rc; uint64_t token; xce_job_state_entry_t e; xce_job_state_entry_t* entries[1]; xce_job_state_report_t report; 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); TEST_ASSERT_NOT_NULL(c.jobmgr); 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)); { 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*)"q"; wal_submit.user = (char*)"tester"; memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = TWR_JOBID_REPORT_FINISHTIME_REPLAY; wal_jdef.state = JOB_STAT_RUN; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, twr_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); } { job_submit_data_t submit; xce_job_t jdef; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = TWR_JOBID_REPORT_FINISHTIME_REPLAY; jdef.state = JOB_STAT_RUN; jdef.submit = &submit; j = makeJob(c.jobmgr, &jdef); TEST_ASSERT_NOT_NULL(j); } j->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_FINISHTIME_REPLAY, "resync-finishtime-host"); w = twr_make_test_worker("resync-finishtime-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&c, w); TEST_ASSERT_EQUAL_INT(1, rc); token = w->resync->round_token; e = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_FINISHTIME_REPLAY, .status = JobStateStatus_finishedOk, .exit_code = 0, .finish_time_ms = worker_reported_finish_time_s * 1000ULL, .queried = 1 }; entries[0] = &e; report = (xce_job_state_report_t){ .n_total = 1, .pos = 0, .count = 1, .entries = entries }; rc = resync_apply_report(&c, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NULL_MESSAGE(w->resync, "the round's only queried job was answered -- it must resolve"); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); persist_stop(&c.persist); persist_fini(&c.persist); dereference(c.jobmgr); xce_hook_fini(&c.hooksys); } /* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only), * recover from disk -- simulates a controller restart. */ { persist_t* p2 = twr_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*)TWR_JOBID_REPORT_FINISHTIME_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_DONE, j2->job.state, "the finished-success fact must apply to DONE on replay"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(worker_reported_finish_time_s, (uint64_t)j2->job.finish_time, "bug-1747 (Req 4.2): the recovered finish_time must match the " "WORKER-reported finish_time_ms, not reconciliation-processing " "wall-clock time"); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); twr_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); } twr_teardown(); printf("test_apply_report_finished_success_uses_worker_reported_" "finish_time: PASS\n"); } /* Review-caught regression fix (Req 4.8): a job the controller currently * shows PEND (non-terminal, allocs == NULL, unassigned to any host) -- * exactly design.md's own "Documented scope decision" scenario, e.g. its * WAL_JOB_DISPATCH fact never reached durable majority commit before a * controller failover, even though dispatch_a_job_() already sent * xce_startjob_req() to the worker and the worker really is running it -- * must NOT be left stuck at PEND when reported running as an unqueried * extra. It must reconcile via record_job_reconcile_assign_fact() to RUN, * matching the reporting worker's host, with allocs staying NULL (there is * none to fabricate, Req 4.8's own documented scope decision). Mirrors * test_reconcile_assign_survives_restart_replay's rigor: real WAL_JOB_ * SUBMIT seed + real resync_apply_report() call + real WAL flush, followed * by a genuine persist_recover() restart replay -- not just an in-memory * assertion -- so this proves the fix survives a real controller restart, * not merely the live eager-mutation path. A second, unrelated queried job * on the same host keeps the round non-empty (arm_round's own query set * construction only ever finds jobs it already believes are non-terminal * AND assigned to the host -- by definition NOT this PEND/unassigned job, * which is exactly why it only ever arrives as an unqueried extra). */ static void test_apply_report_unqueried_extra_unassigned_pend_job_gets_assigned(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_twr25_lostdispatch_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); twr_setup(); /* Phase 1: real cluster + real disk persist. */ { cluster_t c; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; persist_config_t cfg; job_t* jq; job_t* jlost; worker_t* w; int rc; uint64_t token; xce_job_state_entry_t eq, elost; xce_job_state_entry_t* entries[2]; xce_job_state_report_t report; 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); TEST_ASSERT_NOT_NULL(c.jobmgr); 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 seeds -- jq at RUN (assigned, keeps the * round non-empty), jlost at PEND (unassigned -- the job whose * WAL_JOB_DISPATCH fact never durably committed). */ { 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*)"q"; wal_submit.user = (char*)"tester"; memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = TWR_JOBID_REPORT_LOSTDISPATCH_Q; wal_jdef.state = JOB_STAT_RUN; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, twr_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); memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = TWR_JOBID_REPORT_LOSTDISPATCH; wal_jdef.state = JOB_STAT_PEND; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, twr_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); } /* Seed the in-memory job_t's the same way, bypassing dispatch/ * replay. */ { job_submit_data_t submit; xce_job_t jdef; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = TWR_JOBID_REPORT_LOSTDISPATCH_Q; jdef.state = JOB_STAT_RUN; jdef.submit = &submit; jq = makeJob(c.jobmgr, &jdef); TEST_ASSERT_NOT_NULL(jq); jq->job.allocs = twr_make_test_alloc_for_host( TWR_JOBID_REPORT_LOSTDISPATCH_Q, "resync-lostdispatch-host"); memset(&jdef, 0, sizeof(jdef)); jdef.jobid = TWR_JOBID_REPORT_LOSTDISPATCH; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; jlost = makeJob(c.jobmgr, &jdef); TEST_ASSERT_NOT_NULL(jlost); TEST_ASSERT_NULL_MESSAGE(jlost->job.allocs, "the lost-dispatch-fact job starts PEND and unassigned -- " "no allocation was ever durably recorded for it"); } w = twr_make_test_worker("resync-lostdispatch-host"); TEST_ASSERT_NOT_NULL(w); rc = resync_arm_round(&c, w); TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, "only jq (RUN, assigned) is in the controller's own query set " "-- jlost (PEND, unassigned) is invisible to arm_round's own " "scan, exactly why it can only ever arrive as an unqueried " "extra"); token = w->resync->round_token; eq = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_LOSTDISPATCH_Q, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 1 }; elost = (xce_job_state_entry_t){ .jobid = TWR_JOBID_REPORT_LOSTDISPATCH, .status = JobStateStatus_running, .exit_code = 0, .finish_time_ms = 0, .queried = 0 }; entries[0] = &eq; entries[1] = &elost; report = (xce_job_state_report_t){ .n_total = 2, .pos = 0, .count = 2, .entries = entries }; rc = resync_apply_report(&c, w, token, &report); TEST_ASSERT_EQUAL_INT(0, rc); /* Eager pattern (record_job_reconcile_assign_fact) -- observable * synchronously, before any commit round-trip. */ TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jlost->job.state, "a PEND, unassigned job reported running as an unqueried extra " "must reconcile to RUN via record_job_reconcile_assign_fact -- " "not be left stuck at PEND"); TEST_ASSERT_NULL_MESSAGE(jlost->job.allocs, "Req 4.8: no allocation is fabricated for a reconciliation-" "assigned job"); TEST_ASSERT_NULL_MESSAGE(w->resync, "the round's only QUERIED job (jq) was answered -- it must " "resolve regardless of the unqueried extra"); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); persist_stop(&c.persist); persist_fini(&c.persist); dereference(c.jobmgr); xce_hook_fini(&c.hooksys); } /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk -- * proves the fix survives a real controller restart, not just the live * eager-mutation path. */ { persist_t* p2 = twr_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*)TWR_JOBID_REPORT_LOSTDISPATCH, (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 record_job_reconcile_" "assign_fact()'s real call -- the fix must survive a real " "controller restart, not just the live eager-mutation path"); TEST_ASSERT_NULL_MESSAGE(j2->job.allocs, "Req 4.8: replay must NOT fabricate a resource allocation for " "a reconciliation-assigned job"); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); twr_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); } twr_teardown(); printf("test_apply_report_unqueried_extra_unassigned_pend_job_gets_" "assigned: PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_worker_resync(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_reconcile_pending_applies_releases_allocs_and_notifies); RUN_TEST(test_reconcile_pending_noop_on_already_pending_job); RUN_TEST(test_reconcile_pending_rejects_terminal_job); RUN_TEST(test_reconcile_assign_applies_running_no_allocs_and_notifies); RUN_TEST(test_reconcile_assign_survives_restart_replay); RUN_TEST(test_resync_arm_round_query_set_matches_exactly); RUN_TEST(test_resync_arm_round_no_jobs_is_noop); RUN_TEST(test_resync_arm_round_fresh_token_supersedes_prior); RUN_TEST(test_resync_reconciling_flag_set_after_arm_cleared_after_resolve); RUN_TEST(test_resync_timeout_resolves_round_without_blocking); RUN_TEST(test_apply_report_all_five_outcomes_updates_job_table); RUN_TEST(test_apply_report_duplicate_reapplication_is_noop); RUN_TEST(test_apply_report_stale_round_token_is_discarded); RUN_TEST(test_apply_report_partial_report_does_not_resolve_round); RUN_TEST(test_apply_report_unqueried_extra_already_tracked_is_noop); RUN_TEST( test_apply_report_finished_success_uses_worker_reported_finish_time); RUN_TEST( test_apply_report_unqueried_extra_unassigned_pend_job_gets_assigned); }