test(ha-commit-engine): RED tests for per-LSN majority evaluation (task 1.1)

commit.h umbrella-verbatim interface + inert stub commit.c (test target
only); 7 Unity tests covering per-LSN vs acked_lsn>0 counting, majority
median, monotonicity, append clamp, lagging standby, RETAIN immediacy.
All 7 fail RED as intended. Documented 3 pre-existing masked suite
failures (bug-411) and the per-group exit-code masking root cause.

src/controller/commit.c

@@ -0,0 +1,68 @@
+/*
+ * commit.c — STUB for ha-commit-engine task 1.1 (RED phase).
+ *
+ * Behavioral functions are deliberately inert (-ENOSYS / no-op) so the
+ * task 1.1 tests in testsuite/test_commit_engine.c compile, link, and FAIL
+ * on their assertions. init/fini are real enough that tests exercise the
+ * engine's behavioral contract rather than failing at fixture setup.
+ *
+ * The real implementation lands in task 1.2 (per-LSN majority evaluation,
+ * monotonic/contiguous committed_lsn advancement).
+ */
+#include <errno.h>
+
+#include "commit.h"
+
+int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus)
+{
+    if (ce == NULL || p == NULL || n_peers < 0 || n_vcpus < 1) {
+        return -EINVAL;
+    }
+    pthread_mutex_init(&ce->lock, NULL);
+    pthread_cond_init(&ce->cond, NULL);
+    ce->p              = p;
+    ce->n_peers        = n_peers;
+    ce->n_vcpus        = n_vcpus;
+    ce->inflight_waits = 0;
+    ce->writable_lsn   = 0;
+    return 0;
+}
+
+void commit_engine_fini(commit_engine_t* ce)
+{
+    if (ce == NULL) {
+        return;
+    }
+    pthread_cond_destroy(&ce->cond);
+    pthread_mutex_destroy(&ce->lock);
+    ce->p = NULL;
+}
+
+int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
+                  uint32_t timeout_ms)
+{
+    (void)ce;
+    (void)lsn;
+    (void)mode;
+    (void)timeout_ms;
+    return -ENOSYS; /* stub: real WAIT/RETAIN semantics in task 1.2/2.2 */
+}
+
+void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn)
+{
+    (void)ce;
+    (void)standby_idx;
+    (void)acked_lsn;
+    /* stub: never advances committed_lsn */
+}
+
+void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
+{
+    (void)ce;
+    (void)noop_lsn;
+}
+
+void commit_on_membership_change(commit_engine_t* ce)
+{
+    (void)ce;
+}

src/controller/include/commit.h

@@ -0,0 +1,61 @@
+#pragma once
+/*
+ * commit.h — Commit engine: single authority for per-entry commit evaluation,
+ * committed_lsn advancement, and waiter completion.
+ *
+ * Service interface is umbrella-verbatim from
+ * .kiro/specs/ha-consistency-invariants/design.md (sliced by ha-commit-engine).
+ *
+ * All functions thread-safe; commit_on_ack callable from iomgr (non-blocking).
+ */
+#include <pthread.h>
+#include <stdint.h>
+
+#include "persist.h" /* persist_t (owns committed_lsn atomic + replicator_t registry) */
+
+/* Headroom kept free of blocking commit waits; cap = max(1, n_vcpus - reserved) */
+#define COMMIT_WAIT_RESERVED 2
+
+typedef enum { COMMIT_MODE_WAIT, COMMIT_MODE_RETAIN } commit_mode_t;
+
+/*
+ * Commit engine state — in-memory only (commit state volatile, Raft-conformant).
+ * Per-standby view delegated to standby_conn_t.acked_lsn / in_catch_up via
+ * p->repl (no duplication).
+ */
+typedef struct commit_engine {
+    pthread_mutex_t lock;
+    pthread_cond_t  cond; /* broadcast on committed_lsn advance / membership change */
+    persist_t*      p;    /* back-reference: committed_lsn store + repl registry */
+    int             n_peers;
+    int             n_vcpus;
+    int             inflight_waits;        /* <= max(1, n_vcpus - COMMIT_WAIT_RESERVED) */
+    uint64_t        writable_lsn;          /* own-term gate; 0 = no gate */
+    uint64_t        scratch[MAX_STANDBYS]; /* sorted matchIndex scratch array */
+} commit_engine_t;
+
+/* Lifecycle. init validates n_peers >= 0, n_vcpus >= 1, non-NULL persist_t;
+ * returns -EINVAL otherwise. Call after n_peers and n_vcpus are final. */
+int  commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus);
+void commit_engine_fini(commit_engine_t* ce);
+
+/* Register entry at append time. RETAIN returns 0 immediately.
+ * WAIT blocks up to timeout_ms: 0 = committed; -ETIMEDOUT = not committed
+ * (entry remains in log; outcome resolves per umbrella 2.5); -XCE_ERR_NOT_READY =
+ * admission cap exhausted (only cause). An unmet own-term gate does NOT fail
+ * fast: the gate is part of the completion predicate, so WAIT blocks within
+ * timeout_ms until the gate opens, surfacing -ETIMEDOUT if it never does. */
+int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
+                  uint32_t timeout_ms);
+
+/* 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);
+
+/* Promotion: set own-term gate to the no-op entry's LSN.
+ * (API ships here; no caller in this spec — unset = no gate.) */
+void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn);
+
+/* Quorum-membership change (standby add/remove/catch-up transition):
+ * re-evaluate pending entries; standbys in catch-up are excluded. */
+void commit_on_membership_change(commit_engine_t* ce);

src/controller/meson.build

@@ -119,6 +119,8 @@ controller_unit_test_src = [
   'testsuite/test_ha_connection_reuse.c',
   'testsuite/test_ha_rdhup_integration.c',
   'testsuite/test_ha_liveness.c',
+  'testsuite/test_commit_engine.c',
+  'commit.c',
   'startup_validation.c',
   'worker.c',
   'persist.c',

src/controller/testsuite/controller_test.c

@@ -50,6 +50,7 @@ extern void test_ha_repl_ack(void);
 extern void test_ha_connection_reuse(void);
 extern void test_ha_rdhup_integration(void);
 extern void test_ha_liveness(void);
+extern void test_commit_engine(void);
 
 void setUp(void) { return; }
 
@@ -169,6 +170,7 @@ int main(int argc, char* argv[])
     TEST(test_ha_connection_reuse);
     TEST(test_ha_rdhup_integration);
     TEST(test_ha_liveness);
+    TEST(test_commit_engine);
 
     return UNITY_END();
 }

src/controller/testsuite/test_commit_engine.c

@@ -0,0 +1,270 @@
+/*
+ * 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.
+ *
+ * Requirements: 1.2, 2.1, 2.2, 2.3, 5.1
+ */
+#include <errno.h>
+#include <stdatomic.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <time.h>
+#include <unity/unity.h>
+
+#include "commit.h"
+#include "persist.h"
+#include "repl.h"
+
+/* ── 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.
+ */
+static void engine_setup(int n_peers, int n_standbys, uint64_t appended_lsn)
+{
+    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);
+
+    TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, 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");
+}
+
+/* ── 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);
+    UNITY_END();
+}