wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/controller/testsuite/test_commit_engine.c

1634 lines
68 KiB
C
Raw Blame History

/*
* test_commit_engine.c — RED-phase tests for the commit engine
* (ha-commit-engine task 1.1: per-LSN majority evaluation and
* commit-index integrity).
*
* The engine is constructed standalone against a stub persist_t +
* replicator_t (XCE_TESTING pattern): the persist_t is a zeroed stack
* struct whose only live parts are committed_lsn, current_lsn, and the
* embedded replicator registry. Standby ACK watermarks are written
* directly into standby_conn_t.acked_lsn before feeding commit_on_ack,
* exactly as repl_on_ack does in production.
*
* Cluster model: n_peers = 3 means primary (self) + 2 standbys.
* Majority for entry N = self + at least n_peers/2 standbys with
* acked_lsn >= N (Raft §5.3, umbrella-verbatim commit rule).
*
* These cases FAIL against the task-1.1 stub commit.c and go green with
* the task-1.2 implementation.
*
* Task 2.1 extends the file with COMMIT_MODE_WAIT blocking-wait semantics,
* admission control, and own-term-gate cases. They FAIL (RED) against the
* task-1 engine (WAIT is -ENOSYS) and go green with task 2.2. Contract:
* 0 — entry committed (majority ACK, gate satisfied)
* -ETIMEDOUT — predicate unmet at deadline (incl. a gate that never
* opens: the own-term gate is part of the WAIT
* completion predicate, NOT an admission fast-reject)
* -XCE_ERR_NOT_READY — admission cap exhausted (the ONLY cause);
* cap = max(1, n_vcpus - COMMIT_WAIT_RESERVED)
* Blocking waits run on pthreads; ACKs are fed from the main thread via the
* registry + commit_on_ack. All threads are joined BEFORE any assertion so
* Unity's longjmp never strands a live thread (ASAN/LSan clean).
*
* Task 3.1 extends the file with membership re-evaluation cases
* (commit_on_membership_change): catch-up exclusion against a pending
* waiter, catch-up completion waking a pending waiter, standby removal
* below majority dying at the waiter's own deadline (no forced early-fail —
* design §Commit Engine), and the non-blocking contract of
* commit_on_membership_change itself.
*
* Task 4.1 extends the file with write-admission quorum accounting:
* repl_active_standby_count (registry truth: ep != NULL && !in_catch_up)
* and ha_dispatch_gate write-quorum behavior. RED against the inert repl.c
* stub (returns -1); green with the task-4.2 implementation.
*
* Task 5.1 extends the file with integration tests for the ACK feed and the
* job-submit sync-append path: repl_on_ack with a wired engine
* (repl_set_commit_engine) updates acked_lsn AND triggers commit evaluation;
* the four membership mutators (repl_add_standby / repl_remove_standby /
* repl_start_catchup / repl_complete_catchup) trigger re-evaluation; and
* persist_wal_sync_append over a real persist_t (tmpdir fixture, pattern
* from test_p2_integration.c) honors the POST-wiring contract: 0 only on
* per-entry majority, -ETIMEDOUT on stale ACKs covering only an earlier
* entry, committed_lsn unmoved past the entry on timeout, and n_peers=1
* returning 0 with committed_lsn advanced (replaces -ENODEV degraded path).
* RED against the inert repl_set_commit_engine stub and the unwired
* production sync-append path; green with tasks 5.2 + 5.3.
*
* Requirements: 1.1, 1.2, 1.3, 2.1, 2.2, 2.3, 3.1, 3.2, 3.3, 4.1, 4.2, 4.3,
* 4.4, 5.1, 6.1, 6.2, 6.3, 7.1, 7.2, 7.3
*/
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "xce/errnox.h"
#include "commit.h"
#include "controller.capnp.h" /* OPCode_jobsub (task 4.1 gate cases) */
#include "ha.h" /* ha_state_t, ha_dispatch_gate fixture */
#include "persist.h"
#include "repl.h"
#include "typemap.h" /* wal_catchup_entry_t (task 5.1 send hook) */
#include "wal_types.h" /* WAL_JOB_SUBMIT (task 5.1 sync-append cases) */
/* ha_dispatch_gate — defined in ha.c (same pattern as test_ha_repl_ack.c) */
int ha_dispatch_gate(ha_state_t *ha, int opcode);
/* ── Fixture ──────────────────────────────────────────────────────────────
* Static: persist_t embeds large path buffers, and statics survive Unity's
* longjmp on assertion failure (pattern from test_repl_majority.c). */
static persist_t g_p;
static commit_engine_t g_ce;
/* Dummy endpoints — the replicator only compares pointers */
static int g_eps[MAX_STANDBYS];
/*
* Arrange a fresh engine over a stub persist_t + replicator_t.
* appended_lsn = highest locally appended WAL entry; persist invariant is
* committed_lsn <= current_lsn - 1, so current_lsn = appended_lsn + 1.
* n_vcpus drives the WAIT admission cap: max(1, n_vcpus - COMMIT_WAIT_RESERVED).
*/
static void engine_setup_vcpus(int n_peers, int n_standbys, uint64_t appended_lsn,
int n_vcpus)
{
int i;
char name[16];
memset(&g_p, 0, sizeof(g_p));
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_p.repl));
for (i = 0; i < n_standbys; i++) {
snprintf(name, sizeof(name), "sb%d", i);
/* repl_add_standby returns the slot index (>= 0) on success */
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);
/* ha-commit-durability 2.2: the commit majority now caps self at
* durable_lsn, not the appended tip. By default treat every appended
* entry as already fdatasync'd locally (durable_lsn == appended tip), so
* the pre-durability test semantics ("self covers every appended entry")
* are preserved. Tests that exercise the self-cap lower durable_lsn
* explicitly via set_durable() after setup. */
atomic_store(&g_p.durable_lsn, appended_lsn);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, n_vcpus));
}
/* Set the local durable watermark (highest LSN fdatasync'd locally). Mirrors
* what the persist group-commit flusher does after a successful fdatasync;
* here it is driven directly so the commit-engine self-cap can be tested in
* isolation from real disk I/O. */
static void set_durable(uint64_t durable_lsn)
{
atomic_store(&g_p.durable_lsn, durable_lsn);
}
/* Drive the registered durable-advance callback exactly as the persist flusher
* does after advancing durable_lsn and releasing p->lock: bump durable_lsn,
* then invoke the hook so the commit engine re-evaluates + wakes WAIT callers.
* Asserts a callback is actually registered (it is, from commit_engine_init). */
static void advance_durable_and_wake(uint64_t durable_lsn)
{
atomic_store(&g_p.durable_lsn, durable_lsn);
TEST_ASSERT_NOT_NULL(g_p.on_durable_advance);
g_p.on_durable_advance(g_p.durable_cb_ctx);
}
static void engine_setup(int n_peers, int n_standbys, uint64_t appended_lsn)
{
engine_setup_vcpus(n_peers, n_standbys, appended_lsn, 8);
}
static void engine_teardown(void)
{
commit_engine_fini(&g_ce);
repl_fini(&g_p.repl);
}
/* Register appended entries 1..up_to with the engine (RETAIN = bookkeeping
* only, no blocking). Return codes asserted in the dedicated RETAIN test. */
static void append_entries(uint64_t up_to)
{
uint64_t lsn;
for (lsn = 1; lsn <= up_to; lsn++) {
(void)commit_submit(&g_ce, lsn, COMMIT_MODE_RETAIN, 0);
}
}
/* Standby ACK watermark arrives: update the registry (what repl_on_ack
* does in production) and feed the engine. */
static void standby_ack(int idx, uint64_t acked_lsn)
{
atomic_store(&g_p.repl.standbys[idx].acked_lsn, acked_lsn);
commit_on_ack(&g_ce, idx, acked_lsn);
}
static uint64_t committed(void)
{
return atomic_load(&g_p.committed_lsn);
}
static uint64_t mono_ms(void)
{
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
}
/* ── Test 1: per-LSN evaluation — ACK for LSN M never commits N > M ───────
* Kills the `acked_lsn > 0` ("ACKed anything ever") counting. (Req 1.2) */
static void test_ack_for_lower_lsn_does_not_commit_higher_entry(void)
{
/* Arrange: 3-member cluster, entries 1..3 appended */
engine_setup(3, 2, 3);
append_entries(3);
/* Act: standby 0 ACKs only LSN M=1 */
standby_ack(0, 1);
/* Assert: entry N=3 is NOT committed (dishonest counting would have
* counted "ACKed anything" and committed 3); entry 1 IS committed
* (self + standby 0 cover it) */
TEST_ASSERT_TRUE_MESSAGE(committed() < 3,
"ACK for LSN 1 must not commit entry 3");
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_ack_for_lower_lsn_does_not_commit_higher_entry: PASS\n");
}
/* ── Test 2: ACK watermark >= N from one standby plus self commits N ──────
* (Req 1.2, 2.1) */
static void test_ack_at_entry_plus_self_commits_entry(void)
{
/* Arrange: 3-member cluster, entries 1..2 appended */
engine_setup(3, 2, 2);
append_entries(2);
/* Act: standby 0's watermark covers entry 2 (self + sb0 = 2/3) */
standby_ack(0, 2);
/* Assert: entry 2 committed */
TEST_ASSERT_EQUAL_UINT64(2, committed());
engine_teardown();
printf("test_ack_at_entry_plus_self_commits_entry: PASS\n");
}
/* ── Test 3: cumulative / out-of-order watermarks → majority median ───────
* committed_lsn always advances to the highest LSN covered by self + a
* majority of standby watermarks, contiguously. (Req 2.1, 2.3) */
static void test_out_of_order_watermarks_advance_to_majority_median(void)
{
/* Arrange: 3-member cluster, entries 1..5 appended */
engine_setup(3, 2, 5);
append_entries(5);
/* Act + Assert, step by step (cumulative watermarks, mixed order) */
/* sb1 ACKs watermark 2 → matchIndex {self=5, sb0=0, sb1=2}: median 2 */
standby_ack(1, 2);
TEST_ASSERT_EQUAL_UINT64(2, committed());
/* sb0 ACKs watermark 4 → {self=5, sb0=4, sb1=2}: median 4 — entry 5
* still lacks majority and must NOT be committed */
standby_ack(0, 4);
TEST_ASSERT_EQUAL_UINT64(4, committed());
/* sb0 ACKs watermark 5 → {self=5, sb0=5, sb1=2}: median 5 */
standby_ack(0, 5);
TEST_ASSERT_EQUAL_UINT64(5, committed());
engine_teardown();
printf("test_out_of_order_watermarks_advance_to_majority_median: PASS\n");
}
/* ── Test 4: committed_lsn never exceeds the highest appended LSN ─────────
* (Req 2.3) */
static void test_committed_lsn_never_exceeds_appended_log(void)
{
/* Arrange: 3-member cluster, only entries 1..3 appended locally */
engine_setup(3, 2, 3);
append_entries(3);
/* Act: both standbys report watermarks beyond our appended log */
standby_ack(0, 10);
standby_ack(1, 10);
/* Assert: clamped to the highest appended LSN */
TEST_ASSERT_EQUAL_UINT64(3, committed());
engine_teardown();
printf("test_committed_lsn_never_exceeds_appended_log: PASS\n");
}
/* ── Test 5: committed_lsn is monotonic — never moves backwards ───────────
* (Req 2.2) */
static void test_committed_lsn_never_regresses(void)
{
/* Arrange: 3-member cluster, entries 1..4 appended; commit up to 4 */
engine_setup(3, 2, 4);
append_entries(4);
standby_ack(0, 4);
TEST_ASSERT_EQUAL_UINT64(4, committed());
/* Act: a regressing watermark arrives (stale/duplicate ACK) — the
* majority median over current registry state would now be 1 */
standby_ack(0, 1);
/* Assert: commit index never decreases */
TEST_ASSERT_EQUAL_UINT64(4, committed());
engine_teardown();
printf("test_committed_lsn_never_regresses: PASS\n");
}
/* ── Test 6: one far-lagging standby does not block majority commit ───────
* (Req 5.1) */
static void test_lagging_standby_does_not_block_majority_commit(void)
{
/* Arrange: 3-member cluster, entries 1..3 appended; standby 1 is far
* behind and never ACKs (watermark stays 0) */
engine_setup(3, 2, 3);
append_entries(3);
/* Act: standby 0 alone covers entry 3 — self + sb0 = majority */
standby_ack(0, 3);
/* Assert: full replication is not a confirmation precondition */
TEST_ASSERT_EQUAL_UINT64(3, committed());
engine_teardown();
printf("test_lagging_standby_does_not_block_majority_commit: PASS\n");
}
/* ── Test 7: RETAIN mode returns 0 immediately without blocking ───────────
* (interface contract; commit happens later via ACKs, Req 2.1) */
static void test_retain_mode_returns_zero_immediately(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: 3-member cluster, entry 1 appended, no ACKs outstanding */
engine_setup(3, 2, 1);
/* Act: RETAIN submit with a large timeout that must NOT be honored */
t0 = mono_ms();
rc = commit_submit(&g_ce, 1, COMMIT_MODE_RETAIN, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: immediate success, no blocking, and no commit without ACKs */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"RETAIN submit must not block on the timeout");
TEST_ASSERT_EQUAL_UINT64(0, committed());
engine_teardown();
printf("test_retain_mode_returns_zero_immediately: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* ha-commit-durability task 2.2 — self counts toward the commit majority only
* when durable. The PRIMARY's self-contribution is capped at durable_lsn (the
* highest LSN fdatasync'd locally), NOT the appended tip. (Req 1.2, 1.4)
* ════════════════════════════════════════════════════════════════════════ */
/* ── Test 2.2a: self-cap equals durable_lsn, not the appended tip ─────────
* Appended tip is 5 but only LSN 3 is durable locally. A standby ACKs the
* full tip (5). committed_lsn must stop at 3 — self only covers up to its
* durable watermark, so entries 4,5 are NOT committed despite the standby
* majority covering them. Kills the highest_appended_lsn self-cap. */
static void test_self_cap_equals_durable_not_appended_tip(void)
{
/* Arrange: 3-member cluster, entries 1..5 appended, but durable only to 3 */
engine_setup(3, 2, 5);
append_entries(5);
set_durable(3);
/* Act: standby 0 ACKs the full appended tip (5) — self + sb0 majority
* covers 5, but self is only durable to 3. */
standby_ack(0, 5);
/* Assert: committed caps at the durable self-contribution (3), NOT 5. */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
3, committed(),
"self contribution must cap at durable_lsn, not the appended tip");
engine_teardown();
printf("test_self_cap_equals_durable_not_appended_tip: PASS\n");
}
/* ── Test 2.2b: committed_lsn holds over an entry until it is durable ──────
* The core observable (Req 1.2/1.4): with a full standby majority already
* present, committed_lsn does NOT advance over entry 5 until entry 5 is
* durable locally. As durable_lsn steps up, committed tracks it — never
* passing the durable watermark. */
static void test_committed_holds_until_durable_then_advances(void)
{
/* Arrange: 3-member cluster, entries 1..5 appended; nothing durable yet. */
engine_setup(3, 2, 5);
append_entries(5);
set_durable(0);
/* A full standby majority covers the entire tip up front. */
standby_ack(0, 5);
standby_ack(1, 5);
/* Assert: with self durable to 0, NOTHING commits despite the majority. */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
0, committed(),
"no entry may commit while self is not durable for it");
/* Durable advances to 2 → committed may advance to 2 (self+majority). */
set_durable(2);
commit_on_ack(&g_ce, 0, 5); /* re-evaluate (an ACK re-fires evaluation) */
TEST_ASSERT_EQUAL_UINT64(2, committed());
/* Durable advances to 5 → committed reaches the tip. */
set_durable(5);
commit_on_ack(&g_ce, 0, 5);
TEST_ASSERT_EQUAL_UINT64(5, committed());
engine_teardown();
printf("test_committed_holds_until_durable_then_advances: PASS\n");
}
/* ── Test 2.2c: single-member self-cap — standalone still gated on durable ─
* n_peers=1 self-quorum: committed advances on self alone, but ONLY up to
* durable_lsn. An appended-but-unsynced entry is not committable even with
* no peers. (Req 1.2, 6.3) */
static void test_single_member_self_cap_gated_on_durable(void)
{
/* Arrange: single-member, entries 1..3 appended, durable only to 1. */
engine_setup(1, 0, 3);
append_entries(3);
set_durable(1);
/* Act: drive an evaluation (membership change re-evaluates). */
commit_on_membership_change(&g_ce);
/* Assert: self-quorum commits only the durable prefix (1), not the tip. */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
1, committed(),
"single-member commit must cap at durable_lsn");
/* Durable catches up → committed reaches the tip. */
set_durable(3);
commit_on_membership_change(&g_ce);
TEST_ASSERT_EQUAL_UINT64(3, committed());
engine_teardown();
printf("test_single_member_self_cap_gated_on_durable: PASS\n");
}
/* NOTE: test_durable_advance_wake_completes_blocked_wait (the liveness case
* for the durable-advance wake) is defined AFTER the waiter-thread helpers
* (spawn_waiters/poll_inflight/join_waiters), further down this file. */
/* ════════════════════════════════════════════════════════════════════════
* Task 2.1 — COMMIT_MODE_WAIT blocking semantics, admission control,
* own-term gate. RED against the task-1 engine (WAIT = -ENOSYS).
* ════════════════════════════════════════════════════════════════════════ */
/* Per-waiter pthread context. Static array: survives Unity's longjmp. */
typedef struct {
uint64_t lsn;
uint32_t timeout_ms;
int rc;
uint64_t elapsed_ms;
} wait_arg_t;
#define MAX_WAITERS 7
static wait_arg_t g_wargs[MAX_WAITERS];
static pthread_t g_wtids[MAX_WAITERS];
/* Count of wait threads that have RETURNED from commit_submit — lets polls
* exit fast in the RED phase, where WAIT returns -ENOSYS immediately. */
static atomic_int g_waiters_done;
static void* wait_thread_fn(void* p)
{
wait_arg_t* a = (wait_arg_t*)p;
uint64_t t0 = mono_ms();
a->rc = commit_submit(&g_ce, a->lsn, COMMIT_MODE_WAIT, a->timeout_ms);
a->elapsed_ms = mono_ms() - t0;
atomic_fetch_add(&g_waiters_done, 1);
return NULL;
}
static void spawn_waiters(int n, uint64_t lsn, uint32_t timeout_ms)
{
int i;
atomic_store(&g_waiters_done, 0);
for (i = 0; i < n; i++) {
g_wargs[i].lsn = lsn;
g_wargs[i].timeout_ms = timeout_ms;
g_wargs[i].rc = -9999;
g_wargs[i].elapsed_ms = 0;
TEST_ASSERT_EQUAL_INT(
0, pthread_create(&g_wtids[i], NULL, wait_thread_fn, &g_wargs[i]));
}
}
/* Join ALL spawned threads before any assertion: a failing TEST_ASSERT
* longjmps out of the test, and a still-running waiter would then trip
* ASAN/LSan and corrupt the static fixture for the next case. */
static void join_waiters(int n)
{
int i;
for (i = 0; i < n; i++) {
pthread_join(g_wtids[i], NULL);
}
}
static void sleep_ms(uint32_t ms)
{
struct timespec ts;
ts.tv_sec = ms / 1000u;
ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
nanosleep(&ts, NULL);
}
static int inflight_waits(void)
{
int v;
pthread_mutex_lock(&g_ce.lock);
v = g_ce.inflight_waits;
pthread_mutex_unlock(&g_ce.lock);
return v;
}
/* Poll until `want` waiters are blocked inside the engine, all `n_spawned`
* waiters have already returned (RED fast path), or the deadline expires.
* Returns the last observed inflight count. */
static int poll_inflight(int want, int n_spawned, uint32_t deadline_ms)
{
uint64_t t0 = mono_ms();
int v = inflight_waits();
while (v < want && atomic_load(&g_waiters_done) < n_spawned &&
mono_ms() - t0 < deadline_ms) {
sleep_ms(5);
v = inflight_waits();
}
return v;
}
/* ── Test 2.2d: durable-advance wake completes a blocked WAIT promptly ─────
* Liveness for change (B): a COMMIT_MODE_WAIT caller whose standby majority
* already covers its entry, but whose entry is not yet durable locally, must
* complete PROMPTLY when the flusher advances durable_lsn and fires the
* durable-advance callback — NOT only when its timeout expires. The standby
* ACK arrives BEFORE the durable advance, so commit_on_ack alone cannot
* complete it; the durable-advance wake must. (Req 1.2, 1.4) */
static void test_durable_advance_wake_completes_blocked_wait(void)
{
uint64_t t0, elapsed_ms;
/* Arrange: 3-member cluster, entry 1 appended, NOT durable yet. */
engine_setup(3, 2, 1);
append_entries(1);
set_durable(0);
/* Full standby majority ACKs entry 1 up front — but self is not durable,
* so this ACK does NOT commit it (committed holds at 0). */
standby_ack(0, 1);
TEST_ASSERT_EQUAL_UINT64(0, committed());
/* A WAIT caller blocks on entry 1 with a long timeout (5s). It cannot
* complete on the ACK (already delivered) — only the durable-advance wake
* can release it before the timeout. */
spawn_waiters(1, 1, 5000);
(void)poll_inflight(1, 1, 2000); /* ensure the waiter is actually blocked */
/* Act: the "flusher" advances durable_lsn to 1 and fires the wake. */
t0 = mono_ms();
advance_durable_and_wake(1);
join_waiters(1);
elapsed_ms = mono_ms() - t0;
/* Assert: the waiter completed with success, PROMPTLY (well under the 5s
* timeout) — proving the durable-advance wake, not the deadline. */
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_wargs[0].rc,
"WAIT must complete on durable-advance, not time out");
TEST_ASSERT_EQUAL_UINT64(1, committed());
TEST_ASSERT_TRUE_MESSAGE(
elapsed_ms < 2000,
"durable-advance wake must complete the wait promptly, not on timeout");
engine_teardown();
printf("test_durable_advance_wake_completes_blocked_wait: PASS\n");
}
/* ── Test 8: WAIT blocks until majority ACK, then returns 0 ───────────────
* (Req 1.1) */
static void test_wait_blocks_until_majority_ack_then_returns_zero(void)
{
/* Arrange: 3-member cluster, entry 1 appended, no ACKs yet */
engine_setup(3, 2, 1);
spawn_waiters(1, 1, 2000);
/* Act: hold the waiter for >= 120 ms, then a standby ACK covers it */
sleep_ms(120);
standby_ack(0, 1);
join_waiters(1);
/* Assert: success only after the ACK — the wait really blocked */
TEST_ASSERT_EQUAL_INT(0, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 100,
"WAIT must block until the majority ACK arrives");
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_wait_blocks_until_majority_ack_then_returns_zero: PASS\n");
}
/* ── Test 9: WAIT deadline → -ETIMEDOUT, committed_lsn unmoved past entry ─
* (Req 2.1, 4.4) */
static void test_wait_times_out_with_committed_lsn_unmoved_past_entry(void)
{
/* Arrange: 3-member cluster, entries 1..2 appended; majority covers
* only entry 1 — entry 2 can never commit within the wait */
engine_setup(3, 2, 2);
standby_ack(0, 1);
/* Act: WAIT on entry 2 with a short deadline; no further ACKs */
spawn_waiters(1, 2, 250);
join_waiters(1);
/* Assert: timeout surfaced, deadline honored, commit index never
* advanced past the unacknowledged entry */
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 200,
"WAIT must hold until its deadline");
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_wait_times_out_with_committed_lsn_unmoved_past_entry: PASS\n");
}
/* ── Test 10: n_peers=1 → WAIT returns 0 immediately, committed advances ──
* Standalone: self-ACK constitutes majority. (Req 6.3) */
static void test_wait_single_member_returns_immediately_and_advances(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: single-member cluster, entry 1 appended */
engine_setup(1, 0, 1);
/* Act: WAIT with a large timeout that must NOT be consumed */
t0 = mono_ms();
rc = commit_submit(&g_ce, 1, COMMIT_MODE_WAIT, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: immediate local confirmation + committed_lsn advance */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"single-member WAIT must not block");
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_wait_single_member_returns_immediately_and_advances: PASS\n");
}
/* ── Test 11: n_peers=3, ZERO standby ACKs → -ETIMEDOUT, never success ────
* Regression for the zero-active-standbys success short-circuit.
* (Req 4.3, 4.4) */
static void test_wait_zero_standby_acks_never_succeeds(void)
{
/* Arrange: 3-member cluster, entry 1 appended, standbys never ACK */
engine_setup(3, 2, 1);
/* Act */
spawn_waiters(1, 1, 250);
join_waiters(1);
/* Assert: failure result — zero peer ACKs are never a majority */
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].rc != 0,
"zero standby ACKs must never be confirmed");
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, g_wargs[0].rc);
TEST_ASSERT_EQUAL_UINT64(0, committed());
engine_teardown();
printf("test_wait_zero_standby_acks_never_succeeds: PASS\n");
}
/* ── Test 12: admission cap — 7th concurrent WAIT rejected, 6 admitted ────
* n_vcpus=8, COMMIT_WAIT_RESERVED=2 → cap = 6. The 7th gets
* -XCE_ERR_NOT_READY fast; the 6 admitted proceed; while the cap is
* saturated, commit_on_ack still advances committed_lsn and completes the
* admitted waiters. (Req 7.1, 7.2, 7.3) */
static void test_wait_admission_cap_rejects_seventh_concurrent_wait(void)
{
int i, rc7, observed_inflight;
uint64_t t0, elapsed7_ms;
/* Arrange: 3-member cluster, entry 1 appended, cap = max(1, 8-2) = 6 */
engine_setup_vcpus(3, 2, 1, 8);
spawn_waiters(6, 1, 5000);
observed_inflight = poll_inflight(6, 6, 2000);
/* Act: 7th concurrent WAIT from this thread while 6 are blocked */
t0 = mono_ms();
rc7 = commit_submit(&g_ce, 1, COMMIT_MODE_WAIT, 5000);
elapsed7_ms = mono_ms() - t0;
/* ACK arrives while the cap is still saturated (the 6 are blocked) */
standby_ack(0, 1);
join_waiters(6);
/* Assert (after all joins): exactly 6 admitted... */
TEST_ASSERT_EQUAL_INT_MESSAGE(6, observed_inflight,
"all 6 admitted waits must block concurrently");
/* ...7th rejected fast with the retriable admission error... */
TEST_ASSERT_EQUAL_INT(-XCE_ERR_NOT_READY, rc7);
TEST_ASSERT_TRUE_MESSAGE(elapsed7_ms < 500,
"cap rejection must not block");
/* ...ACK processing at saturation advanced committed_lsn (7.3)... */
TEST_ASSERT_EQUAL_UINT64(1, committed());
/* ...and completed every admitted waiter with success */
for (i = 0; i < 6; i++) {
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wargs[i].rc,
"admitted waits must proceed to commit");
}
engine_teardown();
printf("test_wait_admission_cap_rejects_seventh_concurrent_wait: PASS\n");
}
/* ── Test 13: cap floor max(1, n_vcpus - reserved) for tiny n_vcpus ───────
* n_vcpus=2, reserved=2 → cap = max(1, 0) = 1: one wait is ALWAYS
* admitted; the 2nd concurrent one is rejected. (Req 7.1, 7.2) */
static void test_wait_admission_cap_floor_holds_for_tiny_vcpus(void)
{
int rc2;
uint64_t t0, elapsed2_ms;
/* Arrange: 3-member cluster, entry 1 appended, cap = 1 */
engine_setup_vcpus(3, 2, 1, 2);
spawn_waiters(1, 1, 5000);
(void)poll_inflight(1, 1, 2000);
/* Act: 2nd concurrent WAIT while the floor slot is occupied */
t0 = mono_ms();
rc2 = commit_submit(&g_ce, 1, COMMIT_MODE_WAIT, 5000);
elapsed2_ms = mono_ms() - t0;
standby_ack(0, 1);
join_waiters(1);
/* Assert: the floor admits at least one waiter to completion... */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wargs[0].rc,
"cap floor must admit at least one wait");
/* ...and the wait beyond the floor is rejected fast */
TEST_ASSERT_EQUAL_INT(-XCE_ERR_NOT_READY, rc2);
TEST_ASSERT_TRUE_MESSAGE(elapsed2_ms < 500, "cap rejection must not block");
engine_teardown();
printf("test_wait_admission_cap_floor_holds_for_tiny_vcpus: PASS\n");
}
/* ── Test 14: writable_lsn unset (0) → no gating ──────────────────────────
* Gate defaults to 0 = no gate: WAIT on an already-committed entry returns
* 0 immediately with no own-term condition applied. (Req 1.1) */
static void test_wait_no_gate_when_writable_lsn_unset(void)
{
int rc;
uint64_t t0, elapsed_ms;
/* Arrange: 3-member cluster, entry 1 already majority-committed;
* commit_set_writable_lsn never called */
engine_setup(3, 2, 1);
standby_ack(0, 1);
TEST_ASSERT_EQUAL_UINT64(1, committed());
/* Act */
t0 = mono_ms();
rc = commit_submit(&g_ce, 1, COMMIT_MODE_WAIT, 5000);
elapsed_ms = mono_ms() - t0;
/* Assert: immediate success — no gate predicate when unset */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"unset writable_lsn must not gate committed waits");
engine_teardown();
printf("test_wait_no_gate_when_writable_lsn_unset: PASS\n");
}
/* ── Test 15: gate G > entry → waiter completes only once committed >= G ──
* The own-term gate is part of the WAIT completion predicate: the waiter
* stays blocked after its OWN entry commits, completing only when
* committed_lsn reaches the gate. (Req 1.1; design §Commit Engine) */
static void test_wait_gate_blocks_until_committed_reaches_gate(void)
{
int done_after_entry_commit;
/* Arrange: 3-member cluster, entries 1..3 appended, gate G=3 > entry 1 */
engine_setup(3, 2, 3);
commit_set_writable_lsn(&g_ce, 3);
spawn_waiters(1, 1, 3000);
/* Act: entry 1 gains majority — gate (3) still unmet */
sleep_ms(50);
standby_ack(0, 1);
sleep_ms(150);
done_after_entry_commit = atomic_load(&g_waiters_done);
/* Gate opens: committed_lsn reaches G */
standby_ack(0, 3);
join_waiters(1);
/* Assert (after join): waiter held while committed_lsn < G... */
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, done_after_entry_commit,
"waiter must not complete while committed_lsn < writable_lsn");
/* ...then completed successfully once committed_lsn >= G */
TEST_ASSERT_EQUAL_INT(0, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 180,
"waiter must block past its own entry's commit");
TEST_ASSERT_EQUAL_UINT64(3, committed());
engine_teardown();
printf("test_wait_gate_blocks_until_committed_reaches_gate: PASS\n");
}
/* ── Test 16: gate never opens → -ETIMEDOUT, NOT -XCE_ERR_NOT_READY ───────
* CORRECTED contract: -XCE_ERR_NOT_READY is admission-cap-only. An unmet
* gate blocks within the timeout and surfaces -ETIMEDOUT if it never
* opens — never a NOT_READY fast-reject. (Req 1.1; design §Commit Engine) */
static void test_wait_gate_never_opening_times_out_not_notready(void)
{
/* Arrange: 3-member cluster, entry 1 committed, gate G=5 unreachable
* (only entry 1 is appended) */
engine_setup(3, 2, 1);
commit_set_writable_lsn(&g_ce, 5);
standby_ack(0, 1);
TEST_ASSERT_EQUAL_UINT64(1, committed());
/* Act */
spawn_waiters(1, 1, 250);
join_waiters(1);
/* Assert: gate is a blocking predicate, not an admission fast-reject */
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].rc != -XCE_ERR_NOT_READY,
"unmet gate must NOT fast-reject with NOT_READY");
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 200,
"gated WAIT must block until its deadline");
engine_teardown();
printf("test_wait_gate_never_opening_times_out_not_notready: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 3.1 — membership re-evaluation (commit_on_membership_change).
* (Req 4.1, 4.2; design §Commit Engine: re-evaluate under the engine mutex
* against current registry state + broadcast; no forced early-fail.)
* ════════════════════════════════════════════════════════════════════════ */
/* ── Test 17: catch-up standby's watermark never counts for a pending entry
* A pending WAIT must not complete off an ACK from a standby still in
* catch-up — it dies at its own deadline with committed_lsn unmoved.
* (Req 4.1) */
static void test_catchup_standby_ack_does_not_count_toward_pending_entry(void)
{
/* Arrange: 3-member cluster, entry 1 appended; the ONLY ACKing standby
* is in catch-up */
engine_setup(3, 2, 1);
atomic_store(&g_p.repl.standbys[0].in_catch_up, 1);
spawn_waiters(1, 1, 250);
/* Act: the catch-up standby's watermark covers the entry */
sleep_ms(50);
standby_ack(0, 1);
join_waiters(1);
/* Assert: never success — catch-up ACKs are excluded from majority */
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].rc != 0,
"catch-up standby ACK must not commit a pending entry");
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 200,
"WAIT must hold to its deadline, not early-complete");
TEST_ASSERT_EQUAL_UINT64(0, committed());
engine_teardown();
printf("test_catchup_standby_ack_does_not_count_toward_pending_entry: PASS\n");
}
/* ── Test 18: clearing in_catch_up + membership change completes the waiter
* The standby's watermark already covers the entry; once it leaves
* catch-up, commit_on_membership_change must re-evaluate against current
* registry state and wake the pending waiter to success. (Req 4.2) */
static void test_catchup_clear_then_membership_change_completes_waiter(void)
{
int done_while_in_catchup;
/* Arrange: 3-member cluster, entry 1 appended; standby 0 in catch-up
* with a watermark that covers the entry (excluded while catching up) */
engine_setup(3, 2, 1);
atomic_store(&g_p.repl.standbys[0].in_catch_up, 1);
standby_ack(0, 1);
TEST_ASSERT_EQUAL_UINT64(0, committed());
spawn_waiters(1, 1, 2000);
/* Act: waiter stays blocked while the standby is in catch-up... */
sleep_ms(120);
done_while_in_catchup = atomic_load(&g_waiters_done);
/* ...then catch-up completes: clear the flag and signal membership */
atomic_store(&g_p.repl.standbys[0].in_catch_up, 0);
commit_on_membership_change(&g_ce);
join_waiters(1);
/* Assert (after join): held during catch-up... */
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, done_while_in_catchup,
"waiter must not complete while the covering standby is in catch-up");
/* ...then woken to success by the membership re-evaluation */
TEST_ASSERT_EQUAL_INT(0, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 100,
"waiter must block until the membership change");
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_catchup_clear_then_membership_change_completes_waiter: PASS\n");
}
/* ── Test 19: removal below majority — pending WAIT dies at ITS deadline ──
* Standby removal that makes majority unreachable must never surface
* success; per the design (no forced early-fail — reconnection within the
* deadline may still commit), the wait fails with -ETIMEDOUT at its own
* deadline, and committed_lsn never regresses. (Req 4.1, 4.2) */
static void test_standby_removal_below_majority_wait_fails_at_deadline(void)
{
/* Arrange: 3-member cluster, entries 1..2 appended; entry 1 committed */
engine_setup(3, 2, 2);
standby_ack(0, 1);
TEST_ASSERT_EQUAL_UINT64(1, committed());
spawn_waiters(1, 2, 400);
/* Act: both standbys drop — majority for entry 2 is now unreachable */
sleep_ms(50);
repl_remove_standby(&g_p.repl, &g_eps[0]);
repl_remove_standby(&g_p.repl, &g_eps[1]);
commit_on_membership_change(&g_ce);
join_waiters(1);
/* Assert: never success... */
TEST_ASSERT_TRUE_MESSAGE(
g_wargs[0].rc != 0,
"WAIT must not succeed after removal makes majority unreachable");
/* ...fails with -ETIMEDOUT at its own deadline (no forced early-fail) */
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, g_wargs[0].rc);
TEST_ASSERT_TRUE_MESSAGE(g_wargs[0].elapsed_ms >= 350,
"removal must not abort the wait before its deadline");
/* ...and the commit index never regresses on membership shrink */
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_standby_removal_below_majority_wait_fails_at_deadline: PASS\n");
}
/* ── Test 20: commit_on_membership_change returns without blocking ────────
* iomgr-callable contract: bounded mutex hold only — it must return
* immediately even while a WAIT is pending (it never waits for waiter
* completion or I/O). (Req 4.2; design §External Dependencies) */
static void test_membership_change_returns_without_blocking(void)
{
uint64_t t0, elapsed_ms;
/* Arrange: 3-member cluster, entry 1 appended, one pending waiter */
engine_setup(3, 2, 1);
spawn_waiters(1, 1, 2000);
sleep_ms(50);
/* Act: membership event with the waiter still blocked */
t0 = mono_ms();
commit_on_membership_change(&g_ce);
elapsed_ms = mono_ms() - t0;
/* Complete the waiter so teardown joins cleanly */
standby_ack(0, 1);
join_waiters(1);
/* Assert (after join): the membership call itself never blocked... */
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 500,
"commit_on_membership_change must not block");
/* ...and the pending waiter still completed normally via the ACK */
TEST_ASSERT_EQUAL_INT(0, g_wargs[0].rc);
TEST_ASSERT_EQUAL_UINT64(1, committed());
engine_teardown();
printf("test_membership_change_returns_without_blocking: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 4.1 — write-admission quorum accounting.
* (Req 3.1, 3.2, 3.3, 6.1, 6.2; design §Quorum Accounting)
*
* repl_active_standby_count is the registry-truth helper ha_has_write_quorum
* adopts in task 4.2: it counts only slots with ep != NULL && !in_catch_up.
* The catch-up exclusion AT THE GATE is asserted through the helper (tests
* 22-25): wiring ha_dispatch_gate to the registry requires the
* ha_state_t.repl back-pointer that task 4.2 adds to ha.h — out of this
* task's boundary. The ha_dispatch_gate cases (tests 26-28) pin the
* standby_endpoints fallback behavior that task 4.2 must preserve when the
* back-pointer is NULL (legacy contexts) and the stateless re-evaluation /
* single-member contracts (6.2, n_peers=1 always true).
* ════════════════════════════════════════════════════════════════════════ */
/* Standalone registry for count tests — static: survives Unity's longjmp */
static replicator_t g_reg;
/* Gate fixture — static for the same reason; standby_endpoints zarray is
* heap-backed per the test_ha_connection_reuse.c pattern */
static ha_state_t g_gate_ha;
static ha_config_t g_gate_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
};
/* Arrange a CTRL_PRIMARY ha_state_t with n_endpoints connected standby
* endpoints (what raft_cb_become_leader / on_standby_connected populate). */
static void gate_setup(int n_peers, int n_endpoints)
{
int i;
TEST_ASSERT_EQUAL_INT(0, ha_init(&g_gate_ha, &g_gate_cfg, 0));
g_gate_ha.n_peers = n_peers;
atomic_store_explicit(&g_gate_ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
g_gate_ha.standby_endpoints = calloc(1, sizeof(zarray_t));
TEST_ASSERT_NOT_NULL(g_gate_ha.standby_endpoints);
zarray_init(g_gate_ha.standby_endpoints, 4, NULL);
for (i = 0; i < n_endpoints; i++) {
zarray_store(g_gate_ha.standby_endpoints, &g_eps[i]);
}
}
static void gate_teardown(void)
{
if (g_gate_ha.standby_endpoints) {
zarray_fini(g_gate_ha.standby_endpoints);
free(g_gate_ha.standby_endpoints);
g_gate_ha.standby_endpoints = NULL;
}
ha_fini(&g_gate_ha);
}
/* ── Test 21: empty / NULL registry counts zero active standbys ───────────
* Disconnected (never-connected) members never count toward write
* admission. (Req 3.3) */
static void test_active_standby_count_empty_registry_is_zero(void)
{
int count_empty, count_null;
/* Arrange: freshly initialized registry, no standbys */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
/* Act */
count_empty = repl_active_standby_count(&g_reg);
count_null = repl_active_standby_count(NULL);
/* Assert: zero connected members → zero active count */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, count_empty,
"empty registry must count zero active standbys");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, count_null,
"NULL registry must count zero, not crash");
repl_fini(&g_reg);
printf("test_active_standby_count_empty_registry_is_zero: PASS\n");
}
/* ── Test 22: count includes ONLY ep != NULL && !in_catch_up slots ────────
* (Req 3.1, 3.2, 3.3) */
static void test_active_standby_count_excludes_catchup_and_disconnected(void)
{
/* Arrange: 3 connected standbys */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
TEST_ASSERT_EQUAL_INT(0, repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000));
TEST_ASSERT_EQUAL_INT(1, repl_add_standby(&g_reg, &g_eps[1], "sb1", 7001));
TEST_ASSERT_EQUAL_INT(2, repl_add_standby(&g_reg, &g_eps[2], "sb2", 7002));
TEST_ASSERT_EQUAL_INT_MESSAGE(3, repl_active_standby_count(&g_reg),
"all connected, none catching up → count 3");
/* Act + Assert: catch-up entry excludes a standby (3.2)... */
repl_start_catchup(&g_reg, &g_eps[1], 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(2, repl_active_standby_count(&g_reg),
"in_catch_up standby must be excluded");
/* ...and disconnect excludes another (3.3); the freed slot keeps its
* disconnect bookkeeping but ep == NULL must not count */
repl_remove_standby(&g_reg, &g_eps[2]);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, repl_active_standby_count(&g_reg),
"disconnected standby must be excluded");
repl_fini(&g_reg);
printf("test_active_standby_count_excludes_catchup_and_disconnected: PASS\n");
}
/* ── Test 23: n_peers=3, only connected standby in catch-up → NO quorum ───
* Gate decision (active >= n_peers/2) must be false: the catch-up standby
* cannot receive new log entries, so the write quorum does not exist.
* This is the case current ha_has_write_quorum gets WRONG — it counts
* standby_endpoints->count (1) including the catch-up member. (Req 3.2, 6.1) */
static void test_active_count_catchup_only_standby_denies_write_quorum(void)
{
int n_peers = 3;
int active;
/* Arrange: one connected standby, in catch-up */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
TEST_ASSERT_EQUAL_INT(0, repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000));
repl_start_catchup(&g_reg, &g_eps[0], 0);
/* Act */
active = repl_active_standby_count(&g_reg);
/* Assert: zero active → quorum (active >= n_peers/2 = 1) unmet */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, active,
"catch-up-only standby must yield zero active standbys");
TEST_ASSERT_TRUE_MESSAGE(active < n_peers / 2,
"write quorum must be DENIED when the only standby is in catch-up");
repl_fini(&g_reg);
printf("test_active_count_catchup_only_standby_denies_write_quorum: PASS\n");
}
/* ── Test 24: n_peers=3, one ACTIVE standby → write quorum met ────────────
* (Req 3.1) */
static void test_active_count_one_active_standby_grants_write_quorum(void)
{
int n_peers = 3;
int active;
/* Arrange: one connected, non-catch-up standby */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
TEST_ASSERT_EQUAL_INT(0, repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000));
/* Act */
active = repl_active_standby_count(&g_reg);
/* Assert: one active standby satisfies active >= n_peers/2 = 1 */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, active,
"one connected active standby must count as 1");
TEST_ASSERT_TRUE_MESSAGE(active >= n_peers / 2,
"write quorum must be GRANTED with one active standby (n_peers=3)");
repl_fini(&g_reg);
printf("test_active_count_one_active_standby_grants_write_quorum: PASS\n");
}
/* ── Test 25: stateless re-evaluation — count recovers with NO reset call ─
* Catch-up completion and reconnect (registry restored) must each restore
* the count purely from registry state; there is no quorum-reset API.
* (Req 6.2) */
static void test_active_count_recovers_after_catchup_and_reconnect(void)
{
/* Arrange: one standby, pushed into catch-up → count 0 */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
TEST_ASSERT_EQUAL_INT(0, repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000));
repl_start_catchup(&g_reg, &g_eps[0], 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, repl_active_standby_count(&g_reg),
"catch-up standby must not count");
/* Act + Assert: catch-up completion restores the count... */
repl_complete_catchup(&g_reg, &g_eps[0]);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, repl_active_standby_count(&g_reg),
"catch-up completion must restore the active count with no reset call");
/* ...disconnect drops it... */
repl_remove_standby(&g_reg, &g_eps[0]);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, repl_active_standby_count(&g_reg),
"disconnect must drop the active count");
/* ...and reconnect (registry restored) recovers it — stateless */
TEST_ASSERT_TRUE_MESSAGE(
repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000) >= 0,
"reconnect re-add must succeed");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, repl_active_standby_count(&g_reg),
"reconnect must restore the active count with no reset call");
repl_fini(&g_reg);
printf("test_active_count_recovers_after_catchup_and_reconnect: PASS\n");
}
/* ── Test 26: n_peers=1 → write quorum ALWAYS true (self is majority) ─────
* Gate-level: single-member primary admits writes with no standbys at all.
* Must stay green before AND after task 4.2 (quorum condition unchanged). */
static void test_gate_single_member_always_admits_writes(void)
{
int rc;
/* Arrange: single-member primary, zero standby endpoints */
gate_setup(1, 0);
/* Act */
rc = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* Assert: self-majority — writes always admitted */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"n_peers=1 primary must always admit state-changing ops");
gate_teardown();
printf("test_gate_single_member_always_admits_writes: PASS\n");
}
/* ── Test 27: n_peers=3, ZERO connected standbys → write rejected ─────────
* Gate-level fast rejection with the retriable not-ready error. (Req 6.1, 3.3) */
static void test_gate_rejects_writes_with_zero_connected_standbys(void)
{
int rc;
/* Arrange: 3-member primary, no connected standbys */
gate_setup(3, 0);
/* Act */
rc = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* Assert: cluster cannot currently commit */
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc,
"zero connected standbys must reject writes with NOT_READY");
gate_teardown();
printf("test_gate_rejects_writes_with_zero_connected_standbys: PASS\n");
}
/* ── Test 28: gate re-accepts after reconnection with NO reset call ───────
* The gate is stateless per request: once the standby endpoint registry is
* restored, the very next dispatch is admitted — no operator intervention,
* no reset API. (Req 6.2) */
static void test_gate_reaccepts_after_reconnect_without_reset(void)
{
int rc_lost, rc_restored;
/* Arrange: 3-member primary that has lost all standbys */
gate_setup(3, 0);
rc_lost = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* Act: a standby reconnects (endpoint registry restored) — nothing else */
zarray_store(g_gate_ha.standby_endpoints, &g_eps[0]);
rc_restored = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* Assert: rejected while below quorum... */
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc_lost,
"writes must be rejected while below quorum");
/* ...admitted immediately after restoration — stateless re-evaluation */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_restored,
"gate must re-accept immediately after reconnect, no reset call");
gate_teardown();
printf("test_gate_reaccepts_after_reconnect_without_reset: PASS\n");
}
/* ── Test 29: WIRED gate — registry truth excludes catch-up at the gate ───
* Task 4.2 (reviewer-mandated RED for the gate half): n_peers=3 primary with
* the ha_state_t.repl back-pointer wired to a registry whose ONLY connected
* standby is in catch-up. standby_endpoints still lists the endpoint (as
* on_standby_connected would), so the legacy count says 1 — but the registry
* truth says 0 active. The gate must reject the write with NOT_READY; once
* catch-up clears, the very next dispatch is admitted (stateless, Req 6.2).
* (Req 3.1, 3.2, 6.1, 6.2) */
static void test_gate_wired_registry_catchup_standby_rejects_then_admits(void)
{
int rc_catchup, rc_cleared;
/* Arrange: 3-member primary; endpoint registered in standby_endpoints
* AND in the replicator registry, where it is marked in catch-up */
gate_setup(3, 1);
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_reg));
TEST_ASSERT_EQUAL_INT(0, repl_add_standby(&g_reg, &g_eps[0], "sb0", 7000));
repl_start_catchup(&g_reg, &g_eps[0], 0);
g_gate_ha.repl = &g_reg; /* task 5.3 wires this at init; set directly here */
/* Act: dispatch while the only standby is catching up... */
rc_catchup = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* ...then catch-up completes — nothing else changes */
repl_complete_catchup(&g_reg, &g_eps[0]);
rc_cleared = ha_dispatch_gate(&g_gate_ha, OPCode_jobsub);
/* Assert: catch-up-only standby → no write quorum at the gate */
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc_catchup,
"wired gate must reject writes when the only standby is in catch-up");
/* ...and clearing catch-up re-admits with no reset call */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_cleared,
"wired gate must admit writes once catch-up completes");
g_gate_ha.repl = NULL;
repl_fini(&g_reg);
gate_teardown();
printf("test_gate_wired_registry_catchup_standby_rejects_then_admits: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 5.1 — integration: ACK feed and the job-submit sync-append path.
* (Req 1.1, 1.2, 1.3, 2.1, 4.2, 4.4, 6.3; design §ACK Feed,
* §Job-Submit Adapter)
*
* Tests 30-31 wire the engine into the stub-fixture replicator via
* repl_set_commit_engine and drive the PRODUCTION entry points
* (repl_on_ack, membership mutators) instead of commit_on_ack directly.
* Tests 32-34 run persist_wal_sync_append against a real persist_t
* (tmpdir fixture, test_p2_integration.c pattern) with an engine wired
* over it, asserting the post-task-5.3 return contract.
* ════════════════════════════════════════════════════════════════════════ */
/* Test-injectable send hook (defined in repl.c under XCE_TESTING) — keeps
* repl_broadcast from dereferencing the dummy endpoints. */
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int ce_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
(void)ep;
(void)entries;
(void)n_entries;
return 0; /* pretend success, drop the data */
}
/* Heap persist over a private tmpdir (test_p2_integration.c pattern).
* Static pointer: survives Unity's longjmp on assertion failure. */
static persist_t *g_sp;
static persist_t *sync_persist_setup(const char *suffix, uint32_t sync_timeout_ms)
{
persist_t *p;
persist_config_t cfg;
char tmpdir[256];
p = calloc(1, sizeof(persist_t));
TEST_ASSERT_NOT_NULL(p);
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = sync_timeout_ms;
snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_ce51_%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
persist_set_workdir(p, tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(p, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(p));
return p;
}
static void sync_persist_teardown(persist_t *p)
{
if (p == NULL) {
return;
}
persist_stop(p);
persist_fini(p);
free(p);
}
/* Background ACK through the PRODUCTION feed (repl_on_ack), not the
* commit_on_ack shortcut — exercises the task-5.2 wiring. */
typedef struct {
replicator_t *r;
void *ep;
uint64_t lsn;
uint32_t delay_ms;
} repl_ack_arg_t;
static repl_ack_arg_t g_rack;
static void *repl_ack_thread_fn(void *arg)
{
repl_ack_arg_t *a = (repl_ack_arg_t *)arg;
if (a->delay_ms > 0) {
sleep_ms(a->delay_ms);
}
repl_on_ack(a->r, a->ep, a->lsn);
return NULL;
}
/* ── Test 30: repl_on_ack with a wired engine feeds commit evaluation ─────
* The production ACK entry point must BOTH update the registry watermark
* AND trigger engine evaluation: majority → committed_lsn advances and the
* pending waiter wakes with success. (Req 1.1, 4.2) */
static void test_repl_on_ack_wired_updates_watermark_and_commits(void)
{
uint64_t watermark;
/* Arrange: 3-member cluster, entry 1 appended; engine wired into the
* replicator the way cluster init will (repl_set_commit_engine) */
engine_setup(3, 2, 1);
repl_set_commit_engine(&g_p.repl, &g_ce);
append_entries(1);
spawn_waiters(1, 1, 1000);
/* Act: standby 0 ACKs entry 1 through the PRODUCTION feed */
sleep_ms(80);
repl_on_ack(&g_p.repl, &g_eps[0], 1);
join_waiters(1);
/* Assert: the registry watermark was updated (pre-existing contract)... */
watermark = atomic_load(&g_p.repl.standbys[0].acked_lsn);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, watermark,
"repl_on_ack must update the standby's acked_lsn");
/* ...AND the wired engine evaluated the majority: commit advanced... */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, committed(),
"wired repl_on_ack must trigger commit evaluation (majority -> advance)");
/* ...and the pending waiter woke with success */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wargs[0].rc,
"waiter must complete once the wired ACK feed reaches majority");
engine_teardown();
printf("test_repl_on_ack_wired_updates_watermark_and_commits: PASS\n");
}
/* ── Test 31: all four membership mutators trigger re-evaluation ──────────
* For each of repl_add_standby / repl_start_catchup / repl_complete_catchup
* / repl_remove_standby: a majority-covering watermark is planted DIRECTLY
* in the registry (no ACK feed), so committed_lsn can only advance if the
* mutator itself triggers commit_on_membership_change re-evaluation.
* (Req 4.2) */
static void test_membership_mutators_trigger_commit_reevaluation(void)
{
/* Arrange: 3-member cluster, entries 1..4 appended, ONE standby (sb0)
* registered and wired; sb0's watermark alone forms majority with self */
engine_setup(3, 1, 4);
repl_set_commit_engine(&g_p.repl, &g_ce);
append_entries(4);
/* Act + Assert per mutator. Plant the watermark silently, confirm no
* advancement happened, then attribute the advance to the mutator. */
/* repl_add_standby (standby connect) re-evaluates */
atomic_store(&g_p.repl.standbys[0].acked_lsn, 1);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, committed(),
"planted watermark alone must not advance committed_lsn");
TEST_ASSERT_TRUE(repl_add_standby(&g_p.repl, &g_eps[1], "sb1", 7001) >= 0);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, committed(),
"repl_add_standby must trigger commit re-evaluation");
/* repl_start_catchup (catch-up entry) re-evaluates */
atomic_store(&g_p.repl.standbys[0].acked_lsn, 2);
TEST_ASSERT_EQUAL_UINT64(1, committed());
repl_start_catchup(&g_p.repl, &g_eps[1], 0);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(2, committed(),
"repl_start_catchup must trigger commit re-evaluation");
/* repl_complete_catchup (catch-up completion) re-evaluates */
atomic_store(&g_p.repl.standbys[0].acked_lsn, 3);
TEST_ASSERT_EQUAL_UINT64(2, committed());
repl_complete_catchup(&g_p.repl, &g_eps[1]);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(3, committed(),
"repl_complete_catchup must trigger commit re-evaluation");
/* repl_remove_standby (disconnect) re-evaluates */
atomic_store(&g_p.repl.standbys[0].acked_lsn, 4);
TEST_ASSERT_EQUAL_UINT64(3, committed());
repl_remove_standby(&g_p.repl, &g_eps[1]);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, committed(),
"repl_remove_standby must trigger commit re-evaluation");
engine_teardown();
printf("test_membership_mutators_trigger_commit_reevaluation: PASS\n");
}
/* ── Test 32: sync append returns 0 on per-entry majority (wired) ─────────
* Contract-pinning positive case: a standby watermark covering THE entry
* commits it — return 0, sync status 0, committed_lsn at the entry.
* (Req 1.1, 1.2) */
static int g_ep_sa, g_ep_sb;
static void test_sync_append_wired_commits_on_per_entry_majority(void)
{
pthread_t th;
int rc, sync_rc;
/* Arrange: real persist, 2 standbys, engine n_peers=3 wired. Under the
* reserved-0 contract a fresh store starts current_lsn=1, so the sync entry
* lands at LSN 1 with no priming append needed (LSN 0 is the reserved
* sentinel — never a real entry). */
g_sp = sync_persist_setup("maj", 2000);
TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sa, "sb_a", 7100) >= 0);
TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sb, "sb_b", 7101) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 3, 8));
repl_set_commit_engine(&g_sp->repl, &g_ce);
repl_broadcast_send_hook = ce_drop_send_hook;
/* Act: standby A's ACK covering the sync entry (LSN 1) arrives mid-wait */
g_rack = (repl_ack_arg_t){ &g_sp->repl, &g_ep_sa, 1, 30 };
TEST_ASSERT_EQUAL_INT(0,
pthread_create(&th, NULL, repl_ack_thread_fn, &g_rack));
sync_rc = -1;
rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc);
pthread_join(th, NULL);
repl_broadcast_send_hook = NULL;
/* Assert: per-entry majority → confirmed, commit index at the entry */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc,
"sync status must be 0 on per-entry majority");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
1, atomic_load(&g_sp->committed_lsn),
"committed_lsn must advance to the majority-acked entry");
commit_engine_fini(&g_ce);
sync_persist_teardown(g_sp);
g_sp = NULL;
printf("test_sync_append_wired_commits_on_per_entry_majority: PASS\n");
}
/* ── Test 33: stale-ACK regression — earlier-entry ACKs never commit ──────
* Both standbys' watermarks cover only an EARLIER entry. The dishonest
* `acked_lsn > 0` counting confirms the new entry off those stale ACKs;
* the engine contract is -ETIMEDOUT with committed_lsn unmoved past the
* entry: connectivity / old ACKs never substitute for per-entry
* acknowledgment. (Req 1.2, 1.3, 2.1, 4.4) */
static void test_sync_append_wired_stale_acks_time_out(void)
{
int rc, sync_rc;
/* Arrange: real persist (short sync timeout), 2 standbys, engine
* n_peers=3 wired; entries 0..1 appended and BOTH standbys ACK
* watermark 1 BEFORE the new sync entry exists */
g_sp = sync_persist_setup("stale", 60);
TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sa, "sb_a", 7110) >= 0);
TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sb, "sb_b", 7111) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 3, 8));
repl_set_commit_engine(&g_sp->repl, &g_ce);
repl_broadcast_send_hook = ce_drop_send_hook;
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "p", 1));
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "q", 1));
repl_on_ack(&g_sp->repl, &g_ep_sa, 1);
repl_on_ack(&g_sp->repl, &g_ep_sb, 1);
/* Act: sync append entry LSN 2 — no ACK for it ever arrives */
sync_rc = 0;
rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc);
repl_broadcast_send_hook = NULL;
/* Assert: stale ACKs (watermark 1 > 0) must NOT confirm entry 2... */
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
"stale earlier-entry ACKs must never confirm a new entry");
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, sync_rc);
/* ...and the commit index never advances past the unacked entry */
TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g_sp->committed_lsn) < 2,
"committed_lsn must stay unmoved past the entry on timeout");
commit_engine_fini(&g_ce);
sync_persist_teardown(g_sp);
g_sp = NULL;
printf("test_sync_append_wired_stale_acks_time_out: PASS\n");
}
/* ── Test 34: n_peers=1 → sync append succeeds with committed advanced ────
* Standalone self-quorum replaces the old `-ENODEV` degraded path: the
* single-member primary confirms locally (sync status 0, NOT -ENODEV) and
* committed_lsn advances to the entry. (Req 6.3) */
static void test_sync_append_wired_single_member_commits_locally(void)
{
int rc, sync_rc;
/* Arrange: real persist, NO standbys, engine n_peers=1 wired. Under the
* reserved-0 contract the sync entry lands at LSN 1 with no priming append
* (fresh store starts current_lsn=1; LSN 0 is the reserved sentinel). */
g_sp = sync_persist_setup("solo", 2000);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 1, 8));
repl_set_commit_engine(&g_sp->repl, &g_ce);
/* Act */
sync_rc = -1;
rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc);
/* Assert: local confirmation, no degraded -ENODEV signal... */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc,
"n_peers=1 must self-commit (0), not report -ENODEV degraded mode");
/* ...and the standalone commit index advances to the entry */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
1, atomic_load(&g_sp->committed_lsn),
"n_peers=1 sync append must advance committed_lsn to the entry");
commit_engine_fini(&g_ce);
sync_persist_teardown(g_sp);
g_sp = NULL;
printf("test_sync_append_wired_single_member_commits_locally: PASS\n");
}
/* ── Test group entry point ──────────────────────────────────────────────── */
void test_commit_engine(void)
{
UNITY_BEGIN();
RUN_TEST(test_ack_for_lower_lsn_does_not_commit_higher_entry);
RUN_TEST(test_ack_at_entry_plus_self_commits_entry);
RUN_TEST(test_out_of_order_watermarks_advance_to_majority_median);
RUN_TEST(test_committed_lsn_never_exceeds_appended_log);
RUN_TEST(test_committed_lsn_never_regresses);
RUN_TEST(test_lagging_standby_does_not_block_majority_commit);
RUN_TEST(test_retain_mode_returns_zero_immediately);
/* ha-commit-durability task 2.2 — self counts toward majority only when durable */
RUN_TEST(test_self_cap_equals_durable_not_appended_tip);
RUN_TEST(test_committed_holds_until_durable_then_advances);
RUN_TEST(test_single_member_self_cap_gated_on_durable);
RUN_TEST(test_durable_advance_wake_completes_blocked_wait);
/* Task 2.1 — WAIT semantics, admission control, own-term gate */
RUN_TEST(test_wait_blocks_until_majority_ack_then_returns_zero);
RUN_TEST(test_wait_times_out_with_committed_lsn_unmoved_past_entry);
RUN_TEST(test_wait_single_member_returns_immediately_and_advances);
RUN_TEST(test_wait_zero_standby_acks_never_succeeds);
RUN_TEST(test_wait_admission_cap_rejects_seventh_concurrent_wait);
RUN_TEST(test_wait_admission_cap_floor_holds_for_tiny_vcpus);
RUN_TEST(test_wait_no_gate_when_writable_lsn_unset);
RUN_TEST(test_wait_gate_blocks_until_committed_reaches_gate);
RUN_TEST(test_wait_gate_never_opening_times_out_not_notready);
/* Task 3.1 — membership re-evaluation */
RUN_TEST(test_catchup_standby_ack_does_not_count_toward_pending_entry);
RUN_TEST(test_catchup_clear_then_membership_change_completes_waiter);
RUN_TEST(test_standby_removal_below_majority_wait_fails_at_deadline);
RUN_TEST(test_membership_change_returns_without_blocking);
/* Task 4.1 — write-admission quorum accounting */
RUN_TEST(test_active_standby_count_empty_registry_is_zero);
RUN_TEST(test_active_standby_count_excludes_catchup_and_disconnected);
RUN_TEST(test_active_count_catchup_only_standby_denies_write_quorum);
RUN_TEST(test_active_count_one_active_standby_grants_write_quorum);
RUN_TEST(test_active_count_recovers_after_catchup_and_reconnect);
RUN_TEST(test_gate_single_member_always_admits_writes);
RUN_TEST(test_gate_rejects_writes_with_zero_connected_standbys);
RUN_TEST(test_gate_reaccepts_after_reconnect_without_reset);
/* Task 4.2 — wired gate: registry truth at ha_dispatch_gate */
RUN_TEST(test_gate_wired_registry_catchup_standby_rejects_then_admits);
/* Task 5.1 — integration: ACK feed + job-submit sync-append path */
RUN_TEST(test_repl_on_ack_wired_updates_watermark_and_commits);
RUN_TEST(test_membership_mutators_trigger_commit_reevaluation);
RUN_TEST(test_sync_append_wired_commits_on_per_entry_majority);
RUN_TEST(test_sync_append_wired_stale_acks_time_out);
RUN_TEST(test_sync_append_wired_single_member_commits_locally);
UNITY_END();
}
View git blame Copy permalink