test(ha-config-bootstrap): add RED-phase tests for repl_wait_majority_ack and pending_ack_count (task 7.1)

.kiro/specs/ha-config-bootstrap/tasks.md

@@ -148,7 +148,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A);
   - _Boundary: Term Durability (ha.c)_
 
 - [ ] 7. Change E: Majority-commit WAL replication
-- [ ] 7.1 Write failing tests for repl_broadcast and repl_wait_majority_ack
+- [x] 7.1 Write failing tests for repl_broadcast and repl_wait_majority_ack
   - In `src/controller/testsuite/test_repl_majority.c`: test 2/3 ACKs → `repl_wait_majority_ack` returns 0; test timeout returns `-ETIMEDOUT`; test single-node (`n_active == 0`) returns 0 immediately; test `in_catch_up=1` standby excluded from quorum denominator; test `pending_ack_count > max` removes standby from active set; test `OP_walAck` vs `OP_upgradeConfigAck` dispatched separately via `ack_op`
   - `make test` fails
   - _Requirements: 5.1, 5.4, 5.5, 5.8, 8.2_

src/controller/meson.build

@@ -106,6 +106,7 @@ controller_unit_test_src = [
   'testsuite/test_ha_bootstrap.c',
   'testsuite/test_ha_bootstrap_support.c',
   'testsuite/test_ha_durability.c',
+  'testsuite/test_repl_majority.c',
   'startup_validation.c',
   'worker.c',
   'persist.c',

src/controller/testsuite/controller_test.c

@@ -38,6 +38,7 @@ extern void test_ha_startup(void);
 extern void test_ha_election(void);
 extern void test_ha_bootstrap(void);
 extern void test_ha_durability(void);
+extern void test_repl_majority(void);
 
 void setUp(void) { return; }
 
@@ -145,6 +146,7 @@ int main(int argc, char* argv[])
     TEST(test_ha_election);
     TEST(test_ha_bootstrap);
     TEST(test_ha_durability);
+    TEST(test_repl_majority);
 
     return UNITY_END();
 }

src/controller/testsuite/test_repl_majority.c

