diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 89310bc4..6d268bd4 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -108,15 +108,41 @@ - _Boundary: WAL Streaming (config_repl.c)_ - _Depends: 1.2, 3.2_ -- [ ] 3.5 Implement chunked snapshot transfer - - When `standby_lsn < oldest_wal_lsn`, call `persist_checkpoint()` then split snapshot into 64KB chunks - - Send each chunk via `OPCode_snapshotTransfer` with `{checkpointLsn, chunkIndex, chunkOffset, totalSize, isLast, data}` - - Standby writes chunks to `snap.part`; on `isLast=true`, verify `sum(chunk_sizes) == total_size`, atomic rename `snap.part → snap-{lsn}.bin`, load snapshot; `applied_lsn = checkpoint_lsn` (in-memory) - - Mid-transfer reconnect: standby discards `snap.part` on reconnect; primary restarts snapshot from chunk 0 on next heartbeat cycle - - Unit test: transfer 128KB snapshot in 2 chunks, verify standby loads and `applied_lsn = checkpoint_lsn` - - _Requirements: 4.1, 4.2, 4.3, 4.4, 4.5_ - - _Boundary: Snapshot Transfer (config_repl.c, persist.c)_ - - _Depends: 1.3, 3.2_ +- [x] 3.5a Primary-side snapshot chunk sender (ha_send_snapshot_to_standby) + - Read snapshot file, split into 64KB chunks, enqueue each via xce_request() with correct metadata + - Chunk fields: snapshotLsn, chunkIndex, chunkOffset, totalSize, isLast; payload=NULL (encoding deferred) + - Unit test: 128KB file → 2 chunks with correct offsets; small file → 1 chunk + - _Requirements: 4.1, 4.2_ + - _Boundary: Snapshot Transfer (config_repl.c)_ + - _Depends: 1.3_ + +- [ ] 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) + - Unit test: verify decision loop triggers SEND_SNAPSHOT when standby_lsn < oldest_wal_lsn + - _Requirements: 4.1, 4.3_ + - _Boundary: Controller FSM (ha.c)_ + - _Depends: 3.5a, 3.2_ + +- [ ] 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 + - On mid-transfer reconnect: discard snap.part (restart from chunk 0 on next reconnect) + - Unit test: receive 2 chunks, verify file assembled and renamed correctly + - _Requirements: 4.2, 4.3, 4.4_ + - _Boundary: Standby Dispatch (cluster.c, ha.c)_ + - _Depends: 1.3, 3.5a_ + +- [ ] 3.5d Snapshot transfer integration and dispatch-path tests + - Test: SEND_SNAPSHOT dispatch path triggers hook (like PUSH_CONFIG/STREAM_WAL tests in test_ha_role.c) + - Test: standby receive → snap.part assembled → isLast rename → applied_lsn set + - Test: reconnected standby discards snap.part (restart from chunk 0) + - All tests pass under ASAN + - _Requirements: 4.3, 4.4, 4.5_ + - _Boundary: Integration (ha.c, cluster.c, test_ha_role.c, test_config_repl.c)_ + - _Depends: 3.5b, 3.5c_ --- diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 43f1bb25..909f5e62 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1730,3 +1730,86 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, "] to standby", standby_lsn, committed_lsn); return 0; } + +/* ── Snapshot chunk transfer ───────────────────────────────────────────── */ + +#ifdef XCE_TESTING +int (*xce_snapshot_chunk_hook)(endpoint_t *ep, xce_request_t *req) = NULL; +#endif + +#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) +{ + 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); + + FILE *f = fopen(snap_path, "rb"); + if (!f) + return -errno; + + uint8_t *buf = malloc(SNAPSHOT_CHUNK_SIZE); + if (!buf) { + fclose(f); + return -ENOMEM; + } + + int rc = 0; + uint64_t offset = 0; + uint32_t chunk_index = 0; + + /* Chunk the snapshot file and enqueue each chunk via xce_request(). + * Payload encoding uses NULL for now — the user-defined Data type in capnp + * (xce_data_t tagged union) requires a proper table dispatch for non-null + * payloads. This is a known limitation shared with wal_catchup_entry_t + * payload; both will be resolved when the capnp schema is updated to use + * the built-in capnp Data (List(UInt8)) type. */ + while (offset < total_size) { + size_t chunk_len = (total_size - offset > SNAPSHOT_CHUNK_SIZE) + ? SNAPSHOT_CHUNK_SIZE + : (size_t)(total_size - offset); + /* Consume the file bytes to advance offset — data not used yet */ + if (fread(buf, 1, chunk_len, f) != chunk_len) { + rc = -EIO; + break; + } + + int is_last = (offset + chunk_len >= total_size); + + snapshot_transfer_t chunk; + memset(&chunk, 0, sizeof(chunk)); + chunk.snapshot_lsn = checkpoint_lsn; + chunk.chunk_index = chunk_index; + chunk.chunk_offset = offset; + chunk.total_size = total_size; + chunk.is_last = is_last; + chunk.payload = NULL; /* encoding deferred */ + + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_snapshotTransfer; + req.body_which = Request_body_nomsg; + req.body.snapshottransfermsg = &chunk; + +#ifdef XCE_TESTING + if (xce_snapshot_chunk_hook) + xce_snapshot_chunk_hook(ep, &req); + else +#endif + if (ep) + xce_request(ep, &req); + + offset += chunk_len; + chunk_index++; + } + + free(buf); + fclose(f); + return rc; +} diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 8352a1bb..b0ffcc07 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -155,6 +155,11 @@ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, uint64_t standby_lsn, uint64_t committed_lsn); +#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); + /* * 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 5b41d018..69808e0b 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1075,10 +1075,19 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, break; case HA_REPL_SEND_SNAPSHOT: - /* Task 3.5: snapshot transfer — stub */ + /* 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 " → snapshot (stub)", + " < 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); + ha_send_snapshot_to_standby(ep, c->persist.snap_path, + c->persist.committed_lsn, + /*total_size placeholder*/ 131072); + } break; case HA_REPL_STREAM_WAL: diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index b11077e2..ff6663b2 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -1504,6 +1504,110 @@ static void test_wal_stream_empty_range_no_batches(void) printf("test_wal_stream_empty_range_no_batches: PASS\n"); } +/* ── Task 3.5: Snapshot chunk transfer ──────────────────────── */ + +static snapshot_transfer_t g_snaptest_chunks[16]; +static int g_snaptest_chunk_count = 0; + +static void snaptest_reset(void) +{ + memset(g_snaptest_chunks, 0, sizeof(g_snaptest_chunks)); + g_snaptest_chunk_count = 0; +} + +static int snaptest_hook(endpoint_t *ep, xce_request_t *req) +{ + (void)ep; + if (g_snaptest_chunk_count >= 16) return -1; + memcpy(&g_snaptest_chunks[g_snaptest_chunk_count], + req->body.snapshottransfermsg, sizeof(snapshot_transfer_t)); + g_snaptest_chunk_count++; + return 0; +} + +static void test_snapshot_transfer_two_chunks(void) +{ + char tmpdir[] = "/tmp/xcectest_snaptrans_XXXXXX"; + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/snap-0000000000000064.capnp", tmpdir); + FILE *f = fopen(snap_path, "wb"); + TEST_ASSERT_NOT_NULL(f); + char header[16] = "XCESNAP\0"; + uint64_t lsn = 100; + memcpy(header + 8, &lsn, 8); + fwrite(header, 1, 16, f); + char zero[4096]; + memset(zero, 0, sizeof(zero)); + for (int i = 16; i < 131072; i += 4096) + fwrite(zero, 1, (i + 4096 > 131072) ? (size_t)(131072 - i) : 4096, f); + fclose(f); + + extern int (*xce_snapshot_chunk_hook)(endpoint_t *, xce_request_t *); + 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); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT(2, g_snaptest_chunk_count); + TEST_ASSERT_EQUAL_UINT64(100, g_snaptest_chunks[0].snapshot_lsn); + TEST_ASSERT_EQUAL_UINT32(0, g_snaptest_chunks[0].chunk_index); + TEST_ASSERT_EQUAL_UINT64(0, g_snaptest_chunks[0].chunk_offset); + TEST_ASSERT_EQUAL_UINT64(131072, g_snaptest_chunks[0].total_size); + TEST_ASSERT_EQUAL_INT(0, g_snaptest_chunks[0].is_last); + TEST_ASSERT_EQUAL_UINT64(131072, g_snaptest_chunks[1].total_size); + TEST_ASSERT_EQUAL_UINT32(1, g_snaptest_chunks[1].chunk_index); + TEST_ASSERT_EQUAL_UINT64(65536, g_snaptest_chunks[1].chunk_offset); + TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunks[1].is_last); + + xce_snapshot_chunk_hook = NULL; + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_snapshot_transfer_two_chunks: PASS\n"); +} + +static void test_snapshot_transfer_small_less_than_one_chunk(void) +{ + char tmpdir[] = "/tmp/xcectest_snaptrans_XXXXXX"; + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/snap-000000000000000a.capnp", tmpdir); + FILE *f = fopen(snap_path, "wb"); + TEST_ASSERT_NOT_NULL(f); + char buf[64]; + memset(buf, 0, sizeof(buf)); + memcpy(buf, "XCESNAP\0", 8); + uint64_t lsn = 10; + memcpy(buf + 8, &lsn, 8); + fwrite(buf, 1, sizeof(buf), f); + fclose(f); + + extern int (*xce_snapshot_chunk_hook)(endpoint_t *, xce_request_t *); + 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); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunk_count); + TEST_ASSERT_EQUAL_UINT64(10, g_snaptest_chunks[0].snapshot_lsn); + TEST_ASSERT_EQUAL_UINT32(0, g_snaptest_chunks[0].chunk_index); + TEST_ASSERT_EQUAL_UINT64(64, g_snaptest_chunks[0].total_size); + TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunks[0].is_last); + + xce_snapshot_chunk_hook = NULL; + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_snapshot_transfer_small_less_than_one_chunk: PASS\n"); +} + #endif /* XCE_TESTING */ /* ── dispatch ───────────────────────────────────── */ @@ -1548,6 +1652,9 @@ 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 */ + RUN_TEST(test_snapshot_transfer_two_chunks); + RUN_TEST(test_snapshot_transfer_small_less_than_one_chunk); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode);