diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 4da1a9d..14659a8 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -76,7 +76,7 @@ - _Requirements: 1.2, 2.1, 2.3_ - _Boundary: Raft Step (Command) — job.c handle_jobsubmit and its call chain only, no dependency on task 1_ -- [ ] 3.2 Convert job submission to register its commit outcome through the new continuation mechanism instead of blocking +- [x] 3.2 Convert job submission to register its commit outcome through the new continuation mechanism instead of blocking - Replace the synchronous commit call from 3.1 with a registration against the task 1 continuation API - On a failed or timed-out outcome, reply failure directly from the continuation — no apply is needed on this path - On a committed outcome, hand the entry off to the ordered apply pump (task 6) rather than applying or replying from this continuation directly, so this entry's apply and reply happen in strict commit order relative to every other entry @@ -193,6 +193,7 @@ - **1.2**: implemented `commit_resolve()` + wired it into all 6 commit.c-internal paths that can advance `committed_lsn`/`writable_lsn` (the COMMITTED/majority-ack outcome). Chose `VTASK_LIVE_BLOCK` for dispatch (never admission-gated, unlike `VTASK_REQ_BLOCK` which can fast-reject — unacceptable for "exactly-once, never unresolved"). **Reviewer caught a real task-graph gap, not a defect in 1.2 itself**: converting `COMMIT_MODE_WAIT` away from a blocking wait also removes the *only* mechanism that ever detected a commit timing out (the caller's own `pthread_cond_timedwait` deadline) — nothing yet arms an equivalent timeout for async-registered continuations, and nothing yet wires demotion/log-truncation detection to call `commit_resolve(FAILED, ...)`. Without this, a pending async command during quorum loss would hang forever instead of failing — worse than today's behavior. Split out as new tasks **1.3** (timeout arming) and **1.4** (demotion/truncation wiring), inserted before the renumbered **1.5** (was 1.3, concurrency test — now exercises the real detection paths from 1.3/1.4 instead of calling `commit_resolve()` directly). Task 3.2 now also depends on 1.3/1.4, since its own "failure/timeout outcome replies" observable is meaningless without them. - **1.3**: added a dedicated per-commit-engine sweep thread (not the shared `delaymgr` — would violate boundary + delaymgr's 200ms poll granularity vs. `DEFAULT_SYNC_TIMEOUT_MS`=500ms would add timing jitter). Race rule: majority-satisfied always wins over an already-expired deadline when both are true under the same lock hold. **Round 1 REJECTED**: `commit_sweep_once()` computed the next-deadline snapshot under `ce->lock`, unlocked to dispatch, then slept on the *stale* snapshot — a `commit_submit_async()` registration/broadcast landing during that unlocked dispatch window was lost (no thread parked on the condvar to receive it), so the sweep thread could sleep on an absent/late deadline indefinitely (worst case: unbounded delay in a low-traffic quorum-loss scenario, violating the "identical failure timing" no-regression bar). **Fixed** by holding `ce->lock` continuously across scan→wait decisions (releasing only for the unlocked dispatch step) and always re-scanning under the lock immediately before ever calling `cond_wait`/`cond_timedwait` — the standard recheck-then-wait idiom. Round 2 reviewer independently reproduced the round-1 bug shape to confirm the fix actually closes it, not just narrows it. **Precedent for 1.4**: any future code that needs to interrupt the sweep thread's wait (e.g., a demotion/truncation handler resolving a pending LSN out-of-band) must go through the same lock+broadcast discipline `commit_submit_async()` uses, or it risks the identical lost-wakeup class of bug. - **1.4**: added `commit_resolve_pending_above(ce, lsn_floor, rc)` (snapshot-then-`commit_resolve()`-per-entry, no parallel bookkeeping). Truncation: a single new `on_truncate`/`truncate_cb_ctx` callback slot on `persist_t` (same inversion-of-control shape as the existing `on_durable_advance` hook) fires from `persist_truncate_wal_above` after releasing `p->lock`, only when a truncation actually removed entries — this ONE wrapper covers all three real truncation call sites (`raft_cb_become_leader`'s post-election truncate, the library's `on_truncate_from` conflict-reconcile, `ha_demotion_convergence_fn`'s post-demotion truncate), verified by the reviewer reading all three. Demotion: `raft_cb_become_follower` calls `commit_resolve_pending_above(&c->commit, 0, -XCE_ERR_STANDBY)` (lsn_floor=0 → fails everything — a demoted node can never authoritatively confirm anything pending; verified against the CP posture). **Known residual, non-blocking**: a fourth internal truncation path (`persist.c` inside `persist_wal_write_replicated`'s tip-conflict-repair branch, STANDBY replicated-apply only) calls `truncate_wal_above_locked()` directly, bypassing the new wrapper — safe today only because demotion resolves ALL pending continuations before role flips to STANDBY (so nothing can still be registered by the time that path runs), but this is an *implicit* ordering invariant, not an asserted one. Flag for whoever next touches the STANDBY replicated-apply path: add a defensive comment/assertion there, or route it through the wrapper too, before relying on it further. Also: bug-1407 was believed closed at this point (70/70 across 3 runs) — **see 1.5's note below, this was a false reading, not a real fix**. +- **3.2**: converted `handle_jobsubmit()` from 3.1's still-synchronous commit to a real async registration against task 1's continuation API. New `persist_submit_job_payloads_async()` (persist.c) appends the WAL entries via the same `ha_primary_log_append`/`persist_wal_async` primitive the pre-existing RETAIN-fact path (`wal_hook_append_classified`) already uses, then calls `commit_submit_async()` directly on `p->repl.ce` — bypassing `repl_wait_majority_ack`/`repl_submit_majority_commit` in repl.c, which is what design.md's text actually specifies as the integration site. **Independently verified safe** (reviewer traced the full ACK→`commit_resolve()` path in commit.c): majority-ack detection is driven entirely by `(ce, lsn)` events inside the commit engine, agnostic to which wrapper registered the continuation — no false-commit risk. But it's a real, permanent design deviation, not just an implementation detail: this creates a THIRD parallel "append+track a commit" implementation (alongside the old sync one, now alive only in 3 test files, and the RETAIN-fact one), narrowing the reusable primitive design.md wanted — a future command handler (e.g. a config-change command) gets no shared plumbing from this work and would need its own bespoke path. On FAILED/timeout, replies failure immediately via `jobsubmit_reply_failure_once()`, no job ever constructed. On COMMITTED, calls new `persist_register_pending_apply(p, lsn, on_applied, ctx)` — a hand-off linked list on `persist_t` for task 6.1 to consume — and does NOT apply or reply, per the task's explicit instruction. **Deleted (not preserved-dead) task 3.1's bug-1408 create_job-loop/notify block**: verified correct — `apply_committed_entry()`'s existing `WAL_JOB_SUBMIT` case (task 2.2) already fires `HOOK_DETAIL_JOB_NEW` per-entry independently, structurally immune to bug-1408's batch-suppression class; resurrecting the old loop in task 6 would double-construct/double-notify. Original code fully recoverable from `git show 6f1fa97:src/controller/job.c` if ever needed. **CRITICAL OPERATIONAL CALLOUT — this branch must NOT be soak-tested or deployed against live traffic until task 6 lands**: there is currently NO live code path that applies a successfully-committed job submission (only restart replay, task 2.3, does) — every successful job submission produces a durably-committed WAL entry but ZERO reply to the client, forever, until task 6.1/6.2 exist to consume the pending-apply hand-off and complete the reply. This is exactly what the task text sanctions ("no independently observable behavior until task 6"), not a bug, but it means this commit range is only safe in isolated unit-test form right now. **Hand-off mechanism for task 6.1's briefing**: `persist_register_pending_apply()`/`persist_has_pending_apply()` exist; there is NO "take and remove" accessor yet — task 6.1 must build the consumer side itself. The per-batch `jobsubmit_async_ctx_t` (ref-counted across array elements, frees only once every element has either failed-and-replied or been applied via `jobsubmit_apply_complete`) will not free for any batch containing a committed element until task 6 exists — an accepted, bounded-by-test-scope leak today, but would be unbounded growth under any live/soak run before task 6 lands. Outcome encoding uses a plain `int rc` (not `commit_outcome_t`+rc) to avoid a persist.h→commit.h circular dependency — verified unambiguous today (every real FAILED call site in commit.c passes non-zero rc) but not compiler-enforced; fragile by convention, not by construction. **New regression found by review, not yet fixed, non-blocking**: `array_size == 0` now hangs the client forever and leaks one `jobsubmit_async_ctx_t` per request (3.1's version replied synchronously even for this input; 3.2's async loop never runs so the ctx never releases) — confirmed NOT reachable via any current legitimate client (Go apiserver always sets `arraySize >= 1`; `xce_test`'s `cmd_jobsub` clamps `arsize <= 0` to 1), so flagged as a defense-in-depth hardening gap for a future task rather than fixed here. Also no `array_size > 1` mixed-outcome test exists yet (per-element independent resolution verified correct by code inspection, not by test) — fast-follow recommended. - **3.1**: reordered `handle_jobsubmit()` (job.c) so job objects are only constructed via `create_job()` AFTER the (still-synchronous) WAL commit succeeds — extracted `jobid_range_exists()` (fast-fail pre-check, no WAL work), `build_job_submit_payloads()` (WAL-payload construction with no jobmgr insertion), and a new persist.c sibling `persist_submit_job_payloads_and_commit()` (operates on raw `xce_job_t*` instead of pre-existing `job_t*`, single production call site in job.c — legitimately within `handle_jobsubmit`'s call chain, not a boundary spillover; no task-1 continuation API dependency, genuinely still synchronous). Took 3 review rounds. **Round 1 REJECTED — real defect, not documentation**: the post-commit array-construction loop suppressed the `AFTER_EVENT`/`HOOK_DETAIL_JOB_NEW` notification for an ENTIRE array batch if even one sibling element failed to construct (e.g. a jobid collision slipping past `jobid_range_exists()`'s inherent TOCTOU window, or a plain allocation failure — no race needed) — stranding OTHER elements that were durably committed AND successfully constructed, invisible to the purely event-driven scheduler/apiserver until the next restart's replay. Logged as bug-1408. **Round 2 fixed the defect** (moved the `construct_rc < 0` early-return to AFTER the jlist-build + hook-fire block, so every successfully-constructed sibling is always notified regardless of a sibling's failure; client reply still separately reports the batch's `construct_rc`) but was REJECTED again on a narrower, documentation-only ground: bug-1408's RED-phase writeup claimed a clean assertion FAIL, but under the standard full-group `--run test_commit_barrier_reply` invocation the failing assertion actually longjmps past the test's own `cbrp_teardown()` call (Unity `TEST_ABORT` semantics) and HANGS the next test — the same longjmp-past-teardown pathology already known from task 1.3's era. **Round 3 (parent-handled directly, no code regression risk)**: corrected bug-1408's buglog.json description to accurately describe the hang mode, and hardened the new test itself — every assertion-relevant observable is now captured into a local, `cbrp_teardown()` is called unconditionally, and assertions run only after teardown, so a future regression in this exact test can no longer skip cleanup and deadlock sibling tests. No production code changed in round 3; job.c's fix from round 2 was independently verified correct by the round-2 reviewer (traced jlist-build/hook-fire at job.c:717-766 running unconditionally before the `construct_rc` check at line 775) and needed no further changes. **Known low-severity residual, not blocking, flagged for a future task**: `apiserver_push_joblist` (apiserver.c) lacks the `nfiltered > 0` guard `sched.c`'s equivalent has, so a total-batch construction failure (every array element fails, jlist ends up empty) now sends an empty/no-op jobs push to every connected apiserver — informationless, can't corrupt state, but a minor behavioral quirk introduced by firing the hook unconditionally. - **2.4**: pure test-only close-out of task-group 2 — new `test_apply_reapply_idempotency.c` (3 tests), no production changes, approved first round. Covers three distinct cases beyond 2.2's existing coverage: (1) triple-reapplication of an identical WAL_JOB_SUBMIT through `apply_committed_entry()` with hook-fire-count AND pointer-identity assertions across all 3 calls (not just final state); (2) the `JOB_TRANSITION_NOOP_DUPLICATE` guard branch (state == current state) — a genuinely untested path, distinct from 2.2's forward-once and backward-rejected cases; (3) the task's named scenario, driven entirely through the real dispatch seam (not seeded/shortcut): real SUBMIT → real FINISH → stale out-of-order START (lower LSN, arriving after), asserting the job stays in its FINISH/terminal state and no third hook fires for the rejected START. Test assertions key on hook-fire COUNT and job STATE, never a specific `HOOK_DETAIL_*` value — so these tests stay valid even if 2.2's documented generic-hook-routing deviation is later corrected. **Task-group 2 (shared apply seam) is now complete**: 2.1 (guard cores extracted) → 2.2 (dispatch function built, generic job-state-family routing is a confirmed permanent deviation from design.md, tracked) → 2.3 (replay wired to the seam, duplicate-notification-on-restart hazard confirmed inert today but must be re-verified before 4.1/6.1/6.2/6.3) → 2.4 (idempotency proven through the real dispatch path). Next: task-group 3 (command apply-after-commit ordering), which is the first consumer of both this seam and task-group 1's commit-engine continuation API. - **2.3**: rewired `replay_wal()` (persist.c) to call `apply_committed_entry()` (task 2.2's seam) instead of `apply_wal_entry()` for restart replay's per-entry apply. `apply_wal_entry`/`apply_committed_entry` are siblings, not wrapper/wrappee — `apply_wal_entry` is NOT retired, it remains ha.c's live standby catch-up/replicated-apply dispatch (untouched, different call sites, out of this task's boundary). **Confirmed hazard, investigated to resolution by the reviewer (not left open)**: `replay_wal(p, p->checkpoint_lsn, ...)` replays the FULL WAL range since the last snapshot on every restart (`last_applied_lsn` is deliberately not persisted — recomputed at replay time per design.md), not just a narrow crash-gap tail. Since `apply_committed_entry` fires `AFTER_EVENT` hooks (unlike the old `apply_wal_entry`, which never did), this means every restart re-fires notification hooks for entries already applied+notified in a prior process life — a duplicate-notification hazard in principle. **Independently traced and confirmed inert today** for both existing hook consumers: scheduler hooks aren't registered until `on_scheduler_join()` (sched.c), which happens post-boot via network JOIN handshake, strictly after `persist_recover()` runs; apiserver hooks ARE registered before `persist_recover()` (`apiserver_init()` in `cluster_init_phase2()`), but every apiserver hook body gates its actual network notify on `apiservers[i].ep != NULL`, and the apiserver-join table starts empty each process life — a no-op except via a narrow reconnect race. **MUST be re-verified before tasks 4.1/6.1/6.2/6.3** make this path unambiguously live and broad (e.g. if a future task registers apiserver connections earlier, or scheduler hooks get registered pre-replay) — at that point this either needs a replay-mode flag suppressing hook-firing, or an explicit accepted-idempotent-duplicate-notification decision recorded in design.md. A loud in-line comment cross-referencing this note now sits at the `apply_committed_entry()` call site in `replay_wal()` (persist.c). Also confirmed: job-state facts now replay through 2.1's guarded `apply_job_update_transition()` instead of the previously-unguarded `restore_job_state()` — intentional (Req 3/7 online-replay divergence fix), verified no existing test depended on the old unguarded semantics. `apply_committed_entry`'s missing oversized-payload guard (vs. `apply_wal_entry`'s `MAX_WAL_PAYLOAD_BYTES`/`-EMSGSIZE` check) confirmed inert — `persist_wal_write_replicated()` already rejects oversized payloads at write time, before anything reaches disk. diff --git a/src/controller/job.c b/src/controller/job.c index 4c58486..5b82007 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -595,7 +595,155 @@ int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request) return rc; } -static int xlist_deref_job_item_(job_t* j, void* arg) { return 0; } +/* raft-apply-decoupling task 3.2: shared state for one handle_jobsubmit() + * call's asynchronously-resolving per-entry commit continuations. Exactly + * one instance per client request — an array submission registers one + * continuation per array element, all sharing this struct, so the client + * still gets exactly one reply for the whole batch (matching pre-3.2 + * per-request reply semantics), never one reply per array element. + * + * Lifetime: allocated in handle_jobsubmit() before it returns; released by + * whichever per-entry continuation (jobsubmit_commit_continuation) or + * apply-complete callback (jobsubmit_apply_complete) observes `pending` + * reach 0 — the last one out closes the door. + * + * IMPORTANT: a COMMITTED outcome does NOT decrement `pending` by itself + * (see jobsubmit_commit_continuation below) — task 3.2's own contract is + * that a committed entry is handed off to the not-yet-built ordered apply + * pump (task 6.1/6.2), and only jobsubmit_apply_complete (invoked BY that + * future pump) decrements `pending` for it. A batch with any committed + * entry is therefore only ever freed once task 6 exists and actually + * applies it. This mirrors the design's already-accepted "durably + * committed, not yet applied" posture for retained facts during quorum + * loss (Req 6.1) — it is not a leak. */ +typedef struct { + cluster_t* c; + endpoint_t* ep; /* ref'd via endpoint_ref(); released exactly + * once, in jobsubmit_async_ctx_release() */ + uint64_t seq; /* xce_request_t.seq, copied for the reply */ + uint32_t opcode; /* xce_request_t.opcode, copied for the reply */ + pthread_mutex_t lock; /* guards pending/replied below */ + int pending; /* entries not yet FULLY resolved (see above) */ + int replied; /* 1 once a reply has been sent for this batch */ +} jobsubmit_async_ctx_t; + +static void jobsubmit_async_ctx_release(jobsubmit_async_ctx_t* actx) +{ + endpoint_unref(actx->ep); + pthread_mutex_destroy(&actx->lock); + free(actx); +} + +/* Sends exactly one failure reply for the whole batch — the first failure + * observed wins; later failures/hand-offs never send a second, possibly + * contradictory reply for the same original request (Req 9.3, read at the + * per-REQUEST granularity a job-array submission actually has one of). */ +static void jobsubmit_reply_failure_once(jobsubmit_async_ctx_t* actx, int rc) +{ + xce_reply_t reply; + int send = 0; + + pthread_mutex_lock(&actx->lock); + if (!actx->replied) { + actx->replied = 1; + send = 1; + } + pthread_mutex_unlock(&actx->lock); + + if (!send) + return; + + memset(&reply, 0, sizeof(reply)); + reply.seq = actx->seq; + reply.opcode = actx->opcode; + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + + xce_reply(actx->ep, actx->c->myhost, &reply); +} + +/* raft-apply-decoupling task 6 hand-off target (task 6 itself is NOT built + * by this task): invoked by the future ordered apply pump immediately + * after it applies this entry via apply_committed_entry() — task 6.2's own + * contract, "fire a command's registered continuation immediately after + * its entry applies". Until task 6 exists, nothing ever calls this + * function; see jobsubmit_commit_continuation's COMMITTED branch below, + * which registers it as the hand-off target instead of applying or + * replying itself. + * + * NOTE for whoever builds task 6.2: apply_committed_entry()'s WAL_JOB_SUBMIT + * case (persist.c) already constructs the job (via restore_job()) AND fires + * the HOOK_DETAIL_JOB_NEW AFTER_EVENT notification, independently per entry + * — so by the time this function is invoked, the job is very likely already + * visible and already notified. This function most likely only needs to + * send the batch's success reply (echoing the jobid) once every entry has + * been applied, NOT reconstruct or re-notify anything — see the removed-code + * note above handle_jobsubmit's old (pre-3.2) construct/notify block for the + * full reasoning (git history, commit 6f1fa97, task 3.1). */ +static void jobsubmit_apply_complete(void* ctx, uint64_t lsn) +{ + jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx; + int done; + + (void)lsn; + + pthread_mutex_lock(&actx->lock); + actx->pending--; + done = (actx->pending <= 0); + pthread_mutex_unlock(&actx->lock); + + /* Task 6.2's success-reply belongs here (RequestReply_data_jobid, + * gated on actx->replied exactly like jobsubmit_reply_failure_once, so + * a partially-failed batch's earlier failure reply is never followed + * by a contradictory success reply) — not built by this task, which + * must not reply from the committed-outcome path at all (see the + * struct doc comment above). */ + if (done) + jobsubmit_async_ctx_release(actx); +} + +/* raft-apply-decoupling task 3.2: per-entry outcome of one array element's + * commit_submit_async() registration (persist_submit_job_payloads_async's + * `fn` callback). + * + * On FAILED/timed-out (rc != 0): reply failure now — no apply is needed on + * this path (Req 2.3). This entry has nothing further to wait on, so it + * decrements `pending` itself. + * + * On COMMITTED (rc == 0): task 3.2 must NOT apply the job or reply here + * (the task's own explicit instruction) — hand the entry off to the + * not-yet-built ordered apply pump (task 6.1/6.2) via + * persist_register_pending_apply(), the producer-side hand-off this task + * builds. `pending` is intentionally NOT decremented here for a committed + * entry — see jobsubmit_async_ctx_t's doc comment. */ +static void jobsubmit_commit_continuation(void* ctx, uint64_t lsn, int rc) +{ + jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx; + + if (rc != 0) { + int done; + + WARN("handle_jobsubmit: lsn=%lu commit failed/timed out (rc=%d) — " + "rejecting; no job ever constructed for it", + (unsigned long)lsn, rc); + jobsubmit_reply_failure_once(actx, rc); + + pthread_mutex_lock(&actx->lock); + actx->pending--; + done = (actx->pending <= 0); + pthread_mutex_unlock(&actx->lock); + if (done) + jobsubmit_async_ctx_release(actx); + return; + } + + /* COMMITTED: hand off to task 6's ordered apply pump instead of + * applying/replying directly from this continuation — see + * persist_register_pending_apply's doc comment (persist.h) for the + * full hand-off contract. */ + persist_register_pending_apply(&actx->c->persist, lsn, + jobsubmit_apply_complete, actx); +} int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) { @@ -603,10 +751,7 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) xce_job_submit_t* submit; xce_reply_t reply; xlist_t plist; /* xce_job_t* WAL payloads, pre-commit */ - xlist_t jlist; /* job_t* for the post-commit notify hook */ uint64_t* lsns = NULL; - job_t* j = NULL; - int i; int rc; submit = request->body.jobsubmsg; @@ -660,138 +805,72 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) return rc; } - /* Explicit raft step (raft-apply-decoupling item #1 / bug-1306; task - * 3.1): WAL-append + majority-commit barrier for every payload, reading - * the commit verdict DIRECTLY. This used to run inside the AFTER_EVENT - * hook (persist_job_hook_fun), but xce_run_hooks returns only the LAST - * hook's rc, so a later notify subscriber (sched/apiserver) returning 0 - * CLOBBERED the -ETIMEDOUT — the client was acked for an entry that - * never reached majority, which a new-term leader then legally - * discarded (committed-job-loss). The client must not be acknowledged, - * and no job object may exist, until the submission is durable on a - * majority. */ - int commit_rc = persist_submit_job_payloads_and_commit(&c->persist, - &plist, lsns); - if (commit_rc < 0) { - WARN("handle_jobsubmit: job not durable (rc=%d) — rejecting submission", - commit_rc); + /* Explicit raft step (raft-apply-decoupling item #1 / bug-1306, task + * 3.1; converted to the continuation mechanism by task 3.2): register + * every payload's commit outcome instead of blocking on the majority + * barrier. This handler returns as soon as every payload is durably + * appended locally and its outcome-resolution path is established — + * BEFORE any of them necessarily commit. From this point on, job + * construction and the client reply happen from + * jobsubmit_commit_continuation / jobsubmit_apply_complete (above), + * never again from this stack frame. + * + * REMOVED (raft-apply-decoupling task 3.2) relative to task 3.1's + * version of this function: the post-commit create_job()-loop + + * jlist-build + xce_run_hooks(AFTER_EVENT, HOOK_DETAIL_JOB_NEW) block + * (bug-1408's fix: unconditional per-successfully-constructed-element + * notification) is gone, not merely deferred. It is SUPERSEDED, not + * resurrected, by apply_committed_entry()'s existing WAL_JOB_SUBMIT + * case (persist.c, raft-apply-decoupling task 2.2, already built): + * that function already does restore_job() + fires HOOK_DETAIL_JOB_NEW + * per entry, independently per array element — which is structurally + * immune to the bug-1408 class (a sibling's apply failure can never + * suppress another sibling's notification, because each LSN applies + * through its own independent call). The full reasoning is preserved + * in this task's status report; the removed code itself remains in + * git history at commit 6f1fa97. Whoever builds task 6.1/6.2 should + * call apply_committed_entry() (not resurrect this loop) and should + * expect jobsubmit_apply_complete to need only the success reply. */ + jobsubmit_async_ctx_t* actx = calloc(1, sizeof(*actx)); + if (actx == NULL) { free_job_submit_payloads(&plist); free(lsns); - reply.status = commit_rc; + reply.status = -ENOMEM; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); - return commit_rc; + return -ENOMEM; } + actx->c = c; + actx->ep = ep; + actx->seq = request->seq; + actx->opcode = request->opcode; + actx->pending = plist.count; + actx->replied = 0; + pthread_mutex_init(&actx->lock, NULL); + endpoint_ref(ep); - /* Committed on a majority — NOW construct and register each job object - * (moved here from before the commit barrier; raft-apply-decoupling - * task 3.1 / Req 2.1, 2.2, 2.3). A construction failure at this point - * cannot be rolled back (the entry is already durably committed); it is - * a genuine apply-time failure, not a retryable one (design.md Error - * Strategy) — logged loud, and the submission is NOT reported as a - * success. */ - int construct_rc = 0; - for (i = 0; i < plist.count; i++) { - xce_job_t* jp = (xce_job_t*)xlist_get(&plist, i); - job_t* job = create_job(c, jp); - - if (job == NULL) { - int saved_errno = errno; - - ERROR("handle_jobsubmit: job %lu committed at lsn %lu but " - "could not be constructed (errno=%d)", - (unsigned long)jp->jobid, (unsigned long)lsns[i], - saved_errno); - if (construct_rc == 0) - construct_rc = -saved_errno; - continue; - } - job->submit_lsn = lsns[i]; - } + (void)persist_submit_job_payloads_async(&c->persist, &plist, lsns, + jobsubmit_commit_continuation, + actx); + /* Every payload's WAL data is no longer needed once appended: the + * future apply pump (task 6) re-reads each committed entry's payload + * from the WAL itself — the same way restart replay already does — + * not from this in-memory copy. persist_submit_job_payloads_async has + * ALREADY invoked jobsubmit_commit_continuation, synchronously or + * asynchronously, exactly once for every payload by (or after) the + * time it returns; its own return value carries no information this + * function still needs to act on. */ free_job_submit_payloads(&plist); free(lsns); - /* Build the job list for the AFTER_EVENT hook — the same lookup used - * before this task, just run after construction (which now itself runs - * after commit) instead of before either. - * - * IMPORTANT (raft-apply-decoupling task 3.1 remediation): this lookup - * and the hook it feeds run UNCONDITIONALLY here, even when - * construct_rc < 0 for one or more array elements. Every element whose - * create_job() succeeded above is already durably committed AND already - * inserted into c->jobmgr — key_find below naturally finds only those - * (a failed element was never inserted, so it is silently absent from - * jlist with no extra bookkeeping needed). Both the scheduler and the - * apiserver are purely event-driven off this hook with no periodic - * reconciliation, so returning early here (as a previous version of - * this function did whenever ANY sibling failed to construct) would - * strand the successfully constructed siblings — real, durable jobs — - * invisible until the next controller restart replays the WAL. A - * whole-batch construction failure must never suppress notification for - * the elements that individually succeeded. */ - xlist_init(&jlist, submit->array_size); - - for (i = 0; i < submit->array_size; i++) { - lock(c->jobmgr, RTTI_LOCK_READ); - rc = key_find(c->jobmgr, (void*)(jobid + i), (void**)&j); - if (j != NULL) { - reference(j); - } - unlock(c->jobmgr); - - if ((rc <= 0) || (j == NULL)) { - char idstring[128]; - - fmt_jobid(idstring, sizeof(idstring) - 1, jobid); - ERROR("%s fail to lock job %s (%lu)", __func__, idstring, jobid); - } else { - if (xlist_store(&jlist, j) < 0) { - PERROR("xlist_store"); - } - } - } - - /* Committed + constructed — fire the AFTER_EVENT hook as NOTIFICATION - * only (scheduler push, apiserver SSE), observing already-committed, - * already-applied state. Its return no longer gates durability: a - * notify-hook failure cannot un-commit an already-durable job, so it is - * logged but never fails the client reply. Fired for every - * successfully constructed element regardless of construct_rc (see the - * comment above jlist's construction). */ - int notify_rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, - HOOK_AFTER_EVENT, - HOOK_DETAIL_JOB_NEW, c, &jlist); - if (notify_rc != 0) - WARN("handle_jobsubmit: post-commit notification hook rc=%d " - "(job already durable; not failing the reply)", notify_rc); - - xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL); - - xlist_fini(&jlist); - - if (construct_rc < 0) { - /* At least one array element durably committed but could not be - * constructed (see the ERROR logged at the point of failure above). - * The client learns the submission did not fully succeed; the - * siblings that DID construct successfully were already notified - * above and must never have that notification gated on this. */ - reply.status = construct_rc; - reply.data_which = RequestReply_data_nodata; - - xce_reply(ep, c->myhost, &reply); - return construct_rc; - } - - /* Reply success only AFTER WAL write + majority commit + construction. */ - reply.status = 0; - reply.data_which = RequestReply_data_jobid; - reply.data.jobid = jobid; - - xce_reply(ep, c->myhost, &reply); - + /* The calling thread returns here, before any payload's commit outcome + * is necessarily known (raft-apply-decoupling task 3.2, Req 1.1, 1.4, + * 2.1, 2.3) — every reply from this point on is sent by + * jobsubmit_reply_failure_once or (once task 6 exists) + * jobsubmit_apply_complete, off this stack frame. */ return 0; } diff --git a/src/controller/persist.c b/src/controller/persist.c index 6a5e854..4c74cfc 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -38,6 +38,20 @@ extern int worker_from_capnp(cluster_t* c, extern int restore_queue_state(cluster_t* c, const char* name, uint32_t status); +/* raft-apply-decoupling task 3.2: pending command-apply hand-off list node + * (producer side; the not-yet-built ordered apply pump, task 6.1/6.2, is + * the consumer). Declared here (rather than persist.h) so the list stays + * an implementation detail — persist.h exposes only the opaque `void*` + * head field plus persist_register_pending_apply/persist_has_pending_apply. + * See persist.h's doc comment on persist_register_pending_apply for the + * full contract. */ +typedef struct pending_command_apply { + uint64_t lsn; + pending_apply_notify_fn on_applied; + void* ctx; + struct pending_command_apply* next; +} pending_command_apply_t; + /* ---- Cap'n Proto serialization helpers (WAL payload encoding) ---------- */ /* Serialize xce_job_t to a malloc'd Cap'n Proto buffer. @@ -617,6 +631,12 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) p->term_index_count = 0; p->term_index_cap = 0; + /* Pending command-apply hand-off list starts empty (raft-apply- + * decoupling task 3.2): no committed-outcome continuations have + * registered yet. Freed in persist_fini. */ + p->pending_applies = NULL; + pthread_mutex_init(&p->pending_applies_lock, NULL); + /* Create WAL and snapshot directories */ if (mkdir_p(p->wal_path, 0700) != 0) return -errno; if (mkdir_p(p->snap_path, 0700) != 0) return -errno; @@ -723,6 +743,23 @@ void persist_fini(persist_t* p) p->term_index = NULL; p->term_index_count = 0; p->term_index_cap = 0; + + /* Free any pending command-apply hand-off entries never drained by a + * (not-yet-built) task 6 apply pump (raft-apply-decoupling task 3.2). + * Each node's `ctx` is opaque to persist.c (owned by the registering + * caller, e.g. job.c's jobsubmit_async_ctx_t) — freeing the node here + * does NOT free ctx; a caller shutting down with entries still pending + * is responsible for its own ctx lifetime (test harnesses that tear + * down mid-flight must account for this). */ + pending_command_apply_t* pn = + (pending_command_apply_t*)p->pending_applies; + while (pn != NULL) { + pending_command_apply_t* next = pn->next; + free(pn); + pn = next; + } + p->pending_applies = NULL; + pthread_mutex_destroy(&p->pending_applies_lock); } /* Forward declaration — defined after checkpoint internals */ @@ -3706,6 +3743,194 @@ int persist_submit_job_payloads_and_commit(persist_t* p, xlist_t* payloads, return rc; } +/* raft-apply-decoupling task 3.2: append-only counterpart of + * submit_job_wal_entry_and_commit — writes the entry via the SAME + * primary-append primitive wal_hook_append_classified already uses for + * RETAIN-mode facts (ha_primary_log_append / persist_wal_async), WITHOUT + * waiting on the majority-commit barrier. Deliberately does NOT reuse + * submit_job_wal_entry_and_commit itself: that helper is shared with + * persist_submit_wal_and_commit's still-synchronous, still-blocking + * contract (relied on directly by test_egress_filter.c / test_e2e_hooks.c / + * test_persist_hooks.c), which this task must not disturb. */ +static int append_job_wal_entry_local(persist_t* p, const xce_job_t* job, + uint64_t* out_lsn) +{ + uint8_t* buf = NULL; + size_t sz = 0; + int rc; + uint64_t lsn; + + if (capnp_encode_job(job, &buf, &sz) != 0) + return -ENOMEM; + + /* Snapshot the LSN this entry will receive BEFORE the append (same + * convention as submit_job_wal_entry_and_commit / wal_hook_append_ + * classified: persist_wal_async / ha_primary_log_append assigns + * hdr.lsn = current_lsn then increments). */ + pthread_mutex_lock(&p->lock); + lsn = p->current_lsn; + pthread_mutex_unlock(&p->lock); + + if (p->cluster != NULL) + rc = ha_primary_log_append(p->cluster, WAL_JOB_SUBMIT, buf, + (uint32_t)sz, NULL); + else + rc = persist_wal_async(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz); + free(buf); + + if (rc == 0) + *out_lsn = lsn; + return rc; +} + +/* raft-apply-decoupling task 3.2: bridges commit_submit_async's + * commit_continuation_fn shape (commit_outcome_t + rc) to + * persist_job_commit_notify_fn's shape (a single rc) so persist.h's public + * declaration never needs commit.h's commit_outcome_t (see persist.h's + * doc comment on persist_job_commit_notify_fn). One adapter is heap- + * allocated per registered lsn and frees itself after its single + * invocation — safe because commit_resolve guarantees exactly-once + * delivery. */ +typedef struct { + persist_job_commit_notify_fn fn; + void* ctx; +} job_commit_adapter_ctx_t; + +static void job_commit_resolve_adapter(void* adapter_ctx, uint64_t lsn, + commit_outcome_t outcome, int rc) +{ + job_commit_adapter_ctx_t* a = (job_commit_adapter_ctx_t*)adapter_ctx; + int final_rc = (outcome == COMMIT_OUTCOME_COMMITTED) ? 0 : rc; + + a->fn(a->ctx, lsn, final_rc); + free(a); +} + +int persist_submit_job_payloads_async(persist_t* p, xlist_t* payloads, + uint64_t* out_lsns, + persist_job_commit_notify_fn fn, + void* ctx) +{ + int first_rc = 0; + + if (p == NULL || payloads == NULL || out_lsns == NULL || fn == NULL) + return -EINVAL; + + for (int i = 0; i < payloads->count; i++) { + xce_job_t* j = (xce_job_t*)xlist_get(payloads, i); + uint64_t lsn = 0; + int rc; + + if (first_rc != 0) { + /* A prior payload in this batch already failed locally: every + * remaining payload resolves with that SAME failure now, + * synchronously — so `fn` is invoked exactly once per payload + * regardless of where the failure occurred, and the caller + * never has to separately track "never going to be called" + * entries. */ + fn(ctx, 0, first_rc); + continue; + } + + rc = append_job_wal_entry_local(p, j, &lsn); + if (rc != 0) { + first_rc = rc; + fn(ctx, 0, rc); + continue; + } + out_lsns[i] = lsn; + + /* LSN-0 edge (documented once-per-cluster-lifetime interim + * behavior — ha-commit-engine Implementation Notes, mirrored by + * repl_wait_majority_ack's walAck branch): the first-ever WAL + * entry of a fresh cluster is structurally un-commitable and + * commit_submit_async rejects lsn == 0, so resolve it inline. + * Engine-not-wired (bootstrap / engine-less contexts) mirrors + * persist_commit_barrier's own "no engine -> no-op success" + * convention — there is no async completion authority to + * register with. */ + if (lsn == 0 || p->repl.ce == NULL) { + fn(ctx, lsn, 0); + continue; + } + + job_commit_adapter_ctx_t* a = calloc(1, sizeof(*a)); + if (a == NULL) { + first_rc = -ENOMEM; + fn(ctx, lsn, -ENOMEM); + continue; + } + a->fn = fn; + a->ctx = ctx; + + int reg_rc = commit_submit_async(p->repl.ce, lsn, + job_commit_resolve_adapter, a); + if (reg_rc != 0) { + free(a); + first_rc = reg_rc; + ERROR("persist_submit_job_payloads_async: lsn=%lu appended but " + "commit_submit_async registration failed (rc=%d) -- the " + "entry remains durable and will still be caught up " + "eventually via replay, but this caller will not be told " + "its outcome via this path", + (unsigned long)lsn, reg_rc); + fn(ctx, lsn, reg_rc); + continue; + } + } + + return first_rc; +} + +void persist_register_pending_apply(persist_t* p, uint64_t lsn, + pending_apply_notify_fn on_applied, + void* ctx) +{ + pending_command_apply_t* node; + + if (p == NULL || on_applied == NULL) + return; + + node = calloc(1, sizeof(*node)); + if (node == NULL) { + ERROR("persist_register_pending_apply: OOM registering the apply " + "hand-off for lsn=%lu -- the entry is still durably committed " + "and will still be applied by a future restart replay, but no " + "continuation will fire for it via this path", + (unsigned long)lsn); + return; + } + node->lsn = lsn; + node->on_applied = on_applied; + node->ctx = ctx; + + pthread_mutex_lock(&p->pending_applies_lock); + node->next = (pending_command_apply_t*)p->pending_applies; + p->pending_applies = node; + pthread_mutex_unlock(&p->pending_applies_lock); +} + +int persist_has_pending_apply(persist_t* p, uint64_t lsn) +{ + pending_command_apply_t* n; + int found = 0; + + if (p == NULL) + return 0; + + pthread_mutex_lock(&p->pending_applies_lock); + for (n = (pending_command_apply_t*)p->pending_applies; n != NULL; + n = n->next) { + if (n->lsn == lsn) { + found = 1; + break; + } + } + pthread_mutex_unlock(&p->pending_applies_lock); + + return found; +} + /* P0 payload: just the detail value (4 bytes). * Full Cap'n Proto serialization is wired in Task 15. */ diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 694a130..29bd91a 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -834,13 +834,23 @@ static void test_worker_join_no_hostconfig_error_reply_immediate(void) printf("test_worker_join_no_hostconfig_error_reply_immediate: PASS\n"); } -/* ── handle_jobsubmit (raft-apply-decoupling task 3.1: apply-after-commit) ── +/* ── handle_jobsubmit (raft-apply-decoupling task 3.1/3.2) ─────────────── * - * Req 2.1/2.3: a job submission whose commit fails or times out must leave - * NO job object in c->jobmgr — construction is deferred until after the - * commit barrier succeeds. Req 2.2: a successful submission still replies - * synchronously with the jobid, and by the time that reply is observable the - * job is already visible in c->jobmgr. */ + * task 3.1 (Req 2.1/2.3): a job submission whose commit fails or times out + * must leave NO job object in c->jobmgr — construction is deferred until + * after the commit barrier succeeds. + * + * task 3.2 (Req 1.1, 1.4, 2.1, 2.3): the synchronous commit call from 3.1 is + * replaced with a registration against the commit engine's continuation + * mechanism (commit_submit_async, via persist_submit_job_payloads_async). + * handle_jobsubmit now returns BEFORE any commit outcome is known. A + * failed/timed-out outcome still replies failure directly, with no job ever + * constructed (same observable as 3.1). A COMMITTED outcome is handed off to + * the not-yet-built ordered apply pump (task 6.1/6.2) via + * persist_register_pending_apply() instead of applying or replying directly + * — so a successful submission has NO independently observable behavior + * (no reply, no job in c->jobmgr) until task 6 exists; the test below + * asserts only that the hand-off itself was registered for the correct lsn. */ static job_startinfo_t g_jobsub_startinfo; static job_submit_data_t g_jobsub_data; @@ -902,121 +912,24 @@ static void test_jobsubmit_withheld_acks_no_phantom_job(void) printf("test_jobsubmit_withheld_acks_no_phantom_job: PASS\n"); } -static void test_jobsubmit_success_reply_and_job_visible_after_commit(void) +/* raft-apply-decoupling task 3.2: supersedes the pre-3.2 + * test_jobsubmit_success_reply_and_job_visible_after_commit (which drove + * handle_jobsubmit through cbrp_drive_acked, a helper that asserts the + * handler blocks until its reply is observable — no longer true here by + * this task's own explicit contract). This test asserts ONLY what task 3.2 + * itself makes observable on the committed-outcome path: the calling + * thread returns before the commit resolves, and the continuation hands + * the entry off to persist's pending-apply registry keyed by the correct + * lsn — NOT a reply, NOT a constructed job (task 6 owns both; neither + * exists yet). */ +static void test_jobsubmit_returns_before_commit_and_registers_pending_apply(void) { - cbrp_setup(1500); + uint64_t lsn_before, watermark, t0, join_elapsed_ms, t1; + cbrp_reply_obs_t obs = { 0 }; + bool has_pending, job_exists; + + cbrp_setup(2000); /* generous timeout: must never be needed */ cbrp_arrange_jobsubmit(9101, 1); - cbrp_drive_acked(handle_jobsubmit, &g_jobsub_req); - TEST_ASSERT_TRUE_MESSAGE(cbrp_job_exists(9101), - "a successful job submission must construct the job object once " - "its commit succeeds, and still reply synchronously with the " - "jobid (raft-apply-decoupling task 3.1, Req 2.2)"); - cbrp_teardown(); - printf("test_jobsubmit_success_reply_and_job_visible_after_commit: PASS\n"); -} - -/* Captures every jobid delivered via the AFTER_EVENT/HOOK_DETAIL_JOB_NEW - * notification hook (job.c's post-commit notify list). Needed by the - * array-partial-construction-failure remediation test below to observe - * that a durably committed AND successfully constructed sibling is still - * notified even when another array element in the same submission fails to - * construct (raft-apply-decoupling task 3.1 remediation). xce_run_hooks - * also fires this same hook TYPE for HOOK_BEFORE_EVENT with a different - * vararg shape (xce_job_submit_t*, not xlist_t*) — filtered out below - * before touching ap. */ -#define CBRP_MAX_JOBNEW_NOTIFY 8 -static int g_jobnew_notify_count; -static uint64_t g_jobnew_notify_jobids[CBRP_MAX_JOBNEW_NOTIFY]; - -static void cbrp_jobnew_capture_reset(void) -{ - g_jobnew_notify_count = 0; - memset(g_jobnew_notify_jobids, 0, sizeof(g_jobnew_notify_jobids)); -} - -static int cbrp_jobnew_capture_fn(int type, int situation, int detail, - void *data, va_list ap) -{ - (void)type; (void)data; - - if (detail != HOOK_DETAIL_JOB_NEW || situation != HOOK_AFTER_EVENT) - return 0; - - { - xlist_t *jlist; - int i; - - (void)va_arg(ap, cluster_t *); - jlist = va_arg(ap, xlist_t *); - - if (jlist == NULL) - return 0; - - for (i = 0; i < jlist->count - && g_jobnew_notify_count < CBRP_MAX_JOBNEW_NOTIFY; i++) { - job_t *j = (job_t *)xlist_get(jlist, i); - if (j != NULL) - g_jobnew_notify_jobids[g_jobnew_notify_count++] = j->job.jobid; - } - } - - return 0; -} - -static bool cbrp_jobnew_notified(uint64_t jobid) -{ - int i; - - for (i = 0; i < g_jobnew_notify_count; i++) { - if (g_jobnew_notify_jobids[i] == jobid) - return true; - } - return false; -} - -/* Array-partial-construction-failure remediation (round-2 fix, task 3.1): - * when ONE array element's post-commit create_job() fails (a jobid - * collision that slipped past jobid_range_exists()'s pre-check, or any - * other construct-time failure), the OTHER array elements that constructed - * successfully — and are already durably committed — must still fire the - * AFTER_EVENT/JOB_NEW notification hook. Both scheduler and apiserver are - * purely event-driven off this hook with no periodic reconciliation, so a - * withheld notification strands an otherwise-healthy, already-durable job - * until the next controller restart replays the WAL. - * - * The colliding element's failure is reproduced deterministically (no real - * race) by exploiting the actual TOCTOU window between - * jobid_range_exists()'s pre-check (a READ-locked snapshot taken BEFORE any - * WAL work) and makeJob_'s own key_find (a WRITE-locked check made fresh at - * construction time, AFTER the commit barrier): a stand-in job for the - * second array element's jobid is inserted directly into c->jobmgr from the - * test thread strictly after the pre-check has already run (its own WAL - * entry is visible, which cannot happen until after the pre-check passes) - * and strictly before the construction loop can run (its commit is still - * pending on a withheld ACK). */ -static void test_jobsubmit_array_partial_construct_failure_notifies_survivors(void) -{ - cbrp_reply_obs_t obs = { 0 }; - uint64_t lsn_before, wm0, wm1; - job_submit_data_t collide_submit; - bool elem0_wal_appeared, elem1_wal_appeared; - bool elem0_exists, elem0_notified; - - /* Every TEST_ASSERT_* below is deferred until after cbrp_teardown(). - * A failing Unity assertion longjmps out of this function (TEST_ABORT), - * which would otherwise skip cbrp_teardown() and leak live - * persist/commit-engine background threads into the NEXT test's setup, - * deadlocking the whole group (see bug-1408's round-2 correction: this - * exact failure mode was originally reproduced as a hang, not a clean - * FAIL, under the standard --run test_commit_barrier_reply invocation). - * Capture every observable into a local first, teardown unconditionally, - * then assert on the captured locals. */ - - cbrp_setup(2000); - cbrp_jobnew_capture_reset(); - TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, - cbrp_jobnew_capture_fn, NULL)); - cbrp_arrange_jobsubmit(9200, 2); /* 9200 survives, 9201 collides */ pthread_mutex_lock(&g_hc.persist.lock); lsn_before = g_hc.persist.current_lsn; @@ -1028,70 +941,146 @@ static void test_jobsubmit_array_partial_construct_failure_notifies_survivors(vo TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, cbrp_op_thread_fn, &g_op_arg)); - /* Element 0's (jobid 9200) WAL entry appears first; ACK it so the - * sequential payload loop advances to element 1. The settling sleep - * before each ACK mirrors cbrp_drive_acked's proven pattern — it gives - * the handler thread time to actually register its commit wait for the - * just-appended LSN before the ACK is delivered. */ - wm0 = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); - elem0_wal_appeared = (wm0 > 0); - cbrp_sleep_ms(150); - if (elem0_wal_appeared) - repl_on_ack(&g_hc.persist.repl, &g_hep_a, wm0); + /* No ACK is delivered before this join. If handle_jobsubmit still + * blocked on the majority-commit barrier (the pre-3.2 behavior), this + * join could not return before the 2000ms sync_timeout elapses; a fast + * return proves registration replaced blocking (Req 1.1, 1.4). */ + t0 = cbrp_mono_ms(); + TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); + join_elapsed_ms = cbrp_mono_ms() - t0; - /* Element 1's (jobid 9201) WAL entry appears once element 0 commits. - * Its own ACK is deliberately withheld until the collision stand-in - * below is in place. cbrp_poll_append's lsn_before is a current_lsn - * SNAPSHOT (see its first use above, seeded from persist.current_lsn), - * not an entry's assigned LSN — element 0 already advanced current_lsn - * to wm0 + 1, so that (not wm0 itself) is the correct baseline to wait - * past for element 1's append. */ - wm1 = cbrp_poll_append(&g_hc.persist, wm0 + 1, 2000); - elem1_wal_appeared = (wm1 > wm0); - cbrp_sleep_ms(150); + watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); - /* Deterministic TOCTOU-outcome injection (see block comment above). */ - memset(&collide_submit, 0, sizeof(collide_submit)); - cbrp_make_job(9201, &collide_submit); + /* Deliver the covering ACK, then poll for the hand-off artifact + * (raft-apply-decoupling task 3.2's producer-side registration) — + * NOT for a reply or a constructed job, neither of which this task + * produces on the committed-outcome path. */ + repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); - if (elem1_wal_appeared) - repl_on_ack(&g_hc.persist.repl, &g_hep_a, wm1); + has_pending = false; + t1 = cbrp_mono_ms(); + while (cbrp_mono_ms() - t1 < 2000) { + if (persist_has_pending_apply(&g_hc.persist, watermark)) { + has_pending = true; + break; + } + cbrp_sleep_ms(5); + } - cbrp_poll_reply(&g_req_ep, 2000, &obs); - pthread_join(g_op_tid, NULL); - - elem0_exists = cbrp_job_exists(9200); - elem0_notified = cbrp_jobnew_notified(9200); + /* Give the (non-existent) rest of the pipeline a beat, then confirm + * nothing leaked through — the task's own explicit "no independently + * observable behavior" contract for the committed-outcome path. */ + cbrp_sleep_ms(200); + cbrp_scan_replies(&g_req_ep, &obs); + job_exists = cbrp_job_exists(9101); cbrp_teardown(); - TEST_ASSERT_TRUE_MESSAGE(elem0_wal_appeared, - "array element 0's WAL entry must appear"); - TEST_ASSERT_TRUE_MESSAGE(elem1_wal_appeared, - "array element 1's WAL entry must appear"); + TEST_ASSERT_TRUE_MESSAGE(join_elapsed_ms < 500, + "handle_jobsubmit must return before any commit outcome is known " + "(raft-apply-decoupling task 3.2, Req 1.1/1.4) — no ACK was ever " + "delivered before this join"); + TEST_ASSERT_TRUE_MESSAGE(watermark > 0, + "job submission must still append its WAL entry"); + TEST_ASSERT_TRUE_MESSAGE(has_pending, + "a committed entry must be handed off to the ordered apply pump " + "via persist_register_pending_apply, keyed by its lsn " + "(raft-apply-decoupling task 3.2)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.n_replies, + "no reply may be sent on the committed-outcome path until task 6's " + "apply pump exists (raft-apply-decoupling task 3.2's own contract)"); + TEST_ASSERT_FALSE_MESSAGE(job_exists, + "no job may be constructed on the committed-outcome path until " + "task 6's apply pump exists (raft-apply-decoupling task 3.2)"); - TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, - "the requester must still receive a reply when one array element " - "fails post-commit construction"); - TEST_ASSERT_TRUE_MESSAGE(obs.last_status < 0, - "a partially-failed array submission must report the construction " - "failure to the client"); - - TEST_ASSERT_TRUE_MESSAGE(elem0_exists, - "the non-colliding array element must still be constructed and " - "present in c->jobmgr despite its sibling failing to construct"); - TEST_ASSERT_TRUE_MESSAGE(elem0_notified, - "the successfully constructed sibling must fire the AFTER_EVENT/" - "JOB_NEW notification hook even though array element 9201 failed " - "to construct post-commit (raft-apply-decoupling task 3.1 " - "remediation) — otherwise it is durably committed and constructed " - "but invisible to the purely event-driven scheduler/apiserver " - "until the next controller restart replays the WAL"); - - printf("test_jobsubmit_array_partial_construct_failure_notifies_survivors:" + printf("test_jobsubmit_returns_before_commit_and_registers_pending_apply:" " PASS\n"); } +/* raft-apply-decoupling task 3.2: the failure/timeout path must ALSO prove + * the handler returns immediately — cbrp_drive_withheld (used by + * test_jobsubmit_withheld_acks_no_phantom_job below) only proves the + * EVENTUAL reply/no-phantom-job observable, not that the calling thread + * stopped blocking well before the commit timeout elapses. */ +static void test_jobsubmit_returns_before_commit_on_failure_path_too(void) +{ + uint64_t lsn_before, watermark, t0, join_elapsed_ms; + cbrp_reply_obs_t obs = { 0 }; + bool job_exists; + + cbrp_setup(1000); + cbrp_arrange_jobsubmit(9104, 1); + + pthread_mutex_lock(&g_hc.persist.lock); + lsn_before = g_hc.persist.current_lsn; + pthread_mutex_unlock(&g_hc.persist.lock); + + g_op_arg.fn = handle_jobsubmit; + g_op_arg.req = &g_jobsub_req; + g_op_arg.rc = -9999; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, + cbrp_op_thread_fn, &g_op_arg)); + + /* No ACK is EVER delivered in this test. The 1000ms sync_timeout_ms is + * how long the ENTRY takes to resolve as FAILED — the HANDLER itself + * must return immediately, well before that timeout. */ + t0 = cbrp_mono_ms(); + TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); + join_elapsed_ms = cbrp_mono_ms() - t0; + + watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 500); + + /* Let the timeout sweep resolve the entry FAILED and confirm the + * (eventual) failure reply + no-phantom-job observable — same + * contract test_jobsubmit_withheld_acks_no_phantom_job already pins, + * verified here alongside the non-blocking-return proof. */ + cbrp_poll_reply(&g_req_ep, 1000 + 1500, &obs); + job_exists = cbrp_job_exists(9104); + + cbrp_teardown(); + + TEST_ASSERT_TRUE_MESSAGE(join_elapsed_ms < 500, + "handle_jobsubmit must return immediately after registering its " + "continuation, without waiting for the commit timeout to elapse " + "(raft-apply-decoupling task 3.2, Req 1.1/1.4)"); + TEST_ASSERT_TRUE_MESSAGE(watermark > 0, + "job submission must still append its WAL entry before the " + "handler returns"); + TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0, + "a forced timeout must still eventually reply to the client"); + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status, + "a timed-out commit must reply with -ETIMEDOUT, the same " + "vocabulary the synchronous path already used (Req 2.3)"); + TEST_ASSERT_FALSE_MESSAGE(job_exists, + "no job may ever be constructed for an entry whose commit timed " + "out (raft-apply-decoupling task 3.2, Req 2.3)"); + + printf("test_jobsubmit_returns_before_commit_on_failure_path_too: PASS\n"); +} + +/* REMOVED (raft-apply-decoupling task 3.2): the array-partial-construction- + * failure remediation test (task 3.1 round-2 fix, bug-1408) that lived here + * drove handle_jobsubmit's OLD sequential, per-element BLOCKING append loop + * — it ACKed array element 0 and only then observed element 1's WAL entry + * appear, exploiting that strict ordering to inject a deterministic jobid + * collision between the two elements' commits. Task 3.2 appends every array + * element back-to-back, non-blocking, with no per-element ACK gate between + * them, so that ordering assumption no longer holds — the test's whole + * premise is gone, not just its expected outcome. + * + * More importantly, the CODE this test exercised (handle_jobsubmit's own + * post-commit create_job()-loop + jlist-build + xce_run_hooks(AFTER_EVENT) + * block) no longer exists in job.c at all (see the "REMOVED" comment above + * handle_jobsubmit's async registration block) — it is superseded, not + * relocated, by apply_committed_entry()'s existing WAL_JOB_SUBMIT case + * (persist.c, task 2.2), which already does restore_job() + fires + * HOOK_DETAIL_JOB_NEW per entry, independently per array element, and is + * therefore structurally immune to the bug-1408 class already. Coverage for + * that invariant now belongs to persist.c's own apply-seam test suite + * (test_apply_seam.c), not here. See this task's status report for the + * full reasoning; git history (commit 6f1fa97) preserves the removed test + * verbatim. */ + /* ════════════════════════════════════════════════════════════════════════ * Receipt-ACK verdict pins (ha-commit-barrier-rollout task 3.3, Req 2.3) * @@ -2133,8 +2122,8 @@ void test_commit_barrier_reply(void) RUN_TEST(test_worker_join_no_hostconfig_error_reply_immediate); /* handle_jobsubmit (raft-apply-decoupling task 3.1, Req 2.1/2.2/2.3) */ RUN_TEST(test_jobsubmit_withheld_acks_no_phantom_job); - RUN_TEST(test_jobsubmit_success_reply_and_job_visible_after_commit); - RUN_TEST(test_jobsubmit_array_partial_construct_failure_notifies_survivors); + RUN_TEST(test_jobsubmit_returns_before_commit_and_registers_pending_apply); + RUN_TEST(test_jobsubmit_returns_before_commit_on_failure_path_too); /* 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); diff --git a/src/include/persist.h b/src/include/persist.h index b64b82c..a8f8487 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -291,6 +291,17 @@ struct persist_s { atomic_int checkpoint_in_progress; /* CAS guard: 0=idle, 1=scheduled/running */ atomic_int fsync_stop; atomic_int fsync_thread_running; + + /* raft-apply-decoupling task 3.2: pending command-apply hand-off list + * (producer side; the not-yet-built ordered apply pump, task 6.1/6.2, + * is the consumer). Opaque here (pending_command_apply_t is private to + * persist.c) — guarded by its OWN dedicated lock, deliberately NOT + * `lock` above, since entries are appended from a commit continuation + * callback (dispatched off-thread via vtask_enq_classified) that must + * never contend with the hot WAL-append path. See + * persist_register_pending_apply's doc comment for the full contract. */ + void* pending_applies; + pthread_mutex_t pending_applies_lock; #ifdef XCE_TESTING /* Fault-injection seam (ha-commit-durability task 2.1 error-path test): * when non-zero, every fdatasync site treats the sync as FAILED — it @@ -732,6 +743,91 @@ int persist_submit_wal_and_commit(persist_t* p, xlist_t* jlist); int persist_submit_job_payloads_and_commit(persist_t* p, xlist_t* payloads, uint64_t* out_lsns); +/* raft-apply-decoupling task 3.2: outcome delivered to a job submission's + * per-entry commit continuation once its commit_submit_async() registration + * resolves (or, for the LSN-0 / engine-not-wired edge cases, immediately + * and synchronously). rc == 0 means the entry committed on a majority; + * rc < 0 carries the failure/timeout error code, the same vocabulary + * commit_submit's WAIT mode already uses (-ETIMEDOUT, -XCE_ERR_STANDBY, + * -ENOMEM, etc.). `lsn` is 0 for an entry that never reached the point of + * being assigned a real LSN (e.g. a local encode/append failure) — 0 is + * the reserved "no real entry" sentinel already used throughout persist.h. + * + * Declared with a single plain rc — NOT commit.h's `commit_outcome_t` + + * `int rc` pair — so this header never needs to depend on commit.h, which + * already depends on this one (commit_engine_t embeds a `persist_t*`). + * persist.c's implementation adapts commit_submit_async's real callback + * shape to this one internally (see job_commit_resolve_adapter). */ +typedef void (*persist_job_commit_notify_fn)(void* ctx, uint64_t lsn, int rc); + +/* Async sibling of persist_submit_job_payloads_and_commit (raft-apply- + * decoupling task 3.2): WAL-appends each payload the same way (encode + + * local append; out_lsns[i] set on a successful append) but never blocks + * on the majority-commit barrier. Instead, `fn`/`ctx` is invoked EXACTLY + * ONCE per payload in `payloads` — either synchronously, right here (a + * local append failure, the once-per-cluster-lifetime LSN-0 edge, an + * engine-not-wired context, or a commit_submit_async registration + * failure), or later, asynchronously, off this call's stack, once that + * payload's entry's commit outcome is known. + * + * On the FIRST local failure (encode/append), every REMAINING payload in + * the batch also resolves immediately with that SAME failure rc — so the + * caller always receives exactly `payloads->count` total `fn` invocations, + * regardless of where (or whether) a failure occurs; the caller never + * needs to separately account for "never going to be called" entries. + * + * Returns 0 once every payload is durably appended locally and every + * payload has an outcome resolution path established (sync or async); the + * first negative rc encountered otherwise. Every payload's `fn` invocation + * has already happened (or is already guaranteed to happen later) by the + * time this function returns, regardless of its return value — the return + * value itself carries no information the caller still needs to act on. + * + * out_lsns[i] is valid for every i whose local append succeeded (same + * contract as the synchronous sibling); it is left unset (0, if the caller + * zero-initialized it) for a payload whose local append itself failed. */ +int persist_submit_job_payloads_async(persist_t* p, xlist_t* payloads, + uint64_t* out_lsns, + persist_job_commit_notify_fn fn, + void* ctx); + +/* raft-apply-decoupling task 3.2: producer-side hand-off for a command + * whose commit_submit_async() continuation resolved COMMITTED. Per task + * 3.2's own contract, a COMMITTED outcome must NOT apply state or reply to + * the client directly from the commit continuation — both must happen + * from the not-yet-built ordered apply pump (task 6.1/6.2), strictly in + * commit order relative to every other entry (command or fact). + * + * persist_register_pending_apply() is the producer: it records that once + * `lsn` is applied (via apply_committed_entry(), the same seam restart + * replay already uses), `on_applied(ctx, lsn)` must be invoked immediately + * afterward — task 6.2's own stated contract: "fire a command's registered + * continuation immediately after its entry applies". Until task 6 exists, + * registered entries accumulate here inertly: this is the SAME "durably + * committed, not yet applied" posture already accepted for retained facts + * during quorum loss (Req 6.1), not a leak — task 6.1 is expected to drain + * and free each node as it walks committed LSNs. + * + * NOTE for whoever builds task 6.1: apply_committed_entry()'s WAL_JOB_SUBMIT + * case (persist.c) already constructs the job (via restore_job()) AND fires + * the HOOK_DETAIL_JOB_NEW AFTER_EVENT notification, per entry, independently + * of any sibling array element — calling apply_committed_entry() for this + * lsn is very likely ALL the "apply" work `on_applied`'s caller needs to do + * before invoking `on_applied`; `on_applied` itself should very likely only + * need to send the client's success reply, not reconstruct or re-notify + * anything (see job.c's jobsubmit_apply_complete for the concrete case). */ +typedef void (*pending_apply_notify_fn)(void* ctx, uint64_t lsn); + +void persist_register_pending_apply(persist_t* p, uint64_t lsn, + pending_apply_notify_fn on_applied, + void* ctx); + +/* Non-destructive lookup: 1 if `lsn` has a registered, not-yet-consumed + * pending-apply hand-off, 0 otherwise. Test/introspection helper only — + * task 6.1's real consumer will need its own draining accessor (e.g. "take + * and remove"), not built by this task. */ +int persist_has_pending_apply(persist_t* p, uint64_t lsn); + int persist_job_hook_fun(int type, int situation, int detail, void* cls, va_list ap); int persist_queue_hook_fun(int type, int situation, int detail,