forked from lei.chen/myxce
feat(ha-repl-hardening): primary snapshot chunk sender (task 3.5a)
ha_send_snapshot_to_standby reads snapshot file, splits into 64KB chunks, sends each via OPCode_snapshotTransfer with correct metadata (snapshotLsn, chunkIndex, chunkOffset, totalSize, isLast). Payload encoding deferred. Hook seam for unit testing. 2 tests: 128KB/2-chunks, small-file/1-chunk. Tasks 3.5b-3.5d remain.
This commit is contained in:
parent
6b69a47c50
commit
834ab4d5e7
5 changed files with 241 additions and 11 deletions
|
|
@ -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_
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue