wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-wal-replication): primary WalAckReply handler (task 4.1)

Browse source 
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).
This commit is contained in:
Rongsong Shen 2026-06-10 09:46:22 +00:00
parent 566c91cd8b
commit 861b746c84
2 changed files with 115 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/controller/ha.c
View file

@ -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);
}

97
src/controller/testsuite/test_repl_unit.c
View file

@ -1,5 +1,6 @@
#include <assert.h>
#include <errno.h>
#include <stdatomic.h>
#include <stdint.h>
#include <string.h>
#include <stdio.h>
@ -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();
}