From 6e4fd1aa81ca24cb21180335f0d4dd2570d21e01 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 18:14:12 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20P2-5=20repl=5Fstart/complete=5Fcatchup,?= =?UTF-8?q?=20repl=5Fneeds=5Fsnapshot=20=E2=80=94=20standby=20catch-up=20s?= =?UTF-8?q?tate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the catch-up state machine as testable pure-logic functions: - repl_start_catchup(): marks standby in_catch_up=1, stores join_applied_lsn - repl_complete_catchup(): clears in_catch_up=0, broadcasts ack_cond so any blocked repl_wait_ack() re-checks the ACK condition immediately - repl_needs_snapshot(): pure predicate — returns 1 when standby LSN predates oldest surviving WAL segment (snapshot transfer required), 0 otherwise Also fixes pre-existing linker error: repl.c was missing from controller_src (xce_controller production binary), causing undefined references to repl_* symbols introduced in P2-4. 7 new TDD tests in test_repl_catchup.c — all pass. Existing tests green. --- src/controller/meson.build | 2 + src/controller/repl.c | 46 +++++++ src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_repl_catchup.c | 134 +++++++++++++++++++ src/include/repl.h | 12 ++ 5 files changed, 196 insertions(+) create mode 100644 src/controller/testsuite/test_repl_catchup.c diff --git a/src/controller/meson.build b/src/controller/meson.build index e27333dc..3b8971ac 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -18,6 +18,7 @@ controller_src = [ 'apiserver.c', 'myconf.c', 'persist.c', + 'repl.c', ] incdir = include_directories('.', '../include', '../lib', @@ -90,6 +91,7 @@ controller_unit_test_src = [ 'testsuite/test_repl_unit.c', 'testsuite/test_repl_apply.c', 'testsuite/test_persist_repl.c', + 'testsuite/test_repl_catchup.c', 'persist.c', 'ha.c', 'repl.c', diff --git a/src/controller/repl.c b/src/controller/repl.c index 7b6b2814..2c142060 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -296,3 +296,49 @@ uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, return min_lsn; } + +/* ============================================================ + * Catch-up state management (P2-5) + * ============================================================ */ + +void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn) +{ + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep == ep) { + r->standbys[i].join_applied_lsn = join_applied_lsn; + atomic_store_explicit(&r->standbys[i].in_catch_up, 1, + memory_order_release); + break; + } + } + mylock_unlock(&r->lock); +} + +void repl_complete_catchup(replicator_t *r, repl_ep_t ep) +{ + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep == ep) { + atomic_store_explicit(&r->standbys[i].in_catch_up, 0, + memory_order_release); + break; + } + } + mylock_unlock(&r->lock); + + /* Broadcast so any blocked repl_wait_ack() re-checks the ACK condition */ + pthread_mutex_lock(&r->ack_lock); + pthread_cond_broadcast(&r->ack_cond); + pthread_mutex_unlock(&r->ack_lock); +} + +int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn, + uint64_t oldest_wal_lsn) +{ + (void)r; + /* No WAL at all: empty standby can start from LSN 0 without a snapshot */ + if (oldest_wal_lsn == 0) return 0; + /* Standby LSN predates oldest surviving WAL segment — full snapshot required */ + return (standby_lsn < oldest_wal_lsn) ? 1 : 0; +} diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8b22bc49..651a4f19 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -26,6 +26,7 @@ extern void test_p1_integration(void); extern void test_repl_unit(void); extern void test_repl_apply(void); extern void test_persist_repl(void); +extern void test_repl_catchup(void); void setUp(void) { return; } @@ -121,6 +122,7 @@ int main(int argc, char* argv[]) TEST(test_repl_unit); TEST(test_repl_apply); TEST(test_persist_repl); + TEST(test_repl_catchup); return UNITY_END(); } diff --git a/src/controller/testsuite/test_repl_catchup.c b/src/controller/testsuite/test_repl_catchup.c new file mode 100644 index 00000000..86a25177 --- /dev/null +++ b/src/controller/testsuite/test_repl_catchup.c @@ -0,0 +1,134 @@ +/* + * test_repl_catchup.c — P2-5: standby catch-up state machine tests. + * Tests repl_start_catchup, repl_complete_catchup, repl_needs_snapshot. + * No live network I/O — pure state machine logic. + */ +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "repl.h" + +static void test_start_catchup_sets_in_catch_up(void) +{ + replicator_t r; + int ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&ep, "sb", 7001); + + repl_start_catchup(&r, (void *)&ep, /*join_applied_lsn=*/5); + + /* in_catch_up must be set */ + TEST_ASSERT_EQUAL_INT( + 1, + (int)atomic_load_explicit(&r.standbys[0].in_catch_up, + memory_order_acquire)); + /* join_applied_lsn must be stored */ + TEST_ASSERT_EQUAL_UINT64(5, r.standbys[0].join_applied_lsn); + + repl_fini(&r); + printf("test_start_catchup_sets_in_catch_up: PASS\n"); +} + +static void test_complete_catchup_clears_in_catch_up(void) +{ + replicator_t r; + int ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&ep, "sb", 7001); + repl_start_catchup(&r, (void *)&ep, 5); + + repl_complete_catchup(&r, (void *)&ep); + + TEST_ASSERT_EQUAL_INT( + 0, + (int)atomic_load_explicit(&r.standbys[0].in_catch_up, + memory_order_acquire)); + + repl_fini(&r); + printf("test_complete_catchup_clears_in_catch_up: PASS\n"); +} + +static void test_wait_ack_excludes_catching_up_standby(void) +{ + replicator_t r; + int ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&ep, "sb", 7001); + repl_start_catchup(&r, (void *)&ep, 5); /* mark as catching up */ + + /* repl_wait_ack must return -ENODEV (no non-catching-up standbys) */ + int rc = repl_wait_ack(&r, 10, 50); + TEST_ASSERT_EQUAL_INT(-ENODEV, rc); + + repl_fini(&r); + printf("test_wait_ack_excludes_catching_up_standby: PASS\n"); +} + +static void test_complete_catchup_allows_ack(void) +{ + replicator_t r; + int ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&ep, "sb", 7001); + repl_start_catchup(&r, (void *)&ep, 5); + repl_complete_catchup(&r, (void *)&ep); + + /* After catch-up completes, the standby's ACK must count */ + repl_on_ack(&r, (void *)&ep, 20); + int rc = repl_wait_ack(&r, 10, 50); + TEST_ASSERT_EQUAL_INT(0, rc); + + repl_fini(&r); + printf("test_complete_catchup_allows_ack: PASS\n"); +} + +static void test_repl_needs_snapshot_no_wal(void) +{ + /* When standby LSN is 0 and no WAL exists, needs_snapshot returns 0 + * (empty state = fresh standby, can start from empty — no snapshot needed) */ + replicator_t r; + repl_init(&r); + int needs = repl_needs_snapshot(&r, /*standby_lsn=*/0, /*oldest_wal_lsn=*/0); + TEST_ASSERT_EQUAL_INT(0, needs); + repl_fini(&r); + printf("test_repl_needs_snapshot_no_wal: PASS\n"); +} + +static void test_repl_needs_snapshot_standby_behind_wal(void) +{ + /* Standby LSN 5, oldest WAL segment starts at LSN 10 — snapshot required */ + replicator_t r; + repl_init(&r); + int needs = repl_needs_snapshot(&r, /*standby_lsn=*/5, /*oldest_wal_lsn=*/10); + TEST_ASSERT_EQUAL_INT(1, needs); + repl_fini(&r); + printf("test_repl_needs_snapshot_standby_behind_wal: PASS\n"); +} + +static void test_repl_needs_snapshot_standby_within_wal(void) +{ + /* Standby LSN 15, oldest WAL starts at LSN 10 — WAL replay sufficient */ + replicator_t r; + repl_init(&r); + int needs = repl_needs_snapshot(&r, /*standby_lsn=*/15, /*oldest_wal_lsn=*/10); + TEST_ASSERT_EQUAL_INT(0, needs); + repl_fini(&r); + printf("test_repl_needs_snapshot_standby_within_wal: PASS\n"); +} + +void test_repl_catchup(void) +{ + RUN_TEST(test_start_catchup_sets_in_catch_up); + RUN_TEST(test_complete_catchup_clears_in_catch_up); + RUN_TEST(test_wait_ack_excludes_catching_up_standby); + RUN_TEST(test_complete_catchup_allows_ack); + RUN_TEST(test_repl_needs_snapshot_no_wal); + RUN_TEST(test_repl_needs_snapshot_standby_behind_wal); + RUN_TEST(test_repl_needs_snapshot_standby_within_wal); +} diff --git a/src/include/repl.h b/src/include/repl.h index 62ac6528..a4c3ee2e 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -58,3 +58,15 @@ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms); /* WAL trim guard */ uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s); + +/* Catch-up state management */ + +/* Mark standby as entering catch-up mode; stores join_applied_lsn */ +void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn); + +/* Mark standby catch-up complete (called on OP_replCatchUpDone receipt) */ +void repl_complete_catchup(replicator_t *r, repl_ep_t ep); + +/* Returns 1 if standby needs a full snapshot transfer (its LSN predates oldest WAL). + * oldest_wal_lsn == 0 means no WAL exists — no snapshot needed for LSN 0 standbys. */ +int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn, uint64_t oldest_wal_lsn);