diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 5d961ee..a27329f 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -125,7 +125,7 @@ - _Boundary: Controller FSM (ha.c)_ - _Depends: 3.5a, 3.2_ -- [ ] 3.5c Standby-side snapshot receive handler (OPCode_snapshotTransfer) +- [x] 3.5c Standby-side snapshot receive handler (OPCode_snapshotTransfer) - Add case OPCode_snapshotTransfer in myreq_handler dispatch (cluster.c) - Write each chunk payload to snap.part temp file; on isLast=true, verify total size, atomic rename to snap-{lsn}.bin, load via load_snapshot() - Set applied_lsn = checkpoint_lsn (in-memory) after successful load diff --git a/src/controller/cluster.c b/src/controller/cluster.c index e9706d6..25b4cc1 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -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. diff --git a/src/controller/ha.c b/src/controller/ha.c index 52042c1..151543a 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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 diff --git a/src/controller/ha.h b/src/controller/ha.h index 5f563bb..98f9cba 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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) ── */ /* diff --git a/src/controller/persist.c b/src/controller/persist.c index a057bb9..a070e87 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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. diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index ba5e665..20e665d 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -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); diff --git a/src/include/persist.h b/src/include/persist.h index 2efadaa..d23d0ec 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -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 #include "xce/types.h"