feat(ha-repl-hardening): standby snapshot receive handler (task 3.5c)

ha_handle_snapshot_chunk writes chunks to snap.part, on isLast verifies
total_size, atomic renames to snap-{lsn}.bin, loads via new
persist_load_snapshot wrapper, sets applied_lsn=checkpoint_lsn.
3 tests: 2-chunk assembly, size mismatch, null args. Wired into
myreq_handler OPCode_snapshotTransfer dispatch.
This commit is contained in:
Rongsong Shen 2026-06-05 01:43:05 +00:00
parent 318098c727
commit cbd91fca43
7 changed files with 361 additions and 2 deletions

View file

@ -173,6 +173,35 @@ static int handle_wal_catchup_entry(cluster_t *c, xce_request_t *req)
}
/* handle_snapshot_transfer: dispatch handler for OPCode_snapshotTransfer.
* Delegates to ha_handle_snapshot_chunk (ha.c).
* On mid-transfer reconnect the caller (iomgr reconnect path) must delete
* snap.part before the next transfer starts — this handler does not track
* cross-chunk state itself (Req 4.4).
* No reply is sent — primary learns of completion via next heartbeat reply. */
static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req)
{
snapshot_transfer_t *chunk =
(snapshot_transfer_t *)req->body.snapshottransfermsg;
/* body may be NULL if sender used Request_body_nomsg; skip gracefully */
if (!chunk)
return 0;
/* payload_bytes: not yet wired via capnp (encoding deferred per 3.5a).
* Pass NULL/0 — ha_handle_snapshot_chunk handles NULL payload gracefully. */
int rc = ha_handle_snapshot_chunk(&c->ha, &c->persist,
c->persist.snap_path,
chunk,
/*payload_bytes=*/NULL,
/*payload_len=*/0);
if (rc != 0)
ERROR("handle_snapshot_transfer: chunk %u rc=%d",
chunk->chunk_index, rc);
/* Return 0 — errors are logged; primary retries on next ctrlhello (Req 4.4) */
return 0;
}
static int myreq_handler(vtask_t* task)
{
vtask_data_t* data = (vtask_data_t*)task->data;
@ -382,6 +411,13 @@ static int myreq_handler(vtask_t* task)
* learns of progress via the next heartbeat reply's stateReport. */
rc = handle_wal_catchup_entry(c, &request);
break;
case OPCode_snapshotTransfer:
/* Task 3.5c: standby receives one snapshot chunk from the primary.
* Writes chunk payload to snap.part; on isLast renames and loads.
* No reply — primary learns of completion via next heartbeat reply.
* Req 4.2, 4.3, 4.4. */
rc = handle_snapshot_transfer(c, &request);
break;
case OPCode_upgradeConfig:
#if CONF_ENABLE_UPGRADE_CONFIG
/* OPCode_upgradeConfig is sent BY the primary TO standbys to push config.

View file

@ -918,6 +918,95 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
return 0;
}
/* ha_handle_snapshot_chunk — standby snapshot receive handler (Task 3.5c).
*
* Each call appends payload_bytes to snap_dir/snap.part.
* When chunk->is_last is non-zero:
* 1. Verify total bytes written matches chunk->total_size.
* 2. Atomically rename snap.part → snap-{snapshot_lsn:016x}.bin.
* 3. Load the snapshot via persist_load_snapshot().
* 4. Set ha->applied_lsn = p->checkpoint_lsn (in-memory).
*
* On mid-transfer reconnect callers discard snap.part externally (Req 4.4).
* Returns 0 on success, negative errno on error.
* Requirements: 4.2, 4.3, 4.4 (ha-repl-hardening spec)
*/
int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
const char *snap_dir,
snapshot_transfer_t *chunk,
const uint8_t *payload_bytes,
size_t payload_len)
{
if (!ha || !p || !snap_dir || !chunk)
return -EINVAL;
char part_path[4096 + 16];
snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir);
/* Append chunk payload to snap.part */
if (payload_bytes && payload_len > 0) {
FILE *f = fopen(part_path, "ab");
if (!f) {
int err = errno;
ERROR("ha_handle_snapshot_chunk: cannot open snap.part: %s",
strerror(err));
return -err;
}
size_t written = fwrite(payload_bytes, 1, payload_len, f);
fclose(f);
if (written != payload_len) {
ERROR("ha_handle_snapshot_chunk: short write to snap.part "
"(%zu of %zu bytes)", written, payload_len);
return -EIO;
}
}
if (!chunk->is_last)
return 0;
/* isLast=true: verify size, rename, load, set applied_lsn */
/* Verify total file size equals chunk->total_size */
struct stat st;
if (stat(part_path, &st) != 0) {
int err = errno;
ERROR("ha_handle_snapshot_chunk: stat(snap.part) failed: %s",
strerror(err));
return -err;
}
if ((uint64_t)st.st_size != chunk->total_size) {
ERROR("ha_handle_snapshot_chunk: size mismatch — "
"written=%" PRIu64 " expected=%" PRIu64,
(uint64_t)st.st_size, chunk->total_size);
return -EIO;
}
/* Atomic rename snap.part → snap-{lsn:016x}.bin */
char final_path[4096 + 32];
snprintf(final_path, sizeof(final_path),
"%s/snap-%016" PRIx64 ".bin", snap_dir, chunk->snapshot_lsn);
if (rename(part_path, final_path) != 0) {
int err = errno;
ERROR("ha_handle_snapshot_chunk: rename snap.part → %s failed: %s",
final_path, strerror(err));
return -err;
}
/* Load the snapshot — sets p->checkpoint_lsn and p->current_lsn */
int rc = persist_load_snapshot(p, final_path, chunk->snapshot_lsn);
if (rc != 0) {
ERROR("ha_handle_snapshot_chunk: persist_load_snapshot failed rc=%d", rc);
return rc;
}
/* Advance applied_lsn to checkpoint_lsn (in-memory only — Req 4.2) */
atomic_store_explicit(&ha->applied_lsn, p->checkpoint_lsn,
memory_order_release);
INFO("HA: snapshot loaded lsn=%" PRIu64 " applied_lsn set",
p->checkpoint_lsn);
return 0;
}
/* ha_nop_reply_state_report — populate *sr from ha state and wire into reply.
* Reads ha->applied_lsn (atomic) and myconf->config_version_tag, stores them
* in *sr, then sets reply_out->data_which = stateReportMsg and

View file

@ -361,6 +361,30 @@ void ha_process_standby_state(ha_state_t *ha, int peer_idx,
uint64_t reported_tag, uint64_t reported_lsn,
uint64_t committed_tag, uint64_t committed_lsn);
/* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */
/*
* ha_handle_snapshot_chunk: receive one snapshot transfer chunk on the standby.
*
* Each chunk's payload bytes are written (or appended) to snap.part in snap_dir.
* When chunk->is_last is non-zero:
* 1. Verify total bytes written == chunk->total_size.
* 2. Atomically rename snap.part → snap-{snapshot_lsn:016x}.bin.
* 3. Load the snapshot via persist_load_snapshot().
* 4. Set ha->applied_lsn = p->checkpoint_lsn (in-memory only).
*
* On mid-transfer reconnect (caller resets state), snap.part is discarded.
* Returns 0 on success, negative errno on error.
*
* Requirements: 4.2, 4.3, 4.4 (ha-repl-hardening spec)
* Boundary: Standby Dispatch (cluster.c, ha.c)
*/
int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
const char *snap_dir,
snapshot_transfer_t *chunk,
const uint8_t *payload_bytes,
size_t payload_len);
/* ── HA Config Bootstrap & Term Durability (ha-config-bootstrap spec) ── */
/*

View file

@ -676,6 +676,12 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
return 0;
}
/* Public wrapper for load_snapshot — exposed via persist.h for ha.c (Task 3.5c). */
int persist_load_snapshot(persist_t *p, const char *path, uint64_t lsn)
{
return load_snapshot(p, path, lsn);
}
/* P2-2: Standby WAL replay — apply a single replicated entry to in-memory state.
*
* No hooks fire on standbys; the hook system is empty until promotion.

View file

@ -1608,6 +1608,201 @@ static void test_snapshot_transfer_small_less_than_one_chunk(void)
printf("test_snapshot_transfer_small_less_than_one_chunk: PASS\n");
}
/* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */
/* Helper: initialise a minimal ha_state_t for snapshot receive tests */
static void snap_recv_ha_init(ha_state_t *ha)
{
memset(ha, 0, sizeof(*ha));
ha_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
cfg.heartbeat_timeout_ms = 10000;
cfg.election_timeout_ms = 1500;
cfg.committing_timeout_ms = 10000;
ha_init(ha, &cfg, 0);
}
/* Helper: write a minimal valid snapshot file to path (XCESNAP header + body) */
static void write_test_snapshot(const char *path, uint64_t lsn, size_t body_size)
{
FILE *f = fopen(path, "wb");
TEST_ASSERT_NOT_NULL_MESSAGE(f, "Failed to create test snapshot file");
char header[16];
memcpy(header, "XCESNAP\0", 8);
memcpy(header + 8, &lsn, 8);
fwrite(header, 1, 16, f);
if (body_size > 16) {
char *zeros = calloc(1, body_size - 16);
fwrite(zeros, 1, body_size - 16, f);
free(zeros);
}
fclose(f);
}
/* Test 3.5c-A: receive 2 chunks, verify snap.part assembled and renamed correctly,
* applied_lsn set to checkpoint_lsn.
* Requirements: 4.2, 4.3 */
static void test_snapshot_recv_two_chunks_assembles_and_renames(void)
{
char tmpdir[] = "/tmp/xcectest_snaprecv_XXXXXX";
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
/* Create snap_dir subdirectory (matches persist->snap_path layout) */
char snap_dir[512];
snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir);
TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700));
/* Build a fake 128-byte snapshot body (two 64-byte chunks).
* payload_bytes contains the raw file bytes for each chunk.
* We skip writing an actual snapshot source file — the receiver
* writes snap.part and renames it; we use the resulting file to verify. */
uint64_t snap_lsn = 100;
size_t total_size = 128;
uint8_t chunk0[64];
uint8_t chunk1[64];
memset(chunk0, 0xAA, 64);
memset(chunk1, 0xBB, 64);
/* Embed XCESNAP header in chunk0 so load_snapshot can parse it */
memcpy(chunk0, "XCESNAP\0", 8);
memcpy(chunk0 + 8, &snap_lsn, 8);
persist_t *p = calloc(1, sizeof(persist_t));
TEST_ASSERT_NOT_NULL(p);
persist_config_t pcfg = persist_config_defaults();
persist_set_workdir(p, tmpdir, "ctrl-snaprecv");
persist_init(p, &pcfg, NULL);
persist_start(p);
ha_state_t ha;
snap_recv_ha_init(&ha);
/* Chunk 0 */
snapshot_transfer_t c0;
memset(&c0, 0, sizeof(c0));
c0.snapshot_lsn = snap_lsn;
c0.chunk_index = 0;
c0.chunk_offset = 0;
c0.total_size = (uint64_t)total_size;
c0.is_last = 0;
int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, chunk0, 64);
TEST_ASSERT_EQUAL_INT(0, rc);
/* snap.part must exist after first chunk */
char part_path[600];
snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir);
struct stat st;
TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(part_path, &st),
"snap.part should exist after chunk 0");
TEST_ASSERT_EQUAL_UINT64(64, (uint64_t)st.st_size);
/* Chunk 1 (isLast=true) */
snapshot_transfer_t c1;
memset(&c1, 0, sizeof(c1));
c1.snapshot_lsn = snap_lsn;
c1.chunk_index = 1;
c1.chunk_offset = 64;
c1.total_size = (uint64_t)total_size;
c1.is_last = 1;
rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c1, chunk1, 64);
TEST_ASSERT_EQUAL_INT(0, rc);
/* snap.part must be gone (renamed) */
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, stat(part_path, &st),
"snap.part should be gone after isLast");
/* Final file snap-{lsn:016x}.bin must exist */
char final_path[600];
snprintf(final_path, sizeof(final_path),
"%s/snap-%016" PRIx64 ".bin", snap_dir, snap_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(final_path, &st),
"snap-{lsn}.bin should exist after isLast");
TEST_ASSERT_EQUAL_UINT64((uint64_t)total_size, (uint64_t)st.st_size);
/* applied_lsn must be set to checkpoint_lsn (= snap_lsn = 100) */
uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn,
memory_order_acquire);
TEST_ASSERT_EQUAL_UINT64(snap_lsn, got_lsn);
ha_fini(&ha);
persist_stop(p);
persist_fini(p);
free(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_snapshot_recv_two_chunks_assembles_and_renames: PASS\n");
}
/* Test 3.5c-B: size mismatch on isLast returns error (verify size == total_size).
* Requirements: 4.3 */
static void test_snapshot_recv_size_mismatch_fails(void)
{
char tmpdir[] = "/tmp/xcectest_snaprecv2_XXXXXX";
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
char snap_dir[512];
snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir);
TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700));
persist_t *p = calloc(1, sizeof(persist_t));
TEST_ASSERT_NOT_NULL(p);
persist_config_t pcfg = persist_config_defaults();
persist_set_workdir(p, tmpdir, "ctrl-snaprecv2");
persist_init(p, &pcfg, NULL);
persist_start(p);
ha_state_t ha;
snap_recv_ha_init(&ha);
uint64_t snap_lsn = 200;
uint8_t chunk0[64];
memset(chunk0, 0xCC, 64);
/* Claim total_size=200 but only send 64 bytes with is_last=true */
snapshot_transfer_t c0;
memset(&c0, 0, sizeof(c0));
c0.snapshot_lsn = snap_lsn;
c0.chunk_index = 0;
c0.chunk_offset = 0;
c0.total_size = 200; /* mismatch — only 64 bytes will be written */
c0.is_last = 1;
int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, chunk0, 64);
/* Must fail — size mismatch */
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, rc, "size mismatch must return error");
/* applied_lsn must remain 0 (not advanced on failure) */
uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn,
memory_order_acquire);
TEST_ASSERT_EQUAL_UINT64(0, got_lsn);
ha_fini(&ha);
persist_stop(p);
persist_fini(p);
free(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_snapshot_recv_size_mismatch_fails: PASS\n");
}
/* Test 3.5c-C: NULL snap_dir or NULL chunk returns error gracefully */
static void test_snapshot_recv_null_args_fail(void)
{
ha_state_t ha;
snap_recv_ha_init(&ha);
int rc = ha_handle_snapshot_chunk(&ha, NULL, NULL, NULL, NULL, 0);
TEST_ASSERT_NOT_EQUAL(0, rc);
ha_fini(&ha);
printf("test_snapshot_recv_null_args_fail: PASS\n");
}
#endif /* XCE_TESTING */
/* ── dispatch ───────────────────────────────────── */
@ -1652,9 +1847,13 @@ void test_config_repl(void)
RUN_TEST(test_wal_stream_500_entries_two_batches);
RUN_TEST(test_wal_stream_partial_batch);
RUN_TEST(test_wal_stream_empty_range_no_batches);
/* Task 3.5: snapshot chunk transfer */
/* Task 3.5: snapshot chunk transfer (sender side) */
RUN_TEST(test_snapshot_transfer_two_chunks);
RUN_TEST(test_snapshot_transfer_small_less_than_one_chunk);
/* Task 3.5c: standby snapshot receive handler */
RUN_TEST(test_snapshot_recv_two_chunks_assembles_and_renames);
RUN_TEST(test_snapshot_recv_size_mismatch_fails);
RUN_TEST(test_snapshot_recv_null_args_fail);
#endif
/* Task A: raw capnp shims */
RUN_TEST(test_replconfig_req_ssl_encode);

View file

@ -152,6 +152,11 @@ int persist_worker_hook_fun(int type, int situation, int detail,
* negative errno on error (oversized payload, NULL arguments). */
int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload);
/* Load a snapshot file from path, set p->checkpoint_lsn and p->current_lsn.
* lsn hint is informational — the actual LSN is read from the file header.
* Returns 0 on success, -errno or -EINVAL/-EIO on error. */
int persist_load_snapshot(persist_t* p, const char* path, uint64_t lsn);
/* GC retention predicates (P0-5a / P0-8 / P0-13) */
#include <stdbool.h>
#include "xce/types.h"