/* * test_commit_resolve.c — raft-apply-decoupling tasks 1.1 + 1.2. * * Keystone conversion, registration half (task 1.1): the commit engine gains * commit_submit_async(), a non-blocking sibling of * commit_submit(..., COMMIT_MODE_WAIT, ...) that registers a continuation * (callback + context) against an appended LSN instead of parking the * calling thread. The task-1.1 cases below cover ONLY the registration * mechanism — a registered continuation is never invoked in those tests, by * design (resolution did not exist yet). * * Asserts (task 1.1): * (1) commit_submit_async returns immediately (0), before the LSN * commits, and does NOT invoke the continuation inline. * (2) Bad-argument rejection: -EINVAL for NULL engine, LSN 0, NULL fn. * (3) The SAME admission-cap discipline commit_submit's WAIT mode uses * (wait_admit_locked / wait_cap / inflight_waits) governs * registration: cap exhaustion fast-rejects with -XCE_ERR_NOT_READY, * never blocks. * (4) That admission budget is SHARED with COMMIT_MODE_WAIT — a parked * blocking waiter and a pending async registration draw from the * same inflight_waits pool. * (5) Distinct concurrent registrations occupy distinct per-LSN slots. * * Requirements (task 1.1): 1.3, 1.4, 9.1, 9.2 * * Resolution half (task 1.2): commit_resolve() delivers a registered * continuation's outcome exactly once, dispatched off the calling thread via * vtask_enq_classified (VTASK_LIVE_BLOCK), never inline on the ACK/iomgr * path in the real (pool-wired) case. Covers the four Observable triggers * from tasks.md 1.2 -- {majority ack, timeout, demotion, truncation} -- plus * idempotency, unregistered-lsn safety, admission-slot release, and the * fail-loud inline fallback when no vtaskmgr pool is wired. * * On "timeout"/"demotion"/"truncation": commit_resolve() is a generic * outcome+rc delivery primitive (design §Commit Engine Resolve) -- it does * not itself detect any of these three conditions. Tracing the existing * codebase (persist.c truncate path, ha.c raft_cb_become_follower demotion * path, repl_wait_majority_ack's error vocabulary) found NO existing * distinct rc for demotion or truncation anywhere today: commit_submit's * ONLY negative "did not commit" outcome is -ETIMEDOUT, because the current * blocking-WAIT model doesn't distinguish these causes either -- a * demoted/truncated pending entry simply never satisfies its predicate and * dies at the caller's own deadline. Wiring an ACTUAL demotion/truncation * detector to call commit_resolve() belongs to later tasks that touch * ha.c/persist.c/repl.c (out of this task's commit.c-only boundary; see * task 3.2 in tasks.md). These tests therefore "force" each condition the * way a future detector will: by calling commit_resolve(FAILED, rc) * directly, using representative existing rc vocabulary (-ETIMEDOUT for * timeout, -XCE_ERR_STANDBY -- "op is primary-only" -- for demotion, -ENOENT * for a log-truncated-away entry) to prove commit_resolve delivers ANY * 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 #include #include #include #include #include #include #include "xce/errnox.h" #include "commit.h" #include "repl.h" #include "vtaskmgr.h" /* ── Fixture ────────────────────────────────────────────────────────────── * Static: persist_t embeds large path buffers, and statics survive Unity's * longjmp on assertion failure (pattern from test_commit_engine.c / * test_commit_admission_pool.c). */ static persist_t g_p; static commit_engine_t g_ce; static int g_eps[MAX_STANDBYS]; /* dummy endpoints — pointer-only */ /* Arrange a fresh engine over a stub persist_t + replicator_t, mirroring * test_commit_engine.c's engine_setup_vcpus. n_vcpus drives the local * admission cap: max(1, n_vcpus - COMMIT_WAIT_RESERVED). */ static void resolve_setup(int n_peers, int n_standbys, uint64_t appended_lsn, int n_vcpus) { char name[16]; memset(&g_p, 0, sizeof(g_p)); TEST_ASSERT_EQUAL_INT(0, repl_init(&g_p.repl)); for (int i = 0; i < n_standbys; i++) { snprintf(name, sizeof(name), "sb%d", i); TEST_ASSERT_EQUAL_INT( i, repl_add_standby(&g_p.repl, &g_eps[i], name, 7000 + (uint32_t)i)); } g_p.current_lsn = appended_lsn + 1; atomic_store(&g_p.committed_lsn, 0); atomic_store(&g_p.durable_lsn, appended_lsn); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, n_vcpus)); } static void resolve_teardown(void) { commit_engine_fini(&g_ce); repl_fini(&g_p.repl); } /* ── Pool-wired fixture (task 1.2) ─────────────────────────────────────────── * A real, RUNNING vtaskmgr_t wired via commit_set_pool, mirroring * test_commit_admission_pool.c's pool_setup but additionally calling * vtaskmgr_start so enqueued vtasks are actually picked up and executed — * required to observe commit_resolve's real (non-fallback) off-thread * dispatch, not just the admission-accounting seam. */ static vtaskmgr_t g_mgr; static void resolve_pool_setup(int n_peers, int n_standbys, uint64_t appended_lsn, int n_vcpus) { resolve_setup(n_peers, n_standbys, appended_lsn, n_vcpus); TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_mgr, n_vcpus)); TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_mgr)); commit_set_pool(&g_ce, &g_mgr); } static void resolve_pool_teardown(void) { commit_engine_fini(&g_ce); vtaskmgr_fini(&g_mgr); repl_fini(&g_p.repl); } static uint64_t mono_ms_r(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u; } /* Spin-poll (bounded) until *calls reaches want; avoids correctness-by-sleep * on the assertion path (pattern from test_vtask_admission.c wait_atomic_ge / * test_commit_engine.c poll_inflight). */ static int poll_calls(_Atomic int* calls, int want, uint32_t deadline_ms) { uint64_t t0 = mono_ms_r(); int v; for (;;) { v = atomic_load(calls); if (v >= want || mono_ms_r() - t0 >= deadline_ms) { return v; } usleep(500); } } /* Continuation recorder — proves (non-)invocation and captures the args a * commit_resolve() delivers, including the thread it ran on (task 1.2: proves * dispatch is off the calling thread, never inline on the real pool-wired * path). */ typedef struct { _Atomic int calls; uint64_t last_lsn; commit_outcome_t last_outcome; 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, int rc) { 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; r->last_ctx = ctx; r->last_thread = pthread_self(); atomic_fetch_add(&r->calls, 1); } /* ------------------------------------------------------------------ * * (1) Registration returns immediately, before the LSN commits, and * * never invokes the continuation inline. * * ------------------------------------------------------------------ */ static void test_commit_submit_async_returns_immediately_before_commit(void) { continuation_recorder_t rec = {0}; /* n_peers=3, no standby ACKs delivered: LSN 1 can never commit on its * own (majority unreachable), so a blocking WAIT would park here. */ resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, /*n_vcpus=*/8); int rc = commit_submit_async(&g_ce, /*lsn=*/1, recorder_fn, &rec); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, rc, "registration must succeed and return without blocking"); TEST_ASSERT_EQUAL_UINT64_MESSAGE( 0, atomic_load(&g_p.committed_lsn), "the LSN must still be uncommitted at the point of registration"); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, atomic_load(&rec.calls), "the continuation must NOT be invoked by registration alone"); resolve_teardown(); printf("test_commit_submit_async_returns_immediately_before_commit: " "PASS\n"); } /* ------------------------------------------------------------------ * * (2) Bad-argument rejection. * * ------------------------------------------------------------------ */ static void test_commit_submit_async_rejects_invalid_args(void) { continuation_recorder_t rec = {0}; resolve_setup(/*n_peers=*/1, /*n_standbys=*/0, /*appended_lsn=*/1, /*n_vcpus=*/8); TEST_ASSERT_EQUAL_INT(-EINVAL, commit_submit_async(NULL, 1, recorder_fn, &rec)); TEST_ASSERT_EQUAL_INT( -EINVAL, commit_submit_async(&g_ce, /*lsn=*/0, recorder_fn, &rec)); TEST_ASSERT_EQUAL_INT(-EINVAL, commit_submit_async(&g_ce, /*lsn=*/1, NULL, &rec)); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, atomic_load(&rec.calls), "no rejected registration may invoke the continuation"); resolve_teardown(); printf("test_commit_submit_async_rejects_invalid_args: PASS\n"); } /* ------------------------------------------------------------------ * * (3) Admission cap governs registration exactly as it governs WAIT: * * n_vcpus=2, COMMIT_WAIT_RESERVED=2 -> local cap = max(1,0) = 1. * * ------------------------------------------------------------------ */ static void test_commit_submit_async_respects_admission_cap(void) { continuation_recorder_t rec1 = {0}, rec2 = {0}; resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/2, /*n_vcpus=*/2); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec1)); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_ce.inflight_waits, "registration consumes an admission slot"); int rc = commit_submit_async(&g_ce, 2, recorder_fn, &rec2); TEST_ASSERT_EQUAL_INT_MESSAGE( -XCE_ERR_NOT_READY, rc, "over-cap registration must fast-reject, never block"); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, g_ce.inflight_waits, "a rejected registration must not consume a slot"); TEST_ASSERT_EQUAL_INT(0, atomic_load(&rec1.calls)); TEST_ASSERT_EQUAL_INT(0, atomic_load(&rec2.calls)); resolve_teardown(); printf("test_commit_submit_async_respects_admission_cap: PASS\n"); } /* ------------------------------------------------------------------ * * (4) The admission budget is SHARED with COMMIT_MODE_WAIT: a parked * * blocking waiter occupies the same pool an async registration * * draws from, and releasing it frees the slot for either caller. * * ------------------------------------------------------------------ */ typedef struct { uint64_t lsn; uint32_t timeout_ms; int rc; } waiter_arg_t; static void* waiter_thread(void* a) { waiter_arg_t* w = (waiter_arg_t*)a; w->rc = commit_submit(&g_ce, w->lsn, COMMIT_MODE_WAIT, w->timeout_ms); return NULL; } static uint64_t mono_ms(void) { struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); return (uint64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000; } /* Poll ce->inflight_waits until it reaches `want` or the deadline elapses; * returns the last observed value (pattern from test_commit_admission_pool.c * poll_inflight — avoids a fixed sleep race against the background thread's * admission acquire). */ static int poll_inflight(int want, uint32_t deadline_ms) { uint64_t t0 = mono_ms(); int v; for (;;) { pthread_mutex_lock(&g_ce.lock); v = g_ce.inflight_waits; pthread_mutex_unlock(&g_ce.lock); if (v >= want || mono_ms() - t0 >= deadline_ms) { return v; } usleep(500); } } static void test_commit_submit_async_shares_admission_budget_with_wait(void) { continuation_recorder_t rec = {0}; pthread_t tid; waiter_arg_t w = {.lsn = 1, .timeout_ms = 4000, .rc = -9999}; /* n_vcpus=2 -> local cap = 1 (same sizing as test 3 above). */ resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/2, /*n_vcpus=*/2); TEST_ASSERT_EQUAL_INT(0, pthread_create(&tid, NULL, waiter_thread, &w)); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, poll_inflight(1, 3000), "the blocking WAIT must occupy the sole admission slot"); int rc = commit_submit_async(&g_ce, /*lsn=*/2, recorder_fn, &rec); TEST_ASSERT_EQUAL_INT_MESSAGE( -XCE_ERR_NOT_READY, rc, "async registration must fast-reject while a blocking WAIT holds " "the shared admission slot"); /* Release the blocking waiter via a covering majority ACK (self + 1 * standby = 2 > 3/2), freeing the slot for the pending registration. */ repl_on_ack(&g_p.repl, &g_eps[0], 1); commit_on_ack(&g_ce, 0, 1); TEST_ASSERT_EQUAL_INT(0, pthread_join(tid, NULL)); TEST_ASSERT_EQUAL_INT(0, w.rc); rc = commit_submit_async(&g_ce, /*lsn=*/2, recorder_fn, &rec); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, rc, "registration must succeed once the shared slot is freed"); resolve_teardown(); printf("test_commit_submit_async_shares_admission_budget_with_wait: " "PASS\n"); } /* ------------------------------------------------------------------ * * (5) Distinct concurrent registrations occupy distinct per-LSN slots. * * ------------------------------------------------------------------ */ static void test_commit_submit_async_tracks_distinct_lsns(void) { continuation_recorder_t rec1 = {0}, rec2 = {0}; resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/2, /*n_vcpus=*/8); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec1)); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 2, recorder_fn, &rec2)); TEST_ASSERT_EQUAL_INT(2, g_ce.inflight_waits); /* White-box: the two registrations must land in distinct slots, each * carrying its own lsn/fn/ctx (the per-LSN bookkeeping the task * requires, not a single-shot overwrite). */ int found_lsn1 = 0, found_lsn2 = 0; for (int i = 0; i < g_ce.n_continuations; i++) { if (g_ce.continuations[i].lsn == 1) { found_lsn1 = 1; TEST_ASSERT_EQUAL_PTR(recorder_fn, g_ce.continuations[i].fn); TEST_ASSERT_EQUAL_PTR(&rec1, g_ce.continuations[i].ctx); } else if (g_ce.continuations[i].lsn == 2) { found_lsn2 = 1; TEST_ASSERT_EQUAL_PTR(recorder_fn, g_ce.continuations[i].fn); TEST_ASSERT_EQUAL_PTR(&rec2, g_ce.continuations[i].ctx); } } TEST_ASSERT_TRUE_MESSAGE(found_lsn1, "lsn=1 registration must be tracked"); TEST_ASSERT_TRUE_MESSAGE(found_lsn2, "lsn=2 registration must be tracked"); resolve_teardown(); printf("test_commit_submit_async_tracks_distinct_lsns: PASS\n"); } /* ════════════════════════════════════════════════════════════════════════ * Task 1.2 — commit_resolve(): majority-commit and failure-path resolution. * ════════════════════════════════════════════════════════════════════════ */ /* ------------------------------------------------------------------ * * (6) Majority ack: a registered continuation is resolved COMMITTED, * * exactly once, dispatched OFF the ACK-driving thread (real, * * pool-wired path — never inline). * * ------------------------------------------------------------------ */ static void test_commit_resolve_majority_ack_dispatches_committed_off_thread(void) { continuation_recorder_t rec = {0}; pthread_t self = pthread_self(); /* 3-member cluster, entry 1 appended+durable, no ACKs yet. */ 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)); TEST_ASSERT_EQUAL_INT(1, g_ce.inflight_waits); /* Majority ACK arrives on THIS thread (mirrors commit_on_ack running on * the iomgr thread in production). */ repl_on_ack(&g_p.repl, &g_eps[0], 1); commit_on_ack(&g_ce, 0, 1); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, poll_calls(&rec.calls, 1, 2000), "commit_resolve must deliver COMMITTED once the majority ack lands"); TEST_ASSERT_EQUAL_UINT64(1, rec.last_lsn); TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_COMMITTED, rec.last_outcome); TEST_ASSERT_EQUAL_INT(0, rec.last_rc); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&rec.calls), "the continuation must be invoked exactly once"); TEST_ASSERT_FALSE_MESSAGE( pthread_equal(self, rec.last_thread), "the continuation must run off the calling (ACK-driving) thread, " "never inline"); TEST_ASSERT_EQUAL_INT_MESSAGE( 0, g_ce.inflight_waits, "resolve must release the continuation's admission slot"); resolve_pool_teardown(); printf("test_commit_resolve_majority_ack_dispatches_committed_off_thread: " "PASS\n"); } /* ------------------------------------------------------------------ * * (7) Failure outcomes: commit_resolve delivers ANY caller-supplied * * outcome/rc faithfully, off-thread, for the three failure causes * * tasks.md 1.2 names — timeout, demotion, truncation. commit.c has * * no existing distinct detector for demotion/truncation (see file * * header); these calls simulate a future detector invoking the * * generic resolve primitive directly. * * ------------------------------------------------------------------ */ static void test_commit_resolve_delivers_failed_outcome_for_timeout(void) { continuation_recorder_t rec = {0}; pthread_t self = pthread_self(); 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)); commit_resolve(&g_ce, 1, COMMIT_OUTCOME_FAILED, -ETIMEDOUT); TEST_ASSERT_EQUAL_INT(1, poll_calls(&rec.calls, 1, 2000)); 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_FALSE(pthread_equal(self, rec.last_thread)); TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits); resolve_pool_teardown(); printf("test_commit_resolve_delivers_failed_outcome_for_timeout: PASS\n"); } static void test_commit_resolve_delivers_failed_outcome_for_demotion(void) { continuation_recorder_t rec = {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)); /* -XCE_ERR_STANDBY ("op is primary-only") is the existing vocabulary code * closest to "this controller is no longer primary" — a future * ha.c demotion detector's natural choice of rc. */ commit_resolve(&g_ce, 1, COMMIT_OUTCOME_FAILED, -XCE_ERR_STANDBY); TEST_ASSERT_EQUAL_INT(1, poll_calls(&rec.calls, 1, 2000)); TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_FAILED, rec.last_outcome); TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rec.last_rc); TEST_ASSERT_EQUAL_INT(1, atomic_load(&rec.calls)); resolve_pool_teardown(); printf("test_commit_resolve_delivers_failed_outcome_for_demotion: PASS\n"); } static void test_commit_resolve_delivers_failed_outcome_for_truncation(void) { continuation_recorder_t rec = {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)); /* -ENOENT ("no such entry") represents a pending lsn whose log tail was * truncated out from under it — a future persist.c truncation detector's * natural choice of rc. */ commit_resolve(&g_ce, 1, COMMIT_OUTCOME_FAILED, -ENOENT); TEST_ASSERT_EQUAL_INT(1, poll_calls(&rec.calls, 1, 2000)); TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_FAILED, rec.last_outcome); TEST_ASSERT_EQUAL_INT(-ENOENT, rec.last_rc); TEST_ASSERT_EQUAL_INT(1, atomic_load(&rec.calls)); resolve_pool_teardown(); printf("test_commit_resolve_delivers_failed_outcome_for_truncation: PASS\n"); } /* ------------------------------------------------------------------ * * (8) Idempotency: a second resolve() for an already-resolved lsn is a * * no-op — the FIRST outcome/rc stands, no second invocation. * * ------------------------------------------------------------------ */ static void test_commit_resolve_second_resolve_is_noop(void) { continuation_recorder_t rec = {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)); commit_resolve(&g_ce, 1, COMMIT_OUTCOME_FAILED, -ETIMEDOUT); TEST_ASSERT_EQUAL_INT(1, poll_calls(&rec.calls, 1, 2000)); /* A second resolve for the SAME lsn, with a DIFFERENT outcome/rc, must * be a silent no-op: no second invocation, and the recorded outcome/rc * from the first (and only) delivery must stand unchanged. */ commit_resolve(&g_ce, 1, COMMIT_OUTCOME_COMMITTED, 0); usleep(50 * 1000); /* give a wrongly-dispatched second call a chance to land */ TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&rec.calls), "a second resolve for an already-resolved lsn must not re-invoke"); TEST_ASSERT_EQUAL_INT_MESSAGE( COMMIT_OUTCOME_FAILED, rec.last_outcome, "the first delivery's outcome must stand, not be overwritten"); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rec.last_rc); resolve_pool_teardown(); printf("test_commit_resolve_second_resolve_is_noop: PASS\n"); } /* ------------------------------------------------------------------ * * (9) Resolving an lsn with no registered continuation is a safe * * no-op — never crashes, never invokes anything. * * ------------------------------------------------------------------ */ static void test_commit_resolve_unregistered_lsn_is_safe_noop(void) { resolve_pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, /*n_vcpus=*/4); commit_resolve(&g_ce, /*lsn=*/1, COMMIT_OUTCOME_COMMITTED, 0); commit_resolve(NULL, 1, COMMIT_OUTCOME_FAILED, -ETIMEDOUT); commit_resolve(&g_ce, /*lsn=*/0, COMMIT_OUTCOME_FAILED, -ETIMEDOUT); /* No crash is the primary assertion; give any errant dispatch a moment * then confirm nothing was admitted/released unexpectedly. */ usleep(50 * 1000); TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits); resolve_pool_teardown(); printf("test_commit_resolve_unregistered_lsn_is_safe_noop: PASS\n"); } /* ------------------------------------------------------------------ * * (10) Fail-loud inline fallback: with no vtaskmgr pool wired (the * * "commit-only unit fixtures" case, commit.h field comment), the * * continuation must STILL be resolved — never silently dropped — * * via the documented inline fallback. * * ------------------------------------------------------------------ */ static void test_commit_resolve_falls_back_to_inline_without_pool(void) { continuation_recorder_t rec = {0}; /* Plain resolve_setup: no commit_set_pool call, so g_ce.pool == NULL. */ resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1, /*n_vcpus=*/4); TEST_ASSERT_NULL(g_ce.pool); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec)); commit_resolve(&g_ce, 1, COMMIT_OUTCOME_FAILED, -ETIMEDOUT); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&rec.calls), "without a pool the continuation must still resolve, via the " "fail-loud inline fallback (never silently dropped)"); TEST_ASSERT_EQUAL_INT(COMMIT_OUTCOME_FAILED, rec.last_outcome); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rec.last_rc); TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits); resolve_teardown(); 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); RUN_TEST(test_commit_submit_async_rejects_invalid_args); RUN_TEST(test_commit_submit_async_respects_admission_cap); RUN_TEST(test_commit_submit_async_shares_admission_budget_with_wait); RUN_TEST(test_commit_submit_async_tracks_distinct_lsns); RUN_TEST(test_commit_resolve_majority_ack_dispatches_committed_off_thread); RUN_TEST(test_commit_resolve_delivers_failed_outcome_for_timeout); RUN_TEST(test_commit_resolve_delivers_failed_outcome_for_demotion); RUN_TEST(test_commit_resolve_delivers_failed_outcome_for_truncation); 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); }