wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(ha-commit-engine): RED tests for WAIT mode, admission cap, own-term gate (task 2.1)

Browse source 
9 new cases (8-16): block-until-majority, deadline timeout, standalone
fast path, zero-ACK regression, n_vcpus-relative cap with floor, ACK
progress at saturation, gate as blocking predicate (NOT_READY is
admission-cap-only per corrected contract). Main-thread-only assertions
after joins; 16 tests / 9 RED failures; baseline unchanged.
This commit is contained in:
Rongsong Shen 2026-06-11 14:56:31 +00:00
parent eae4767d7d
commit d91f446e9e
4 changed files with 405 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/ha-commit-engine/tasks.md
View file

@ -18,7 +18,7 @@
- _Requirements: 1.2, 2.1, 2.2, 2.3, 5.1_
- [ ] 2. WAIT mode, self-quorum, zero-peer rule, admission cap, own-term gate storage
- [ ] 2.1 Write RED tests for blocking-wait semantics and admission control
- [x] 2.1 Write RED tests for blocking-wait semantics and admission control
- Tests in `test_commit_engine.c`: WAIT blocks until majority ACK then returns 0; WAIT returns `-ETIMEDOUT` at deadline with `committed_lsn` unmoved past the entry; n_peers=1 → WAIT returns 0 immediately and `committed_lsn` advances to the entry; n_peers=3 with zero standby ACKs → `-ETIMEDOUT`, never success (short-circuit regression); with n_vcpus=8 and reserved=2, a 7th concurrent WAIT gets `-XCE_ERR_NOT_READY` while the 6 admitted waits proceed; cap floor `max(1, n_vcpus - reserved)` holds for tiny n_vcpus; while the cap is saturated, `commit_on_ack` still advances `committed_lsn` and completes admitted waiters; `writable_lsn` unset (0) → no gating; `commit_set_writable_lsn(G)` with G > entry → waiter completes only once `committed_lsn >= G`
- Observable completion: new cases FAIL (red) against the task-1 engine
- _Requirements: 1.1, 4.3, 4.4, 6.3, 7.1, 7.2, 7.3_
@ -85,3 +85,4 @@
- (1.1) Full-suite output under default ASAN dies at exit via LeakSanitizer (pre-existing leaks in worker_join/rdhup_integration/start_election_async/liveness groups) without flushing stdout — use ASAN_OPTIONS=detect_leaks=0 for full-run logs or run groups standalone with -t.
- (1.1) repl_add_standby returns the SLOT INDEX, not 0 (bug-410 note for fixture writers).
- (1.1) commit.c is registered ONLY in controller_unit_test_src for now; production target linkage lands with task 5.3.
- (2.1) WAIT/cap/gate tests (8-16) read `g_ce.inflight_waits` under `g_ce.lock` — task 2.2 must keep that field accurate while waiters block. Timing floors have ~20ms slack vs sleeps (reviewer advisory: watch for flake on loaded boxes). Unity assertions are main-thread-only after joins — keep that discipline in future threaded tests.

6
.wolf/anatomy.md
View file

@ -1,6 +1,6 @@
# anatomy.md
> Auto-maintained by OpenWolf. Last scanned: 2026-06-11T14:36:56.598Z
> Auto-maintained by OpenWolf. Last scanned: 2026-06-11T14:56:17.297Z
> Files: 511 tracked | Anatomy hits: 0 | Misses: 0
## ./
@ -262,7 +262,7 @@
- `requirements.md` — Requirements Document (~2715 tok)
- `research.md` — Research Log: ha-commit-engine (~1345 tok)
- `spec.json` (~140 tok)
- `tasks.md` — Implementation Plan: ha-commit-engine (~2947 tok)
- `tasks.md` — Implementation Plan: ha-commit-engine (~3031 tok)
## .kiro/specs/ha-config-bootstrap/
@ -716,4 +716,4 @@
## src/controller/testsuite/
- `controller_test.c` — include <argp.h> (~1323 tok)
- `test_commit_engine.c` — RED tests for per-LSN majority + commit-index integrity (Reqs 1.2, 2.1-2.3, 5.1): 7 cases against stub persist_t+replicator_t (~2459 tok)
- `test_commit_engine.c` — Commit engine tests: tasks 1.x majority/commit-index (green) + task 2.1 WAIT blocking/admission-cap/own-term-gate RED cases (~13200 tok)

7
.wolf/memory.md
View file

@ -873,3 +873,10 @@
| 07:22 | Created src/controller/commit.c | — | ~2108 |
| 2026-06-11 | ha-commit-engine task 1.2: implemented commit engine core (majority median, monotonic clamped committed_lsn, RETAIN) — 7/7 tests green | src/controller/commit.c | success | ~9k |
| 07:36 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~14 |
| 07:41 | Edited src/controller/testsuite/test_commit_engine.c | expanded (+17 lines) | ~339 |
| 07:41 | Edited src/controller/testsuite/test_commit_engine.c | modified engine_setup_vcpus() | ~303 |
| 07:43 | Edited src/controller/testsuite/test_commit_engine.c | modified struct() | ~3781 |
| 07:43 | Edited src/controller/testsuite/test_commit_engine.c | 4→3 lines | ~17 |
| 14:47 | task 2.1 RED tests: 9 WAIT/admission/gate cases added (16 total, 7 pass / 9 RED -ENOSYS) | src/controller/testsuite/test_commit_engine.c | RED verified, ASAN clean | ~9k |
| 07:56 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~20 |
| 07:56 | Edited .kiro/specs/ha-commit-engine/tasks.md | 1→2 lines | ~122 |

396
src/controller/testsuite/test_commit_engine.c
View file

@ -17,9 +17,23 @@
* These cases FAIL against the task-1.1 stub commit.c and go green with
* the task-1.2 implementation.
*
* Requirements: 1.2, 2.1, 2.2, 2.3, 5.1
* 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).
*
* Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 4.3, 4.4, 5.1, 6.3, 7.1, 7.2, 7.3
*/
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
@ -27,6 +41,8 @@
#include <time.h>
#include <unity/unity.h>
#include "xce/errnox.h"
#include "commit.h"
#include "persist.h"
#include "repl.h"
@ -44,8 +60,10 @@ 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(int n_peers, int n_standbys, uint64_t appended_lsn)
static void engine_setup_vcpus(int n_peers, int n_standbys, uint64_t appended_lsn,
int n_vcpus)
{
int i;
char name[16];
@ -61,7 +79,12 @@ static void engine_setup(int n_peers, int n_standbys, uint64_t appended_lsn)
g_p.current_lsn = appended_lsn + 1;
atomic_store(&g_p.committed_lsn, 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, 8));
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, n_vcpus));
}
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)
@ -254,6 +277,363 @@ static void test_retain_mode_returns_zero_immediately(void)
printf("test_retain_mode_returns_zero_immediately: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* 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 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");
}
/* ── Test group entry point ──────────────────────────────────────────────── */
void test_commit_engine(void)
@ -266,5 +646,15 @@ void test_commit_engine(void)
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);
/* 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);
UNITY_END();
}