From 861b746c849f045829a6583ff169a5956ca51445 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 10 Jun 2026 09:46:22 +0000 Subject: [PATCH] feat(ha-wal-replication): primary WalAckReply handler (task 4.1) Handle incoming WalAckReply in repl_ack_decode_and_dispatch: extract lsn+success, call repl_on_ack on success to update standby acked_lsn and signal majority-ack waiter. NACK is WARN-logged with no state change. Two unit tests verify positive path (acked_lsn updated, waiter unblocked) and negative path (NACK leaves acked_lsn=0, wait times out). --- src/controller/ha.c | 18 +++++ src/controller/testsuite/test_repl_unit.c | 97 +++++++++++++++++++++++ 2 files changed, 115 insertions(+) diff --git a/src/controller/ha.c b/src/controller/ha.c index 0be34a6c..27a2c84e 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2661,6 +2661,24 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) ha_dispatch_repl_action_for_standby(ha, c, ep, peer_idx, committed_tag, committed_lsn, oldest_wal_lsn); + } else if (rep_s.data_which == RequestReply_data_walackreplymsg) { + /* Task 4.1: WAL ACK reply from standby — extract lsn + success, + * call repl_on_ack to update acked_lsn and signal majority waiter. */ + struct WalAckReply war_raw; + memset(&war_raw, 0, sizeof(war_raw)); + read_WalAckReply(&war_raw, rep_s.data.walackreplymsg); + uint64_t acked_lsn = (uint64_t)war_raw.lsn; + int success = war_raw.success ? 1 : 0; + capn_free(&capn_ctx); + + if (success && c) { + repl_on_ack(&c->persist.repl, ep, acked_lsn); + TRACE("repl_ack_decode_and_dispatch: WAL ACK from standby" + " lsn=%lu success=1", (unsigned long)acked_lsn); + } else if (!success) { + WARN("repl_ack_decode_and_dispatch: WAL NACK from standby" + " lsn=%lu success=0 — ignoring", (unsigned long)acked_lsn); + } } else { capn_free(&capn_ctx); } diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c index 36863f61..290522b2 100644 --- a/src/controller/testsuite/test_repl_unit.c +++ b/src/controller/testsuite/test_repl_unit.c @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -11,6 +12,7 @@ #include "config.h" #include "typemap.h" #include "wal_types.h" +#include "controller.capnp.h" /* OPCode_walAck */ /* ============================================================ * Test seam for repl_broadcast send path (task 1.1) @@ -354,6 +356,98 @@ static void test_repl_add_standby_idempotent(void) { printf("test_repl_add_standby_idempotent: PASS\n"); } +/* ============================================================ + * Task 4.1: WAL ACK reply handler — Req 4.3 + * + * These tests verify the behaviour that repl_ack_decode_and_dispatch + * implements in ha.c: on a successful WalAckReply it calls + * repl_on_ack(&c->persist.repl, ep, acked_lsn), which must: + * a) update standby's acked_lsn + * b) signal ack_cond so repl_wait_majority_ack unblocks + * + * We test repl_on_ack directly (the exact call the handler makes). + * ============================================================ */ + +/* ACK sender thread args reused from above — ack_args struct already defined */ + +/* Test: repl_on_ack (handler ACK path) updates acked_lsn and unblocks + * repl_wait_majority_ack. Req 4.3: when the primary receives a WAL ACK + * from a standby it shall update that standby's acked_lsn in the replicator + * and signal any waiter on the ACK condition variable. */ +static void test_wal_ack_reply_updates_acked_lsn_and_signals(void) +{ + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); + + const uint64_t target_lsn = 77; + + /* Background thread simulates the handler calling repl_on_ack after 30 ms */ + struct ack_args args = { &r, (void *)&dummy_ep, target_lsn, 30 }; + pthread_t th; + pthread_create(&th, NULL, ack_sender_thread, &args); + + /* repl_wait_majority_ack: 1 active standby, quorum=(1+2)/2=1, + * acked_total = 1(self) + 1(standby) = 2 >= 1 → returns 0. */ + int rc = repl_wait_majority_ack(&r, 200, (uint8_t)OPCode_walAck); + pthread_join(th, NULL); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "repl_wait_majority_ack must return 0 after repl_on_ack signals (Req 4.3)"); + + /* Verify acked_lsn was actually updated in the slot */ + int found = 0; + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r.standbys[i].ep == (void *)&dummy_ep) { + uint64_t al = atomic_load_explicit(&r.standbys[i].acked_lsn, + memory_order_acquire); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(target_lsn, al, + "acked_lsn must equal the LSN passed to repl_on_ack (Req 4.3)"); + found = 1; + break; + } + } + TEST_ASSERT_TRUE_MESSAGE(found, "standby slot must be present after repl_on_ack"); + + repl_fini(&r); + printf("test_wal_ack_reply_updates_acked_lsn_and_signals: PASS\n"); +} + +/* Test: NACK path — handler does NOT call repl_on_ack, so acked_lsn stays 0 + * and repl_wait_majority_ack times out. Req 4.3 (negative): only a + * success==1 reply triggers the update; success==0 is a no-op on the + * replicator. */ +static void test_wal_nack_does_not_update_acked_lsn(void) +{ + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); + + /* No repl_on_ack call — simulates handler receiving success==0 NACK */ + + /* acked_lsn must remain 0 */ + uint64_t al = 0; + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r.standbys[i].ep == (void *)&dummy_ep) { + al = atomic_load_explicit(&r.standbys[i].acked_lsn, + memory_order_acquire); + break; + } + } + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, al, + "acked_lsn must remain 0 when NACK path skips repl_on_ack (Req 4.3)"); + + /* repl_wait_majority_ack must time out because no ACK arrived */ + int rc = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_walAck); + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "repl_wait_majority_ack must time out when no ACK arrives (NACK path) (Req 4.3)"); + + repl_fini(&r); + printf("test_wal_nack_does_not_update_acked_lsn: PASS\n"); +} + /* Test: send failure (hook returns -EIO) is tolerated — broadcast returns 0 */ static void test_broadcast_send_failure_continues(void) { @@ -403,5 +497,8 @@ void test_repl_unit(void) { RUN_TEST(test_broadcast_skips_catchup_standby); RUN_TEST(test_broadcast_evicts_overloaded_standby); RUN_TEST(test_broadcast_send_failure_continues); + /* Task 4.1: WAL ACK reply handler — Req 4.3 */ + RUN_TEST(test_wal_ack_reply_updates_acked_lsn_and_signals); + RUN_TEST(test_wal_nack_does_not_update_acked_lsn); UNITY_END(); }