diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index 8a7a7c5..d1db2f8 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.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_ diff --git a/src/controller/meson.build b/src/controller/meson.build index c0d8ac2..17bc7a2 100644 --- a/src/controller/meson.build +++ b/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', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 1e3b823..61d780b 100644 --- a/src/controller/testsuite/controller_test.c +++ b/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(); } diff --git a/src/controller/testsuite/test_repl_majority.c b/src/controller/testsuite/test_repl_majority.c new file mode 100644 index 0000000..9ff1e12 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include +#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(); +}