feat(scheduler-dispatch-commit): add durable WAL_JOB_DISPATCH fact type (task 1.2)
New wal_entry_type WAL_JOB_DISPATCH=12, reusing wal_jobstate_t verbatim (_pad=0, absolute-value semantics), wired into both apply_wal_entry()'s and apply_committed_entry()'s existing job-state fall-through groups per Requirement 2.4. Restart-replay test writes only WAL_JOB_SUBMIT+ WAL_JOB_DISPATCH to a real on-disk WAL and confirms job.state==RUN after a fresh persist_recover(), independent of WAL_JOB_START. Review found apply_wal_entry() has zero production call sites today (persist_recover() and standby catch-up both route through apply_committed_entry() post the 2026-07-10 ha_apply_single_entry refactor) -- design.md's stale characterization corrected; the dual case-label registration itself is unaffected and correct. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
5724d1e90a
commit
8eea74909c
7 changed files with 367 additions and 5 deletions
|
|
@ -109,7 +109,7 @@ sequenceDiagram
|
|||
- `handle_jobdispatch()`'s loop drops its existing per-job `notify_dispatch(c, ep, alloc->jobid, ...)` calls (both the success-path and failure-path ones) — notification is now entirely owned by `dispatch_a_job_()`, preventing the double-notify race found in round 2 of this review (a terminal failure previously would have produced both `jobDispatchFail` and `schedFailJob`).
|
||||
- `src/include/xce/types.h` — `xce_job_t`'s wrapper `job_t` (not the wire type itself) gains `uint32_t dispatch_fail_count;`, following `submit_lsn`'s existing "in-memory only, reset on replay" precedent.
|
||||
- `src/include/wal_types.h` — one new `wal_entry_type` value, `WAL_JOB_DISPATCH = 12`, reusing the existing `wal_jobstate_t{jobid, state, _pad=0}` payload shape (the "legacy: state is absolute" discipline, matching `job.allocs`'s own synchronous-commit precedent — see Data Models).
|
||||
- `src/controller/persist.c` — `WAL_JOB_DISPATCH` is added as a new `case` label to two existing fall-through groups: `apply_wal_entry()`'s (recovery-replay path, `persist.c:3452-3466`) and `apply_committed_entry()`'s (live-apply path, `persist.c:3700-3706`) — both already apply `wal_jobstate_t.state` as an absolute value for every type in their respective groups; no new function or case body.
|
||||
- `src/controller/persist.c` — `WAL_JOB_DISPATCH` is added as a new `case` label to two existing fall-through groups: `apply_wal_entry()`'s (`persist.c:3452-3466`; currently has zero production call sites — see the Components section correction below) and `apply_committed_entry()`'s (the actual live-apply-and-recovery path, `persist.c:3700-3706`) — both already apply `wal_jobstate_t.state` as an absolute value for every type in their respective groups; no new function or case body.
|
||||
- `src/controller/job.c` (same file, no new header) — `xce_classify_dispatch_fail()`, `xce_dispatch_fail_site_t`, and `xce_dispatch_fail_class_t` are declared `static`/file-local next to `dispatch_a_job_()`, its only caller — not in a shared header, since nothing outside this file consumes them (see Controller Dispatch-Failure Classification's placement-correction note).
|
||||
|
||||
### New/Modified Test Files
|
||||
|
|
@ -289,7 +289,7 @@ xce_dispatch_fail_class_t xce_classify_dispatch_fail(xce_dispatch_fail_site_t si
|
|||
**Responsibilities & Constraints**
|
||||
- New `wal_entry_type` value `WAL_JOB_DISPATCH = 12` (next free value in the job-fact range, after `WAL_JOB_ARRAY_SUBMIT = 11`).
|
||||
- Reuses the existing `wal_jobstate_t{jobid, state, _pad}` payload verbatim — no new struct. `_pad` is always `0` for this type (the "legacy: state is absolute value" convention `wal_jobstate_t`'s own header comment documents), matching the synchronous, non-deferred-apply commit style Controller Dispatch Commit uses.
|
||||
- **Two separate functions, both needing the new case** — found by tracing precisely rather than assuming a single switch: `apply_wal_entry()` (`persist.c:3375+`, the recovery/replay-time path reading WAL segments from disk after restart) has its own fall-through group at `persist.c:3452-3466` (`WAL_JOB_START` through `WAL_JOB_RUSAGE`, including `WAL_JOB_FAIL` — this recovery path applies every type as an absolute overwrite, unconditionally). `apply_committed_entry()` (`persist.c:3570+`, the live path invoked immediately when a WAL entry commits during normal operation — the one `raft-apply-decoupling` actually introduced the `_pad`-discriminated routing for) has its *own separate* fall-through group at `persist.c:3700-3706` (`WAL_JOB_START`, `WAL_JOB_FINISH`, `WAL_JOB_STOP`, `WAL_JOB_RESUME`, `WAL_JOB_RUSAGE`, `WAL_JOB_CKPT`, `WAL_JOB_PREASON` — deliberately excluding `WAL_JOB_FAIL`/`WAL_JOB_UPDATE`, which get their own `_pad`-aware handling immediately after). That group's own comment states the exact rationale this design already relies on: these types' payloads are "deterministic/state-independent... an absolute `state` baked at receipt time never goes stale" — `WAL_JOB_DISPATCH` fits this group by the same reasoning (Research Log: no same-job racing-fact concern for the first post-PEND transition). `WAL_JOB_DISPATCH` joins both fall-through lists — no new apply-time function or standalone case body is introduced in either.
|
||||
- **Two separate functions, both needing the new case** — found by tracing precisely rather than assuming a single switch: `apply_wal_entry()` (`persist.c:3375+`) has its own fall-through group at `persist.c:3452-3466` (`WAL_JOB_START` through `WAL_JOB_RUSAGE`, including `WAL_JOB_FAIL` — applies every type as an absolute overwrite, unconditionally). `apply_committed_entry()` (`persist.c:3570+`, the live path invoked immediately when a WAL entry commits during normal operation — the one `raft-apply-decoupling` actually introduced the `_pad`-discriminated routing for) has its *own separate* fall-through group at `persist.c:3700-3706` (`WAL_JOB_START`, `WAL_JOB_FINISH`, `WAL_JOB_STOP`, `WAL_JOB_RESUME`, `WAL_JOB_RUSAGE`, `WAL_JOB_CKPT`, `WAL_JOB_PREASON` — deliberately excluding `WAL_JOB_FAIL`/`WAL_JOB_UPDATE`, which get their own `_pad`-aware handling immediately after). That group's own comment states the exact rationale this design already relies on: these types' payloads are "deterministic/state-independent... an absolute `state` baked at receipt time never goes stale" — `WAL_JOB_DISPATCH` fits this group by the same reasoning (Research Log: no same-job racing-fact concern for the first post-PEND transition). `WAL_JOB_DISPATCH` joins both fall-through lists — no new apply-time function or standalone case body is introduced in either. **Corrected during task 1.2 implementation review**: `apply_wal_entry()` is NOT actually "the recovery/replay-time path" in the current architecture — `persist_recover()`'s real boot-time replay (`replay_wal()`) and the standby catch-up path (post the 2026-07-10 `ha_apply_single_entry` refactor) both call `apply_committed_entry()` exclusively; `apply_wal_entry()` has zero production call sites today. Adding the case to both functions is still correct (matches every sibling `WAL_JOB_*` type's identical dual-registration, harmless dead-code parity, and costs nothing if `apply_wal_entry()` is ever revived) — only the prose description of what `apply_wal_entry()` currently *is* was wrong.
|
||||
- Does not carry or restore `job.allocs`/`job.accounts` — those remain non-durable in-memory fields, unchanged from today's behavior (Boundary: Out of Boundary does not list a new durability guarantee for allocation data itself, only for the state transition).
|
||||
|
||||
**Contracts**: Batch [x] / State [x]
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
- _Requirements: 4.1, 4.2_
|
||||
- _Boundary: Controller Dispatch-Failure Classification (site enum + classify function)_
|
||||
|
||||
- [ ] 1.2 (P) Add a durable record of a job's dispatch-time transition to a running state
|
||||
- [x] 1.2 (P) Add a durable record of a job's dispatch-time transition to a running state
|
||||
- Introduce a new durable fact type recording that a job has transitioned to running state at dispatch-commit time, reusing the existing shape used for other job-state facts
|
||||
- Wire this new fact type into both the normal live-commit apply path and the recovery/restart replay path, so the transition is recovered identically whether observed live or replayed after a restart
|
||||
- Observable completion: appending this fact and then replaying it (simulating a restart) leaves the job's state as running, without requiring the separate worker-confirmed-start fact to also be present
|
||||
|
|
@ -99,3 +99,4 @@
|
|||
- Task 1.1: design.md's "Controller Dispatch-Failure Classification" Service Interface Preconditions text claims a default-less `switch` over `xce_dispatch_fail_site_t` gives "compiler-enforced exhaustiveness... fails to compile" for an unhandled case — verified false for plain C without `-Werror` (this project sets none, confirmed in `src/meson.build`/`src/controller/meson.build`). Implemented instead with a `default:` arm that ERROR-logs and returns a safe `XCE_DISPATCH_FAIL_TERMINAL`, with `assert(0 && ...)` gated to `#ifdef XCE_TESTING` only — mirrors the established, verified-matching precedent `persist.c`'s `wal_classify()` already uses (`persist.c:6030-6037`) for the same "never risk silent misbehavior on an unrecognized case" concern. Any later task/spec touching this same design.md section should correct its wording rather than re-trusting the "fails to compile" claim.
|
||||
- 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.
|
||||
|
|
|
|||
|
|
@ -169,6 +169,7 @@ controller_unit_test_src = [
|
|||
'testsuite/test_apply_seam.c',
|
||||
'testsuite/test_apply_committed_entry.c',
|
||||
'testsuite/test_replay_apply_seam.c',
|
||||
'testsuite/test_wal_job_dispatch_fact.c',
|
||||
'testsuite/test_apply_reapply_idempotency.c',
|
||||
'testsuite/test_apply_correctness_verification.c',
|
||||
'testsuite/test_apply_pump.c',
|
||||
|
|
|
|||
|
|
@ -3458,6 +3458,7 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
|
|||
case WAL_JOB_CKPT:
|
||||
case WAL_JOB_PREASON:
|
||||
case WAL_JOB_RUSAGE:
|
||||
case WAL_JOB_DISPATCH:
|
||||
/* Payload: wal_jobstate_t {jobid, state} */
|
||||
if (payload_len >= sizeof(wal_jobstate_t)) {
|
||||
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
|
||||
|
|
@ -3703,7 +3704,8 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
|
|||
case WAL_JOB_RESUME:
|
||||
case WAL_JOB_RUSAGE:
|
||||
case WAL_JOB_CKPT:
|
||||
case WAL_JOB_PREASON: {
|
||||
case WAL_JOB_PREASON:
|
||||
case WAL_JOB_DISPATCH: {
|
||||
/* Every job-state fact type in this group shares one wire shape —
|
||||
* wal_jobstate_t {jobid, state} — and therefore one guarded core.
|
||||
* Their payloads are deterministic/state-independent (a START is
|
||||
|
|
@ -3716,7 +3718,13 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
|
|||
* finish}_transition() cores (their parameter structs are not
|
||||
* present on this wire format) — start_time/finish_time/run_time
|
||||
* are instead stamped directly below, keyed on wal_type, since
|
||||
* apply_job_update_transition() only ever touches `state`. */
|
||||
* apply_job_update_transition() only ever touches `state`.
|
||||
* WAL_JOB_DISPATCH joins this group for the same reason: it is
|
||||
* synchronously computed at dispatch-commit time under the job's
|
||||
* own write lock, with no other in-flight fact for the same job
|
||||
* (it is the first transition after PEND) — so it never needs
|
||||
* WAL_JOB_FAIL/WAL_JOB_UPDATE's deferred, apply-time-recomputed
|
||||
* handling. */
|
||||
if (len < sizeof(wal_jobstate_t))
|
||||
return 0; /* undersized payload — skip, mirrors apply_wal_entry */
|
||||
|
||||
|
|
|
|||
|
|
@ -83,6 +83,7 @@ extern void test_persist_snapshot_entities(void);
|
|||
extern void test_apply_seam(void);
|
||||
extern void test_apply_committed_entry(void);
|
||||
extern void test_replay_apply_seam(void);
|
||||
extern void test_wal_job_dispatch_fact(void);
|
||||
extern void test_apply_reapply_idempotency(void);
|
||||
extern void test_apply_correctness_verification(void);
|
||||
extern void test_apply_pump(void);
|
||||
|
|
@ -247,6 +248,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_apply_seam);
|
||||
TEST(test_apply_committed_entry);
|
||||
TEST(test_replay_apply_seam);
|
||||
TEST(test_wal_job_dispatch_fact);
|
||||
TEST(test_apply_reapply_idempotency);
|
||||
TEST(test_apply_correctness_verification);
|
||||
TEST(test_apply_pump);
|
||||
|
|
|
|||
341
src/controller/testsuite/test_wal_job_dispatch_fact.c
Normal file
341
src/controller/testsuite/test_wal_job_dispatch_fact.c
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
/* test_wal_job_dispatch_fact.c — scheduler-dispatch-commit task 1.2
|
||||
*
|
||||
* Proves the new WAL_JOB_DISPATCH fact type (wal_types.h) -- the durable
|
||||
* record of a job's dispatch-time transition to a running state (Req 2.4,
|
||||
* design.md "WAL Dispatch Fact") -- is wired into BOTH of persist.c's apply
|
||||
* seams:
|
||||
*
|
||||
* - apply_committed_entry() (live-commit apply / boot replay dispatch)
|
||||
* - apply_wal_entry() (standby catch-up / replicated-apply dispatch)
|
||||
*
|
||||
* and that appending ONLY a WAL_JOB_DISPATCH fact (no accompanying
|
||||
* WAL_JOB_START) and replaying it -- simulating a controller restart via a
|
||||
* real persist_recover() over a real on-disk WAL -- recovers the job as
|
||||
* running (this task's "Observable completion").
|
||||
*
|
||||
* This task does NOT wire any live caller that appends this fact type
|
||||
* (dispatch_a_job_() is untouched -- that's task 3.1's job); every test here
|
||||
* constructs and applies/replays a WAL_JOB_DISPATCH entry directly.
|
||||
*
|
||||
* Fixtures mirror this suite's established conventions exactly:
|
||||
* - wjd_setup/wjd_teardown: bare cluster_t (jobmgr + hooksys) + in-memory
|
||||
* persist_t, mirroring test_apply_committed_entry.c's tace_setup and
|
||||
* test_job_array_apply.c's tajd_setup.
|
||||
* - wjd_make_fresh_persist/wjd_teardown_persist: real disk-backed
|
||||
* persist_t + persist_recover(), mirroring test_replay_apply_seam.c's
|
||||
* trr_make_fresh_persist/trr_teardown_persist.
|
||||
*/
|
||||
#include <rtti.h>
|
||||
|
||||
#include <capnp_c.h>
|
||||
#include <errno.h>
|
||||
#include <stdarg.h>
|
||||
#include <stdbool.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 "hook.h"
|
||||
#include "jobj.h"
|
||||
#include "persist.h"
|
||||
#include "wal_types.h"
|
||||
#include "xce/cluster_def.h"
|
||||
#include "xce/types.h"
|
||||
#include "xstring.h"
|
||||
|
||||
/* jobj.c's jobsub table and the process-global xstring table are shared
|
||||
* across every test file in the controller_unit_test binary — guard
|
||||
* idempotently, mirroring every sibling fixture in this suite. */
|
||||
static int g_wjd_jobobj_ready;
|
||||
|
||||
static cluster_t g_wjd_c;
|
||||
static persist_t g_wjd_p;
|
||||
|
||||
/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
|
||||
static int g_wjd_hook_fires;
|
||||
|
||||
static int wjd_spy_hook(int type, int situation, int detail, void* data,
|
||||
va_list ap)
|
||||
{
|
||||
(void)type; (void)detail; (void)data; (void)ap;
|
||||
if (situation == HOOK_AFTER_EVENT)
|
||||
g_wjd_hook_fires++;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void wjd_setup(void)
|
||||
{
|
||||
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 32,
|
||||
.delta = 16 };
|
||||
|
||||
if (makeString((char*)"wjd-probe") == NULL)
|
||||
xstring_table_init(64);
|
||||
|
||||
if (!g_wjd_jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
||||
g_wjd_jobobj_ready = 1;
|
||||
}
|
||||
|
||||
memset(&g_wjd_c, 0, sizeof(g_wjd_c));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
xce_hook_init(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
||||
|
||||
g_wjd_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
||||
TEST_ASSERT_NOT_NULL(g_wjd_c.jobmgr);
|
||||
|
||||
memset(&g_wjd_p, 0, sizeof(g_wjd_p));
|
||||
g_wjd_p.cluster = &g_wjd_c;
|
||||
|
||||
g_wjd_hook_fires = 0;
|
||||
}
|
||||
|
||||
static void wjd_teardown(void)
|
||||
{
|
||||
xce_hook_fini(&g_wjd_c.hooksys);
|
||||
if (g_wjd_c.jobmgr)
|
||||
dereference(g_wjd_c.jobmgr);
|
||||
}
|
||||
|
||||
/* Look up a job_t wrapper by jobid (NULL if absent). */
|
||||
static job_t* wjd_find_job(uint64_t jobid)
|
||||
{
|
||||
job_t* j = NULL;
|
||||
lock(g_wjd_c.jobmgr, RTTI_LOCK_READ);
|
||||
int found = key_find(g_wjd_c.jobmgr, (void*)jobid, (void**)&j);
|
||||
unlock(g_wjd_c.jobmgr);
|
||||
return (found > 0) ? j : NULL;
|
||||
}
|
||||
|
||||
/* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay
|
||||
* entirely — for seeding fact-transition tests, mirroring
|
||||
* test_apply_committed_entry.c's tace_seed_job. */
|
||||
static job_submit_data_t g_wjd_seed_submit;
|
||||
|
||||
static job_t* wjd_seed_job(uint64_t jobid, uint32_t state)
|
||||
{
|
||||
xce_job_t jdef;
|
||||
|
||||
memset(&g_wjd_seed_submit, 0, sizeof(g_wjd_seed_submit));
|
||||
g_wjd_seed_submit.queue = (char*)"q";
|
||||
g_wjd_seed_submit.user = (char*)"tester";
|
||||
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = jobid;
|
||||
jdef.state = state;
|
||||
jdef.submit = &g_wjd_seed_submit;
|
||||
|
||||
return makeJob(g_wjd_c.jobmgr, &jdef);
|
||||
}
|
||||
|
||||
/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */
|
||||
|
||||
static void test_wjd_apply_committed_entry_transitions_pend_to_run(void)
|
||||
{
|
||||
wjd_setup();
|
||||
|
||||
job_t* j = wjd_seed_job(71001, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
wal_jobstate_t js = { .jobid = 71001, .state = JOB_STAT_RUN, ._pad = 0 };
|
||||
|
||||
xce_add_hook(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjd_spy_hook,
|
||||
NULL);
|
||||
|
||||
int rc = apply_committed_entry(&g_wjd_p, WAL_JOB_DISPATCH, 100, &js,
|
||||
sizeof(js));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
|
||||
"WAL_JOB_DISPATCH must apply via the same absolute-value guarded "
|
||||
"core as WAL_JOB_START/etc, joining their fall-through group");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjd_hook_fires,
|
||||
"a forward WAL_JOB_DISPATCH transition must fire exactly one "
|
||||
"notification, matching every other type in its shared group");
|
||||
|
||||
wjd_teardown();
|
||||
printf("test_wjd_apply_committed_entry_transitions_pend_to_run: PASS\n");
|
||||
}
|
||||
|
||||
/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */
|
||||
|
||||
static void
|
||||
test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook(void)
|
||||
{
|
||||
wjd_setup();
|
||||
|
||||
job_t* j = wjd_seed_job(71002, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
wal_jobstate_t js = { .jobid = 71002, .state = JOB_STAT_RUN, ._pad = 0 };
|
||||
|
||||
xce_add_hook(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjd_spy_hook,
|
||||
NULL);
|
||||
|
||||
int rc = apply_wal_entry(&g_wjd_p, WAL_JOB_DISPATCH, 101, &js,
|
||||
sizeof(js));
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
|
||||
"WAL_JOB_DISPATCH must apply via restore_job_state() through "
|
||||
"apply_wal_entry's catch-up dispatch, joining the same "
|
||||
"absolute-value fall-through group as WAL_JOB_START/etc");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjd_hook_fires,
|
||||
"apply_wal_entry's catch-up dispatch must never fire a notification "
|
||||
"hook, matching every other type in its group");
|
||||
|
||||
wjd_teardown();
|
||||
printf("test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ── Restart-replay Observable (this task's completion criterion) ───────── */
|
||||
|
||||
static persist_t* wjd_make_fresh_persist(const char* tmpdir)
|
||||
{
|
||||
persist_t* p = calloc(1, sizeof(persist_t));
|
||||
if (!p) return NULL;
|
||||
|
||||
persist_config_t cfg = persist_config_defaults();
|
||||
cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */
|
||||
|
||||
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 wjd_teardown_persist(persist_t* p)
|
||||
{
|
||||
persist_stop(p);
|
||||
persist_fini(p);
|
||||
free(p);
|
||||
}
|
||||
|
||||
/* Encode helper mirroring test_apply_committed_entry.c's tace_encode_job /
|
||||
* test_replay_apply_seam.c's trr_encode_job (capnp_encode_job is file-static
|
||||
* in persist.c). */
|
||||
static int wjd_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;
|
||||
}
|
||||
|
||||
/* Appends ONLY WAL_JOB_SUBMIT (jobid PEND) + WAL_JOB_DISPATCH -- no
|
||||
* WAL_JOB_START anywhere in this log -- then replays it via a fresh
|
||||
* persist_t's persist_recover() (a real restart). This is the task's own
|
||||
* "Observable completion": the dispatch-time transition survives a restart
|
||||
* standing entirely on its own, with no dependence on the separate
|
||||
* worker-confirmed-start fact. */
|
||||
static void
|
||||
test_wjd_restart_replay_recovers_running_without_start_fact(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_wjd_replay_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
/* Phase 1: write the log to real disk. No cluster needed —
|
||||
* persist_wal_async is a physical-write-only path. */
|
||||
{
|
||||
persist_t* p = wjd_make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
|
||||
job_submit_data_t submit;
|
||||
memset(&submit, 0, sizeof(submit));
|
||||
submit.queue = (char*)"q";
|
||||
submit.user = (char*)"tester";
|
||||
|
||||
xce_job_t jdef;
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = 71003;
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = &submit;
|
||||
|
||||
uint8_t* job_payload = NULL;
|
||||
size_t job_plen = 0;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjd_encode_job(&jdef, &job_payload, &job_plen));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
persist_wal_async(p, WAL_JOB_SUBMIT, job_payload,
|
||||
(uint32_t)job_plen));
|
||||
free(job_payload);
|
||||
|
||||
/* The dispatch-time fact — no WAL_JOB_START anywhere in this log. */
|
||||
wal_jobstate_t js = { .jobid = 71003, .state = JOB_STAT_RUN,
|
||||
._pad = 0 };
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
persist_wal_async(p, WAL_JOB_DISPATCH, &js, sizeof(js)));
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
|
||||
wjd_teardown_persist(p);
|
||||
}
|
||||
|
||||
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk —
|
||||
* simulates a controller restart. */
|
||||
{
|
||||
persist_t* p2 = wjd_make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p2);
|
||||
|
||||
wjd_setup();
|
||||
p2->cluster = &g_wjd_c;
|
||||
|
||||
int rc = persist_recover(p2);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
job_t* j = wjd_find_job(71003);
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(j,
|
||||
"restart replay must construct the job via WAL_JOB_SUBMIT");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
|
||||
"restart replay must recover the job as RUNNING from the "
|
||||
"WAL_JOB_DISPATCH fact alone, with no WAL_JOB_START present "
|
||||
"in the log at all");
|
||||
|
||||
wjd_teardown();
|
||||
wjd_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_wjd_restart_replay_recovers_running_without_start_fact: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ───────────────────────────────────────────── */
|
||||
|
||||
void test_wal_job_dispatch_fact(void)
|
||||
{
|
||||
UnitySetTestFile(__FILE__);
|
||||
RUN_TEST(test_wjd_apply_committed_entry_transitions_pend_to_run);
|
||||
RUN_TEST(test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook);
|
||||
RUN_TEST(test_wjd_restart_replay_recovers_running_without_start_fact);
|
||||
}
|
||||
|
|
@ -114,6 +114,15 @@ typedef enum {
|
|||
* the whole array (base spec + index range),
|
||||
* distinct from WAL_JOB_SUBMIT's single-job
|
||||
* format, which stays unchanged. */
|
||||
WAL_JOB_DISPATCH = 12, /* scheduler-dispatch-commit: durable record of a
|
||||
* job's synchronous dispatch-time transition to
|
||||
* running, distinct from (and independent of)
|
||||
* WAL_JOB_START's later worker-confirmed-start
|
||||
* fact. Payload is wal_jobstate_t{jobid, state,
|
||||
* _pad=0} -- always the legacy absolute-value
|
||||
* form (no racing same-job fact to defer against,
|
||||
* since this is the first transition after a job
|
||||
* leaves PEND). */
|
||||
WAL_QUEUE_CTRL = 20,
|
||||
WAL_WORKER_CTRL = 30,
|
||||
WAL_NOOP = 40, /* Internal: empty payload, replay no-op; appended only at promotion */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue