test(ha-commit-barrier-rollout): receipt-ACK verdict pins (task 3.3)
4 green pins: handle_jobdispatch replies receipt-only with zero WAL appends through a real 2-alloc dispatch cycle (notify outcomes intact, reply-precedes-processing); set/clr_preason and do_update_op rusage stay receipt-style with RETAIN-pending fact appends ACK-committed. Umbrella Reply-Ordering Contract cited per pin. DISCOVERY bug-451 (reviewer-verified): RUSAGE hook va_list type confusion — 4-byte OOB read + garbage state in WAL_JOB_RUSAGE payloads; fix deferred to 7.1. Census 488; zero failures.
This commit is contained in:
parent
facdb90917
commit
7103fa643d
6 changed files with 865 additions and 327 deletions
|
|
@ -43,7 +43,7 @@
|
|||
- Observable completion: all task 3.1 tests PASS; `handle_jobsubmit` untouched (reference implementation)
|
||||
- _Requirements: 1.1, 1.2, 1.3, 1.4_
|
||||
|
||||
- [ ] 3.3 Pin the receipt-ACK verdicts with tests
|
||||
- [x] 3.3 Pin the receipt-ACK verdicts with tests
|
||||
- Tests: `handle_jobdispatch` (job.c:459-491) replies immediately AND dispatch processing appends no WAL entries (classification guard stays silent through a dispatch cycle; `notify_dispatch` outcomes unaffected); `do_set_preason`/`do_clr_preason` and the `do_update_op` rusage path remain receipt-style replies whose subsequent fact appends are RETAIN-registered
|
||||
- Document the verdict in the test comments with the umbrella Reply-Ordering Contract reference
|
||||
- Observable completion: verdict tests PASS without modifying those handlers; any future WAL append from dispatch processing trips the classification guard in CI
|
||||
|
|
@ -112,6 +112,7 @@
|
|||
- (1.1) wal_entry_type has 13 values (wal_types.h:44-58), no WAL_NOOP yet (lands in ha-single-leader-role, which must add a classification row + sweep entry — the totality sweep will catch it).
|
||||
- (1.1) Suite baseline is ZERO failures since ha-commit-engine 6.1. detect_leaks=0 remains the harness norm (6 inventoried pre-existing leak groups).
|
||||
- (1.2) CENSUS METHOD: only ~9 of 48 groups emit Unity summary blocks; the rest report per-test PASS lines only — count distinct Unity results (~465) and grep FAIL lines, never sum summary blocks. commit_submit(RETAIN) early-returns BEFORE taking ce->lock (commit.c:233-239) — registration is lock-free, safe with entity write locks held. wal_hook_append_classified is the single classified-append seam; appended_lsn read = current_lsn - 1 under p->lock (benign race, commented).
|
||||
- (3.3) **bug-451 DISCOVERED (reviewer-verified byte-precise, NOT fixed — boundary)**: update_job_rusages fires HOOK_DETAIL_JOB_RUSAGE_UPDATE with (cluster_t*, xce_job_rusages_t*) (job.c:1014-1015) but persist_job_hook_fun's default arm (persist.c:1355-1359) reads vararg 2 as job_t* and dereferences j->job.state at offset 56 — one past the end of the exactly-56-byte xce_job_rusages_t → 4-byte OOB read + garbage state field in every WAL_JOB_RUSAGE payload (jobid lines up by accident). Production fix candidate for task 7.1's sweep or a follow-up: persist arm needs a RUSAGE-specific va_list signature. Census now 488.
|
||||
- (3.2) bug-447 FIXED via rc normalization inside the three ctrl fns (real errors/misses preserved; reviewer-verified per-fn). Semantics change shipped intentionally: idempotent no-op control actions now return 0 (was leaked 1); queue/host mutations now append WAL + replicate. Join barrier failure → error reply + on_worker_join SKIPPED (coherent reading of design; worker retries via idempotent rejoin; documented at join.c:108-113). PRE-EXISTING ref leak: job_ctrl takes reference(j) (job.c:317) but normal exit never dereferences (bug-448) — out of scope, future fix.
|
||||
- (3.1) **bug-447 CONFIRMED (reviewer-verified to librtti disassembly)**: key_find returns 1 on found (unnormalized from xkvmgr_find/kvtab_find); queue_ctrl (queue.c:61, gate :106) and worker_ctrl (worker.c:644, gate :678) leak rc=1 → AFTER_EVENT never fires on the found path → **queue/host control silently unpersisted+unreplicated in production today**; job_ctrl appends via mark_job_* hooks but still replies status 1. Fix point = 3.2's handler rc normalization (key_find is in binary librtti, out of repo). RED tests pin it.
|
||||
- (3.1) Test link gained job.c/queue.c/join.c (test-only meson change) + 8 inert link stubs in test_commit_barrier_reply.c (none on tested paths — reviewer-traced). Reply capture = zeroed stack endpoint ep->outq inspection (test_config_repl.c:2210 precedent; endpoint_update_epoll no-ops when iomgr NULL). RED-window orphaned fsync threads self-resolve at GREEN (reviewer finding 5).
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
# anatomy.md
|
||||
|
||||
> Auto-maintained by OpenWolf. Last scanned: 2026-06-12T10:00:34.785Z
|
||||
> Files: 509 tracked | Anatomy hits: 0 | Misses: 0
|
||||
> Auto-maintained by OpenWolf. Last scanned: 2026-06-12T11:16:47.642Z
|
||||
> Files: 513 tracked | Anatomy hits: 0 | Misses: 0
|
||||
|
||||
## ./
|
||||
|
||||
|
|
@ -253,7 +253,7 @@
|
|||
- `requirements.md` — Requirements Document (~3181 tok)
|
||||
- `research.md` — Research & Design Decisions: ha-commit-barrier-rollout (~3672 tok)
|
||||
- `spec.json` (~143 tok)
|
||||
- `tasks.md` — Implementation Plan: ha-commit-barrier-rollout (~4337 tok)
|
||||
- `tasks.md` — Implementation Plan: ha-commit-barrier-rollout (~4653 tok)
|
||||
|
||||
## .kiro/specs/ha-commit-engine/
|
||||
|
||||
|
|
@ -706,9 +706,13 @@
|
|||
|
||||
## src/controller/
|
||||
|
||||
- `meson.build` (~1130 tok)
|
||||
- `job.c` — include <errno.h> (~8196 tok)
|
||||
- `join.c` — include "config.h" (~2530 tok)
|
||||
- `meson.build` (~1134 tok)
|
||||
- `queue.c` — include <errno.h> (~1451 tok)
|
||||
- `worker.c` — include <errno.h> (~5124 tok)
|
||||
|
||||
## src/controller/testsuite/
|
||||
|
||||
- `controller_test.c` — include <argp.h> (~1368 tok)
|
||||
- `test_commit_barrier_reply.c` — Task 3.1 RED integration tests: commit-first-then-reply per opcode (jobctrl/queuectrl/hostctrl/worker-join) via endpoint outq reply capture + pthread handler runner; pins the key_find rc==1 hook-skip gap (~8639 tok)
|
||||
- `test_commit_barrier_reply.c` — Task 3.1 commit-first-then-reply integration tests + task 3.3 receipt-ACK verdict pins (jobdispatch/preason/rusage); cluster harness w/ real persist, wired commit engine, stub standbys, reply-capture endpoint, notify/get_ep_worker capture stubs (~17500 tok)
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -937,3 +937,14 @@
|
|||
| 11:00 | Task 3.2 ha-commit-barrier-rollout: commit barrier in 4 handlers + bug-447 rc normalization in queue_ctrl/worker_ctrl/job_ctrl | src/controller/{job,queue,worker,join}.c | 12/12 barrier-reply tests PASS, full suite 0 FAIL, no sanitizer reports | ~60k |
|
||||
| 03:28 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~32 |
|
||||
| 03:28 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | 1→2 lines | ~172 |
|
||||
| 03:48 | Edited src/controller/testsuite/test_commit_barrier_reply.c | modified HOSTS() | ~231 |
|
||||
| 03:48 | Edited src/controller/testsuite/test_commit_barrier_reply.c | added 2 condition(s) | ~685 |
|
||||
| 03:48 | Edited src/controller/testsuite/test_commit_barrier_reply.c | modified cbrp_make_job() | ~213 |
|
||||
| 03:49 | Edited src/controller/testsuite/test_commit_barrier_reply.c | 4→6 lines | ~64 |
|
||||
| 03:50 | Edited src/controller/testsuite/test_commit_barrier_reply.c | modified reference() | ~4973 |
|
||||
| 03:50 | Edited src/controller/testsuite/test_commit_barrier_reply.c | 6→11 lines | ~154 |
|
||||
| 03:51 | Edited src/controller/testsuite/test_commit_barrier_reply.c | 3→4 lines | ~22 |
|
||||
| 03:52 | Edited src/controller/meson.build | 12→13 lines | ~48 |
|
||||
| 11:06 | Task 3.3 receipt-ACK verdict pins: 4 GREEN pins (jobdispatch no-WAL receipt ACK, set/clr_preason, do_update_op rusage RETAIN) added to test_commit_barrier_reply.c; opupd.c added to test link | src/controller/testsuite/test_commit_barrier_reply.c, src/controller/meson.build | suite green 488 distinct PASS / 0 FAIL; discovered bug-451 (rusage hook va-arg type confusion) | ~95k |
|
||||
| 04:16 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~14 |
|
||||
| 04:16 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | 1→2 lines | ~184 |
|
||||
|
|
|
|||
|
|
@ -127,6 +127,7 @@ controller_unit_test_src = [
|
|||
'job.c',
|
||||
'queue.c',
|
||||
'join.c',
|
||||
'opupd.c',
|
||||
'startup_validation.c',
|
||||
'worker.c',
|
||||
'persist.c',
|
||||
|
|
|
|||
|
|
@ -5,6 +5,14 @@
|
|||
* timeout → -ETIMEDOUT error reply), 1.3 (barrier rejection → error reply,
|
||||
* never success), 1.4 (no success-interpretable reply pre-commit).
|
||||
*
|
||||
* ALSO HOSTS (task 3.3, Req 2.3): the receipt-ACK verdict pins for
|
||||
* handle_jobdispatch, do_set_preason / do_clr_preason, and the
|
||||
* do_update_op rusage path — see the "Receipt-ACK verdict pins" section.
|
||||
* These land GREEN by design (pins of current correct behavior, like the
|
||||
* 3.1 validation pins); they live here, not in a new file, because they
|
||||
* reuse this file's entire harness: cluster + real persist + wired engine
|
||||
* + stub standby registry + reply-capture endpoint + the job.c link.
|
||||
*
|
||||
* Drives the four request-shaped handlers DIRECTLY on a worker pthread
|
||||
* (vtask-context stand-in; production dispatch: cluster.c:325-363) against
|
||||
* a real tmpdir persist_t embedded in a test cluster_t with registered
|
||||
|
|
@ -83,21 +91,66 @@
|
|||
#include "typemap.h"
|
||||
#include "wal_types.h"
|
||||
#include "wobj.h"
|
||||
#include "xcealloc.h"
|
||||
#include "xstring.h"
|
||||
#include "zarray.h"
|
||||
|
||||
/* ── Link stubs ──────────────────────────────────────────────────────────
|
||||
* job.c / join.c reference these cross-module functions (opupd.c, sched.c,
|
||||
* apiserver.c — not part of the unit-test link). None of them sits on the
|
||||
* four code paths under test (jobctrl not-started branch, queue/host ctrl,
|
||||
* worker join); stubbed so the binary links, same precedent as
|
||||
* test_worker_join.c's collect_confvars stub. */
|
||||
* job.c / join.c / opupd.c reference these cross-module functions (dataex.c,
|
||||
* misc.c, sched.c, apiserver.c — not part of the unit-test link). For the
|
||||
* 3.1 paths (jobctrl not-started branch, queue/host ctrl, worker join) all
|
||||
* stubs are inert; for the 3.3 verdict pins two of them gained test
|
||||
* controls with NULL/no-op defaults that preserve the 3.1 behavior:
|
||||
* notify() — captures dispatch-outcome notifications (jobdispatch
|
||||
* verdict pin observes notify_dispatch through it)
|
||||
* get_ep_worker() — returns g_stub_ep_worker (rusage verdict pin needs a
|
||||
* reporting worker; production impl is misc.c:49) */
|
||||
|
||||
static endpoint_t g_req_ep; /* requester endpoint (capture; set up below) */
|
||||
|
||||
/* notify() capture — records each xce_notify_data_t delivery. For
|
||||
* jobDispatchFail events (the shared ok/fail event of notify_dispatch,
|
||||
* job.c) jobid + code are unpacked; reply_already_enqueued snapshots
|
||||
* whether the receipt ACK was already sitting on the requester outq when
|
||||
* the outcome notification fired (single-threaded handler call → exact
|
||||
* ordering observation, no race). */
|
||||
#define CBRP_MAX_NOTIFY 8
|
||||
static int g_notify_count;
|
||||
static int g_notify_events[CBRP_MAX_NOTIFY];
|
||||
static uint64_t g_notify_jobids[CBRP_MAX_NOTIFY];
|
||||
static int g_notify_codes[CBRP_MAX_NOTIFY];
|
||||
static int g_notify_reply_already_enqueued[CBRP_MAX_NOTIFY];
|
||||
|
||||
static void cbrp_notify_capture_reset(void)
|
||||
{
|
||||
g_notify_count = 0;
|
||||
memset(g_notify_events, 0, sizeof(g_notify_events));
|
||||
memset(g_notify_jobids, 0, sizeof(g_notify_jobids));
|
||||
memset(g_notify_codes, 0, sizeof(g_notify_codes));
|
||||
memset(g_notify_reply_already_enqueued, 0,
|
||||
sizeof(g_notify_reply_already_enqueued));
|
||||
}
|
||||
|
||||
static worker_t *g_stub_ep_worker; /* NULL by default (3.1 behavior) */
|
||||
|
||||
/* clang-format off */
|
||||
int notify(cluster_t *c, endpoint_t *e, xce_notify_data_t *d)
|
||||
{ (void)c; (void)e; (void)d; return 0; }
|
||||
{
|
||||
(void)c; (void)e;
|
||||
if (g_notify_count < CBRP_MAX_NOTIFY) {
|
||||
g_notify_events[g_notify_count] = d->event;
|
||||
if (d->event == NotifyData_data_jobDispatchFail) {
|
||||
g_notify_jobids[g_notify_count] = d->data.dispatchfail->jobid;
|
||||
g_notify_codes[g_notify_count] = d->data.dispatchfail->code;
|
||||
}
|
||||
g_notify_reply_already_enqueued[g_notify_count] =
|
||||
!DATAQ_EMPTY(&g_req_ep.outq);
|
||||
}
|
||||
g_notify_count++;
|
||||
return 0;
|
||||
}
|
||||
worker_t *get_ep_worker(cluster_t *c, endpoint_t *e)
|
||||
{ (void)c; (void)e; return NULL; }
|
||||
{ (void)c; (void)e; return g_stub_ep_worker; }
|
||||
int update_host_runusage(xce_job_runusage_t **r, char *h, int n,
|
||||
xce_rusage_t **u)
|
||||
{ (void)r; (void)h; (void)n; (void)u; return 0; }
|
||||
|
|
@ -277,6 +330,8 @@ static void cbrp_setup(uint32_t sync_timeout_ms)
|
|||
memset(&g_req_ep, 0, sizeof(g_req_ep));
|
||||
memset(&g_req_priv, 0, sizeof(g_req_priv));
|
||||
memset(&g_req_from, 0, sizeof(g_req_from));
|
||||
cbrp_notify_capture_reset();
|
||||
g_stub_ep_worker = NULL;
|
||||
|
||||
/* Requester endpoint: zeroed, no iomgr — xce_reply packets stay on
|
||||
* ep->outq (see file header). */
|
||||
|
|
@ -385,21 +440,30 @@ static xce_jobctrl_t g_jobctrl;
|
|||
static xce_request_t g_jobctrl_req;
|
||||
static job_submit_data_t g_job_submit; /* job_init requires submit != NULL */
|
||||
|
||||
static int g_jobobj_ready; /* jobj module tables — once per process */
|
||||
|
||||
/* Create a pending (never-started) job in g_hc.jobmgr. submit may carry
|
||||
* queue/user/start_info (deep-copied by makeJob via makeJobsub). */
|
||||
static void cbrp_make_job(uint64_t jobid, job_submit_data_t *submit)
|
||||
{
|
||||
xce_job_t jdef;
|
||||
|
||||
if (!g_jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64));
|
||||
g_jobobj_ready = 1;
|
||||
}
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = jobid;
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = submit;
|
||||
TEST_ASSERT_NOT_NULL(makeJob(g_hc.jobmgr, &jdef));
|
||||
}
|
||||
|
||||
static void cbrp_arrange_jobctrl(uint64_t jobid, int create_job)
|
||||
{
|
||||
if (create_job) {
|
||||
xce_job_t jdef;
|
||||
static int jobobj_ready; /* jobj module tables — once per process */
|
||||
if (!jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64));
|
||||
jobobj_ready = 1;
|
||||
}
|
||||
memset(&g_job_submit, 0, sizeof(g_job_submit));
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = jobid;
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = &g_job_submit;
|
||||
TEST_ASSERT_NOT_NULL(makeJob(g_hc.jobmgr, &jdef));
|
||||
cbrp_make_job(jobid, &g_job_submit);
|
||||
}
|
||||
memset(&g_jobctrl, 0, sizeof(g_jobctrl));
|
||||
g_jobctrl.action = JobControl_u_siginfo;
|
||||
|
|
@ -768,6 +832,418 @@ static void test_worker_join_no_hostconfig_error_reply_immediate(void)
|
|||
printf("test_worker_join_no_hostconfig_error_reply_immediate: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Receipt-ACK verdict pins (ha-commit-barrier-rollout task 3.3, Req 2.3)
|
||||
*
|
||||
* Umbrella Reply-Ordering Contract reference (ha-consistency-invariants
|
||||
* design.md §Write-Path Classification ▸ Reply-Ordering Contract):
|
||||
* handle_jobdispatch's early reply is an ACK-of-receipt and MAY STAND,
|
||||
* because dispatch outcomes are fact-shaped — per-job results are
|
||||
* delivered via notify_dispatch, and the resulting job-state changes
|
||||
* arrive later as worker-reported facts. The same verdict covers
|
||||
* do_set_preason / do_clr_preason (scheduler-origin WAL_JOB_PREASON
|
||||
* facts, RETAIN) and the do_update_op rusage path (worker-origin
|
||||
* WAL_JOB_UPDATE + WAL_JOB_RUSAGE facts, RETAIN): their replies
|
||||
* acknowledge RECEIPT of the report, never commitment of any
|
||||
* state-changing log entry.
|
||||
*
|
||||
* These are PINS of current correct behavior — green on landing, no RED
|
||||
* phase exists (task-stated design, like the 3.1 validation pins). They
|
||||
* run with ACKs withheld throughout, so any future regression that makes
|
||||
* one of these handlers wait on the commit barrier trips the latency
|
||||
* bound, and any future WAL append from dispatch processing trips either
|
||||
* the unchanged-LSN pin or (for an unclassified type) the XCE_TESTING
|
||||
* abort in wal_classify — the classification guard stays silent through
|
||||
* a clean dispatch cycle.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* ── handle_jobdispatch fixture ──────────────────────────────────────── */
|
||||
|
||||
static endpoint_t g_wkr_disp_ep; /* dispatch worker endpoint */
|
||||
static job_submit_data_t g_disp_submit;
|
||||
static job_startinfo_t g_disp_startinfo;
|
||||
static xce_job_dispatch_t g_dispatchmsg;
|
||||
static xce_job_alloc_t *g_dispatch_allocs[2];
|
||||
static xce_request_t g_dispatch_req;
|
||||
|
||||
#define CBRP_DISP_JOBID 60u
|
||||
|
||||
/* Full dispatch fixture: queue (get_queue_exec), pending job with a real
|
||||
* submit (dispatch_a_job_ reads submit->queue and submit->start_info),
|
||||
* worker with a live no-iomgr endpoint (the startjob request must be
|
||||
* enqueueable), and a 2-job dispatch message exercising BOTH outcomes:
|
||||
* allocs[0] — dispatchable to wkr-disp → notify_dispatch(ok, code 0)
|
||||
* allocs[1] — nhosts == 0 → notify_dispatch(fail, -EINVAL)
|
||||
* The allocs are heap-built via xcealloc helpers because the handler
|
||||
* takes ownership (MOVE_PTR + free_JobAlloc_ptr per job). */
|
||||
static void cbrp_arrange_jobdispatch(void)
|
||||
{
|
||||
xce_queue_t qdef;
|
||||
xce_worker_t wdef;
|
||||
hostattrs_t *attrs = NULL;
|
||||
worker_t *w = NULL;
|
||||
|
||||
memset(&qdef, 0, sizeof(qdef));
|
||||
qdef.queue = (char *)"qd";
|
||||
TEST_ASSERT_NOT_NULL(makeQueue(g_hc.qmgr, &qdef));
|
||||
|
||||
memset(&g_disp_startinfo, 0, sizeof(g_disp_startinfo));
|
||||
g_disp_startinfo.cmdline = (char *)"/bin/true";
|
||||
memset(&g_disp_submit, 0, sizeof(g_disp_submit));
|
||||
g_disp_submit.queue = (char *)"qd";
|
||||
g_disp_submit.user = (char *)"tester";
|
||||
g_disp_submit.start_info = &g_disp_startinfo;
|
||||
cbrp_make_job(CBRP_DISP_JOBID, &g_disp_submit);
|
||||
|
||||
attrs = calloc(1, sizeof(*attrs));
|
||||
TEST_ASSERT_NOT_NULL(attrs);
|
||||
attrs->hosttype = makeString("LINUX");
|
||||
attrs->hostmodel = makeString("x86");
|
||||
attrs->cpufactor = 1.0f;
|
||||
attrs->maxcpus = 4;
|
||||
memset(&wdef, 0, sizeof(wdef));
|
||||
wdef.name = (char *)"wkr-disp";
|
||||
wdef.attrs = attrs;
|
||||
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
|
||||
|
||||
/* dispatch gate: worker must have an endpoint and status 0 */
|
||||
memset(&g_wkr_disp_ep, 0, sizeof(g_wkr_disp_ep));
|
||||
g_wkr_disp_ep.fd = -1;
|
||||
lock(g_hc.workermgr, RTTI_LOCK_READ);
|
||||
key_find(g_hc.workermgr, (void *)"wkr-disp", (void **)&w);
|
||||
unlock(g_hc.workermgr);
|
||||
TEST_ASSERT_NOT_NULL(w);
|
||||
w->ep = &g_wkr_disp_ep;
|
||||
|
||||
g_dispatch_allocs[0] = make_job_alloc(
|
||||
CBRP_DISP_JOBID, 1,
|
||||
make_host_allocs(1,
|
||||
make_host_alloc((char *)"wkr-disp", 1,
|
||||
make_allocs(1,
|
||||
make_alloc((char *)"slots", NULL, 0, 1.0f, 0)))));
|
||||
TEST_ASSERT_NOT_NULL(g_dispatch_allocs[0]);
|
||||
g_dispatch_allocs[1] = make_job_alloc(CBRP_DISP_JOBID, 0, NULL);
|
||||
TEST_ASSERT_NOT_NULL(g_dispatch_allocs[1]);
|
||||
|
||||
memset(&g_dispatchmsg, 0, sizeof(g_dispatchmsg));
|
||||
g_dispatchmsg.njobs = 2;
|
||||
g_dispatchmsg.allocs = g_dispatch_allocs;
|
||||
memset(&g_dispatch_req, 0, sizeof(g_dispatch_req));
|
||||
g_dispatch_req.opcode = OPCode_jobdispatch;
|
||||
g_dispatch_req.seq = 15;
|
||||
g_dispatch_req.from = &g_req_from;
|
||||
g_dispatch_req.body_which = Request_body_dispatchmsg;
|
||||
g_dispatch_req.body.dispatchmsg = &g_dispatchmsg;
|
||||
}
|
||||
|
||||
/* VERDICT PIN (Req 2.3): handle_jobdispatch (job.c:468-500, re-verified
|
||||
* 2026-06-12) replies status=0 IMMEDIATELY — an ACK-of-receipt per the
|
||||
* umbrella Reply-Ordering Contract (see section header) — and a full
|
||||
* dispatch cycle (one successful + one failed allocation) appends NO WAL
|
||||
* entries: allocation results return via notify_dispatch and job-state
|
||||
* changes arrive later as worker facts. ACKs are withheld for the whole
|
||||
* test, so a regression that adds a barrier wait trips the latency bound
|
||||
* and any WAL append from dispatch processing trips the unchanged-LSN
|
||||
* assert (an unclassified type would abort in wal_classify first). */
|
||||
static void test_jobdispatch_receipt_ack_no_wal_append(void)
|
||||
{
|
||||
cbrp_reply_obs_t obs = { 0 };
|
||||
uint64_t lsn_before, lsn_after, t0, elapsed_ms;
|
||||
int rc;
|
||||
|
||||
cbrp_setup(3000);
|
||||
cbrp_arrange_jobdispatch();
|
||||
cbrp_notify_capture_reset(); /* drop any arrange-time noise */
|
||||
|
||||
pthread_mutex_lock(&g_hc.persist.lock);
|
||||
lsn_before = g_hc.persist.current_lsn;
|
||||
pthread_mutex_unlock(&g_hc.persist.lock);
|
||||
|
||||
/* Act: synchronous — the receipt ACK must never block on commit */
|
||||
t0 = cbrp_mono_ms();
|
||||
rc = handle_jobdispatch(&g_hc, &g_req_ep, &g_dispatch_req);
|
||||
elapsed_ms = cbrp_mono_ms() - t0;
|
||||
|
||||
pthread_mutex_lock(&g_hc.persist.lock);
|
||||
lsn_after = g_hc.persist.current_lsn;
|
||||
pthread_mutex_unlock(&g_hc.persist.lock);
|
||||
cbrp_scan_replies(&g_req_ep, &obs);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handle_jobdispatch rc");
|
||||
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
|
||||
"receipt ACK must be latency-bounded — no commit barrier wait "
|
||||
"(Req 2.3: the dispatch reply confirms receipt only)");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
|
||||
"scheduler must receive exactly the one receipt ACK");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status,
|
||||
"the receipt ACK reports acceptance of the message (status 0)");
|
||||
|
||||
/* THE classification pin: dispatch processing appends no WAL entries,
|
||||
* so there is nothing the receipt ACK could have falsely confirmed */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
|
||||
"a dispatch cycle must append NO WAL entries (Req 2.3 — any "
|
||||
"future append from dispatch processing must move behind the "
|
||||
"barrier or be re-classified; the wal_classify XCE_TESTING "
|
||||
"guard backstops unclassified types)");
|
||||
|
||||
/* notify_dispatch outcomes unaffected: one ok + one fail, in order */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_notify_count,
|
||||
"both allocations must produce their per-job outcome notification");
|
||||
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
|
||||
g_notify_events[0]);
|
||||
TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
|
||||
g_notify_events[1]);
|
||||
TEST_ASSERT_EQUAL_UINT64(CBRP_DISP_JOBID, g_notify_jobids[0]);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_notify_codes[0],
|
||||
"allocs[0] dispatch to the live worker must succeed (code 0)");
|
||||
TEST_ASSERT_EQUAL_UINT64(CBRP_DISP_JOBID, g_notify_jobids[1]);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, g_notify_codes[1],
|
||||
"allocs[1] (nhosts == 0) must fail with -EINVAL");
|
||||
|
||||
/* Ordering: the receipt ACK was already enqueued when the FIRST
|
||||
* dispatch outcome fired — reply precedes processing (ACK-of-receipt,
|
||||
* not ACK-of-outcome) */
|
||||
TEST_ASSERT_TRUE_MESSAGE(g_notify_reply_already_enqueued[0],
|
||||
"the receipt ACK must be enqueued before any dispatch outcome "
|
||||
"notification (reply-before-processing is the verdict)");
|
||||
|
||||
/* drain the startjob request enqueued on the worker endpoint */
|
||||
{
|
||||
cbrp_reply_obs_t wdrain = { 0 };
|
||||
cbrp_scan_replies(&g_wkr_disp_ep, &wdrain);
|
||||
}
|
||||
cbrp_teardown();
|
||||
printf("test_jobdispatch_receipt_ack_no_wal_append: PASS\n");
|
||||
}
|
||||
|
||||
/* ── do_set_preason / do_clr_preason fixtures ────────────────────────── */
|
||||
|
||||
static xce_pendreason_t g_pendreason;
|
||||
static xce_pendreason_t *g_pendreasons[1];
|
||||
static xce_preason_set_t g_preason_set;
|
||||
static xce_request_t g_preason_req;
|
||||
static uint64_t g_clr_jobs[1];
|
||||
static xce_clear_reason_t g_clrreason;
|
||||
static xce_request_t g_clrpreason_req;
|
||||
|
||||
/* set_preason MOVEs the inner xce_job_preason_t into the job (job.c
|
||||
* MOVE_PTR(job->pend_reasons, ...)) and the job/do_clr_preason frees it
|
||||
* later — so it must be heap-allocated; the carrier structs stay static. */
|
||||
static void cbrp_arrange_set_preason(uint64_t jobid)
|
||||
{
|
||||
xce_job_preason_t *jpr = calloc(1, sizeof(*jpr));
|
||||
|
||||
TEST_ASSERT_NOT_NULL(jpr);
|
||||
jpr->main_reason = 7; /* arbitrary scheduler pend reason */
|
||||
|
||||
memset(&g_pendreason, 0, sizeof(g_pendreason));
|
||||
g_pendreason.jobid = jobid;
|
||||
g_pendreason.reasons = jpr;
|
||||
g_pendreasons[0] = &g_pendreason;
|
||||
memset(&g_preason_set, 0, sizeof(g_preason_set));
|
||||
g_preason_set.nreasons = 1;
|
||||
g_preason_set.reasons = g_pendreasons;
|
||||
memset(&g_preason_req, 0, sizeof(g_preason_req));
|
||||
g_preason_req.opcode = OPCode_setpreason;
|
||||
g_preason_req.seq = 16;
|
||||
g_preason_req.from = &g_req_from;
|
||||
g_preason_req.body_which = Request_body_preasonmsg;
|
||||
g_preason_req.body.preasonmsg = &g_preason_set;
|
||||
}
|
||||
|
||||
static void cbrp_arrange_clr_preason(uint64_t jobid)
|
||||
{
|
||||
g_clr_jobs[0] = jobid;
|
||||
memset(&g_clrreason, 0, sizeof(g_clrreason));
|
||||
g_clrreason.njobs = 1;
|
||||
g_clrreason.jobs = g_clr_jobs;
|
||||
memset(&g_clrpreason_req, 0, sizeof(g_clrpreason_req));
|
||||
g_clrpreason_req.opcode = OPCode_clrpreason;
|
||||
g_clrpreason_req.seq = 17;
|
||||
g_clrpreason_req.from = &g_req_from;
|
||||
g_clrpreason_req.body_which = Request_body_clrpreasonmsg;
|
||||
g_clrpreason_req.body.clrpreasonmsg = &g_clrreason;
|
||||
}
|
||||
|
||||
/* Shared receipt-style fact-handler driver: with ACKs withheld the
|
||||
* handler must (a) reply status 0 latency-bounded — receipt only, no
|
||||
* barrier wait — (b) append exactly expect_appends fact entries that
|
||||
* remain PENDING (lsn > committed_lsn: the reply confirmed nothing about
|
||||
* commitment), and (c) once a covering ACK arrives the retained suffix
|
||||
* commits with no redelivery (the RETAIN contract observable per spec
|
||||
* Implementation Notes 1.1: registration-free, pending purely by
|
||||
* lsn > committed_lsn, ACK-driven advance). */
|
||||
static void cbrp_drive_receipt_fact(cbrp_op_fn_t fn, xce_request_t *req,
|
||||
uint64_t expect_appends)
|
||||
{
|
||||
cbrp_reply_obs_t obs = { 0 };
|
||||
uint64_t lsn_before, lsn_after, watermark, t0, elapsed_ms;
|
||||
uint64_t deadline;
|
||||
int rc;
|
||||
|
||||
pthread_mutex_lock(&g_hc.persist.lock);
|
||||
lsn_before = g_hc.persist.current_lsn;
|
||||
pthread_mutex_unlock(&g_hc.persist.lock);
|
||||
|
||||
/* Act: synchronous — receipt-style handlers must never block */
|
||||
t0 = cbrp_mono_ms();
|
||||
rc = fn(&g_hc, &g_req_ep, req);
|
||||
elapsed_ms = cbrp_mono_ms() - t0;
|
||||
|
||||
pthread_mutex_lock(&g_hc.persist.lock);
|
||||
lsn_after = g_hc.persist.current_lsn;
|
||||
pthread_mutex_unlock(&g_hc.persist.lock);
|
||||
watermark = lsn_after - 1; /* last appended entry */
|
||||
cbrp_scan_replies(&g_req_ep, &obs);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "fact handler rc");
|
||||
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
|
||||
"receipt-style reply must be latency-bounded — no commit barrier "
|
||||
"wait with ACKs withheld (facts have no commit timeout, Req 2.3)");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies,
|
||||
"the reporter must receive exactly one receipt reply");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status,
|
||||
"the receipt reply reports acceptance (status 0)");
|
||||
|
||||
/* the fact appends happened ... */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + expect_appends, lsn_after,
|
||||
"the handler must append exactly its fact entries");
|
||||
/* ... and are RETAIN-pending: the receipt reply coexists with the
|
||||
* uncommitted suffix, proving it confirmed receipt, not commitment */
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
atomic_load(&g_hc.persist.committed_lsn) < watermark,
|
||||
"fact entries must remain pending (lsn > committed_lsn) — the "
|
||||
"receipt reply must not imply commitment (Req 2.3)");
|
||||
TEST_ASSERT_TRUE_MESSAGE(obs.committed_at_obs < watermark,
|
||||
"the receipt reply was observable while the fact entries were "
|
||||
"still uncommitted — receipt-only semantics pinned");
|
||||
|
||||
/* RETAIN completion: a covering ACK (self + 1 standby = majority of
|
||||
* n_peers=3) advances committed_lsn over the retained suffix with no
|
||||
* redelivery from the reporter */
|
||||
repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark);
|
||||
deadline = cbrp_mono_ms() + 2000;
|
||||
while (atomic_load(&g_hc.persist.committed_lsn) < watermark
|
||||
&& cbrp_mono_ms() < deadline)
|
||||
cbrp_sleep_ms(5);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
atomic_load(&g_hc.persist.committed_lsn) >= watermark,
|
||||
"a covering ACK must commit the retained fact entries "
|
||||
"(RETAIN registration contract)");
|
||||
}
|
||||
|
||||
/* VERDICT PIN (Req 2.3): do_set_preason (job.c:1062) replies status=0
|
||||
* receipt-style BEFORE applying the scheduler-reported pend reasons; the
|
||||
* subsequent WAL_JOB_PREASON fact append (HOOK_DETAIL_SETPREASON →
|
||||
* persist_job_hook_fun, RETAIN per wal_classify) stays pending until
|
||||
* ACK-covered. Umbrella Reply-Ordering Contract: receipt-only ACK for a
|
||||
* fact-shaped delivery (see section header). */
|
||||
static void test_set_preason_receipt_reply_fact_retained(void)
|
||||
{
|
||||
cbrp_setup(3000);
|
||||
cbrp_arrange_jobctrl(70, 1); /* job 70 exists; jobctrl req unused */
|
||||
cbrp_arrange_set_preason(70);
|
||||
cbrp_drive_receipt_fact(do_set_preason, &g_preason_req, 1);
|
||||
cbrp_teardown();
|
||||
printf("test_set_preason_receipt_reply_fact_retained: PASS\n");
|
||||
}
|
||||
|
||||
/* VERDICT PIN (Req 2.3): do_clr_preason (job.c:1090) — same receipt-style
|
||||
* verdict as do_set_preason; the clear is also a scheduler-origin
|
||||
* WAL_JOB_PREASON fact (HOOK_DETAIL_CLRPREASON, RETAIN). */
|
||||
static void test_clr_preason_receipt_reply_fact_retained(void)
|
||||
{
|
||||
cbrp_reply_obs_t drain = { 0 };
|
||||
|
||||
cbrp_setup(3000);
|
||||
cbrp_arrange_jobctrl(71, 1);
|
||||
/* seed pend reasons so the clear exercises the real erase path */
|
||||
cbrp_arrange_set_preason(71);
|
||||
TEST_ASSERT_EQUAL_INT(0, do_set_preason(&g_hc, &g_req_ep,
|
||||
&g_preason_req));
|
||||
cbrp_scan_replies(&g_req_ep, &drain); /* drop the seeding reply */
|
||||
|
||||
cbrp_arrange_clr_preason(71);
|
||||
cbrp_drive_receipt_fact(do_clr_preason, &g_clrpreason_req, 1);
|
||||
cbrp_teardown();
|
||||
printf("test_clr_preason_receipt_reply_fact_retained: PASS\n");
|
||||
}
|
||||
|
||||
/* ── do_update_op rusage path fixture ────────────────────────────────── */
|
||||
|
||||
/* KNOWN PRE-EXISTING MISMATCH (discovered while pinning, reported in the
|
||||
* task evidence — NOT fixed here, production is out of boundary):
|
||||
* update_job_rusages fires the AFTER_EVENT HOOK_DETAIL_JOB_RUSAGE_UPDATE
|
||||
* hook with (cluster_t*, xce_job_rusages_t*) [job.c:1015-1016], but
|
||||
* persist_job_hook_fun's default arm reads the second vararg as a job_t*
|
||||
* [persist.c:1354-1359] and dereferences j->job.state at offset 56 —
|
||||
* exactly sizeof(xce_job_rusages_t) — an out-of-bounds read on an
|
||||
* exactly-sized allocation (the WAL_JOB_RUSAGE payload's state field is
|
||||
* garbage). The padded box below keeps THIS pin's read in-bounds under
|
||||
* ASAN without masking the report; the jobid field (offset 0 in both
|
||||
* types) lines up by accident. */
|
||||
static union {
|
||||
xce_job_rusages_t r;
|
||||
unsigned char pad[sizeof(xce_job_rusages_t) + 64];
|
||||
} g_rusages_box;
|
||||
|
||||
static xce_update_data_t g_updatemsg;
|
||||
static xce_request_t g_update_req;
|
||||
|
||||
static void cbrp_arrange_update_rusage(uint64_t jobid)
|
||||
{
|
||||
xce_worker_t wdef;
|
||||
hostattrs_t *attrs = NULL;
|
||||
worker_t *w = NULL;
|
||||
|
||||
/* reporting worker — get_ep_worker (stubbed) must return it */
|
||||
attrs = calloc(1, sizeof(*attrs));
|
||||
TEST_ASSERT_NOT_NULL(attrs);
|
||||
attrs->hosttype = makeString("LINUX");
|
||||
attrs->hostmodel = makeString("x86");
|
||||
attrs->cpufactor = 1.0f;
|
||||
attrs->maxcpus = 4;
|
||||
memset(&wdef, 0, sizeof(wdef));
|
||||
wdef.name = (char *)"wkr-rus";
|
||||
wdef.attrs = attrs;
|
||||
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
|
||||
lock(g_hc.workermgr, RTTI_LOCK_READ);
|
||||
key_find(g_hc.workermgr, (void *)"wkr-rus", (void **)&w);
|
||||
unlock(g_hc.workermgr);
|
||||
TEST_ASSERT_NOT_NULL(w);
|
||||
g_stub_ep_worker = w;
|
||||
|
||||
memset(&g_rusages_box, 0, sizeof(g_rusages_box));
|
||||
g_rusages_box.r.jobid = jobid; /* count/npids/npgids 0 — stubs no-op */
|
||||
memset(&g_updatemsg, 0, sizeof(g_updatemsg));
|
||||
g_updatemsg.table = Update_data_jobrusages;
|
||||
g_updatemsg.data.jobrusages = &g_rusages_box.r;
|
||||
memset(&g_update_req, 0, sizeof(g_update_req));
|
||||
g_update_req.opcode = OPCode_update;
|
||||
g_update_req.seq = 18;
|
||||
g_update_req.from = &g_req_from;
|
||||
g_update_req.body_which = Request_body_updatemsg;
|
||||
g_update_req.body.updatemsg = &g_updatemsg;
|
||||
}
|
||||
|
||||
/* VERDICT PIN (Req 2.3): the do_update_op rusage path (opupd.c:69 →
|
||||
* do_update_jobrusage → update_job_rusages) replies receipt-style — the
|
||||
* reply carries the LOCAL application result, never a commit barrier
|
||||
* result — and its two fact appends (HOOK_DETAIL_JOB_UPDATE →
|
||||
* WAL_JOB_UPDATE, HOOK_DETAIL_JOB_RUSAGE_UPDATE → WAL_JOB_RUSAGE, both
|
||||
* RETAIN per wal_classify) stay pending until ACK-covered. Umbrella
|
||||
* Reply-Ordering Contract: receipt-only ACK for fact-shaped delivery. */
|
||||
static void test_update_op_rusage_receipt_reply_facts_retained(void)
|
||||
{
|
||||
cbrp_setup(3000);
|
||||
cbrp_arrange_jobctrl(72, 1); /* job 72 exists; jobctrl req unused */
|
||||
cbrp_arrange_update_rusage(72);
|
||||
cbrp_drive_receipt_fact(do_update_op, &g_update_req, 2);
|
||||
g_stub_ep_worker = NULL;
|
||||
cbrp_teardown();
|
||||
printf("test_update_op_rusage_receipt_reply_facts_retained: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ──────────────────────────────────────────── */
|
||||
|
||||
void test_commit_barrier_reply(void)
|
||||
|
|
@ -796,5 +1272,10 @@ void test_commit_barrier_reply(void)
|
|||
RUN_TEST(test_worker_join_withheld_acks_replies_error);
|
||||
RUN_TEST(test_worker_join_success_reply_only_after_commit);
|
||||
RUN_TEST(test_worker_join_no_hostconfig_error_reply_immediate);
|
||||
/* Receipt-ACK verdict pins (task 3.3, Req 2.3) — green on landing */
|
||||
RUN_TEST(test_jobdispatch_receipt_ack_no_wal_append);
|
||||
RUN_TEST(test_set_preason_receipt_reply_fact_retained);
|
||||
RUN_TEST(test_clr_preason_receipt_reply_fact_retained);
|
||||
RUN_TEST(test_update_op_rusage_receipt_reply_facts_retained);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue