diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 709e006..c096a61 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -16,7 +16,7 @@ - _Requirements: 1.3, 9.3, 9.4_ - _Note: real timeout/demotion/truncation **detection** (as opposed to the generic resolve primitive) is 1.3/1.4 — see Implementation Notes_ -- [ ] 1.3 Arm a timeout for registered async continuations +- [x] 1.3 Arm a timeout for registered async continuations - `commit_submit_async()` no longer parks a thread that would otherwise notice its own `pthread_cond_timedwait` deadline expire — converting away from the blocking wait removes the ONLY mechanism that ever detected a commit timing out. Add an explicit timeout (e.g., a delaymgr-driven check, or a deadline recorded at registration and swept periodically) that resolves a still-pending registered continuation as FAILED with `-ETIMEDOUT` if majority is not reached within the caller-supplied timeout - Preserve the existing default timeout value and error code so callers observe identical failure timing/vocabulary to today's blocking path — this is a "no regression" requirement (11.1), not just a new capability - Observable: an async-registered continuation for an LSN that never reaches majority is resolved as FAILED with `-ETIMEDOUT` within the configured timeout window, with no thread blocked to detect it, verified by a test that lets real time (or a mocked clock) elapse past the deadline without ever calling `commit_resolve()` manually @@ -191,3 +191,4 @@ - **1.1**: `commit_submit_async()` reuses the exact `wait_admit_locked`/`wait_release_locked`/`inflight_waits` admission-cap discipline as `COMMIT_MODE_WAIT` — task 1.2's `commit_resolve()` MUST call `wait_release_locked()` when resolving an async-registered entry, or the shared admission cap leaks. Continuation registry is `calloc(n_vcpus, ...)`, sized so `n_continuations >= wait_cap(ce)` always holds; `lsn == 0` is the free-slot sentinel. - **1.1 review**: bare `controller_unit_test` (no `--run` filter) segfaults deterministically in this environment (SSL_CTX_free/iomgr_fini crash, `test_ha_vep_teardown.c`) — reproduced identically on the pre-task base commit, confirmed pre-existing and unrelated. Logged as bug-1406. Also `test_commit_advance_retry_clears_push_before_commit_fence` (`test_config_repl.c`) fails independently of this feature, also reproduced on base — logged as bug-1407, flagged because it's adjacent to `commit_advance_locked`, which task 1.2 modifies. **Always validate with a filtered `--run ` invocation or the tightly-coupled commit-engine cluster, not the bare unfiltered binary, until 1406/1407 are separately fixed.** - **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. diff --git a/src/controller/commit.c b/src/controller/commit.c index 1ea75e0..915affa 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -26,6 +26,29 @@ * §Commit Engine). n_peers <= 1 self-quorum advances committed_lsn * through the single advancement path and completes immediately. * + * raft-apply-decoupling task 1.3: commit_submit_async() registers a + * continuation instead of parking a caller thread, which removes the ONLY + * mechanism (the caller's own pthread_cond_timedwait deadline) that ever + * detected a commit timing out. commit_engine_t therefore owns a dedicated + * timeout-sweep thread (commit_sweep_worker, started in commit_engine_init / + * stopped+joined in commit_engine_fini) that sleeps on the SAME ce->cond + * used everywhere else in this file, waking either on a broadcast (new + * registration, majority advance, membership change, gate change) or at the + * earliest pending continuation's deadline, whichever comes first. + * + * WHY A DEDICATED THREAD, NOT THE SHARED CONTROLLER delaymgr: this task's + * boundary is commit.c/commit.h only (no cluster.c/ha.c wiring changes), and + * the shared delaymgr's production polling precision (200ms, cluster.c) is + * coarse relative to DEFAULT_SYNC_TIMEOUT_MS (500ms) — routing through it + * would measurably loosen the "identical failure timing" bar (Req 11.1). + * Reusing ce->lock/ce->cond and the exact wait_deadline() computation + * commit_submit's blocking path already uses gives the async path the same + * precision as the blocking path it replaces, with a single thread + * regardless of how many continuations are pending (unlike the old + * one-parked-thread-per-waiter model). This thread is private to the commit + * engine and never touches iomgr or delaymgr, so it does not interact with + * either thread's non-blocking contract (.kiro/steering/ha-async-io.md). + * * Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 4.3, 4.4, 5.1, 6.3, 7.1, 7.2, 7.3 */ #include @@ -50,6 +73,12 @@ * that section for the full definition and contract. */ static void commit_resolve_pending(commit_engine_t* ce); +/* Forward declaration: raft-apply-decoupling task 1.3. commit_engine_init + * (below) starts the timeout-sweep thread; its entry function is defined in + * the Timeout Sweep section further down this file (alongside + * commit_sweep_scan_locked, the locked-scan helper it drives). */ +static void* commit_sweep_worker(void* arg); + /* ============================================================ * Internal helpers * ============================================================ */ @@ -267,6 +296,21 @@ static void wait_deadline(struct timespec* dl, uint32_t timeout_ms) } } +/* raft-apply-decoupling task 1.3: three-way compare of two CLOCK_REALTIME + * timespecs (the same clock wait_deadline() uses). Used by the timeout-sweep + * thread to detect an expired deadline and to track the earliest of several + * pending ones. */ +static int ts_cmp(const struct timespec* a, const struct timespec* b) +{ + if (a->tv_sec != b->tv_sec) { + return (a->tv_sec < b->tv_sec) ? -1 : 1; + } + if (a->tv_nsec != b->tv_nsec) { + return (a->tv_nsec < b->tv_nsec) ? -1 : 1; + } + return 0; +} + /* Durable-advance wake (ha-commit-durability task 2.2): invoked by the persist * group-commit flusher AFTER it advances durable_lsn and AFTER it releases * p->lock. Now that the commit majority caps self at durable_lsn, an advance of @@ -342,6 +386,30 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp * we only register a function + ctx, preserving the commit→persist (never * the reverse) dependency direction. */ persist_set_durable_cb(p, commit_on_durable_advance, ce); + + /* raft-apply-decoupling task 1.3: start the dedicated timeout-sweep + * thread (see file header for why this is a private thread rather than + * the shared controller delaymgr). A creation failure is treated as a + * hard init failure -- like the calloc failure above -- because a + * commit engine that cannot detect async timeouts cannot honor its + * documented contract (Req 9.4: every registered continuation must + * eventually resolve). */ + ce->sweep_stop = 0; + ce->sweep_thread = 0; /* explicit sentinel: pthread_create leaves this + * unspecified on failure (POSIX), and ce may be + * caller-reused memory from a prior lifecycle. */ + { + int prc = pthread_create(&ce->sweep_thread, NULL, commit_sweep_worker, ce); + if (prc != 0) { + persist_set_durable_cb(p, NULL, NULL); + free(ce->continuations); + ce->continuations = NULL; + ce->n_continuations = 0; + pthread_cond_destroy(&ce->cond); + pthread_mutex_destroy(&ce->lock); + return -prc; + } + } return 0; } @@ -364,6 +432,20 @@ void commit_engine_fini(commit_engine_t* ce) if (ce == NULL) { return; } + /* raft-apply-decoupling task 1.3: stop + join the timeout-sweep thread + * BEFORE tearing down lock/cond/continuations -- it reads all three. + * Signal under the lock and broadcast to wake it from either + * pthread_cond_wait (no pending continuations) or pthread_cond_timedwait + * (a pending deadline); the join guarantees it has fully exited before + * anything it touches is freed or destroyed. */ + if (ce->sweep_thread != 0) { + pthread_mutex_lock(&ce->lock); + ce->sweep_stop = 1; + pthread_cond_broadcast(&ce->cond); + pthread_mutex_unlock(&ce->lock); + pthread_join(ce->sweep_thread, NULL); + ce->sweep_thread = 0; + } /* Unregister the durable-advance wake BEFORE tearing down the mutex/cond: * otherwise the flusher could fire commit_on_durable_advance against a * destroyed engine (use-after-free / ASAN). Callers are expected to have @@ -530,7 +612,18 @@ int commit_submit_async(commit_engine_t* ce, uint64_t lsn, ce->continuations[slot].lsn = lsn; ce->continuations[slot].fn = fn; ce->continuations[slot].ctx = ctx; + /* raft-apply-decoupling task 1.3: arm this registration's timeout with + * the SAME deadline computation and SAME default constant + * commit_submit's blocking WAIT path uses, so an async registration + * observes identical failure timing to today's blocking path (Req + * 11.1). The dedicated timeout-sweep thread (commit_sweep_worker) + * resolves this continuation FAILED/-ETIMEDOUT if it is still pending + * once this deadline passes. */ + wait_deadline(&ce->continuations[slot].deadline, DEFAULT_SYNC_TIMEOUT_MS); + /* Wake the sweep thread: a freshly armed deadline may be sooner than + * whatever it is currently sleeping toward. */ + pthread_cond_broadcast(&ce->cond); pthread_mutex_unlock(&ce->lock); return 0; } @@ -708,6 +801,130 @@ static void commit_resolve_pending(commit_engine_t* ce) } } +/* ============================================================ + * Timeout sweep (raft-apply-decoupling task 1.3) + * ============================================================ */ + +/* One bounded, read-only classification pass over the continuation + * registry: for every still-registered lsn, majority-satisfied ALWAYS wins + * over an expired deadline when both are true in this same lock hold -- + * mirroring commit_submit's own last-look-before-timeout tie-break (a + * majority ack landing exactly at the deadline is never spuriously reported + * as a timeout). Populates committed_ready[] and failed_ready[] (via nc and + * nf) and reports the earliest deadline among whatever remains pending via + * next_deadline and have_next. Caller MUST hold ce->lock and does its own + * unlocking -- this function never blocks and never dispatches, so it is + * safe to call as the last thing done before committing to a + * pthread_cond_wait/pthread_cond_timedwait in the SAME lock hold (see + * commit_sweep_worker). */ +static void commit_sweep_scan_locked(commit_engine_t* ce, + uint64_t* committed_ready, int* nc, + uint64_t* failed_ready, int* nf, + struct timespec* next_deadline, + int* have_next) +{ + struct timespec now; + + clock_gettime(CLOCK_REALTIME, &now); /* same clock as wait_deadline() */ + *nc = 0; + *nf = 0; + *have_next = 0; + + for (int i = 0; i < ce->n_continuations; i++) { + uint64_t lsn = ce->continuations[i].lsn; + + if (lsn == 0) { + continue; + } + if (wait_done_locked(ce, lsn)) { + committed_ready[(*nc)++] = lsn; + } else if (ts_cmp(&now, &ce->continuations[i].deadline) >= 0) { + failed_ready[(*nf)++] = lsn; + } else if (!*have_next || + ts_cmp(&ce->continuations[i].deadline, next_deadline) < 0) { + *next_deadline = ce->continuations[i].deadline; + *have_next = 1; + } + } +} + +/* Dedicated timeout-sweep thread, private to this commit_engine_t (see the + * file header for why this is not routed through the shared controller + * delaymgr). Sleeps on ce->cond to the earliest pending continuation + * deadline, or indefinitely when none are pending; a broadcast (new + * registration, majority advance, membership change, gate change, or + * commit_engine_fini's shutdown signal) wakes it early to re-evaluate. + * Exits once ce->sweep_stop is observed under ce->lock. + * + * LOST-WAKEUP FIX: commit_sweep_scan_locked() only classifies -- it never + * dispatches -- so every dispatch (commit_resolve(), which must run + * UNLOCKED per the lock-ordering rule) happens between an unlock and the + * next lock in this loop. A commit_submit_async() registration (or any + * other broadcaster; all of them broadcast while HOLDING ce->lock) landing + * during that unlocked window would find no thread parked in cond_wait to + * receive it -- POSIX condvars do not latch a signal sent to no waiter. + * The fix follows the standard "recheck the wait condition in the SAME + * lock hold that leads directly into the wait call" idiom: whenever this + * loop has something to dispatch, it unlocks, dispatches, RE-LOCKS, and + * loops back to scan again -- it never commits to a wait based on a + * have_next/next_deadline computed before an unlocked dispatch step. Only + * a scan that found NOTHING to dispatch (so no unlock is needed before the + * wait) is trusted to size the wait. */ +static void* commit_sweep_worker(void* arg) +{ + commit_engine_t* ce = (commit_engine_t*)arg; + + pthread_mutex_lock(&ce->lock); + for (;;) { + uint64_t committed_ready[ce->n_continuations]; /* VLA: bounded + * by n_vcpus, + * see the + * commit.h + * sizing + * invariant */ + uint64_t failed_ready[ce->n_continuations]; + int nc, nf, have_next; + struct timespec next_deadline; + + if (ce->sweep_stop) { + break; + } + + commit_sweep_scan_locked(ce, committed_ready, &nc, failed_ready, &nf, + &next_deadline, &have_next); + + if (nc > 0 || nf > 0) { + pthread_mutex_unlock(&ce->lock); + for (int i = 0; i < nc; i++) { + commit_resolve(ce, committed_ready[i], COMMIT_OUTCOME_COMMITTED, 0); + } + for (int i = 0; i < nf; i++) { + WARN("commit: async continuation deadline expired without " + "commit (lsn=%lu committed_lsn=%lu writable_lsn=%lu " + "timeout_ms=%u)", + failed_ready[i], atomic_load(&ce->p->committed_lsn), + ce->writable_lsn, (unsigned)DEFAULT_SYNC_TIMEOUT_MS); + commit_resolve(ce, failed_ready[i], COMMIT_OUTCOME_FAILED, + -ETIMEDOUT); + } + pthread_mutex_lock(&ce->lock); + continue; /* re-scan under the lock before ever waiting */ + } + + /* Nothing to dispatch: this scan just ran in the exact lock hold we + * are about to wait under, so no broadcast between "now" and the + * wait call below can be missed. */ + if (have_next) { + pthread_cond_timedwait(&ce->cond, &ce->lock, &next_deadline); + } else { + pthread_cond_wait(&ce->cond, &ce->lock); + } + } + pthread_mutex_unlock(&ce->lock); + + return NULL; +} + /* ============================================================ * ACK and membership events * ============================================================ */ diff --git a/src/controller/include/commit.h b/src/controller/include/commit.h index ee3bdf9..d98b1df 100644 --- a/src/controller/include/commit.h +++ b/src/controller/include/commit.h @@ -10,6 +10,7 @@ */ #include #include +#include /* struct timespec — per-continuation timeout deadline (task 1.3) */ #include "persist.h" /* persist_t (owns committed_lsn atomic + replicator_t registry) */ #include "vtaskmgr.h" /* vtaskmgr_t — LivenessAdmission blocking-admission seam */ @@ -47,6 +48,15 @@ typedef struct commit_continuation { uint64_t lsn; commit_continuation_fn fn; void* ctx; + /* raft-apply-decoupling task 1.3: absolute CLOCK_REALTIME deadline this + * registration must resolve by if majority is never reached -- computed + * at registration time via the SAME wait_deadline() helper and the SAME + * DEFAULT_SYNC_TIMEOUT_MS constant commit_submit's blocking WAIT path + * uses, so an async registration observes identical failure timing to + * today's blocking path (Req 11.1). Only meaningful while lsn != 0; + * overwritten on every fresh registration of a reused slot. Swept by the + * commit engine's dedicated timeout-sweep thread (see commit.c). */ + struct timespec deadline; } commit_continuation_t; /* @@ -90,6 +100,17 @@ typedef struct commit_engine { * inert -- nothing reads it back. */ commit_continuation_t* continuations; int n_continuations; /* capacity, == n_vcpus */ + /* raft-apply-decoupling task 1.3: dedicated timeout-sweep thread, private + * to this commit_engine_t (NOT the shared controller delaymgr/iomgr -- + * see commit.c file header for why). Sleeps on `cond` to the earliest + * pending continuation deadline (or indefinitely when none are pending) + * and resolves any still-registered continuation FAILED/-ETIMEDOUT once + * its deadline passes. Started in commit_engine_init, stopped+joined in + * commit_engine_fini before `lock`/`cond`/`continuations` are torn down. + * sweep_stop is read/written only under `lock`, alongside the broadcast + * that wakes the thread to observe it. */ + pthread_t sweep_thread; + int sweep_stop; } commit_engine_t; /* Lifecycle. init validates n_peers >= 0, n_vcpus >= 1, non-NULL persist_t; @@ -121,9 +142,18 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, * -XCE_ERR_NOT_READY when the admission cap is exhausted (the same shared * discipline commit_submit's WAIT mode uses -- see wait_admit_locked). * Never blocks: the calling thread returns immediately, before lsn's - * outcome is known. Registering here does not yet cause `fn` to be - * invoked -- delivery is commit_resolve's job, added by a follow-up task - * that wires it onto the majority-commit and failure paths. */ + * outcome is known. + * + * raft-apply-decoupling task 1.3: registration also arms an implicit + * DEFAULT_SYNC_TIMEOUT_MS deadline (identical constant and clock the + * blocking WAIT path uses) -- if lsn has not resolved by then, the commit + * engine's own timeout-sweep thread resolves it FAILED/-ETIMEDOUT with no + * caller action required. A genuine majority ack that lands at or after the + * deadline still wins: the sweep thread always checks the majority + * predicate before the deadline in the same lock hold, mirroring + * commit_submit's own last-look-before-timeout tie-break. There is no + * per-call timeout parameter (unlike commit_submit) -- see design.md + * §Commit Engine Resolve for the fixed signature. */ int commit_submit_async(commit_engine_t* ce, uint64_t lsn, commit_continuation_fn fn, void* ctx); diff --git a/src/controller/testsuite/test_commit_resolve.c b/src/controller/testsuite/test_commit_resolve.c index 314e376..89d2127 100644 --- a/src/controller/testsuite/test_commit_resolve.c +++ b/src/controller/testsuite/test_commit_resolve.c @@ -52,6 +52,21 @@ * caller-supplied outcome/rc faithfully, exactly once, off-thread. * * Requirements (task 1.2): 1.3, 9.3, 9.4 + * + * Timeout-sweep half (task 1.3): commit_submit_async() no longer parks a + * thread that would otherwise notice its own pthread_cond_timedwait deadline + * expire, so a dedicated timeout-sweep thread (private to commit_engine_t, + * started in commit_engine_init / stopped+joined in commit_engine_fini) is + * the only remaining detector of "this registered lsn has been pending too + * long". The cases below prove: (a) a registered continuation for an lsn + * that never reaches majority and is NEVER manually resolved still resolves + * FAILED/-ETIMEDOUT on its own, near the existing DEFAULT_SYNC_TIMEOUT_MS + * default (Req 11.1 -- identical failure timing/vocabulary to today's + * blocking path); and (b) the majority-satisfied outcome always wins a race + * against an already-expired deadline, mirroring commit_submit's own + * last-look-before-timeout tie-break. + * + * Requirements (task 1.3): 1.3, 1.4, 9.3, 9.4 */ #include #include @@ -165,6 +180,11 @@ typedef struct { int last_rc; void* last_ctx; pthread_t last_thread; + /* raft-apply-decoupling task 1.3 lost-wakeup regression: when nonzero, + * recorder_fn sleeps this long BEFORE recording. Zero (the default + * zero-initializer every existing test uses) preserves today's + * immediate-recording behavior. */ + unsigned sleep_us; } continuation_recorder_t; static void recorder_fn(void* ctx, uint64_t lsn, commit_outcome_t outcome, @@ -172,6 +192,9 @@ static void recorder_fn(void* ctx, uint64_t lsn, commit_outcome_t outcome, { continuation_recorder_t* r = (continuation_recorder_t*)ctx; + if (r->sleep_us > 0) { + usleep(r->sleep_us); + } r->last_lsn = lsn; r->last_outcome = outcome; r->last_rc = rc; @@ -589,6 +612,189 @@ static void test_commit_resolve_falls_back_to_inline_without_pool(void) printf("test_commit_resolve_falls_back_to_inline_without_pool: PASS\n"); } +/* ════════════════════════════════════════════════════════════════════════ + * Task 1.3 — timeout-sweep: async continuations resolve FAILED/-ETIMEDOUT + * on their own, with no manual commit_resolve() call. + * ════════════════════════════════════════════════════════════════════════ */ + +/* ------------------------------------------------------------------ * + * (11) A registered continuation for an lsn that never reaches * + * majority resolves FAILED/-ETIMEDOUT on its own, near the * + * existing DEFAULT_SYNC_TIMEOUT_MS default -- no manual * + * commit_resolve() call, no thread parked to detect it. * + * ------------------------------------------------------------------ */ +static void test_commit_submit_async_resolves_failed_on_timeout(void) +{ + continuation_recorder_t rec = {0}; + uint64_t t0; + uint64_t elapsed; + + /* n_peers=3, no standby ACKs ever delivered: lsn=1 can never commit on + * its own. Nothing in this test ever calls commit_resolve() -- the ONLY + * thing that can resolve this continuation is the timeout-sweep + * mechanism built by this task. */ + resolve_pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, + /*n_vcpus=*/4); + + t0 = mono_ms_r(); + TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec)); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, poll_calls(&rec.calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 1500), + "a registered continuation that never reaches majority must resolve " + "FAILED/-ETIMEDOUT on its own within the configured timeout window"); + elapsed = mono_ms_r() - t0; + + TEST_ASSERT_EQUAL_UINT64(1, rec.last_lsn); + TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_FAILED, rec.last_outcome); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rec.last_rc); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&rec.calls)); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, g_ce.inflight_waits, + "the timeout resolve must release the continuation's admission slot"); + + /* Req 11.1 -- identical failure timing to today's blocking path: + * generous bounds absorb scheduler jitter and the sweep thread's own + * bounded scan overhead without masking a mistimed implementation. */ + TEST_ASSERT_TRUE_MESSAGE( + elapsed + 50 >= DEFAULT_SYNC_TIMEOUT_MS, + "must not resolve meaningfully before the configured default " + "timeout"); + TEST_ASSERT_TRUE_MESSAGE( + elapsed < DEFAULT_SYNC_TIMEOUT_MS + 1000, + "must resolve within the configured timeout window, not drastically " + "late"); + + resolve_pool_teardown(); + printf("test_commit_submit_async_resolves_failed_on_timeout: PASS\n"); +} + +/* ------------------------------------------------------------------ * + * (12) Race: majority-satisfied always wins over an already-expired * + * deadline observed in the SAME sweep pass -- mirrors * + * commit_submit's own last-look-before-timeout tie-break. White- * + * box: force the deadline into the past and satisfy the majority * + * predicate directly (bypassing commit_on_ack / the already- * + * tested commit_resolve_pending ack path) so ONLY the timeout- * + * sweep thread's own scan discovers both conditions at once. * + * ------------------------------------------------------------------ */ +static void test_commit_submit_async_majority_wins_race_at_expired_deadline(void) +{ + continuation_recorder_t rec = {0}; + int found = 0; + + resolve_pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, + /*n_vcpus=*/4); + TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec)); + + pthread_mutex_lock(&g_ce.lock); + for (int i = 0; i < g_ce.n_continuations; i++) { + if (g_ce.continuations[i].lsn == 1) { + g_ce.continuations[i].deadline.tv_sec = 0; + g_ce.continuations[i].deadline.tv_nsec = 0; + found = 1; + } + } + /* Satisfy wait_done_locked directly -- no ack path involved. */ + atomic_store(&g_p.committed_lsn, 1); + pthread_cond_broadcast(&g_ce.cond); /* wake the sweep thread promptly */ + pthread_mutex_unlock(&g_ce.lock); + TEST_ASSERT_TRUE_MESSAGE(found, "lsn=1 registration must be tracked"); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, poll_calls(&rec.calls, 1, 2000), + "the sweep thread must independently notice and resolve this lsn"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + COMMIT_OUTCOME_COMMITTED, rec.last_outcome, + "majority-satisfied must win over an already-expired deadline " + "observed in the same scan"); + TEST_ASSERT_EQUAL_INT(0, rec.last_rc); + TEST_ASSERT_EQUAL_INT(1, atomic_load(&rec.calls)); + + resolve_pool_teardown(); + printf("test_commit_submit_async_majority_wins_race_at_expired_deadline: " + "PASS\n"); +} + +/* ------------------------------------------------------------------ * + * (13) Lost-wakeup regression: a registration landing WHILE the sweep * + * thread is mid-dispatch for a previous continuation must not be * + * missed. commit_sweep_scan_locked()'s classification runs under * + * ce->lock, but the actual commit_resolve() dispatch for whatever * + * it found must run UNLOCKED (lock-ordering rule) -- a * + * commit_submit_async() broadcast landing in that unlocked window * + * finds no thread parked in cond_wait to receive it (POSIX * + * condvars do not latch a signal sent to no waiter). If the sweep * + * worker then trusted the have_next/next_deadline it computed * + * BEFORE that dispatch, it would commit to an unbounded * + * pthread_cond_wait() (or a stale deadline) and never notice the * + * new registration until some UNRELATED future broadcast -- * + * unbounded delay instead of DEFAULT_SYNC_TIMEOUT_MS. * + * * + * Deterministic reproduction: use the no-pool inline-dispatch * + * fallback, whose continuation function runs SYNCHRONOUSLY on the * + * thread that called commit_resolve() -- the sweep thread itself, * + * for a timeout-triggered resolve. Continuation A's recorder * + * deliberately sleeps, pinning the sweep thread outside ce->lock * + * for a wide, fully-controlled window (this IS the dispatch * + * window the bug lived in, just stretched from microseconds to * + * ~200ms so a concurrent registration can land inside it with * + * certainty instead of chasing a race). Continuation B is * + * registered mid-sleep and must still resolve within * + * DEFAULT_SYNC_TIMEOUT_MS of ITS OWN registration, not stall on * + * whatever unrelated broadcast happens to come next. * + * ------------------------------------------------------------------ */ +static void test_commit_submit_async_registration_during_dispatch_not_lost(void) +{ + continuation_recorder_t recA = {0}; + continuation_recorder_t recB = {0}; + + /* Plain (no-pool) setup: dispatch_continuation's fail-loud inline + * fallback invokes the continuation function directly, unlocked, on + * the calling thread -- the sweep thread, here. */ + resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, + /*n_vcpus=*/8); + TEST_ASSERT_NULL(g_ce.pool); + + recA.sleep_us = 200 * 1000; /* 200ms: sweep thread is unlocked, "busy + * dispatching" A for this long */ + TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &recA)); + + /* Let A's DEFAULT_SYNC_TIMEOUT_MS deadline pass and the sweep thread + * begin dispatching it (enter recorder_fn's sleep). Comfortably past + * the deadline, comfortably before the 200ms sleep would elapse on its + * own. */ + usleep((DEFAULT_SYNC_TIMEOUT_MS + 100) * 1000); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 0, atomic_load(&recA.calls), + "sanity: A must still be inside its 200ms sleep, not yet recorded, " + "when B is registered below -- otherwise this test isn't actually " + "landing inside the dispatch window"); + + /* Register B while the sweep thread is (with certainty, given the + * 200ms window) unlocked and blocked inside recorder_fn(A)'s sleep -- + * exactly the window a real registration could race in production, + * just widened for a deterministic test. */ + TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 2, recorder_fn, &recB)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_calls(&recA.calls, 1, 2000), + "continuation A must still resolve"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, poll_calls(&recB.calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 2000), + "continuation B, registered while the sweep thread was mid-dispatch " + "for A, must still resolve within the configured timeout window -- " + "not delayed indefinitely waiting for some unrelated future " + "broadcast (the lost-wakeup regression)"); + TEST_ASSERT_EQUAL_UINT64(2, recB.last_lsn); + TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_FAILED, recB.last_outcome); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, recB.last_rc); + TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits); + + resolve_teardown(); + printf("test_commit_submit_async_registration_during_dispatch_not_lost: " + "PASS\n"); +} + void test_commit_resolve(void) { RUN_TEST(test_commit_submit_async_returns_immediately_before_commit); @@ -603,4 +809,7 @@ void test_commit_resolve(void) RUN_TEST(test_commit_resolve_second_resolve_is_noop); RUN_TEST(test_commit_resolve_unregistered_lsn_is_safe_noop); RUN_TEST(test_commit_resolve_falls_back_to_inline_without_pool); + RUN_TEST(test_commit_submit_async_resolves_failed_on_timeout); + RUN_TEST(test_commit_submit_async_majority_wins_race_at_expired_deadline); + RUN_TEST(test_commit_submit_async_registration_during_dispatch_not_lost); }