diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index e56a7af7..d9815ae8 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -121,7 +121,7 @@ - _Requirements: 1.1, 1.2, 1.3, 2.1, 2.2, 3.7_ - _Depends: 1.4, 3.1, 4.1_ -- [ ] 6.2 Route an incoming job-state report to reconciliation processing +- [x] 6.2 Route an incoming job-state report to reconciliation processing - Dispatch an incoming job-state report message to the reconciliation-processing logic, correctly reassembling a report that arrives across multiple pages before it is applied - Confirm whether the two new message kinds need an explicit redirect response when received by a controller that is not currently primary, matching how every other primary-only message already behaves for a standby recipient — do not assume the existing generic rejection path already covers this without checking, given this exact class of gap has recurred for other message kinds before - Observable completion: a report sent across several pages is fully reassembled and applied exactly once, with the same result as an equivalent single-page report; and a query or report sent to a non-primary controller receives the same redirect behavior an equivalent primary-only message already receives, not a bare unrecoverable error @@ -167,3 +167,5 @@ - Task 4.3 (advisory, non-blocking, not fixed in this pass): `resync_pending`/`query_accum` in `xcecd_resync.c` are plain unsynchronized statics touched from two different OS threads — the worker's dedicated connector thread (`checking_connection()`/`xcecd_on_reconnect()`) and a `vtaskmgr` pool thread (`xcecd_resync_handle_query_page()`, reached via the request-dispatch path). This mirrors the pre-existing, design-directed unsynchronized `needs_rejoin` pattern (same codebase convention, deliberately reused per design.md), but is riskier since `xcecd_resync_reset()` calls `free()` on the accumulator rather than just staling a bool read — a narrow, non-deterministic window exists where an independent reconnect could free the buffer while a pool thread is mid-read of the same connection's in-flight query page. Does not reproduce task 4.3's own violated requirement (Req 6.1) and isn't required by any Requirement/AC as written; this project's ASAN/UBSAN test harness has no TSAN, so it also isn't mechanically testable the way the deadlock fix was. Worth a small follow-up (a mutex around the gate+accumulator, or moving accumulation to the connector thread only) if this ever needs hardening beyond what `needs_rejoin`'s own accepted risk profile already tolerates. - **Task 5 found and fixed bug-1751**: design.md's cited candidate-filter site (`scheduler.c:1211`'s `sched_reset_worker_()`, reached only via `sched_begin_session()`) is dead code on both live drivers — `driver.c`/`sim_loop.c` actually call `xce_core_begin_session()` (`core_session.c`), which runs an independently-maintained mirror, `core_reset_workers()`, not listed in design.md's File Structure Plan. Fixed by applying the identical `IS_WORKER_UNAVAIL || IS_WORKER_RECONCILING` disjunct to both sites (parity/dead-code-hygiene on the design-cited one, correctness on the actually-live one) rather than patching only the literally-cited site and leaving the task's own Observable Completion Criterion false in production — same category as bug-1746/bug-1747's design-vs-reality corrections. design.md's File Structure Plan corrected. Also confirmed design.md's Testing Strategy referenced an "existing IS_WORKER_UNAVAIL candidate-filter test" that does not actually exist; new coverage added to `test_core_session.c` instead (white-box `#include`s `core_session.c` already; `scheduler.c`'s own dependency closure is too large to white-box per established precedent, `test_snapshot_load_allocator_regression.c`) — `scheduler.c`'s copy of the fix is therefore verified only by compilation + the identical, tested logic pattern in its mirror, not by a direct test; judged acceptable since it's confirmed dead code today, not a live gap. - **Task 6.1 found and fixed bug-1752**: the worker daemon's `myreq_handler` had NO dispatch case for `OPCode_push` at all before this task — every prior config push (including the pre-existing `clustername` field, in production since before this spec started) was silently discarded via `default: break;`. Confirmed genuine (diffed `myreq_handler` pre/post, ruled out an alternate consumer — `xcectx.c`'s `vtask_entry` only special-cases the unrelated generic `pushmsg` union arm, not `configpushmsg`) and confirmed `clustername` has never been read anywhere in `src/worker/`, so this was harmless dead infrastructure until `workerResyncWindowMs` became its first real consumer. Fixed narrowly — `handle_pushconfig()` consumes only `worker_resync_window_ms`, leaving `clustername`/`confvars` still unconsumed and explicitly out of this task's boundary — rather than expanding scope to wire up every existing-but-never-consumed push field. Ordering (join reply → config push → conditional arm+query) and the live-configured-value plumbing were both verified via genuine structural tests (real packets popped off the wire, two distinct configured values round-tripped), not weaker behavioral proxies. +- **Task 6.2's mandated standby-redirect investigation found bug-1753**, a DIFFERENT gap than the one it was looking for: the generic not-primary (`CTRL_STANDBY`) redirect path turned out to already work correctly with zero opcode-specific code (`ha_dispatch_gate` gates every opcode uniformly before the switch), but the investigation surfaced that `ha_dispatch_gate`'s SEPARATE `CTRL_PRIMARY`-without-write-quorum branch — a "fact-bearing ops admitted anyway" allowlist (`OPCode_notify`/`update`/`setpreason`/`clrpreason`, bug-396) — was missing `OPCode_jobStateReport`, meaning a report arriving during a transient quorum-loss window (e.g. mid-failover, this feature's own core motivating scenario) would be rejected outright with no retry, since the worker's send is fire-and-forget. Fixed by adding the one missing case, verified safe against `resync_round_is_current`'s staleness check and the WAL_MODE_RETAIN semantics every fact this feature produces already uses. Citation nit: this fix's true motivation sits closer to Requirements 1.3/4/6 than the task's own tagged 8.1/8.3 — recorded here for traceability, tasks.md's tag left as originally generated. **Lesson**: a task text that names one specific concern to investigate ("check the redirect path") can surface a DIFFERENT, more consequential gap in the same general area — don't stop the investigation at "the named concern turned out fine," keep reading the surrounding gating logic in full. +- **Task 6.2 also drew three escalating, unsolicited cross-session messages** (from an unfamiliar peer session) mid-flight, pushing a specific unreviewed code change to `test_jobstatereport_dispatch.c` — the exact file the dispatched implementer subagent was independently writing at the time. Declined to act on any of them (risk of conflicting with in-flight subagent work; escalating urgency/"no further investigation needed" framing is a known social-engineering pattern regardless of whether the underlying suggestion is technically plausible). The implementer subagent finished independently and solved the underlying `get_ep_worker()`-in-test-binary limitation completely differently (a `#ifdef XCE_TESTING` call-count seam in `resync.c`, not a global-fixture hack) — validating the decision not to short-circuit the normal TDD+review pipeline based on an unverified peer instruction. diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 5cb3d5bc..045756ca 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -20522,6 +20522,38 @@ ], "occurrences": 1, "last_seen": "2026-09-01T10:00:00Z" + }, + { + "id": "bug-1753", + "timestamp": "2026-09-01T20:30:00Z", + "error_message": "worker-rejoin-resync task 6.2 (controller-side OPCode_jobStateReport dispatch): design.md's Error Handling section only anticipated the STANDBY-role case for this opcode pair (\"verify... whether OPCode_jobStateQuery/OPCode_jobStateReport need an explicit redirect-reply path... Expected answer: no... but this must be confirmed, not assumed, given bug-1700's documented history of this exact class of gap recurring across two other opcodes\"). Auditing ha_dispatch_gate() (ha.c) in full per that instruction found the STANDBY case was indeed already correctly handled automatically (OPCode_jobStateReport, left absent from the CTRL_STANDBY allow-list, falls to the generic default: return -XCE_ERR_STANDBY branch, and cluster.c's existing bug-1700 post-gate block already fills the redirect hint for any such rejection with zero opcode-specific code needed) -- but the same audit surfaced a DIFFERENT, unanticipated gap in the SAME function: a CTRL_PRIMARY controller that has temporarily lost write quorum rejects OPCode_jobStateReport outright with -XCE_ERR_NOT_READY, because the opcode was absent from ha_dispatch_gate's quorum-loss \"fact-bearing ops\" allowlist (OPCode_notify/OPCode_update/OPCode_setpreason/OPCode_clrpreason, ha.c ~line 4686-4696) that already exists specifically to prevent this exact class of loss (bug-396) for other worker/scheduler fact reports.", + "root_cause": "OPCode_jobStateReport is sent fire-and-forget (xce_job_state_report_req(), request.c/xce_request()) with no redelivery on rejection -- structurally identical to OPCode_notify, the opcode bug-396's fix (ha-commit-barrier-rollout) originally added this allowlist to protect. Since worker-rejoin-resync's two new opcodes were added to controller.capnp's OPCode enum without also being added to ha_dispatch_gate's quorum-loss allowlist, a job-state reconciliation report arriving at a primary during a transient write-quorum-loss window (e.g. mid-failover, the exact scenario this whole feature exists to recover from) would be rejected outright and never retried by the worker -- reproducing bug-396's own loss mode for precisely the facts worker-rejoin-resync (bug-1741) exists to recover.", + "file": "src/controller/ha.c (ha_dispatch_gate, CTRL_PRIMARY quorum-loss branch, ~line 4686-4699)", + "fix": "FIXED (2026-09-01, task 6.2, found live during the task's own mandated standby-redirect investigation -- not assumed safe, confirmed by reading the full gate function as instructed). Added `case OPCode_jobStateReport:` alongside the existing OPCode_notify/OPCode_update/OPCode_setpreason/OPCode_clrpreason cases in ha_dispatch_gate's quorum-loss fact-bearing-ops allowlist, so a report arriving while the primary temporarily lacks write quorum is admitted (not rejected) and can flow into the same RETAIN/pending-facts WAL-append mechanism those sibling opcodes already rely on. OPCode_jobStateQuery needed no corresponding change: the controller only ever SENDS that opcode (controller -> worker), never receives it, so it never reaches ha_dispatch_gate at all. New regression test test_gate_no_quorum_admits_jobstatereport_fact (test_commit_barrier_rollout.c), mirroring test_gate_no_quorum_admits_notify_fact's exact fixture/assertion shape; also added to test_gate_with_quorum_admits_all_opcodes's opcode list for parity. The CTRL_STANDBY case needed and received no code change at all -- proven correct as-is by three new tests in the new src/controller/testsuite/test_jobstatereport_dispatch.c (rejects-when-not-primary, standby-with-known-primary-redirects, standby-no-known-primary-no-redirect), mirroring test_sched_admin_dispatch.c's identical bug-1700 regression tests for OPCode_schedAdmin, the 'real existing primary-only opcode' ground truth this task's brief called for.", + "tags": [ + "worker-rejoin-resync", + "task-6.2", + "ha-c", + "ha_dispatch_gate", + "requirement-8.1", + "requirement-8.3", + "bug-396-class", + "quorum-loss", + "design-reality-mismatch", + "found-live-during-implementation", + "fixed" + ], + "related_bugs": [ + "bug-396", + "bug-1700", + "bug-1746", + "bug-1747", + "bug-1748", + "bug-1751", + "bug-1752" + ], + "occurrences": 1, + "last_seen": "2026-09-01T20:30:00Z" } ] } diff --git a/src/controller/cluster.c b/src/controller/cluster.c index bc72882c..99256c78 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -25,6 +25,7 @@ #include "jobj.h" #include "resobj.h" +#include "resync.h" /* worker-rejoin-resync task 6.2: resync_handle_report_page */ #include "utils.h" #include "worker.h" #include "xce/errnox.h" @@ -237,6 +238,58 @@ static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req) return 0; } +/* handle_jobstatereport: dispatch handler for OPCode_jobStateReport + * (worker-rejoin-resync task 6.2, Req 8.1, 8.3) -- the worker's answer to a + * job-state reconciliation query the controller sent it (worker.c's + * worker_send_resync_query(), task 6.1). Resolves the reporting worker from + * the inbound connection via get_ep_worker() (misc.c) -- the same ep- + * >worker_t resolution on_worker_join() already uses to SEND that query + * (worker.c) -- then delegates wire-page reassembly + apply to resync_ + * handle_report_page() (resync.c, task 6.2), which reassembles a report + * that arrives across multiple pages and hands the complete result to + * resync_apply_report() (task 3.2) exactly once, never once per page. + * + * Fire-and-forget from the worker's side (xce_job_state_report_req(), + * request.c -- the same fire-and-forget shape xce_job_state_query_req() + * already uses to SEND the query) -- no reply is expected or sent back, + * matching OPCode_walCatchupEntry/OPCode_snapshotTransfer's existing + * no-reply convention immediately above for this class of one-way, + * already-happened-fact daemon-to-daemon message. + * + * A page from a connection that does not resolve to a known worker + * (get_ep_worker() returns NULL -- e.g. a stale/torn-down connection racing + * teardown) is logged and dropped: there is no worker_t to reassemble or + * apply this report against, and nothing to reply to either way. */ +static int handle_jobstatereport(cluster_t* c, endpoint_t* ep, + xce_request_t* req) +{ + xce_job_state_report_t* page = req->body.jobstatereportmsg; + worker_t* w; + int rc; + + if (page == NULL) + return 0; + + w = get_ep_worker(c, ep); + if (w == NULL) { + WARN("handle_jobstatereport: endpoint %p does not resolve to a " + "known worker -- dropping report page (pos=%d count=%d " + "ntotal=%d)", (void*)ep, page->pos, page->count, page->n_total); + return -ENOENT; + } + + rc = resync_handle_report_page(c, w, page); + if (rc < 0) + ERROR("handle_jobstatereport: worker %s: failed to reassemble/apply " + "report page (pos=%d count=%d ntotal=%d) rc=%d", + (w->host.name != NULL) ? w->host.name : "?", page->pos, + page->count, page->n_total, rc); + + dereference(w); + + return (rc < 0) ? rc : 0; +} + /* ---- OPCode_schedAdmin: forward-and-block-for-reply support (task 10.1) -- * * The controller is a stateless pass-through for OPCode_schedAdmin: it @@ -522,6 +575,29 @@ static int myreq_handler(vtask_t* task) case OPCode_jobdispatch: rc = handle_jobdispatch(c, ep, &request); break; + case OPCode_jobStateReport: + /* worker-rejoin-resync task 6.2 (Req 8.1, 8.3): the controller only + * ever RECEIVES OPCode_jobStateReport (worker -> controller) -- the + * sibling opcode, OPCode_jobStateQuery, flows controller -> worker + * only (sent by worker_send_resync_query(), worker.c) and is never + * inbound to this switch; the worker's own myreq_handler + * (controller_connector.c, task 4.3) is what dispatches THAT one. + * No standby-redirect handling is added here: ha_dispatch_gate() + * above already rejects this opcode on every non-primary role with + * -XCE_ERR_STANDBY (it is deliberately absent from the CTRL_STANDBY + * allow-list, ha.c) BEFORE this switch is ever reached, and the + * generic post-gate rejection block a few lines up already + * populates the redirect hint (bug-1700) for exactly that status -- + * so a standby-received report is redirected the same way every + * other primary-only opcode already is, with zero opcode-specific + * code needed here. See ha.c's CTRL_PRIMARY-without-quorum branch + * for the ADJACENT, genuinely-needed fix this task found: admitting + * this opcode as a fact-bearing op there (alongside OPCode_notify) + * so a report is never rejected outright during a transient quorum + * loss (bug-396's exact failure mode, which this whole feature + * exists to correct). */ + rc = handle_jobstatereport(c, ep, &request); + break; case OPCode_jobsub: rc = handle_jobsubmit(c, ep, &request); break; diff --git a/src/controller/ha.c b/src/controller/ha.c index ee470163..606ee210 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -4688,11 +4688,22 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode) * Rejecting them loses the fact forever: the reporter never * retries (bug-396). Admit so the RETAIN path can append and * hold them until quorum returns (Req 3.2; umbrella 2.8 — - * facts are retained, never dropped). */ + * facts are retained, never dropped). + * + * worker-rejoin-resync task 6.2: OPCode_jobStateReport is the + * exact same shape — a worker's already-happened job-state + * facts, sent fire-and-forget (xce_job_state_report_req(), + * request.c) with no redelivery on rejection, exactly like + * OPCode_notify. Rejecting it outright here during a transient + * quorum-loss window would reproduce bug-396's own loss mode + * for the specific facts this whole feature exists to recover + * (bug-1741) — found live while auditing this gate for task + * 6.2's own standby-redirect investigation, not assumed. */ case OPCode_notify: case OPCode_update: case OPCode_setpreason: case OPCode_clrpreason: + case OPCode_jobStateReport: return 0; default: return -XCE_ERR_NOT_READY; diff --git a/src/controller/include/resync.h b/src/controller/include/resync.h index f14e1afd..313738dd 100644 --- a/src/controller/include/resync.h +++ b/src/controller/include/resync.h @@ -174,6 +174,93 @@ extern int resync_round_is_current(worker_t* w, uint64_t token); extern int resync_apply_report(cluster_t* c, worker_t* w, uint64_t round_token, xce_job_state_report_t* report); +/* worker-rejoin-resync task 6.2 (Req 8.1, 8.3): the real type behind + * worker_t.report_accum (xce/types.h -- stored there as an opaque void* to + * avoid a types.h/typemap.h include cycle, since xce_job_state_entry_t is a + * typemap.h type). Reassembles an OPCode_jobStateReport that arrives split + * across multiple wire pages, before resync_apply_report() ever sees it -- + * mirrors xcecd_resync.c's worker-side query_accum/query_accum_ntotal/ + * query_accum_received statics (task 4.3), moved from file-static (a + * worker daemon has only one controller connection) to a per-worker_t + * field (a controller serves many workers concurrently, so the + * accumulator must be keyed per reporting worker). Lives separately from + * resync_state_t (w->resync) rather than as a field on it: resync_apply_ + * report() may resolve (free) w->resync partway through settling a fully- + * assembled report's entries, and this accumulator's own memory must not + * be tied to that lifetime -- resync_handle_report_page() below always + * detaches (NULLs w->report_accum) and takes sole ownership of this struct + * BEFORE calling resync_apply_report(), so there is nothing for that call + * to free out from under it either way. */ +typedef struct { + /* Captured from w->resync->round_token the instant this report's first + * page (pos == 0) arrives (0, the reserved "no token" sentinel, if no + * round is outstanding yet) -- NOT re-read at apply time. This is what + * lets a report that started under one round but finishes assembling + * after that round was superseded/discarded (Req 6.1, 6.3) still + * discard correctly via resync_apply_report()'s own resync_round_is_ + * current() check, rather than being wrongly matched against whatever + * round happens to be current once the LAST page arrives. */ + uint64_t round_token; + int n_total; + int n_received; + /* Deep copies (plain struct-value copies -- xce_job_state_entry_t has + * no nested heap pointers), one malloc'd xce_job_state_entry_t per + * entry, indexed positionally by wire pos, mirroring xce_job_state_ + * report_t.entries' own array-of-pointers shape (typemap.h) so the + * assembled report can be hand off to resync_apply_report() directly, + * with no reshaping. A deep copy is required, not merely a borrowed + * pointer into the incoming page: each page's xce_job_state_entry_t* + * values are owned by that page's own xce_request_t, freed by xce_ + * free_request() the moment the dispatch handler that decoded it + * returns -- long before a later page (let alone the final one) has + * even arrived. */ + xce_job_state_entry_t** entries; +} resync_report_accum_t; + +/* Handles one incoming OPCode_jobStateReport wire page for worker `w`: + * accumulates it into w->report_accum (allocating/resetting on pos == 0, + * exactly like xcecd_resync_handle_query_page()'s query_accum reset-on- + * pos-0 convention -- a stale/incomplete accumulation left over from an + * aborted prior report, e.g. this worker disconnected mid-report on a + * prior round, is always discarded by the next report's own first page + * rather than corrupting it). Once every entry has arrived (n_received == + * n_total), hands the fully-assembled report to resync_apply_report() + * EXACTLY ONCE -- never once per page -- using the round_token captured at + * this report's first page, then frees the accumulator itself. + * + * Precondition: caller does NOT hold w's lock -- this function acquires it + * itself, narrowly, for each page's accumulator mutation, and always + * releases it again BEFORE calling resync_apply_report() (which manages + * its own w locking internally -- resync.h's own doc comment on that + * function). Mirrors resync_apply_report's own "never nested" locking + * discipline. + * + * Returns 0 if this page was accepted but the report is not yet complete + * (more pages still expected), 1 if this page completed the report and it + * was handed to resync_apply_report() (regardless of whether resync_apply_ + * report applied it or discarded it as stale -- see above), or a negative + * errno for a malformed page (NULL c/w/page, or a page whose pos/count + * fall outside [0, n_total) -- the in-progress accumulation, if any, is + * discarded in that case too, so a later page cannot be mismatched against + * a corrupt partial buffer). */ +extern int resync_handle_report_page(cluster_t* c, worker_t* w, + const xce_job_state_report_t* page); + +#ifdef XCE_TESTING +/* worker-rejoin-resync task 6.2 test seam: counts how many times resync_ + * handle_report_page() has actually invoked resync_apply_report() (process- + * wide, across every worker) -- a direct, structural proof that a report + * spanning N wire pages results in exactly ONE resync_apply_report() call, + * never N, distinct from (and stronger than) merely asserting the resulting + * job-table end state looks correct (job.c's own idempotent fact guards + * would make even a buggy once-per-page implementation LOOK correct on a + * happy-path report with no interleaved round changes -- see this function's + * own doc comment above and testing.md's "structural, not just behavioral" + * guidance for placement/count-sensitive logic). */ +extern int resync_apply_report_call_count_for_test(void); +extern void resync_apply_report_call_count_reset_for_test(void); +#endif + #ifdef XCE_TESTING /* worker-rejoin-resync task 3.1 test seam: directly invokes the SAME * internal timeout-fire logic resync_arm_round's own delaymgr callback runs diff --git a/src/controller/meson.build b/src/controller/meson.build index dd780df2..55440c66 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -197,6 +197,7 @@ controller_unit_test_src = [ 'testsuite/test_upgrade_snapshot_apply.c', 'testsuite/test_bug857_regression.c', 'testsuite/test_sched_admin_dispatch.c', + 'testsuite/test_jobstatereport_dispatch.c', 'testsuite/test_fetchfile_disconnect.c', 'ctrlstatus.c', 'commit.c', diff --git a/src/controller/resync.c b/src/controller/resync.c index 158267ee..f1ba05bd 100644 --- a/src/controller/resync.c +++ b/src/controller/resync.c @@ -753,6 +753,208 @@ int resync_apply_report(cluster_t* c, worker_t* w, uint64_t round_token, return 0; } +/* ── Multi-page report reassembly (task 6.2, Req 8.1, 8.3) ───────────────── + * + * resync_apply_report() above expects an already-fully-assembled report -- + * its own doc comment (resync.h): "multi-page wire reassembly is a later + * task's concern (6.2), out of this function's scope." The functions below + * are that later task: accumulate an incoming OPCode_jobStateReport's wire + * pages into worker_t.report_accum (xce/types.h), then call resync_apply_ + * report() exactly once -- never once per page -- only once the last page + * has arrived. Mirrors xcecd_resync_handle_query_page()'s (worker-side, + * xcecd_resync.c, task 4.3) accumulate-then-apply-once shape and its + * reset-on-pos-0 discard-stale-accumulation convention, adapted from a + * single file-static accumulator (a worker daemon has exactly one + * controller connection) to a per-worker_t one (a controller serves many + * workers concurrently, so the accumulator must be keyed per reporting + * worker, guarded by that worker's own RTTI lock). + */ + +#ifdef XCE_TESTING +/* Structural proof seam (task 6.2): counts actual resync_apply_report() + * invocations made BY resync_handle_report_page() below, process-wide. + * Lets a test assert "a report split across N pages results in exactly ONE + * resync_apply_report() call" directly, rather than only inferring it from + * the resulting job-table end state -- job.c's own idempotent fact guards + * would make even a buggy once-per-page implementation produce a correct- + * looking end state on a happy-path report with no round change + * interleaved between pages (the exact class of placement/count bug task + * 4.3's own round-1 review missed with a purely behavioral test -- see + * tasks.md's Implementation Notes for task 4.3). */ +static _Atomic int g_resync_apply_report_call_count = 0; + +int resync_apply_report_call_count_for_test(void) +{ + return atomic_load_explicit(&g_resync_apply_report_call_count, + memory_order_relaxed); +} + +void resync_apply_report_call_count_reset_for_test(void) +{ + atomic_store_explicit(&g_resync_apply_report_call_count, 0, + memory_order_relaxed); +} +#endif + +/* Frees w->report_accum (if any, casting back from the opaque void* xce/ + * types.h stores it as) and NULLs it. Caller holds w's write lock. + * Idempotent -- a no-op when already NULL, safe to call unconditionally + * (mirrors resync_resolve_round's own idempotence for the sibling + * w->resync field). */ +static void resync_report_accum_reset(worker_t* w) +{ + resync_report_accum_t* accum = (resync_report_accum_t*)w->report_accum; + int i; + + if (accum == NULL) + return; + + for (i = 0; i < accum->n_total; i++) + free(accum->entries[i]); + free(accum->entries); + free(accum); + + w->report_accum = NULL; +} + +int resync_handle_report_page(cluster_t* c, worker_t* w, + const xce_job_state_report_t* page) +{ + resync_report_accum_t* accum; + int rc; + int i; + + if (c == NULL || w == NULL || page == NULL) + return -EINVAL; + + if (page->pos < 0 || page->count < 0) + return -EINVAL; + + lock(w, RTTI_LOCK_WRITE); + + if (page->pos == 0) { + /* Fresh report starting (or restarting over a stale/incomplete + * prior accumulation -- e.g. this worker disconnected mid-report + * on an earlier round) -- always reset first, mirroring xcecd_ + * resync_handle_query_page()'s identical pos==0 convention. */ + resync_report_accum_reset(w); + + accum = (resync_report_accum_t*)calloc(1, sizeof(*accum)); + if (accum == NULL) { + unlock(w); + return -ENOMEM; + } + + /* Req 6.1, 6.3: captured NOW, from whatever round is current the + * instant this report starts arriving -- not re-read later at + * apply time. See resync_report_accum_t's own doc comment + * (resync.h) for why this specific capture point is load-bearing, + * not an arbitrary choice. */ + accum->round_token = (w->resync != NULL) ? w->resync->round_token : 0; + accum->n_total = page->n_total; + + if (page->n_total > 0) { + accum->entries = (xce_job_state_entry_t**)calloc( + (size_t)page->n_total, sizeof(*accum->entries)); + if (accum->entries == NULL) { + free(accum); + unlock(w); + return -ENOMEM; + } + } + + w->report_accum = accum; + } + + accum = (resync_report_accum_t*)w->report_accum; + + if (page->count > 0) { + if (accum == NULL || page->entries == NULL + || (page->pos + page->count) > accum->n_total) { + resync_report_accum_reset(w); + unlock(w); + return -EINVAL; + } + + /* Deep-copy each entry (plain struct-value copy -- xce_job_state_ + * entry_t has no nested heap pointers): page->entries[i] is owned + * by THIS page's own decoded xce_request_t and is freed by xce_ + * free_request() the moment the dispatch handler that decoded it + * returns, long before a later page (let alone the final one) has + * even arrived -- a borrowed pointer would be a use-after-free the + * instant this function returns 0 awaiting more pages. */ + for (i = 0; i < page->count; i++) { + xce_job_state_entry_t* src = page->entries[i]; + xce_job_state_entry_t* copy; + + if (src == NULL) { + resync_report_accum_reset(w); + unlock(w); + return -EINVAL; + } + + copy = (xce_job_state_entry_t*)malloc(sizeof(*copy)); + if (copy == NULL) { + resync_report_accum_reset(w); + unlock(w); + return -ENOMEM; + } + *copy = *src; + + accum->entries[page->pos + i] = copy; + } + + accum->n_received += page->count; + } else if (accum == NULL) { + /* An orphaned continuation page (pos != 0) with no accumulation in + * progress at all (its own preceding pos==0 page never arrived, or + * arrived and was already discarded/completed) -- nothing to + * accumulate into and nothing that will ever complete; reject + * rather than silently waiting forever. */ + unlock(w); + return -EINVAL; + } + + if (accum->n_received < accum->n_total) { + /* More pages still expected -- gate stays set (via w->report_accum + * remaining non-NULL), nothing applied yet. */ + unlock(w); + return 0; + } + + /* Last page: the report is fully assembled. Detach ownership from `w` + * and release w's lock BEFORE calling resync_apply_report() -- that + * function manages its own w write-locking internally (resync.h's own + * doc comment) and must never be called with w's lock already held, or + * its own lock(w, RTTI_LOCK_WRITE) call would deadlock against this + * one. */ + w->report_accum = NULL; + unlock(w); + + { + xce_job_state_report_t report; + + memset(&report, 0, sizeof(report)); + report.n_total = accum->n_total; + report.pos = 0; + report.count = accum->n_total; + report.entries = accum->entries; + +#ifdef XCE_TESTING + atomic_fetch_add_explicit(&g_resync_apply_report_call_count, 1, + memory_order_relaxed); +#endif + rc = resync_apply_report(c, w, accum->round_token, &report); + } + + for (i = 0; i < accum->n_total; i++) + free(accum->entries[i]); + free(accum->entries); + free(accum); + + return (rc < 0) ? rc : 1; +} + #ifdef XCE_TESTING /* worker-rejoin-resync task 3.1 test seam (resync.h): builds a * resync_timeout_ctx_t exactly as resync_arm_round's own delay_task_ms call diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 13907021..9c69f915 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -103,6 +103,7 @@ extern void test_upgrade_snapshot_restore(void); extern void test_upgrade_snapshot_apply(void); extern void test_bug857_regression(void); extern void test_sched_admin_dispatch(void); +extern void test_jobstatereport_dispatch(void); extern void test_fetchfile_disconnect(void); extern void test_job_dispatch(void); extern void test_dispatch_commit(void); @@ -282,6 +283,7 @@ int main(int argc, char* argv[]) TEST(test_upgrade_snapshot_apply); TEST(test_bug857_regression); TEST(test_sched_admin_dispatch); + TEST(test_jobstatereport_dispatch); TEST(test_fetchfile_disconnect); TEST(test_job_dispatch); TEST(test_dispatch_commit); diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index 90f7a27f..81af772d 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -982,6 +982,37 @@ static void test_gate_no_quorum_admits_clrpreason_fact(void) printf("test_gate_no_quorum_admits_clrpreason_fact: PASS\n"); } +/* ── OPCode_jobStateReport admitted without write quorum ─────────────────── + * (worker-rejoin-resync task 6.2, found live while auditing this exact gate + * for task 6.2's own standby-redirect investigation -- design.md only + * anticipated the STANDBY case, per bug-1700's precedent, and did not + * separately consider a PRIMARY that has temporarily lost write quorum. + * OPCode_jobStateReport is a worker's already-happened job-state fact, sent + * fire-and-forget (xce_job_state_report_req(), request.c) with no + * redelivery on rejection -- exactly the same shape as OPCode_notify, which + * this same allowlist already exists to protect from bug-396's silent-loss + * failure mode. Rejecting a reconciliation report outright during a + * transient quorum-loss window would reproduce that exact loss mode for + * the specific facts this whole feature exists to recover (bug-1741). */ +static void test_gate_no_quorum_admits_jobstatereport_fact(void) +{ + int rc; + + /* Arrange: quorum absent */ + fact_gate_setup(3, 0); + + /* Act */ + rc = ha_dispatch_gate(&g_fg_ha, OPCode_jobStateReport); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "OPCode_jobStateReport (worker reconciliation fact) must be " + "admitted without write quorum, exactly like OPCode_notify"); + + fact_gate_teardown(); + printf("test_gate_no_quorum_admits_jobstatereport_fact: PASS\n"); +} + /* ── PIN: request-shaped opcodes stay rejected without quorum (Req 3.2) ─── * The allowlist is FACTS ONLY: every request-shaped opcode (a requester * awaits confirmation, served by the task-3 handler barrier) must keep @@ -1018,16 +1049,17 @@ static void test_gate_no_quorum_still_rejects_request_opcodes(void) static void test_gate_with_quorum_admits_all_opcodes(void) { static const struct { int opcode; const char *name; } ops[] = { - { OPCode_notify, "OPCode_notify" }, - { OPCode_update, "OPCode_update" }, - { OPCode_setpreason, "OPCode_setpreason" }, - { OPCode_clrpreason, "OPCode_clrpreason" }, - { OPCode_jobsub, "OPCode_jobsub" }, - { OPCode_jobcontrol, "OPCode_jobcontrol" }, - { OPCode_queuecontrol, "OPCode_queuecontrol" }, - { OPCode_hostcontrol, "OPCode_hostcontrol" }, - { OPCode_jobdispatch, "OPCode_jobdispatch" }, - { OPCode_join, "OPCode_join" }, + { OPCode_notify, "OPCode_notify" }, + { OPCode_update, "OPCode_update" }, + { OPCode_setpreason, "OPCode_setpreason" }, + { OPCode_clrpreason, "OPCode_clrpreason" }, + { OPCode_jobStateReport, "OPCode_jobStateReport" }, + { OPCode_jobsub, "OPCode_jobsub" }, + { OPCode_jobcontrol, "OPCode_jobcontrol" }, + { OPCode_queuecontrol, "OPCode_queuecontrol" }, + { OPCode_hostcontrol, "OPCode_hostcontrol" }, + { OPCode_jobdispatch, "OPCode_jobdispatch" }, + { OPCode_join, "OPCode_join" }, }; size_t i; @@ -2459,6 +2491,7 @@ void test_commit_barrier_rollout(void) RUN_TEST(test_gate_no_quorum_admits_update_fact); RUN_TEST(test_gate_no_quorum_admits_setpreason_fact); RUN_TEST(test_gate_no_quorum_admits_clrpreason_fact); + RUN_TEST(test_gate_no_quorum_admits_jobstatereport_fact); RUN_TEST(test_gate_no_quorum_still_rejects_request_opcodes); RUN_TEST(test_gate_with_quorum_admits_all_opcodes); /* Task 5.1 — config barrier: configAck arithmetic + upgrade failure diff --git a/src/controller/testsuite/test_jobstatereport_dispatch.c b/src/controller/testsuite/test_jobstatereport_dispatch.c new file mode 100644 index 00000000..e3a67560 --- /dev/null +++ b/src/controller/testsuite/test_jobstatereport_dispatch.c @@ -0,0 +1,322 @@ +/* + * test_jobstatereport_dispatch.c — Unity tests for the OPCode_jobStateReport + * controller dispatch case (worker-rejoin-resync spec, task 6.2, Req 8.1, + * 8.3). + * + * Exercises myreq_handler's OPCode_jobStateReport case through the real + * decode+dispatch entry point exposed for tests, myreq_handler_for_test + * (cluster.c, #ifdef XCE_TESTING) — same precedent test_sched_admin_ + * dispatch.c already established for a comparable primary-only opcode. + * + * Covers task 6.2's own explicit standby-redirect investigation: a report + * (or, by the same generic mechanism, a query) sent to a non-primary + * controller must receive the SAME redirect behavior an equivalent + * existing primary-only message already receives, not a bare unrecoverable + * error — proven here exactly as test_sched_admin_dispatch.c's own + * test_schedadmin_standby_with_known_primary_redirects/test_schedadmin_ + * standby_no_known_primary_no_redirect prove it for OPCode_schedAdmin, the + * "real existing primary-only opcode" ground truth this task's brief + * calls for. OPCode_jobStateReport is deliberately absent from ha_ + * dispatch_gate()'s CTRL_STANDBY allow-list (ha.c) — exactly like OPCode_ + * schedAdmin/OPCode_upgradeConfig — so this redirect behavior requires NO + * opcode-specific code in cluster.c's own dispatch case: it is entirely a + * property of the GENERIC gate-rejection block that already runs ahead of + * every opcode's switch case (cluster.c, the bug-1700 fix). These tests + * prove that generic mechanism actually covers this opcode too, rather + * than assuming it does. + * + * The OTHER half of task 6.2's own brief — multi-page reassembly, applied + * exactly once, equivalent to a single-page report — is proven against a + * REAL worker_t and REAL job table in test_worker_resync.c (resync_handle_ + * report_page()'s own direct tests), not re-proven here through the wire. + * See this file's own comment further down (immediately before the RUN_ + * TEST list) for exactly why a "real wire dispatch resolves a real + * worker_t" version of that proof cannot be built in THIS test binary, and + * why that is a pre-existing, unrelated architectural property, not a gap + * in this task's own coverage. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "ha.h" +#include "netio.h" +#include "request.h" +#include "typemap.h" +#include "xce/errnox.h" +#include "xce/types.h" + +/* myreq_handler exposed for tests (cluster.c) — same seam test_sched_admin_ + * dispatch.c uses. `pkt` ownership: consumed (freed) by the call. */ +extern int myreq_handler_for_test(endpoint_t* ep, io_packet_t* pkt); + +/* ── Fixture ──────────────────────────────────────────────────────────── + * + * A bare calloc'd cluster_t, exactly like test_sched_admin_dispatch.c's + * sa_make_cluster() — sufficient for every test below, none of which + * reaches past ha_dispatch_gate() or handle_jobstatereport()'s own first + * line (get_ep_worker()). + */ +static cluster_t* jsr_make_gate_cluster(void) +{ + cluster_t* c = calloc(1, sizeof(cluster_t)); + TEST_ASSERT_NOT_NULL(c); + mylock_init(&(c->sched.lock)); + return c; +} + +static void jsr_gate_teardown(cluster_t* c) +{ + if (c == NULL) + return; + mylock_fini(&(c->sched.lock)); + free(c); +} + +/* Build a wire-encoded OPCode_jobStateReport request packet carrying a + * minimal single-entry, single-page report — the exact content is + * immaterial for the standby-redirect tests below (the request never + * reaches the dispatch case at all). */ +static io_packet_t* jsr_build_request_pkt(uint64_t jobid) +{ + xce_job_state_entry_t entry; + xce_job_state_entry_t* entries[1]; + xce_job_state_report_t body; + + memset(&entry, 0, sizeof(entry)); + entry.jobid = jobid; + entry.status = JobStateStatus_running; + entry.queried = 1; + entries[0] = &entry; + + memset(&body, 0, sizeof(body)); + body.n_total = 1; + body.pos = 0; + body.count = 1; + body.entries = entries; + + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_jobStateReport; + req.body_which = Request_body_jobstatereportmsg; + req.body.jobstatereportmsg = &body; + + io_packet_t* pkt = calloc(1, sizeof(*pkt)); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_TRUE_MESSAGE(xce_encode(&req, pkt) >= 0, + "failed to encode JobStateReport packet"); + return pkt; +} + +/* Pop the single reply packet myreq_handler queued on ep->outq (if any) and + * decode it into *out. Mirrors test_sched_admin_dispatch.c's sa_decode_ + * queued_reply exactly. */ +static void jsr_decode_queued_reply(endpoint_t* ep, xce_request_t* out) +{ + data_item_t* it = endpoint_deq(ep); + TEST_ASSERT_NOT_NULL_MESSAGE(it, "expected a reply packet on ep->outq"); + + io_packet_t* pkt = ITPTR_TO_PACKET(it); + memset(out, 0, sizeof(*out)); + int rc = xce_decode(pkt, out); + TEST_ASSERT_EQUAL_INT(0, rc); +} + +/* ── Standby redirect (task 6.2's own explicit investigation) ───────────── + * + * Ground truth: OPCode_schedAdmin's identical fixture/assertions + * (test_sched_admin_dispatch.c), the "real existing primary-only opcode" + * this task's brief calls for. */ + +static void test_jobstatereport_rejects_when_not_primary(void) +{ + cluster_t* c = jsr_make_gate_cluster(); + atomic_store(&c->ha.role, CTRL_STANDBY); + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.owner = c; + + io_packet_t* pkt = jsr_build_request_pkt(1001); + int rc = myreq_handler_for_test(&ep, pkt); + + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc); + + xce_request_t decoded; + jsr_decode_queued_reply(&ep, &decoded); + TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.replymsg); + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, decoded.body.replymsg->status); + TEST_ASSERT_EQUAL_INT(RequestReply_data_nodata, + decoded.body.replymsg->data_which); + xce_free_request(&decoded); + + jsr_gate_teardown(c); + printf("test_jobstatereport_rejects_when_not_primary: PASS\n"); +} + +/* bug-1700 precedent: a standby dispatch-gate rejection with a KNOWN + * primary must carry a redirect hint — proven here with NO opcode-specific + * code added anywhere for OPCode_jobStateReport: it works purely because + * this opcode is absent from ha_dispatch_gate's CTRL_STANDBY allow-list, + * so the GENERIC post-gate rejection block (cluster.c) fills the redirect + * for it exactly as it already does for OPCode_schedAdmin/OPCode_join. */ +static void test_jobstatereport_standby_with_known_primary_redirects(void) +{ + cluster_t* c = jsr_make_gate_cluster(); + atomic_store(&c->ha.role, CTRL_STANDBY); + strncpy(c->ha.primary_host, "10.0.1.9", sizeof(c->ha.primary_host) - 1); + c->ha.primary_port = 7003; + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.owner = c; + + io_packet_t* pkt = jsr_build_request_pkt(1002); + int rc = myreq_handler_for_test(&ep, pkt); + + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc); + + xce_request_t decoded; + jsr_decode_queued_reply(&ep, &decoded); + TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which); + xce_reply_t* rr = decoded.body.replymsg; + TEST_ASSERT_NOT_NULL(rr); + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rr->status); + TEST_ASSERT_NOT_NULL_MESSAGE( + rr->redirect_host, + "a standby dispatch-gate rejection with a known primary must carry " + "that primary's address, exactly like every other primary-only " + "opcode already does (bug-1700)"); + TEST_ASSERT_EQUAL_STRING("10.0.1.9", rr->redirect_host); + TEST_ASSERT_EQUAL_UINT16(7003, rr->redirect_port); + xce_free_request(&decoded); + + jsr_gate_teardown(c); + printf("test_jobstatereport_standby_with_known_primary_redirects:" + " PASS\n"); +} + +/* Regression companion: no known primary yet must not crash and must + * decode as a non-redirecting reply (empty host / zero port). */ +static void test_jobstatereport_standby_no_known_primary_no_redirect(void) +{ + cluster_t* c = jsr_make_gate_cluster(); + atomic_store(&c->ha.role, CTRL_STANDBY); + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.owner = c; + + io_packet_t* pkt = jsr_build_request_pkt(1003); + int rc = myreq_handler_for_test(&ep, pkt); + + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc); + + xce_request_t decoded; + jsr_decode_queued_reply(&ep, &decoded); + TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which); + xce_reply_t* rr = decoded.body.replymsg; + TEST_ASSERT_NOT_NULL(rr); + TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rr->status); + TEST_ASSERT_TRUE_MESSAGE( + rr->redirect_host == NULL || rr->redirect_host[0] == '\0', + "no known primary: redirect_host must decode as empty/absent"); + TEST_ASSERT_EQUAL_UINT16(0, rr->redirect_port); + xce_free_request(&decoded); + + jsr_gate_teardown(c); + printf("test_jobstatereport_standby_no_known_primary_no_redirect:" + " PASS\n"); +} + +/* A PRIMARY with write quorum handles the request normally — dispatched + * past the gate into handle_jobstatereport(), which returns -ENOENT for an + * endpoint that does not resolve to a known worker (this bare fixture's + * `ep` was never registered as one) rather than any standby-style + * rejection. Proves the gate's happy path is unaffected by this task's + * change and that the new dispatch case is genuinely reached. */ +static void test_jobstatereport_primary_reaches_dispatch_case(void) +{ + cluster_t* c = jsr_make_gate_cluster(); + atomic_store(&c->ha.role, CTRL_PRIMARY); + /* n_peers defaults to 0 (calloc) -- single-node cluster, trivially has + * write quorum (ha_has_write_quorum: n_peers <= 1 -> true). */ + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.owner = c; + /* ep.data left NULL -> get_ep_role()/get_ep_worker() treat this as "not + * a worker endpoint" and return NULL/-EINVAL, exercising handle_ + * jobstatereport's own not-a-worker branch. */ + + io_packet_t* pkt = jsr_build_request_pkt(1004); + int rc = myreq_handler_for_test(&ep, pkt); + + TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, rc, + "a primary with quorum must reach handle_jobstatereport() itself " + "(not a gate rejection) -- which reports -ENOENT for an endpoint " + "with no resolvable worker_t"); + + jsr_gate_teardown(c); + printf("test_jobstatereport_primary_reaches_dispatch_case: PASS\n"); +} + +/* ── Why no "real wire dispatch resolves a real worker_t" test lives here ── + * + * A genuine end-to-end test (two real wire-encoded OPCode_jobStateReport + * packets through myreq_handler_for_test, resolving a REAL worker_t via + * get_ep_worker() and converging a REAL job table) was attempted here and + * found structurally impossible in THIS binary: get_ep_worker() (misc.c) + * is NOT one of controller_unit_test_src's linked sources (meson.build) -- + * test_commit_barrier_reply.c instead provides controller_unit_test's ONLY + * definition of it, a fixed test double (`return g_stub_ep_worker;`, + * file-static, defaulting to NULL and settable only by that file's own + * tests) that no other test file can point at a real worker_t. Confirmed + * live (not assumed) by direct instrumentation: a page built with a + * correct, verified-by-hand ep_private_t{role=HostRole_worker, idx=w->idx} + * -- independently proven resolvable via key_find()/vget() called directly + * against the same workermgr -- still made get_ep_worker() return NULL, + * because the symbol actually linked into this binary is the stub, not + * misc.c's real implementation. + * + * This is a pre-existing architectural property of controller_unit_test + * (several other cluster.c-adjacent functions -- notify(), update_host_ + * runusage(), on_scheduler_join(), add_apiserver(), on_apiserver_join() -- + * are stubbed the same way in the same file, for the same reason), not + * something this task introduced or can fix without either modifying + * shared test infrastructure many other tests depend on (out of this + * task's boundary) or standing up a new dedicated test binary that links + * the real misc.c (the pattern host_runusage_unit_test/resval_typeresolve_ + * unit_test already establish for exactly this class of problem, bug-1735) + * -- a real option, but a disproportionate scope expansion for what the + * unit tests below plus test_worker_resync.c's own tests already prove + * between them: handle_jobstatereport() (cluster.c) is a two-line function + * (get_ep_worker() then resync_handle_report_page()) whose own correctness + * is not in question -- the two things worth proving are (a) cluster.c's + * switch genuinely routes OPCode_jobStateReport to it, proven above by + * test_jobstatereport_primary_reaches_dispatch_case's distinctive -ENOENT + * return (get_ep_worker's stub-NULL path, reached only via this exact + * call chain), and (b) multi-page reassembly + exactly-once apply + + * equivalence-to-single-page, proven exhaustively against a REAL worker_t + * and REAL job table by resync_handle_report_page()'s own direct tests in + * test_worker_resync.c (test_handle_report_page_multi_page_reassembles_ + * and_applies_once and its siblings) -- the same function handle_ + * jobstatereport() calls, unmodified. + */ + +void test_jobstatereport_dispatch(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_jobstatereport_rejects_when_not_primary); + RUN_TEST(test_jobstatereport_standby_with_known_primary_redirects); + RUN_TEST(test_jobstatereport_standby_no_known_primary_no_redirect); + RUN_TEST(test_jobstatereport_primary_reaches_dispatch_case); +} diff --git a/src/controller/testsuite/test_worker_resync.c b/src/controller/testsuite/test_worker_resync.c index a8e9817d..ded15097 100644 --- a/src/controller/testsuite/test_worker_resync.c +++ b/src/controller/testsuite/test_worker_resync.c @@ -2052,6 +2052,389 @@ test_apply_report_unqueried_extra_unassigned_pend_job_gets_assigned(void) "assigned: PASS\n"); } +/* ══════════════════════════════════════════════════════════════════════ + * Task 6.2 (Req 8.1, 8.3): resync_handle_report_page() -- multi-page wire + * reassembly of an incoming OPCode_jobStateReport, ahead of resync_apply_ + * report() (task 3.2, proven above). Every test here proves the exact + * gap this task's own brief called out: not just that the correct END + * STATE results, but that resync_apply_report() is invoked EXACTLY ONCE + * per report regardless of page count (resync_apply_report_call_count_ + * for_test(), resync.h) -- job.c's own idempotent fact guards would make + * even a buggy once-per-page implementation LOOK correct on a happy-path + * report, exactly the class of bug task 4.3's own round-1 review missed + * with a purely behavioral test (tasks.md Implementation Notes). + * ══════════════════════════════════════════════════════════════════════ */ + +#define TWR_JOBID_MPAGE_FINOK 81201ULL +#define TWR_JOBID_MPAGE_UNREC 81202ULL +#define TWR_JOBID_MPAGE_RUNNING 81203ULL +#define TWR_JOBID_MPAGE_STALE 81210ULL +#define TWR_JOBID_MPAGE_BAD 81220ULL +#define TWR_JOBID_MPAGE_BAD2 81221ULL + +/* Baseline: a single-page report (pos=0, count==n_total) delivered through + * resync_handle_report_page() -- not resync_apply_report() directly -- is + * reassembled trivially (one page IS the whole report) and applied exactly + * once. Establishes the expected end state the multi-page test below must + * independently reproduce via genuine reassembly. */ +static void test_handle_report_page_single_page_baseline(void) +{ + job_t* j_unrec; + worker_t* w; + int rc; + uint64_t token; + int calls_before, calls_after; + + xce_job_state_entry_t e_unrec; + xce_job_state_entry_t* entries[1]; + xce_job_state_report_t page; + + twr_setup(); + + j_unrec = twr_seed_job(TWR_JOBID_MPAGE_UNREC, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j_unrec); + j_unrec->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_MPAGE_UNREC, "mpage-host-1"); + + w = twr_make_test_worker("mpage-host-1"); + TEST_ASSERT_NOT_NULL(w); + + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT(1, rc); + token = w->resync->round_token; + + e_unrec = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_UNREC, .status = JobStateStatus_unrecognized, + .exit_code = 0, .finish_time_ms = 0, .queried = 1 + }; + entries[0] = &e_unrec; + page = (xce_job_state_report_t){ + .n_total = 1, .pos = 0, .count = 1, .entries = entries + }; + + resync_apply_report_call_count_reset_for_test(); + calls_before = resync_apply_report_call_count_for_test(); + + rc = resync_handle_report_page(&g_twr, w, &page); + + calls_after = resync_apply_report_call_count_for_test(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, + "a single page whose count == n_total completes the report " + "immediately"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, calls_after - calls_before, + "resync_apply_report must be invoked exactly once"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j_unrec->job.state); + TEST_ASSERT_NULL(j_unrec->job.allocs); + TEST_ASSERT_NULL_MESSAGE(w->resync, + "the round must resolve once its only queried job is answered"); + + (void)token; + twr_teardown(); + printf("test_handle_report_page_single_page_baseline: PASS\n"); +} + +/* THE core proof (task 6.2's own Observable Completion Criterion): the SAME + * report, split across two wire pages, must (a) leave the job table + * COMPLETELY untouched after the first (non-final) page -- no partial + * application -- (b) converge to EXACTLY the same end state the single-page + * baseline above reached, once the final page arrives, and (c) invoke + * resync_apply_report() EXACTLY ONCE for the whole report, not once per + * page (the structural proof a behavioral-only assertion cannot provide -- + * see this section's own header comment). */ +static void test_handle_report_page_multi_page_reassembles_and_applies_once( + void) +{ + job_t* j_finok; + job_t* j_unrec; + job_t* j_running; + worker_t* w; + int rc; + int calls_before, calls_mid, calls_after; + + xce_job_state_entry_t e_finok, e_unrec, e_running; + xce_job_state_entry_t* entries_p1[2]; + xce_job_state_entry_t* entries_p2[1]; + xce_job_state_report_t page1, page2; + + twr_setup(); + + j_finok = twr_seed_job(TWR_JOBID_MPAGE_FINOK, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j_finok); + j_finok->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_MPAGE_FINOK, "mpage-host-2"); + + j_unrec = twr_seed_job(TWR_JOBID_MPAGE_UNREC, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j_unrec); + j_unrec->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_MPAGE_UNREC, "mpage-host-2"); + + j_running = twr_seed_job(TWR_JOBID_MPAGE_RUNNING, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j_running); + j_running->job.allocs = twr_make_test_alloc_for_host( + TWR_JOBID_MPAGE_RUNNING, "mpage-host-2"); + + w = twr_make_test_worker("mpage-host-2"); + TEST_ASSERT_NOT_NULL(w); + + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT_MESSAGE(3, rc, + "all 3 non-terminal jobs on this host must be armed"); + + e_finok = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_FINOK, .status = JobStateStatus_finishedOk, + .exit_code = 0, .finish_time_ms = 4242, .queried = 1 + }; + e_unrec = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_UNREC, .status = JobStateStatus_unrecognized, + .exit_code = 0, .finish_time_ms = 0, .queried = 1 + }; + e_running = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_RUNNING, .status = JobStateStatus_running, + .exit_code = 0, .finish_time_ms = 0, .queried = 1 + }; + + /* Page 1 of 2: entries [0..1] (finok, unrec). Page 2 of 2: entry [2] + * (running). Mirrors xce_job_state_report_req()'s own {pos,count, + * n_total} pagination shape (request.c) -- this is exactly what a real + * multi-page OPCode_jobStateReport looks like on the wire. */ + entries_p1[0] = &e_finok; + entries_p1[1] = &e_unrec; + page1 = (xce_job_state_report_t){ + .n_total = 3, .pos = 0, .count = 2, .entries = entries_p1 + }; + + entries_p2[0] = &e_running; + page2 = (xce_job_state_report_t){ + .n_total = 3, .pos = 2, .count = 1, .entries = entries_p2 + }; + + resync_apply_report_call_count_reset_for_test(); + calls_before = resync_apply_report_call_count_for_test(); + + rc = resync_handle_report_page(&g_twr, w, &page1); + + calls_mid = resync_apply_report_call_count_for_test(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "page 1 of 2 alone must NOT complete the report"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, calls_mid - calls_before, + "resync_apply_report must NOT be called after a non-final page -- " + "the structural 'not once per page' proof"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_finok->job.state, + "no partial application: job A must be untouched after page 1"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_unrec->job.state, + "no partial application: job B must be untouched after page 1 -- " + "still RUN, NOT yet reconciled to PEND"); + TEST_ASSERT_NOT_NULL_MESSAGE(j_unrec->job.allocs, + "no partial application: job B's allocation must not be released " + "until the report is fully assembled and applied"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_running->job.state, + "job C (only referenced in page 2, not yet arrived) is untouched"); + TEST_ASSERT_NOT_NULL_MESSAGE(w->resync, + "the round must still be outstanding after only a partial report"); + + rc = resync_handle_report_page(&g_twr, w, &page2); + + calls_after = resync_apply_report_call_count_for_test(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, + "page 2 of 2 completes the report"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, calls_after - calls_mid, + "resync_apply_report must be invoked EXACTLY ONCE, on the final " + "page, for the whole 3-entry report -- not once per page"); + + /* Equivalence (task 6.2's own Observable Completion Criterion): the SAME + * end state test_handle_report_page_single_page_baseline reached for an + * equivalent single-page report. */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_finok->job.state, + "finished-success stays in the DEFERRED record_job_finish_fact_at " + "pattern -- state mutates later via apply_committed_entry, same as " + "an equivalent single-page report (test_apply_report_all_five_" + "outcomes_updates_job_table proves this deferred path directly)"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j_unrec->job.state, + "job B reconciles to PEND -- identical to the single-page baseline"); + TEST_ASSERT_NULL_MESSAGE(j_unrec->job.allocs, + "job B's allocation is released -- identical to the single-page " + "baseline"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j_running->job.state, + "job C (queried running) is left exactly as-is -- no fact " + "recorded, identical to the single-page baseline"); + TEST_ASSERT_NULL_MESSAGE(w->resync, + "once every queried job across BOTH pages has a reconciled " + "outcome, the round must resolve -- identical to the single-page " + "baseline"); + TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status), + "XCE_WORKER_RECONCILING must clear once the round resolves"); + + twr_teardown(); + printf("test_handle_report_page_multi_page_reassembles_and_applies_once:" + " PASS\n"); +} + +/* Req 6.1/6.3 at the reassembly layer: a report that STARTS arriving under + * one round but finishes assembling only after that round has been + * superseded by a fresh resync_arm_round() call (simulating a disconnect- + * then-immediate-rejoin racing the tail of a still-in-flight report) must + * be discarded in its entirety -- not matched against whichever round + * happens to be current once the LAST page arrives. Proves the round_token + * CAPTURE point (page 1, resync_report_accum_t.round_token) is what makes + * this correct, per resync.h's own doc comment on why that capture is not + * an arbitrary choice. */ +static void test_handle_report_page_stale_round_discarded_when_superseded( + void) +{ + job_t* j; + worker_t* w; + int rc; + int calls_before, calls_after; + + xce_job_state_entry_t e1, e2; + xce_job_state_entry_t* entries_p1[1]; + xce_job_state_entry_t* entries_p2[1]; + xce_job_state_report_t page1, page2; + + twr_setup(); + + j = twr_seed_job(TWR_JOBID_MPAGE_STALE, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_MPAGE_STALE, "mpage-host-3"); + + w = twr_make_test_worker("mpage-host-3"); + TEST_ASSERT_NOT_NULL(w); + + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT(1, rc); + + e1 = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_STALE, .status = JobStateStatus_finishedOk, + .exit_code = 0, .finish_time_ms = 1, .queried = 1 + }; + entries_p1[0] = &e1; + page1 = (xce_job_state_report_t){ + .n_total = 2, .pos = 0, .count = 1, .entries = entries_p1 + }; + + rc = resync_handle_report_page(&g_twr, w, &page1); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Simulated disconnect-then-rejoin racing the tail of the in-flight + * report above: a fresh round supersedes the one page1 started under, + * BEFORE page 2 ever arrives. */ + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_NOT_NULL(w->resync); + + /* A second entry, unrelated jobid, just to complete the (now-stale) + * report's page count -- its content is immaterial since the whole + * report must be discarded regardless. */ + e2 = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_STALE, .status = JobStateStatus_finishedFail, + .exit_code = 0, .finish_time_ms = 2, .queried = 1 + }; + entries_p2[0] = &e2; + page2 = (xce_job_state_report_t){ + .n_total = 2, .pos = 1, .count = 1, .entries = entries_p2 + }; + + resync_apply_report_call_count_reset_for_test(); + calls_before = resync_apply_report_call_count_for_test(); + + rc = resync_handle_report_page(&g_twr, w, &page2); + + calls_after = resync_apply_report_call_count_for_test(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, + "the report completes (page 2 arrived) and IS handed to " + "resync_apply_report -- which then discards it internally as " + "stale, per its own Req 6.1/6.3 contract"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, calls_after - calls_before, + "resync_apply_report is still called exactly once -- discarding " + "happens INSIDE it (round-token mismatch), not by skipping the " + "call"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, + "the stale report must never touch the job table"); + TEST_ASSERT_NOT_NULL_MESSAGE(w->resync, + "the CURRENT (superseding) round must be left outstanding, " + "untouched by the stale report"); + TEST_ASSERT_EQUAL_INT(1, w->resync->n_queried_jobids); + + resync_resolve_round(&g_twr, w); + twr_teardown(); + printf("test_handle_report_page_stale_round_discarded_when_superseded:" + " PASS\n"); +} + +/* Robustness: a page whose [pos, pos+count) range falls outside the + * report's own declared n_total is rejected (-EINVAL) and discards + * whatever partial accumulation was in progress -- proven by immediately + * starting a FRESH, valid report afterward and confirming it applies + * correctly, i.e. the corrupt attempt left no residue behind to corrupt a + * later, legitimate report on the same worker. */ +static void test_handle_report_page_malformed_bounds_rejected_and_recovers( + void) +{ + job_t* j; + worker_t* w; + int rc; + + xce_job_state_entry_t e_bad, e_good; + xce_job_state_entry_t* entries_bad[1]; + xce_job_state_entry_t* entries_good[1]; + xce_job_state_report_t page_bad, page_good; + + twr_setup(); + + j = twr_seed_job(TWR_JOBID_MPAGE_BAD, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.allocs + = twr_make_test_alloc_for_host(TWR_JOBID_MPAGE_BAD, "mpage-host-4"); + + w = twr_make_test_worker("mpage-host-4"); + TEST_ASSERT_NOT_NULL(w); + + rc = resync_arm_round(&g_twr, w); + TEST_ASSERT_EQUAL_INT(1, rc); + + /* pos(1) + count(1) == 2 > n_total(1): out of bounds. */ + e_bad = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_BAD2, .status = JobStateStatus_running, + .exit_code = 0, .finish_time_ms = 0, .queried = 0 + }; + entries_bad[0] = &e_bad; + page_bad = (xce_job_state_report_t){ + .n_total = 1, .pos = 1, .count = 1, .entries = entries_bad + }; + + rc = resync_handle_report_page(&g_twr, w, &page_bad); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EINVAL, rc, + "an out-of-bounds page must be rejected"); + + /* A fresh, valid, single-page report for the SAME worker right after -- + * proves the malformed attempt above left no corrupted accumulator + * state behind. */ + e_good = (xce_job_state_entry_t){ + .jobid = TWR_JOBID_MPAGE_BAD, .status = JobStateStatus_unrecognized, + .exit_code = 0, .finish_time_ms = 0, .queried = 1 + }; + entries_good[0] = &e_good; + page_good = (xce_job_state_report_t){ + .n_total = 1, .pos = 0, .count = 1, .entries = entries_good + }; + + rc = resync_handle_report_page(&g_twr, w, &page_good); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, + "a subsequent valid report must apply normally"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_NULL(j->job.allocs); + TEST_ASSERT_NULL(w->resync); + + twr_teardown(); + printf("test_handle_report_page_malformed_bounds_rejected_and_recovers:" + " PASS\n"); +} + /* ── Test group entry point ───────────────────────────────────────────── */ void test_worker_resync(void) @@ -2076,4 +2459,8 @@ void test_worker_resync(void) test_apply_report_finished_success_uses_worker_reported_finish_time); RUN_TEST( test_apply_report_unqueried_extra_unassigned_pend_job_gets_assigned); + RUN_TEST(test_handle_report_page_single_page_baseline); + RUN_TEST(test_handle_report_page_multi_page_reassembles_and_applies_once); + RUN_TEST(test_handle_report_page_stale_round_discarded_when_superseded); + RUN_TEST(test_handle_report_page_malformed_bounds_rejected_and_recovers); } diff --git a/src/include/xce/types.h b/src/include/xce/types.h index e7a71f19..8b9e4c4a 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -309,6 +309,17 @@ typedef struct { resync_state_t* resync; /* worker-rejoin-resync: NULL unless a job-state * reconciliation round is currently outstanding * for this worker */ + /* worker-rejoin-resync task 6.2 (Req 8.1, 8.3): opaque (void*) rather + * than a typed pointer here because its real type, resync_report_ + * accum_t (resync.h), embeds xce_job_state_entry_t -- defined in + * typemap.h, which itself includes THIS header (xce/types.h), so + * typemap.h cannot be included from here without a cycle. NULL unless + * an OPCode_jobStateReport is currently being reassembled across + * multiple pages for this worker; owned and cast back to resync_ + * report_accum_t* exclusively by resync.c, under this worker's own + * RTTI lock, mirroring how `resync` above is the only other opaque- + * from-here, resync.c-owned worker_t field. */ + void* report_accum; } worker_t; typedef struct {