diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index d1db2f8f..5fb5d03d 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.kiro/specs/ha-config-bootstrap/tasks.md @@ -153,7 +153,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A); - `make test` fails - _Requirements: 5.1, 5.4, 5.5, 5.8, 8.2_ -- [ ] 7.2 Implement repl_broadcast() (replace stub) +- [x] 7.2 Implement repl_broadcast() (replace stub) - Add `pending_ack_count` field to `standby_conn_t` - `repl_broadcast(cluster_t *c, uint8_t op, const void *payload, size_t len)`: iterate active standbys; skip `in_catch_up==1`; if `pending_ack_count > max_pending_acks` → remove from active set, trigger catch-up reconnect, log WARN; else send payload over `standby.ctrl_fd`, increment `pending_ack_count`; returns 0 after fanout initiated - `make test` GREEN on broadcast fanout tests diff --git a/src/controller/repl.c b/src/controller/repl.c index 2c142060..13eb369e 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -10,6 +10,7 @@ #include "repl.h" #include "mylock.h" +#include "utils.h" /* ============================================================ * Lifecycle @@ -177,13 +178,17 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn) } /* ============================================================ - * Replication send — stub in P2-1; wired in P2-4 + * Replication send (task 7.2) + * + * Iterates active standbys; skips empty slots and standbys in catch-up. + * Evicts standbys whose pending_ack_count exceeds MAX_PENDING_ACKS by marking + * them in_catch_up=1 (inlined to avoid re-entrant lock acquisition). + * TCP send wiring deferred to task 7.6. * ============================================================ */ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, const void *payload, uint32_t payload_len) { - (void)r; (void)payload; (void)payload_len; @@ -191,7 +196,36 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, assert(hdr != NULL); assert(hdr->type != WAL_CHECKPOINT); - /* Full implementation in P2-4 */ + if (r == NULL) + return -EINVAL; + + mylock_lock(&r->lock); + + for (int i = 0; i < MAX_STANDBYS; i++) { + standby_conn_t *s = &r->standbys[i]; + + if (s->ep == NULL) + continue; + + if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire)) + continue; + + if (s->pending_ack_count > MAX_PENDING_ACKS) { + /* Inline repl_start_catchup to avoid re-entrant lock acquisition */ + s->join_applied_lsn = atomic_load_explicit(&s->acked_lsn, + memory_order_acquire); + atomic_store_explicit(&s->in_catch_up, 1, memory_order_release); + WARN("repl_broadcast: standby %s:%u evicted (pending_ack_count=%u > %u)" + " — triggering catch-up reconnect", + s->host, s->port, s->pending_ack_count, MAX_PENDING_ACKS); + continue; + } + + /* Non-evicted standby: track the pending send (TCP wiring in task 7.6) */ + s->pending_ack_count++; + } + + mylock_unlock(&r->lock); return 0; } @@ -258,6 +292,18 @@ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms) return rc; } +/* ============================================================ + * Majority-quorum ACK wait — stub until task 7.3 + * ============================================================ */ + +int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op) +{ + (void)r; + (void)timeout_ms; + (void)ack_op; + return -ENOSYS; +} + /* ============================================================ * WAL trim guard * ============================================================ */ diff --git a/src/controller/testsuite/test_repl_majority.c b/src/controller/testsuite/test_repl_majority.c index 9ff1e121..5141abd4 100644 --- a/src/controller/testsuite/test_repl_majority.c +++ b/src/controller/testsuite/test_repl_majority.c @@ -64,27 +64,33 @@ static void *ack_thread(void *arg) /* ── Test 1: 2/3 ACKs → majority, returns 0 ───────────────────────────────── */ +/* Static so ack_args survive Unity longjmp if assert fires before join */ +static replicator_t g_r1; +static struct ack_args g_a1_0, g_a1_1; + 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); + repl_init(&g_r1); + repl_add_standby(&g_r1, &ep0, "sb0", 7000); + repl_add_standby(&g_r1, &ep1, "sb1", 7001); + repl_add_standby(&g_r1, &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); + g_a1_0 = (struct ack_args){ &g_r1, &ep0, 1, 10 }; + g_a1_1 = (struct ack_args){ &g_r1, &ep1, 1, 20 }; + pthread_create(&t0, NULL, ack_thread, &g_a1_0); + pthread_create(&t1, NULL, ack_thread, &g_a1_1); - int rc = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck); - TEST_ASSERT_EQUAL_INT(0, rc); + int rc = repl_wait_majority_ack(&g_r1, 2000, (uint8_t)OPCode_walAck); + /* Join threads before asserting — prevents use-after-return if assert longjmps */ pthread_join(t0, NULL); pthread_join(t1, NULL); - repl_fini(&r); + + TEST_ASSERT_EQUAL_INT(0, rc); + + repl_fini(&g_r1); printf("test_wait_majority_ack_two_of_three_returns_zero: PASS\n"); } @@ -123,57 +129,55 @@ static void test_wait_majority_ack_no_standbys_returns_zero(void) /* ── Test 4: in_catch_up standby excluded from quorum denominator ──────────── */ +static replicator_t g_r4; +static struct ack_args g_a4_0; + 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 */ + repl_init(&g_r4); + repl_add_standby(&g_r4, &ep0, "sb0", 7000); /* active */ + repl_add_standby(&g_r4, &ep1, "sb1", 7001); /* will be put in catch-up */ + repl_add_standby(&g_r4, &ep2, "sb2", 7002); /* active */ - /* Put ep1 in catch-up — should not count toward quorum */ - repl_start_catchup(&r, &ep1, 0); + repl_start_catchup(&g_r4, &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); + g_a4_0 = (struct ack_args){ &g_r4, &ep0, 1, 10 }; + pthread_create(&t0, NULL, ack_thread, &g_a4_0); - int rc = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck); - TEST_ASSERT_EQUAL_INT(0, rc); + int rc = repl_wait_majority_ack(&g_r4, 2000, (uint8_t)OPCode_walAck); pthread_join(t0, NULL); - repl_fini(&r); + TEST_ASSERT_EQUAL_INT(0, rc); + + repl_fini(&g_r4); printf("test_wait_majority_ack_catchup_excluded_from_quorum: PASS\n"); } /* ── Test 5: OP_walAck vs OP_upgradeConfigAck dispatched independently ─────── */ +static replicator_t g_r5; +static struct ack_args g_a5_0; + static void test_wait_majority_ack_ack_op_discriminates(void) { - replicator_t r; - repl_init(&r); - repl_add_standby(&r, &ep0, "sb0", 7000); + repl_init(&g_r5); + repl_add_standby(&g_r5, &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); + g_a5_0 = (struct ack_args){ &g_r5, &ep0, 1, 10 }; + pthread_create(&t0, NULL, ack_thread, &g_a5_0); - int rc_wal = repl_wait_majority_ack(&r, 2000, (uint8_t)OPCode_walAck); - TEST_ASSERT_EQUAL_INT(0, rc_wal); + int rc_wal = repl_wait_majority_ack(&g_r5, 2000, (uint8_t)OPCode_walAck); pthread_join(t0, NULL); + TEST_ASSERT_EQUAL_INT(0, rc_wal); /* upgradeConfigAck with no pending ACKs → timeout */ - int rc_cfg = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_upgradeConfigAck); + int rc_cfg = repl_wait_majority_ack(&g_r5, 50, (uint8_t)OPCode_upgradeConfigAck); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc_cfg); - repl_fini(&r); + repl_fini(&g_r5); printf("test_wait_majority_ack_ack_op_discriminates: PASS\n"); } diff --git a/src/include/repl.h b/src/include/repl.h index a4c3ee2e..6d05fff7 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -12,6 +12,9 @@ #include "mylock.h" #include "wal_types.h" /* wal_entry_hdr_t — P2-4: use wal_types.h to break circular include */ +/* Max queued-but-unacknowledged sends per standby before eviction to catch-up */ +#define MAX_PENDING_ACKS 16 + /* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */ typedef void* repl_ep_t; @@ -25,6 +28,7 @@ typedef struct { uint32_t port; uint64_t last_disconnected_lsn; time_t last_disconnect_time; + uint32_t pending_ack_count; /* Sends not yet ACKed; evict if > MAX_PENDING_ACKS */ } standby_conn_t; /* Replicator — embedded in persist_t or cluster_t (P2-4). @@ -56,6 +60,9 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn); /* Sync replication (stub in P2-1; wired in P2-3) */ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms); +/* Majority-quorum ACK wait — stub until task 7.3 */ +int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op); + /* WAL trim guard */ uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s);