From 318098c727e2706687a27962a9c82212e8a35520 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 5 Jun 2026 01:25:00 +0000 Subject: [PATCH] fix(ha-repl-hardening): wire snapshot production path (task 3.5b) - SEND_SNAPSHOT uses checkpoint_lsn (not committed_lsn) for file lookup - ha_send_snapshot_to_standby stats file for actual total_size - ha_compute_oldest_wal_lsn scans WAL dir for min segment LSN - Removed hardcoded 131072 and oldest_wal_lsn=0 placeholders --- .kiro/specs/ha-repl-hardening/tasks.md | 2 +- src/controller/config_repl.c | 28 +++++++++++++++++++-- src/controller/config_repl.h | 4 ++- src/controller/ha.c | 19 +++++++------- src/controller/testsuite/test_config_repl.c | 8 +++--- 5 files changed, 43 insertions(+), 18 deletions(-) diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 6d268bd4..5d961ee4 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -116,7 +116,7 @@ - _Boundary: Snapshot Transfer (config_repl.c)_ - _Depends: 1.3_ -- [ ] 3.5b Fix SEND_SNAPSHOT production wiring and oldest_wal_lsn +- [x] 3.5b Fix SEND_SNAPSHOT production wiring and oldest_wal_lsn - Use persist.checkpoint_lsn (not committed_lsn) for snapshot file lookup - Stat the snapshot file to get actual total_size (not hardcoded 131072) - Wire persist.oldest_wal_lsn into ha_evaluate_repl_action call site (remove the hardcoded 0) diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 909f5e62..587818f4 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1740,16 +1740,21 @@ int (*xce_snapshot_chunk_hook)(endpoint_t *ep, xce_request_t *req) = NULL; #define SNAPSHOT_CHUNK_SIZE 65536 int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, - uint64_t checkpoint_lsn, uint64_t total_size) + uint64_t checkpoint_lsn) { if (!snap_dir) return -EINVAL; - /* Find the snapshot file by naming convention */ char snap_path[4096]; snprintf(snap_path, sizeof(snap_path), "%s/snap-%016" PRIx64 ".capnp", snap_dir, checkpoint_lsn); + /* Stat the snapshot file to get total_size */ + struct stat st; + if (stat(snap_path, &st) != 0) + return -errno; + uint64_t total_size = (uint64_t)st.st_size; + FILE *f = fopen(snap_path, "rb"); if (!f) return -errno; @@ -1813,3 +1818,22 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, fclose(f); return rc; } + +uint64_t ha_compute_oldest_wal_lsn(const char *wal_path) +{ + DIR *d = opendir(wal_path); + if (!d) + return 0; + + uint64_t oldest = UINT64_MAX; + struct dirent *ent; + while ((ent = readdir(d)) != NULL) { + uint64_t seg_lsn = 0; + if (sscanf(ent->d_name, "seg-%" SCNx64 ".wal", &seg_lsn) != 1) + continue; + if (seg_lsn < oldest) + oldest = seg_lsn; + } + closedir(d); + return (oldest == UINT64_MAX) ? 0 : oldest; +} diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index b0ffcc07..7a10b04a 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -158,7 +158,9 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, #define SNAPSHOT_CHUNK_SIZE 65536 int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, - uint64_t checkpoint_lsn, uint64_t total_size); + uint64_t checkpoint_lsn); + +uint64_t ha_compute_oldest_wal_lsn(const char *wal_path); /* * handle_ctrlhello: primary-side handler for OP_hello from a joining standby. diff --git a/src/controller/ha.c b/src/controller/ha.c index 69808e0b..52042c1c 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1075,18 +1075,14 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, break; case HA_REPL_SEND_SNAPSHOT: - /* Task 3.5: snapshot transfer. - * Creates a checkpoint in the delaymgr thread (not here — vtask - * must not block), then sends the snapshot in 64KB chunks. */ INFO("HA repl: standby[%d] lsn=%" PRIu64 " < oldest_wal_lsn=%" PRIu64 " → requesting snapshot transfer", peer_idx, standby_lsn, oldest_wal_lsn); if (c && c->persist.snap_path[0]) { extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, - uint64_t, uint64_t); + uint64_t); ha_send_snapshot_to_standby(ep, c->persist.snap_path, - c->persist.committed_lsn, - /*total_size placeholder*/ 131072); + c->persist.checkpoint_lsn); } break; @@ -2249,12 +2245,15 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) sr.config_tag, sr.applied_lsn, committed_tag, committed_lsn); - /* Task 3.3: run the unified decision loop for this standby. - * oldest_wal_lsn=0 until snapshot transfer is wired (Task 3.5). */ + /* Derive oldest_wal_lsn from the WAL directory. + * The oldest retained segment's start LSN is the boundary — + * standbys below it need a snapshot (Task 3.5b). */ uint64_t oldest_wal_lsn = 0; #ifndef XCE_TESTING - if (c) - oldest_wal_lsn = 0; /* Task 3.5: wire persist->oldest_wal_lsn */ + if (c && c->persist.wal_path[0]) { + extern uint64_t ha_compute_oldest_wal_lsn(const char *wal_path); + oldest_wal_lsn = ha_compute_oldest_wal_lsn(c->persist.wal_path); + } #endif ha_dispatch_repl_action_for_standby(ha, c, ep, peer_idx, committed_tag, committed_lsn, diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index ff6663b2..ba5e665d 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -1548,8 +1548,8 @@ static void test_snapshot_transfer_two_chunks(void) xce_snapshot_chunk_hook = snaptest_hook; snaptest_reset(); - extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t, uint64_t); - int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn, 131072); + extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t); + int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(2, g_snaptest_chunk_count); @@ -1591,8 +1591,8 @@ static void test_snapshot_transfer_small_less_than_one_chunk(void) xce_snapshot_chunk_hook = snaptest_hook; snaptest_reset(); - extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t, uint64_t); - int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn, 64); + extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t); + int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunk_count);