/* * 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 * * Concurrency half (task 1.5, closes task-group 1): the tests above prove * each detection path (majority ack, timeout, demotion, truncation) works * one-at-a-time, sequentially. The cases below prove the SAME mechanisms * hold up TOGETHER, under genuine multi-threaded concurrency: real pthreads * racing commit_submit_async's admission gate, and real concurrent * detection paths (a majority-ack-driving thread vs. the engine's own * dedicated sweep thread) resolving a whole burst of pending continuations * without any double-resolve, leak, or hang, and without the admission cap * ever being exceeded. The demotion- and truncation-triggered burst cases * live alongside their existing single/pair-continuation counterparts in * test_demotion_convergence.c and test_persist_truncate.c respectively, * reusing those files' real cluster_t / persist_t fixtures rather than * duplicating them here. * * Requirements (task 1.5): 9.1, 9.2, 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"); } /* ════════════════════════════════════════════════════════════════════════ * Task 1.5 — concurrent in-flight commits: exactly-once resolution and * bounded admission under a GENUINE multi-threaded burst (real pthreads, * not sequential calls dressed up as concurrent). * ════════════════════════════════════════════════════════════════════════ */ /* ------------------------------------------------------------------ * * (14) Admission-cap burst: a genuine multi-threaded burst of * * concurrent registrations must never admit more than `cap` * * simultaneously, even under real contention on ce->lock — not * * just the two-sequential-call shape existing test (3) exercises. * * ------------------------------------------------------------------ */ #define BURST_ADMISSION_N 16 typedef struct { uint64_t lsn; int rc; continuation_recorder_t* rec; } admission_racer_arg_t; static void* admission_racer_thread(void* p) { admission_racer_arg_t* a = (admission_racer_arg_t*)p; a->rc = commit_submit_async(&g_ce, a->lsn, recorder_fn, a->rec); return NULL; } /* Samples ce->inflight_waits at a tight interval while the racer burst is * in flight, recording the highest and lowest values observed — proves the * admission cap is never breached and inflight_waits never goes negative, * under REAL concurrent contention, not just at the two checkpoints a * sequential test would sample. */ typedef struct { _Atomic int stop; int max_seen; int min_seen; } admission_monitor_arg_t; static void* admission_monitor_thread(void* p) { admission_monitor_arg_t* m = (admission_monitor_arg_t*)p; while (!atomic_load(&m->stop)) { pthread_mutex_lock(&g_ce.lock); int v = g_ce.inflight_waits; pthread_mutex_unlock(&g_ce.lock); if (v > m->max_seen) { m->max_seen = v; } if (v < m->min_seen) { m->min_seen = v; } usleep(50); } return NULL; } static void test_commit_submit_async_concurrent_burst_respects_admission_cap(void) { continuation_recorder_t recs[BURST_ADMISSION_N] = {0}; admission_racer_arg_t args[BURST_ADMISSION_N]; pthread_t tids[BURST_ADMISSION_N]; admission_monitor_arg_t mon = {0}; pthread_t mon_tid; const int cap = 4; /* max(1, n_vcpus(6) - COMMIT_WAIT_RESERVED(2)) */ /* n_vcpus=6 -> local admission cap = 4 (same formula the existing * sequential test 3 exercises); this proves it holds under genuine * multi-threaded contention instead of two calls in program order. */ resolve_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/BURST_ADMISSION_N, /*n_vcpus=*/6); TEST_ASSERT_EQUAL_INT( 0, pthread_create(&mon_tid, NULL, admission_monitor_thread, &mon)); for (int i = 0; i < BURST_ADMISSION_N; i++) { args[i].lsn = (uint64_t)(i + 1); args[i].rc = -9999; args[i].rec = &recs[i]; TEST_ASSERT_EQUAL_INT( 0, pthread_create(&tids[i], NULL, admission_racer_thread, &args[i])); } for (int i = 0; i < BURST_ADMISSION_N; i++) { pthread_join(tids[i], NULL); } atomic_store(&mon.stop, 1); pthread_join(mon_tid, NULL); int accepted = 0, rejected = 0; for (int i = 0; i < BURST_ADMISSION_N; i++) { if (args[i].rc == 0) { accepted++; } else if (args[i].rc == -XCE_ERR_NOT_READY) { rejected++; } else { TEST_FAIL_MESSAGE("unexpected rc from a concurrent registration"); } } TEST_ASSERT_EQUAL_INT_MESSAGE( cap, accepted, "exactly `cap` concurrent registrations may be admitted, no more, no " "fewer, even under genuine multi-threaded contention (Req 9.1/9.2)"); TEST_ASSERT_EQUAL_INT(BURST_ADMISSION_N - cap, rejected); TEST_ASSERT_EQUAL_INT_MESSAGE(cap, g_ce.inflight_waits, "inflight_waits must equal exactly the " "accepted count"); TEST_ASSERT_TRUE_MESSAGE( mon.max_seen <= cap, "inflight_waits must never exceed the admission cap, even sampled " "mid-burst under real thread contention"); TEST_ASSERT_TRUE_MESSAGE(mon.min_seen >= 0, "inflight_waits must never go negative"); /* Registration alone, even racing under contention, must never invoke * any continuation. */ for (int i = 0; i < BURST_ADMISSION_N; i++) { TEST_ASSERT_EQUAL_INT(0, atomic_load(&recs[i].calls)); } resolve_teardown(); printf("test_commit_submit_async_concurrent_burst_respects_admission_cap: " "PASS\n"); } /* ------------------------------------------------------------------ * * (15) Concurrent burst resolves exactly once via a MIX of the two * * real detection paths (majority ack + timeout sweep), racing * * each other on ce->lock, with dispatch bounded to a small, fixed * * vtaskmgr worker pool rather than growing with the burst size. * * ------------------------------------------------------------------ */ /* Decouples the continuation registry's capacity (ce_n_vcpus, sized to fit * the whole burst) from the vtaskmgr pool's actual OS worker-thread count * (pool_n_vcpus, deliberately smaller than the burst) — resolve_pool_setup * ties both to the same n_vcpus value, which cannot demonstrate that * dispatch stays bounded to a small worker set instead of growing with the * number of concurrently-resolving continuations (Req 9.1/9.2). */ static void bounded_pool_setup(int n_peers, int n_standbys, uint64_t appended_lsn, int ce_n_vcpus, int pool_n_vcpus) { resolve_setup(n_peers, n_standbys, appended_lsn, ce_n_vcpus); TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_mgr, pool_n_vcpus)); TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_mgr)); commit_set_pool(&g_ce, &g_mgr); } #define BURST_MIX_N 8 typedef struct { uint64_t lsn; int rc; continuation_recorder_t* rec; } mix_register_arg_t; static void* mix_register_thread(void* p) { mix_register_arg_t* a = (mix_register_arg_t*)p; a->rc = commit_submit_async(&g_ce, a->lsn, recorder_fn, a->rec); return NULL; } /* Count pairwise-distinct pthread_t values in threads[0..n) — proves * dispatch is bounded to a small worker set rather than one thread per * resolved continuation (pthread_t is opaque; must compare via * pthread_equal, never ==). */ static int count_distinct_threads(const pthread_t* threads, int n) { int distinct = 0; for (int i = 0; i < n; i++) { int seen_before = 0; for (int j = 0; j < i; j++) { if (pthread_equal(threads[i], threads[j])) { seen_before = 1; break; } } if (!seen_before) { distinct++; } } return distinct; } static void test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_once(void) { continuation_recorder_t recs[BURST_MIX_N] = {0}; mix_register_arg_t args[BURST_MIX_N]; pthread_t tids[BURST_MIX_N]; pthread_t dispatch_threads[BURST_MIX_N]; /* ce_n_vcpus == BURST_MIX_N: the continuation registry has exactly * enough slots for the whole burst, none to spare. pool_n_vcpus=3: a * small, fixed, REAL vtaskmgr worker pool that must service all 8 * resolutions — if dispatch spawned a thread per resolution instead of * reusing the bounded pool, more than 3 distinct threads would appear * below. */ bounded_pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/BURST_MIX_N, /*ce_n_vcpus=*/BURST_MIX_N, /*pool_n_vcpus=*/3); /* Arrange: register the whole burst CONCURRENTLY from real threads. */ for (int i = 0; i < BURST_MIX_N; i++) { args[i].lsn = (uint64_t)(i + 1); args[i].rc = -9999; args[i].rec = &recs[i]; TEST_ASSERT_EQUAL_INT( 0, pthread_create(&tids[i], NULL, mix_register_thread, &args[i])); } for (int i = 0; i < BURST_MIX_N; i++) { pthread_join(tids[i], NULL); } for (int i = 0; i < BURST_MIX_N; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE( 0, args[i].rc, "every concurrent registration must succeed -- the registry has " "exactly enough slots for this burst"); } TEST_ASSERT_EQUAL_INT(BURST_MIX_N, g_ce.inflight_waits); /* Act: satisfy majority for the LOWER half (lsn 1..4) via the REAL * ack-driven path (repl_on_ack + commit_on_ack, exactly what task 1.2 * wired) -- committed_lsn advances contiguously to 4, resolving lsn * 1..4 as COMMITTED in one bulk commit_resolve_pending() scan. The * UPPER half (lsn 5..8) is never ACKed, so it is left entirely to the * real timeout-sweep thread (task 1.3). Both detection paths now race * each other for ce->lock concurrently: this ack-driving thread and the * commit engine's own dedicated sweep thread (which independently scans * the SAME registry and would classify lsn 1..4 as majority-satisfied * too, if its own scan happens to run before this thread's * commit_resolve_pending() call reaches them first -- commit_resolve's * idempotent slot-clear-under-lock is what makes that real race safe). */ repl_on_ack(&g_p.repl, &g_eps[0], BURST_MIX_N / 2); commit_on_ack(&g_ce, 0, BURST_MIX_N / 2); for (int i = 0; i < BURST_MIX_N; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE( 1, poll_calls(&recs[i].calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 2000), "every continuation in the burst must resolve, through one of " "the two real concurrent detection paths"); } for (int i = 0; i < BURST_MIX_N; i++) { /* Exactly-once: no double delivery for any lsn in the burst. */ TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&recs[i].calls), "no continuation in the burst may be resolved more than once"); dispatch_threads[i] = recs[i].last_thread; if (i < BURST_MIX_N / 2) { TEST_ASSERT_EQUAL_INT_MESSAGE( COMMIT_OUTCOME_COMMITTED, recs[i].last_outcome, "the lower half must resolve COMMITTED via the real " "majority-ack path"); TEST_ASSERT_EQUAL_INT(0, recs[i].last_rc); } else { TEST_ASSERT_EQUAL_INT_MESSAGE( COMMIT_OUTCOME_FAILED, recs[i].last_outcome, "the upper half, never acked, must resolve FAILED via the " "real timeout-sweep path"); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, recs[i].last_rc); } } TEST_ASSERT_EQUAL_INT_MESSAGE( 0, g_ce.inflight_waits, "every admission slot in the burst must be released exactly once"); /* Bounded worker usage (Req 9.1/9.2): all BURST_MIX_N resolutions must * be dispatched across no more than the pool's 3 real worker threads -- * proving the burst is serviced by a small, fixed pool rather than one * thread spawned per resolved continuation. */ int distinct = count_distinct_threads(dispatch_threads, BURST_MIX_N); TEST_ASSERT_TRUE_MESSAGE( distinct <= 3, "continuation dispatch for the whole burst must stay within the " "bounded vtaskmgr worker pool, not grow with the burst size"); resolve_pool_teardown(); printf("test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_" "once: PASS\n"); } /* ------------------------------------------------------------------ * * (16, nice-to-have) Genuine two-thread race: a real second thread * * delivers a real majority ACK right at the edge of the * * registration's own (unmanipulated) DEFAULT_SYNC_TIMEOUT_MS * * deadline, racing the commit engine's own dedicated sweep * * thread for ce->lock -- unlike test (12) above, which proves the * * same tie-break with single-thread white-box deadline * * manipulation, this drives it with two REAL OS threads. * * ------------------------------------------------------------------ */ typedef struct { uint64_t lsn; } delayed_ack_arg_t; static void* delayed_ack_thread(void* p) { delayed_ack_arg_t* a = (delayed_ack_arg_t*)p; /* Sleep to just before lsn's own real deadline (150ms of margin absorbs * scheduler jitter under ASAN). commit_advance_locked() broadcasts * ce->cond on every real advance, so this ACK wakes the sweep thread * EARLY -- well before its own timed-out wait would naturally fire -- * and the sweep thread's re-scan finds wait_done_locked() satisfied * first (majority wins, per commit_sweep_scan_locked's own check * order). */ usleep((DEFAULT_SYNC_TIMEOUT_MS - 150) * 1000); repl_on_ack(&g_p.repl, &g_eps[0], a->lsn); commit_on_ack(&g_ce, 0, a->lsn); return NULL; } static void test_commit_resolve_genuine_two_thread_race_majority_wins_over_timeout(void) { continuation_recorder_t rec = {0}; delayed_ack_arg_t arg = {.lsn = 1}; pthread_t tid; 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(0, pthread_create(&tid, NULL, delayed_ack_thread, &arg)); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, poll_calls(&rec.calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 2000), "the raced lsn must still resolve exactly once"); TEST_ASSERT_EQUAL_INT_MESSAGE( COMMIT_OUTCOME_COMMITTED, rec.last_outcome, "a genuine majority ACK landing near the deadline boundary, on a " "real second thread racing the real sweep thread, must still win " "over the timeout -- exactly-once, majority-wins under REAL " "two-thread concurrency (not the white-box simulation in test 12)"); TEST_ASSERT_EQUAL_INT(0, rec.last_rc); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&rec.calls), "exactly one resolution, regardless of which real thread won"); TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits); pthread_join(tid, NULL); resolve_pool_teardown(); printf("test_commit_resolve_genuine_two_thread_race_majority_wins_over_" "timeout: PASS\n"); } void test_commit_resolve(void) { UnitySetTestFile(__FILE__); 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); RUN_TEST(test_commit_submit_async_concurrent_burst_respects_admission_cap); RUN_TEST(test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_once); RUN_TEST(test_commit_resolve_genuine_two_thread_race_majority_wins_over_timeout); }