@@ -0,0 +1,235 @@
+/*
+ * test_repl_majority.c — RED-phase tests for majority-commit replication.
+ *
+ * Tests for:
+ *   repl_wait_majority_ack(replicator_t *, uint32_t timeout_ms, uint8_t ack_op)
+ *   repl_broadcast majority semantics (pending_ack_count, over-limit eviction)
+ *
+ * These functions/fields do not exist yet (task 7.2, 7.3).
+ * This file will fail to link until they are implemented.
+ *
+ * Requirements: 5.1, 5.4, 5.5, 5.8, 8.2
+ */
+#include <errno.h>
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#include <time.h>
+#include <unity/unity.h>
+#include "repl.h"
+#include "config.h"
+#include "controller.capnp.h"  /* OPCode_walAck, OPCode_upgradeConfigAck */
+
+/* MAX_PENDING_ACKS: defined in repl.h after task 7.2; fallback for RED phase */
+#ifndef MAX_PENDING_ACKS
+#define MAX_PENDING_ACKS 16
+#endif
+
+/* ── Forward declarations of functions-under-test (not yet implemented) ───── */
+
+/*
+ * repl_wait_majority_ack: wait until majority of active standbys have ACKed
+ * for the given ack_op. Returns 0 on quorum, -ETIMEDOUT on timeout.
+ * Single-node (no active standbys) returns 0 immediately.
+ */
+extern int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms,
+                                   uint8_t ack_op);
+
+/* ── Helpers ───────────────────────────────────────────────────────────────── */
+
+/* Dummy endpoints — repl.c only compares pointers */
+static int ep0, ep1, ep2;
+
+/* ACK a standby from another thread */
+struct ack_args {
+    replicator_t *r;
+    void         *ep;
+    uint64_t      lsn;
+    uint32_t      delay_ms;
+};
+
+static void *ack_thread(void *arg)
+{
+    struct ack_args *a = arg;
+    if (a->delay_ms > 0) {
+        struct timespec ts = { .tv_sec = 0,
+                               .tv_nsec = (long)a->delay_ms * 1000000L };
+        nanosleep(&ts, NULL);
+    }
+    repl_on_ack(a->r, a->ep, a->lsn);
+    return NULL;
+}
+
+/* ── Test 1: 2/3 ACKs → majority, returns 0 ───────────────────────────────── */
+
+static void test_wait_majority_ack_two_of_three_returns_zero(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    repl_add_standby(&r, &ep0, "sb0", 7000);
+    repl_add_standby(&r, &ep1, "sb1", 7001);
+    repl_add_standby(&r, &ep2, "sb2", 7002);
+
+    /* 2 standbys ACK — majority of 3 active standbys is ceil((3+1)/2) = 2 ACKs + self = quorum */
+    pthread_t t0, t1;
+    struct ack_args a0 = { &r, &ep0, 1, 10 };
+    struct ack_args a1 = { &r, &ep1, 1, 20 };
+    pthread_create(&t0, NULL, ack_thread, &a0);
+    pthread_create(&t1, NULL, ack_thread, &a1);
+
+    int rc = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    pthread_join(t0, NULL);
+    pthread_join(t1, NULL);
+    repl_fini(&r);
+    printf("test_wait_majority_ack_two_of_three_returns_zero: PASS\n");
+}
+
+/* ── Test 2: timeout before majority ACKs → -ETIMEDOUT ────────────────────── */
+
+static void test_wait_majority_ack_timeout_returns_etimedout(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    repl_add_standby(&r, &ep0, "sb0", 7000);
+    repl_add_standby(&r, &ep1, "sb1", 7001);
+    repl_add_standby(&r, &ep2, "sb2", 7002);
+    /* No ACKs arrive — should time out */
+
+    int rc = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_walAck);
+    TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc);
+
+    repl_fini(&r);
+    printf("test_wait_majority_ack_timeout_returns_etimedout: PASS\n");
+}
+
+/* ── Test 3: single-node (no standbys) → returns 0 immediately ────────────── */
+
+static void test_wait_majority_ack_no_standbys_returns_zero(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    /* No standbys added — primary alone is majority */
+
+    int rc = repl_wait_majority_ack(&r, 100, (uint8_t)OPCode_walAck);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    repl_fini(&r);
+    printf("test_wait_majority_ack_no_standbys_returns_zero: PASS\n");
+}
+
+/* ── Test 4: in_catch_up standby excluded from quorum denominator ──────────── */
+
+static void test_wait_majority_ack_catchup_excluded_from_quorum(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    repl_add_standby(&r, &ep0, "sb0", 7000); /* active */
+    repl_add_standby(&r, &ep1, "sb1", 7001); /* will be put in catch-up */
+    repl_add_standby(&r, &ep2, "sb2", 7002); /* active */
+
+    /* Put ep1 in catch-up — should not count toward quorum */
+    repl_start_catchup(&r, &ep1, 0);
+
+    /* With ep1 excluded: 2 active standbys → quorum = ceil((2+1)/2) = 2 (self + 1 ACK) */
+    /* ep0 ACKs → that's 1 standby ACK + self = majority of 2 active */
+    pthread_t t0;
+    struct ack_args a0 = { &r, &ep0, 1, 10 };
+    pthread_create(&t0, NULL, ack_thread, &a0);
+
+    int rc = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    pthread_join(t0, NULL);
+    repl_fini(&r);
+    printf("test_wait_majority_ack_catchup_excluded_from_quorum: PASS\n");
+}
+
+/* ── Test 5: OP_walAck vs OP_upgradeConfigAck dispatched independently ─────── */
+
+static void test_wait_majority_ack_ack_op_discriminates(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    repl_add_standby(&r, &ep0, "sb0", 7000);
+
+    /* ACK arrives for ep0 via repl_on_ack — the ack_op=OPCode_walAck path should
+     * return 0 (quorum reached with 1 standby), while a wait for a different
+     * ack_op with no ACK should time out.
+     * This test verifies the wait function accepts ack_op and routes correctly. */
+    pthread_t t0;
+    struct ack_args a0 = { &r, &ep0, 1, 10 };
+    pthread_create(&t0, NULL, ack_thread, &a0);
+
+    int rc_wal = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck);
+    TEST_ASSERT_EQUAL_INT(0, rc_wal);
+
+    pthread_join(t0, NULL);
+
+    /* upgradeConfigAck with no pending ACKs → timeout */
+    int rc_cfg = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_upgradeConfigAck);
+    TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc_cfg);
+
+    repl_fini(&r);
+    printf("test_wait_majority_ack_ack_op_discriminates: PASS\n");
+}
+
+/* ── Test 6: pending_ack_count > max removes standby from active set ──────── */
+
+static void test_broadcast_over_limit_removes_standby(void)
+{
+    replicator_t r;
+    repl_init(&r);
+    repl_add_standby(&r, &ep0, "sb0", 7000);
+
+    /* Artificially inflate pending_ack_count past the limit.
+     * After overflow the standby should be evicted (in_catch_up=1 or removed). */
+    for (int i = 0; i < r.standbys[0].ep != NULL ? 1 : 0; i++) {
+        /* Access the pending_ack_count field — this will fail to compile if
+         * the field does not exist on standby_conn_t */
+        r.standbys[0].pending_ack_count = MAX_PENDING_ACKS + 1;
+    }
+    /* Also verify the field exists via a direct write */
+    r.standbys[0].pending_ack_count = MAX_PENDING_ACKS + 1;
+
+    /* Broadcast a wal entry — the over-limit standby should be evicted */
+    wal_entry_hdr_t hdr = { .type = WAL_JOB_SUBMIT, .lsn = 1 };
+    int rc = repl_broadcast(&r, &hdr, NULL, 0);
+    /* repl_broadcast itself may return 0 even after eviction */
+    (void)rc;
+
+    /* After broadcast with pending_ack_count > max, standby should be in catch-up */
+    int evicted = 0;
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        if (r.standbys[i].ep == &ep0) {
+            if (atomic_load_explicit(&r.standbys[i].in_catch_up,
+                                     memory_order_acquire))
+                evicted = 1;
+        }
+    }
+    /* Alternatively the standby may have been fully removed */
+    if (!evicted) {
+        evicted = (r.nstandbys == 0);
+    }
+    TEST_ASSERT_TRUE(evicted);
+
+    repl_fini(&r);
+    printf("test_broadcast_over_limit_removes_standby: PASS\n");
+}
+
+/* ── Test group entry point ────────────────────────────────────────────────── */
+
+void test_repl_majority(void)
+{
+    UNITY_BEGIN();
+    RUN_TEST(test_wait_majority_ack_two_of_three_returns_zero);
+    RUN_TEST(test_wait_majority_ack_timeout_returns_etimedout);
+    RUN_TEST(test_wait_majority_ack_no_standbys_returns_zero);
+    RUN_TEST(test_wait_majority_ack_catchup_excluded_from_quorum);
+    RUN_TEST(test_wait_majority_ack_ack_op_discriminates);
+    RUN_TEST(test_broadcast_over_limit_removes_standby);
+    UNITY_END();
+}