From 10d157f8c54aea3845ae620b294b00eccbd0c8a5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 3 Jul 2026 07:55:35 +0000 Subject: [PATCH] feat(raft-apply-decoupling): add commit engine continuation registration (task 1.1) commit_submit_async() lets a WAIT-mode caller register a continuation against an appended LSN instead of blocking on pthread_cond_timedwait, reusing the existing admission-cap discipline (wait_admit_locked/ wait_release_locked) so the bounded-capacity guarantee carries over unchanged. Purely additive: no existing call site (repl_wait_majority_ack) is touched yet -- that switch lands in task 3.2 once commit_resolve() (task 1.2) exists to drive it. --- .kiro/specs/raft-apply-decoupling/tasks.md | 7 +- src/controller/commit.c | 86 +++++ src/controller/include/commit.h | 51 +++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_commit_resolve.c | 311 ++++++++++++++++++ 6 files changed, 457 insertions(+), 1 deletion(-) create mode 100644 src/controller/testsuite/test_commit_resolve.c diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 8eb6950f..b8d35e09 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -1,7 +1,7 @@ # Implementation Plan - [ ] 1. Commit engine event-driven resolve (keystone prerequisite) -- [ ] 1.1 (P) Add per-LSN continuation registration to the commit engine for the synchronous (WAIT) commit mode, replacing the blocking wait +- [x] 1.1 (P) Add per-LSN continuation registration to the commit engine for the synchronous (WAIT) commit mode, replacing the blocking wait - Register a continuation (callback + context) against an appended LSN instead of parking the calling thread - Preserve the existing per-LSN bookkeeping and admission-cap discipline already used by the commit engine - Observable: a unit test can register a continuation for an LSN and the calling thread returns immediately, before the LSN commits @@ -168,3 +168,8 @@ - Observable: a recorded throughput comparison shows no regression relative to the pre-feature baseline - _Requirements: 11.4_ - _Depends: 7.3_ + +## Implementation Notes + +- **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.** diff --git a/src/controller/commit.c b/src/controller/commit.c index 721dbfcf..d81c96cf 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -30,6 +30,7 @@ */ #include #include +#include #include #include @@ -311,6 +312,20 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp ce->raft_commit_index = 0; /* advisory record only; does not gate commit (optA) */ memset(ce->scratch, 0, sizeof(ce->scratch)); + /* raft-apply-decoupling task 1.1: continuation registry, sized to + * n_vcpus (see the field comment in commit.h for the sizing invariant: + * wait_cap(ce) <= n_vcpus always, so n_vcpus slots suffice regardless + * of the blocking-WAIT/async-registration mix sharing inflight_waits). + * calloc zero-fills every slot's lsn to 0, the free-slot sentinel. */ + ce->continuations = (commit_continuation_t*)calloc( + (size_t)n_vcpus, sizeof(commit_continuation_t)); + if (ce->continuations == NULL) { + pthread_cond_destroy(&ce->cond); + pthread_mutex_destroy(&ce->lock); + return -ENOMEM; + } + ce->n_continuations = n_vcpus; + /* Register the durable-advance wake so the flusher can re-trigger commit * evaluation when durable_lsn moves (task 2.2). Persist owns the hook slot; * we only register a function + ctx, preserving the commit→persist (never @@ -346,6 +361,9 @@ void commit_engine_fini(commit_engine_t* ce) if (ce->p != NULL) { persist_set_durable_cb(ce->p, NULL, NULL); } + free(ce->continuations); + ce->continuations = NULL; + ce->n_continuations = 0; pthread_cond_destroy(&ce->cond); pthread_mutex_destroy(&ce->lock); ce->p = NULL; @@ -433,6 +451,74 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, return rc; } +/* Caller holds ce->lock. Returns the index of a free (lsn == 0) slot, or -1 + * if none is free. n_continuations == n_vcpus >= wait_cap(ce) >= + * inflight_waits at all times (the same admission cap governs both this + * registry and COMMIT_MODE_WAIT's blocking parks, sharing one budget), so an + * admitted registration always finds a free slot -- see the commit.h field + * comment for the sizing invariant. */ +static int continuation_find_free_slot_locked(const commit_engine_t* ce) +{ + for (int i = 0; i < ce->n_continuations; i++) { + if (ce->continuations[i].lsn == 0) { + return i; + } + } + return -1; +} + +/* raft-apply-decoupling task 1.1: event-driven registration, the + * non-blocking sibling of commit_submit(..., COMMIT_MODE_WAIT, ...). + * Registers `fn`/`ctx` against `lsn` and returns immediately -- it never + * evaluates or waits on the commit predicate itself. Resolution (invoking + * `fn` on majority-ack / timeout / demotion / truncation) is + * commit_resolve's job, wired onto commit_advance_locked and the existing + * failure paths by a follow-up task; until then a registered slot is + * inert. */ +int commit_submit_async(commit_engine_t* ce, uint64_t lsn, + commit_continuation_fn fn, void* ctx) +{ + int slot, arc; + + if (ce == NULL || ce->p == NULL || lsn == 0 || fn == NULL) { + return -EINVAL; + } + + pthread_mutex_lock(&ce->lock); + + /* Same admission-cap discipline as COMMIT_MODE_WAIT (Req 9.1, 9.2): a + * registered continuation occupies the identical bounded slot a + * blocking wait would have parked in, so the two calling conventions + * share one budget while both exist side by side. */ + arc = wait_admit_locked(ce); + if (arc != 0) { + pthread_mutex_unlock(&ce->lock); + WARN("commit: async registration rejected, admission cap exhausted " + "(pool=%s n_vcpus=%d lsn=%lu)", + ce->pool ? "pool" : "local", ce->n_vcpus, lsn); + return arc; + } + + slot = continuation_find_free_slot_locked(ce); + if (slot < 0) { + /* Unreachable given the sizing invariant above; fail loud rather + * than silently drop the registration if it is ever violated. */ + wait_release_locked(ce); + pthread_mutex_unlock(&ce->lock); + WARN("commit: async registration rejected, no free continuation " + "slot (lsn=%lu, n_continuations=%d) -- sizing invariant " + "violated", lsn, ce->n_continuations); + return -XCE_ERR_NOT_READY; + } + + ce->continuations[slot].lsn = lsn; + ce->continuations[slot].fn = fn; + ce->continuations[slot].ctx = ctx; + + pthread_mutex_unlock(&ce->lock); + return 0; +} + /* ============================================================ * ACK and membership events * ============================================================ */ diff --git a/src/controller/include/commit.h b/src/controller/include/commit.h index 69700b0b..d8729db8 100644 --- a/src/controller/include/commit.h +++ b/src/controller/include/commit.h @@ -23,6 +23,32 @@ typedef enum { COMMIT_MODE_WAIT, COMMIT_MODE_RETAIN } commit_mode_t; +/* raft-apply-decoupling task 1.1: event-driven resolve outcome, delivered to + * a registered continuation in place of a COMMIT_MODE_WAIT caller's blocking + * return value. COMMITTED mirrors commit_submit's rc==0; FAILED mirrors any + * of commit_submit's negative outcomes (timeout, demotion, truncation), with + * the specific cause carried in the continuation's `rc` argument. */ +typedef enum { COMMIT_OUTCOME_COMMITTED, COMMIT_OUTCOME_FAILED } commit_outcome_t; + +/* Invoked exactly once per registered lsn when its outcome becomes known. + * Delivered via commit_resolve (wired by a follow-up task onto + * commit_advance_locked's majority path and the existing timeout/demotion/ + * truncation failure handling), dispatched off the calling thread via + * vtask_enq_classified -- never inline on the replication-ACK or iomgr + * path. */ +typedef void (*commit_continuation_fn)(void* ctx, uint64_t lsn, + commit_outcome_t outcome, int rc); + +/* raft-apply-decoupling task 1.1: one registered COMMIT_MODE_WAIT-equivalent + * continuation. lsn == 0 marks a free slot -- LSN 0 is the reserved + * empty/none sentinel (see persist.h reserve-lsn-0), so it is never a real + * registration and is safe as the "unused" marker. */ +typedef struct commit_continuation { + uint64_t lsn; + commit_continuation_fn fn; + void* ctx; +} commit_continuation_t; + /* * Commit engine state — in-memory only (commit state volatile, Raft-conformant). * Per-standby view delegated to standby_conn_t.acked_lsn / in_catch_up via @@ -52,6 +78,18 @@ typedef struct commit_engine { * non-decreasing, set under ce->lock via commit_set_raft_commit_index. */ uint64_t raft_commit_index; uint64_t scratch[MAX_STANDBYS]; /* sorted matchIndex scratch array */ + /* raft-apply-decoupling task 1.1: per-LSN continuation registry for + * commit_submit_async (event-driven resolve, replacing the + * COMMIT_MODE_WAIT blocking-park model). Sized to n_vcpus at init: the + * SAME admission-cap discipline (wait_admit_locked/wait_cap, the shared + * inflight_waits budget) bounds concurrent registrations to + * wait_cap(ce) <= n_vcpus, so n_vcpus slots always suffice regardless of + * the blocking/async call mix. Resolution (commit_resolve, invoked from + * commit_advance_locked and the timeout/demotion/truncation failure + * paths) is wired by a follow-up task; until then a registered slot is + * inert -- nothing reads it back. */ + commit_continuation_t* continuations; + int n_continuations; /* capacity, == n_vcpus */ } commit_engine_t; /* Lifecycle. init validates n_peers >= 0, n_vcpus >= 1, non-NULL persist_t; @@ -76,6 +114,19 @@ void commit_set_pool(commit_engine_t* ce, vtaskmgr_t* pool); int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, uint32_t timeout_ms); +/* raft-apply-decoupling task 1.1: register a continuation for lsn instead of + * parking the calling thread -- the event-driven equivalent of + * commit_submit(..., COMMIT_MODE_WAIT, ...). Returns 0 on successful + * registration; -EINVAL on bad args (ce/p NULL, lsn == 0, fn == NULL); + * -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. */ +int commit_submit_async(commit_engine_t* ce, uint64_t lsn, + commit_continuation_fn fn, void* ctx); + /* iomgr path: record standby ACK, evaluate majority, advance committed_lsn * contiguously, wake satisfied waiters. Never blocks. */ void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn); diff --git a/src/controller/meson.build b/src/controller/meson.build index 38454c49..072dde16 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -143,6 +143,7 @@ controller_unit_test_src = [ 'testsuite/test_drive_async_config.c', 'testsuite/test_slowlane.c', 'testsuite/test_commit_admission_pool.c', + 'testsuite/test_commit_resolve.c', 'testsuite/test_ha_vep_teardown.c', 'testsuite/test_ha_dispatch_integration.c', 'testsuite/test_ha_core_safety.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 7fae0beb..e2274b5d 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -71,6 +71,7 @@ extern void test_snapdiff(void); extern void test_drive_async_config(void); extern void test_slowlane(void); extern void test_commit_admission_pool(void); +extern void test_commit_resolve(void); extern void test_ha_vep_teardown(void); extern void test_ha_dispatch_integration(void); extern void test_ha_core_safety(void); @@ -216,6 +217,7 @@ int main(int argc, char* argv[]) TEST(test_drive_async_config); TEST(test_slowlane); TEST(test_commit_admission_pool); + TEST(test_commit_resolve); TEST(test_ha_vep_teardown); TEST(test_ha_dispatch_integration); TEST(test_ha_core_safety); diff --git a/src/controller/testsuite/test_commit_resolve.c b/src/controller/testsuite/test_commit_resolve.c new file mode 100644 index 00000000..9eed9fee --- /dev/null +++ b/src/controller/testsuite/test_commit_resolve.c @@ -0,0 +1,311 @@ +/* + * test_commit_resolve.c — raft-apply-decoupling task 1.1. + * + * Keystone conversion, registration half: 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. This file covers ONLY the registration mechanism — + * resolution (invoking the continuation on majority-ack / timeout / + * demotion / truncation) is wired by a follow-up task (1.2) and is not + * exercised here: a registered continuation is never invoked in these + * tests, by design. + * + * Asserts: + * (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: 1.3, 1.4, 9.1, 9.2 + */ +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "xce/errnox.h" + +#include "commit.h" +#include "repl.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); +} + +/* Continuation recorder — proves (non-)invocation and captures the args a + * future commit_resolve() would deliver. Not exercised as a real resolve + * path in this file (task 1.2's job); only used to assert it stays quiet. */ +typedef struct { + _Atomic int calls; + uint64_t last_lsn; + commit_outcome_t last_outcome; + int last_rc; + void* last_ctx; +} 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; + + r->last_lsn = lsn; + r->last_outcome = outcome; + r->last_rc = rc; + r->last_ctx = ctx; + 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"); +} + +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); +}