feat(raft-apply-decoupling): commit engine resolve primitive + majority-ack wiring (task 1.2)
commit_resolve() delivers a registered continuation's outcome exactly once (idempotent re-resolve), dispatched off-thread via vtask_enq_classified(VTASK_LIVE_BLOCK) so it never runs inline on the replication-ACK or iomgr path. Wired into all 6 commit.c-internal paths that can advance committed_lsn/writable_lsn. Review caught a real task-graph gap, not a defect: converting COMMIT_MODE_WAIT away from a blocking wait also removes the only mechanism that ever detected a commit timing out (the caller's own cond_timedwait deadline). Nothing yet arms an equivalent timeout for async-registered continuations, and nothing wires demotion/log-truncation detection to commit_resolve(FAILED, ...) -- left as-is, a pending async command during quorum loss would hang forever instead of failing, which is worse than today. Split into new tasks 1.3 (timeout arming) and 1.4 (demotion/truncation wiring), renumbered the old concurrency test to 1.5, and made 3.2 depend on both since its failure/timeout observable is meaningless without them. design.md's File Structure Plan updated to reflect that ha.c/persist.c need a narrow touch for 1.4's detection hooks.
This commit is contained in:
parent
10d157f8c5
commit
fd7a0746eb
5 changed files with 558 additions and 19 deletions
|
|
@ -44,6 +44,12 @@
|
|||
#include <assert.h>
|
||||
#endif
|
||||
|
||||
/* Forward declaration: raft-apply-decoupling task 1.2. commit_on_durable_advance
|
||||
* (below) is defined ahead of the Continuation Resolution section further down
|
||||
* this file but must call commit_resolve_pending after releasing ce->lock; see
|
||||
* that section for the full definition and contract. */
|
||||
static void commit_resolve_pending(commit_engine_t* ce);
|
||||
|
||||
/* ============================================================
|
||||
* Internal helpers
|
||||
* ============================================================ */
|
||||
|
|
@ -290,6 +296,11 @@ static void commit_on_durable_advance(void* ctx)
|
|||
* broadcasts on a committed_lsn change, so cover the no-change case here. */
|
||||
pthread_cond_broadcast(&ce->cond);
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: the durable watermark moving can newly
|
||||
* satisfy a registered continuation's predicate exactly like it does a
|
||||
* blocked COMMIT_MODE_WAIT caller's -- resolve after ce->lock is released. */
|
||||
commit_resolve_pending(ce);
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
|
|
@ -443,6 +454,11 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
|
|||
gate = ce->writable_lsn;
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: the n_peers<=1 self-quorum branch above
|
||||
* can advance committed_lsn via this same lock section -- resolve any
|
||||
* newly-satisfied registered continuation after ce->lock is released. */
|
||||
commit_resolve_pending(ce);
|
||||
|
||||
if (rc == -ETIMEDOUT) {
|
||||
WARN("commit: WAIT deadline expired without commit "
|
||||
"(lsn=%lu committed_lsn=%lu writable_lsn=%lu timeout_ms=%u)",
|
||||
|
|
@ -519,6 +535,179 @@ int commit_submit_async(commit_engine_t* ce, uint64_t lsn,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* Continuation resolution (raft-apply-decoupling task 1.2)
|
||||
* ============================================================ */
|
||||
|
||||
/* Heap payload for one dispatched continuation invocation. Owned by the
|
||||
* enqueued vtask's vtask_data_t.u.generic; vtask_fini/run_task only free the
|
||||
* vtask_t and its vtask_data_t wrapper, never this payload, so the dispatch
|
||||
* entry function below frees it after invoking fn. */
|
||||
typedef struct {
|
||||
commit_continuation_fn fn;
|
||||
void* ctx;
|
||||
uint64_t lsn;
|
||||
commit_outcome_t outcome;
|
||||
int rc;
|
||||
} commit_resolve_dispatch_t;
|
||||
|
||||
/* vtask entry (run on a pool worker, never the calling thread): invoke the
|
||||
* continuation and free its dispatch payload. */
|
||||
static int commit_resolve_dispatch_entry(vtask_t* task)
|
||||
{
|
||||
commit_resolve_dispatch_t* d =
|
||||
(commit_resolve_dispatch_t*)task->data->u.generic;
|
||||
|
||||
d->fn(d->ctx, d->lsn, d->outcome, d->rc);
|
||||
free(d);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Build + enqueue a generic vtask that invokes fn(ctx, lsn, outcome, rc) off
|
||||
* the calling thread. Caller must NOT hold ce->lock.
|
||||
*
|
||||
* Class VTASK_LIVE_BLOCK, not VTASK_REQ_BLOCK or VTASK_LIVENESS:
|
||||
* - VTASK_REQ_BLOCK is admission-gated (vtask_enq_classified can fast-reject
|
||||
* with -E_XCE_NOT_READY at the pool's blocking cap) -- unacceptable here:
|
||||
* Req 9.4 requires the continuation be resolved exactly once, never left
|
||||
* unresolved, and a rejected enqueue would silently drop it.
|
||||
* - VTASK_LIVENESS is never admission-gated either, but its reservation
|
||||
* (vtaskmgr.h) is sized and reserved for the narrow WAL-ack/heartbeat/
|
||||
* vote receipt+send set -- not general apply/notify continuations.
|
||||
* - VTASK_LIVE_BLOCK is never admission-gated (same delivery guarantee as
|
||||
* LIVENESS) and is documented as "bounded mandatory park ... runs in the
|
||||
* reserved headroom" -- the correct fit for mandatory, must-not-reject
|
||||
* work that is not itself the narrow liveness set.
|
||||
*
|
||||
* On allocation/enqueue failure (OOM, or no vtaskmgr pool wired -- true only
|
||||
* of commit-only unit fixtures; production always wires the pool before any
|
||||
* commit activity, cluster_init_phase2) this fails loud and invokes the
|
||||
* continuation inline as a last resort: "controllers never self-terminate" --
|
||||
* a continuation must never be silently dropped, so a rare fallback inline
|
||||
* call is preferred over an unresolved continuation. */
|
||||
static void dispatch_continuation(commit_engine_t* ce, commit_continuation_fn fn,
|
||||
void* ctx, uint64_t lsn,
|
||||
commit_outcome_t outcome, int rc)
|
||||
{
|
||||
vtask_data_t* data;
|
||||
vtask_t* task;
|
||||
commit_resolve_dispatch_t* d;
|
||||
|
||||
if (fn == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
data = calloc(1, sizeof(*data));
|
||||
task = vtask_alloc();
|
||||
d = calloc(1, sizeof(*d));
|
||||
|
||||
if (ce->pool == NULL || data == NULL || task == NULL || d == NULL) {
|
||||
WARN("commit: continuation dispatch unavailable (pool=%p data=%p "
|
||||
"task=%p d=%p) for lsn=%lu outcome=%d rc=%d -- invoking inline "
|
||||
"(stay alive, never leave a continuation unresolved)",
|
||||
(void*)ce->pool, (void*)data, (void*)task, (void*)d, lsn,
|
||||
(int)outcome, rc);
|
||||
free(data);
|
||||
free(task);
|
||||
free(d);
|
||||
fn(ctx, lsn, outcome, rc);
|
||||
return;
|
||||
}
|
||||
|
||||
d->fn = fn;
|
||||
d->ctx = ctx;
|
||||
d->lsn = lsn;
|
||||
d->outcome = outcome;
|
||||
d->rc = rc;
|
||||
|
||||
data->vtype = VTASK_GENERIC;
|
||||
data->u.generic = d;
|
||||
vtask_init(task, (vtask_func_t)commit_resolve_dispatch_entry, data);
|
||||
|
||||
/* VTASK_LIVE_BLOCK is never admission-gated -- see the function comment
|
||||
* above; this cannot fast-reject the way VTASK_REQ_BLOCK could. */
|
||||
if (vtask_enq_classified(ce->pool, task, VTASK_LIVE_BLOCK) < 0) {
|
||||
WARN("commit: continuation vtask enqueue failed for lsn=%lu -- "
|
||||
"invoking inline (stay alive)", lsn);
|
||||
free(data);
|
||||
free(task);
|
||||
free(d);
|
||||
fn(ctx, lsn, outcome, rc);
|
||||
}
|
||||
}
|
||||
|
||||
/* Resolve lsn's registered continuation (if any) exactly once: find + clear
|
||||
* its slot under ce->lock (idempotent -- a cleared slot's lsn is 0 and never
|
||||
* matches again), release its admission hold (same shared budget as
|
||||
* COMMIT_MODE_WAIT), then dispatch off-thread AFTER ce->lock is released
|
||||
* (lock-ordering rule, design §Commit Engine Resolve). Caller must NOT hold
|
||||
* ce->lock. */
|
||||
void commit_resolve(commit_engine_t* ce, uint64_t lsn, commit_outcome_t outcome,
|
||||
int rc)
|
||||
{
|
||||
commit_continuation_fn fn = NULL;
|
||||
void* ctx = NULL;
|
||||
|
||||
if (ce == NULL || ce->p == NULL || lsn == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
pthread_mutex_lock(&ce->lock);
|
||||
for (int i = 0; i < ce->n_continuations; i++) {
|
||||
if (ce->continuations[i].lsn == lsn) {
|
||||
fn = ce->continuations[i].fn;
|
||||
ctx = ce->continuations[i].ctx;
|
||||
ce->continuations[i].lsn = 0; /* free the slot -- idempotency */
|
||||
ce->continuations[i].fn = NULL;
|
||||
ce->continuations[i].ctx = NULL;
|
||||
wait_release_locked(ce);
|
||||
break;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
if (fn != NULL) {
|
||||
dispatch_continuation(ce, fn, ctx, lsn, outcome, rc);
|
||||
}
|
||||
}
|
||||
|
||||
/* Scan the continuation registry for entries whose WAIT-equivalent completion
|
||||
* predicate (wait_done_locked -- own entry committed AND the own-term gate,
|
||||
* when set, reached) is now satisfied, and resolve each with COMMITTED, rc=0
|
||||
* -- the async-registration mirror of what a woken COMMIT_MODE_WAIT caller's
|
||||
* own re-check observes. Caller must NOT hold ce->lock: this takes the lock
|
||||
* itself for a bounded snapshot scan, then resolves each candidate from an
|
||||
* unlocked context via the self-locking commit_resolve() (satisfying the
|
||||
* dispatch-after-unlock ordering rule). Safe under races: a snapshotted lsn
|
||||
* already resolved by a concurrent caller by the time commit_resolve() runs
|
||||
* is simply a no-op (idempotent), so no continuation is ever double-resolved. */
|
||||
static void commit_resolve_pending(commit_engine_t* ce)
|
||||
{
|
||||
int n = 0;
|
||||
|
||||
if (ce == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
uint64_t ready[ce->n_continuations]; /* VLA: bounded by n_vcpus -- see the
|
||||
* commit.h field-comment sizing
|
||||
* invariant */
|
||||
|
||||
pthread_mutex_lock(&ce->lock);
|
||||
for (int i = 0; i < ce->n_continuations; i++) {
|
||||
uint64_t lsn = ce->continuations[i].lsn;
|
||||
|
||||
if (lsn != 0 && wait_done_locked(ce, lsn)) {
|
||||
ready[n++] = lsn;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
for (int i = 0; i < n; i++) {
|
||||
commit_resolve(ce, ready[i], COMMIT_OUTCOME_COMMITTED, 0);
|
||||
}
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* ACK and membership events
|
||||
* ============================================================ */
|
||||
|
|
@ -538,6 +727,10 @@ void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn)
|
|||
pthread_mutex_lock(&ce->lock);
|
||||
commit_advance_locked(ce);
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: an ACK-driven advance can newly satisfy
|
||||
* a registered continuation -- resolve after ce->lock is released. */
|
||||
commit_resolve_pending(ce);
|
||||
}
|
||||
|
||||
void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
|
||||
|
|
@ -552,6 +745,10 @@ void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
|
|||
* committed_lsn >= writable_lsn). */
|
||||
pthread_cond_broadcast(&ce->cond);
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: the gate opening can newly satisfy a
|
||||
* registered continuation whose own entry already committed. */
|
||||
commit_resolve_pending(ce);
|
||||
}
|
||||
|
||||
/* ha-term-safety commit-ceiling-optA: record the raft library's commit_index
|
||||
|
|
@ -588,6 +785,9 @@ void commit_set_raft_commit_index(commit_engine_t* ce, uint64_t commit_index)
|
|||
commit_advance_locked(ce);
|
||||
pthread_cond_broadcast(&ce->cond);
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: resolve after ce->lock is released. */
|
||||
commit_resolve_pending(ce);
|
||||
}
|
||||
|
||||
void commit_on_membership_change(commit_engine_t* ce)
|
||||
|
|
@ -602,4 +802,7 @@ void commit_on_membership_change(commit_engine_t* ce)
|
|||
commit_advance_locked(ce);
|
||||
pthread_cond_broadcast(&ce->cond);
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: resolve after ce->lock is released. */
|
||||
commit_resolve_pending(ce);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -127,6 +127,28 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
|
|||
int commit_submit_async(commit_engine_t* ce, uint64_t lsn,
|
||||
commit_continuation_fn fn, void* ctx);
|
||||
|
||||
/* raft-apply-decoupling task 1.2: resolve lsn's registered continuation (if
|
||||
* any) with the given outcome/rc, exactly once. Called internally on the
|
||||
* majority-commit path (commit_advance_locked and its callers, via the
|
||||
* wait_done_locked predicate -- mirrors COMMIT_MODE_WAIT's own completion
|
||||
* check) and is also the PUBLIC failure-path hook: a future caller detecting
|
||||
* timeout, demotion, or log-tail truncation for a pending lsn calls this
|
||||
* directly with COMMIT_OUTCOME_FAILED and the appropriate rc (the existing
|
||||
* error-code vocabulary -- e.g. -ETIMEDOUT, -XCE_ERR_STANDBY). Idempotent: a
|
||||
* second resolve() for an lsn with no registered (or already-resolved)
|
||||
* continuation is a silent no-op -- lsn == 0 is the free-slot sentinel, so a
|
||||
* cleared slot never matches again. Never invokes the continuation inline on
|
||||
* the calling thread: dispatch happens off-thread via vtask_enq_classified
|
||||
* (VTASK_LIVE_BLOCK -- never admission-gated, so delivery cannot fast-reject
|
||||
* the way VTASK_REQ_BLOCK could at the pool's blocking cap), falling back to
|
||||
* an inline call ONLY if dispatch is unavailable (OOM, or no vtaskmgr pool
|
||||
* wired -- commit-only unit fixtures), because a continuation must never be
|
||||
* silently left unresolved (Req 9.4; "controllers never self-terminate" —
|
||||
* fail loud and complete the work rather than drop it). Safe to call from any
|
||||
* thread that does NOT already hold ce->lock. */
|
||||
void commit_resolve(commit_engine_t* ce, uint64_t lsn, commit_outcome_t outcome,
|
||||
int rc);
|
||||
|
||||
/* 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);
|
||||
|
|
|
|||
|
|
@ -1,17 +1,15 @@
|
|||
/*
|
||||
* test_commit_resolve.c — raft-apply-decoupling task 1.1.
|
||||
* test_commit_resolve.c — raft-apply-decoupling tasks 1.1 + 1.2.
|
||||
*
|
||||
* Keystone conversion, registration half: the commit engine gains
|
||||
* 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. 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.
|
||||
* 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:
|
||||
* 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.
|
||||
|
|
@ -24,7 +22,36 @@
|
|||
* same inflight_waits pool.
|
||||
* (5) Distinct concurrent registrations occupy distinct per-LSN slots.
|
||||
*
|
||||
* Requirements: 1.3, 1.4, 9.1, 9.2
|
||||
* 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
|
||||
*/
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
|
|
@ -40,6 +67,7 @@
|
|||
|
||||
#include "commit.h"
|
||||
#include "repl.h"
|
||||
#include "vtaskmgr.h"
|
||||
|
||||
/* ── Fixture ──────────────────────────────────────────────────────────────
|
||||
* Static: persist_t embeds large path buffers, and statics survive Unity's
|
||||
|
|
@ -77,15 +105,66 @@ static void resolve_teardown(void)
|
|||
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
|
||||
* 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. */
|
||||
* 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;
|
||||
} continuation_recorder_t;
|
||||
|
||||
static void recorder_fn(void* ctx, uint64_t lsn, commit_outcome_t outcome,
|
||||
|
|
@ -97,6 +176,7 @@ static void recorder_fn(void* ctx, uint64_t lsn, commit_outcome_t outcome,
|
|||
r->last_outcome = outcome;
|
||||
r->last_rc = rc;
|
||||
r->last_ctx = ctx;
|
||||
r->last_thread = pthread_self();
|
||||
atomic_fetch_add(&r->calls, 1);
|
||||
}
|
||||
|
||||
|
|
@ -301,6 +381,214 @@ static void test_commit_submit_async_tracks_distinct_lsns(void)
|
|||
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");
|
||||
}
|
||||
|
||||
void test_commit_resolve(void)
|
||||
{
|
||||
RUN_TEST(test_commit_submit_async_returns_immediately_before_commit);
|
||||
|
|
@ -308,4 +596,11 @@ void test_commit_resolve(void)
|
|||
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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue