lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ha-repl-hardening): wire snapshot production path (task 3.5b)

Browse source 
- 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
This commit is contained in:
Rongsong Shen 2026-06-05 01:25:00 +00:00
parent 834ab4d5e7
commit 318098c727
5 changed files with 43 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/ha-repl-hardening/tasks.md
View file

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

28
src/controller/config_repl.c
View file

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

4
src/controller/config_repl.h
View file

@ -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.

19
src/controller/ha.c
View file

@ -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,

8
src/controller/testsuite/test_config_repl.c
View file

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