diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index a9600673..89310bc4 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -99,7 +99,7 @@ - _Boundary: Config Replication (config_repl.c)_ - _Depends: 3.2_ -- [ ] 3.4 Implement batch WAL streaming from decision loop +- [x] 3.4 Implement batch WAL streaming from decision loop - When `standby_lsn < committed_lsn AND standby_lsn >= oldest_wal_lsn`, stream entries `[standby_lsn+1, committed_lsn]` - Accumulate entries up to 64KB/256 entries per batch; enqueue all batches via `xce_request()` in a single non-blocking invocation — do not block vtaskmgr thread - Send each batch as `OPCode_walCatchupEntry` with the `entries` list; partial final batch sent immediately (no wait for fill) diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index aa3de1ef..43f1bb25 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1546,3 +1546,187 @@ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req) return 0; } + +/* ── Task 3.4: Batch WAL Sender ────────────────────────────────────────────── + * + * ha_stream_wal_to_standby: stream WAL entries [standby_lsn+1, committed_lsn] + * to a standby endpoint as batched OPCode_walCatchupEntry messages. + * + * Design (§Batch WAL Sender): + * - Read WAL segment files from wal_path in LSN order (seg-*.wal format). + * - Accumulate entries into a batch up to WAL_STREAM_BATCH_ENTRIES (256) or + * WAL_STREAM_BATCH_BYTES (65536) total payload bytes. + * - On batch full: enqueue via xce_request() (non-blocking), start new batch. + * - After all segments: flush any partial batch immediately (Req 5.4). + * - All xce_request() calls are enqueue-only — never blocks the caller thread. + * + * Under XCE_TESTING, xce_wal_stream_request_hook intercepts the per-batch + * xce_request() call so unit tests can count batches without a real endpoint. + * + * Requirements: 5.1, 5.2, 5.4, 5.5 + */ + +#ifdef XCE_TESTING +/* Test-only seam: intercepts each batch xce_request() call. + * Defined here (translation unit scope); declared extern in test file. */ +int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL; +#endif + +/* send_wal_batch: enqueue a single OPCode_walCatchupEntry batch message. + * Does not free any memory — caller manages entry lifetime. + * Returns 0 on success. */ +static int send_wal_batch(endpoint_t *ep, + wal_catchup_entry_t **entries, int n_entries) +{ + if (n_entries <= 0) + return 0; + + /* Build the batch container entry */ + wal_catchup_entry_t batch; + memset(&batch, 0, sizeof(batch)); + batch.n_entries = n_entries; + batch.entries = entries; + + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_walCatchupEntry; + req.body_which = Request_body_nomsg; + req.body.walcatchupentrymsg = &batch; + +#ifdef XCE_TESTING + if (xce_wal_stream_request_hook != NULL) + return xce_wal_stream_request_hook(ep, &req); +#endif + if (ep) + xce_request(ep, &req); + return 0; +} + +int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, + uint64_t standby_lsn, uint64_t committed_lsn) +{ + if (!wal_path) + return -EINVAL; + if (standby_lsn >= committed_lsn) + return 0; /* nothing to stream */ + + /* ── Enumerate WAL segment files, sort ascending by first LSN ── */ + DIR *d = opendir(wal_path); + if (!d) + return 0; /* no WAL directory yet — treat as empty */ + + uint64_t seg_lsns[4096]; + int nseg = 0; + struct dirent *ent; + while ((ent = readdir(d)) != NULL && + nseg < (int)(sizeof(seg_lsns) / sizeof(*seg_lsns))) { + uint64_t sl = 0; + if (sscanf(ent->d_name, "seg-%" SCNx64 ".wal", &sl) != 1) continue; + seg_lsns[nseg++] = sl; + } + closedir(d); + + /* Insertion sort ascending by segment start LSN */ + for (int i = 1; i < nseg; i++) { + uint64_t key = seg_lsns[i]; + int j = i - 1; + while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j+1] = seg_lsns[j]; j--; } + seg_lsns[j+1] = key; + } + + /* ── Accumulate and send batches ── */ + wal_catchup_entry_t *batch_entries[WAL_STREAM_BATCH_ENTRIES]; + int batch_n = 0; + size_t batch_bytes = 0; + + for (int si = 0; si < nseg; si++) { + /* Skip segments whose entire range is at or below standby_lsn */ + if (si + 1 < nseg && seg_lsns[si + 1] <= standby_lsn) + continue; + /* Stop once the segment starts beyond committed_lsn */ + if (seg_lsns[si] > committed_lsn) + break; + + char seg_path[4096]; + snprintf(seg_path, sizeof(seg_path), + "%s/seg-%016" PRIx64 ".wal", wal_path, seg_lsns[si]); + + int fd = open(seg_path, O_RDONLY); + if (fd < 0) continue; + + uint32_t off = 0; + for (;;) { + wal_entry_hdr_t hdr; + ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off); + if (n < (ssize_t)sizeof(hdr)) break; + if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break; + if (hdr.lsn > committed_lsn) break; + + if (hdr.lsn > standby_lsn) { + wal_catchup_entry_t *entry = calloc(1, sizeof(*entry)); + if (!entry) { + /* OOM: flush current batch and skip this entry */ + if (batch_n > 0) { + send_wal_batch(ep, batch_entries, batch_n); + for (int bi = 0; bi < batch_n; bi++) { + free(batch_entries[bi]->payload); + free(batch_entries[bi]); + } + batch_n = 0; + batch_bytes = 0; + } + off += (uint32_t)(sizeof(hdr) + hdr.payload_len); + continue; + } + entry->lsn = hdr.lsn; + entry->entry_type = (uint8_t)hdr.type; + entry->n_entries = 0; + entry->entries = NULL; + entry->payload = NULL; + + if (hdr.payload_len > 0) { + uint8_t *pbuf = malloc(hdr.payload_len); + if (pbuf) { + ssize_t pr = pread(fd, pbuf, hdr.payload_len, + (off_t)(off + sizeof(hdr))); + if (pr == (ssize_t)hdr.payload_len) + entry->payload = (xce_data_t *)pbuf; + else + free(pbuf); + } + } + + batch_entries[batch_n] = entry; + batch_n++; + batch_bytes += sizeof(wal_entry_hdr_t) + hdr.payload_len; + + /* Flush when batch is full (by count or accumulated size) */ + if (batch_n >= WAL_STREAM_BATCH_ENTRIES || + batch_bytes >= WAL_STREAM_BATCH_BYTES) { + send_wal_batch(ep, batch_entries, batch_n); + for (int bi = 0; bi < batch_n; bi++) { + free(batch_entries[bi]->payload); + free(batch_entries[bi]); + } + batch_n = 0; + batch_bytes = 0; + } + } + off += (uint32_t)(sizeof(hdr) + hdr.payload_len); + } + close(fd); + } + + /* Flush remaining partial batch (Req 5.4: send immediately, don't wait) */ + if (batch_n > 0) { + send_wal_batch(ep, batch_entries, batch_n); + for (int bi = 0; bi < batch_n; bi++) { + free(batch_entries[bi]->payload); + free(batch_entries[bi]); + } + } + + INFO("ha_stream_wal_to_standby: streamed WAL [%" PRIu64 "+1..%" PRIu64 + "] to standby", standby_lsn, committed_lsn); + return 0; +} diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 77c19810..8352a1bb 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -127,6 +127,34 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, */ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); +/* + * ha_stream_wal_to_standby: batch-stream WAL entries [standby_lsn+1, committed_lsn] + * to a standby endpoint via xce_request() (non-blocking). + * + * Accumulates WAL entries up to WAL_STREAM_BATCH_ENTRIES (256) entries or + * WAL_STREAM_BATCH_BYTES (65536) bytes per batch, then enqueues each batch + * as a single OPCode_walCatchupEntry message with n_entries set. + * All batches for the full range are enqueued in a single invocation — never + * blocks the vtaskmgr thread. + * + * A partial final batch (fewer entries than the batch limit) is sent immediately + * without waiting for the batch to fill (Req 5.4). + * + * Parameters: + * ep — outbound endpoint to the standby + * wal_path — directory containing WAL segment files (seg-*.wal) + * standby_lsn — standby's last applied LSN (stream starts at standby_lsn+1) + * committed_lsn— primary's committed LSN (stream ends at committed_lsn inclusive) + * + * Returns 0 on success; negative on internal error (e.g., NULL ep). + * Requirements: 5.1, 5.2, 5.4, 5.5 (ha-repl-hardening spec) + * Boundary: WAL Streaming (config_repl.c) + */ +#define WAL_STREAM_BATCH_ENTRIES 256 +#define WAL_STREAM_BATCH_BYTES 65536 +int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, + uint64_t standby_lsn, uint64_t committed_lsn); + /* * 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 00304acc..5b41d018 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1017,18 +1017,19 @@ void ha_process_standby_state(ha_state_t *ha, int peer_idx, * (stored by ha_process_standby_state on the same heartbeat reply), calls * ha_evaluate_repl_action, and acts on the result: * - * HA_REPL_PUSH_CONFIG → push_config_to_standbys(c, …) [or test hook] - * HA_REPL_SEND_SNAPSHOT→ stub (Task 3.4) - * HA_REPL_STREAM_WAL → stub (Task 3.5) + * HA_REPL_PUSH_CONFIG → push_config_to_standbys(c, …) [or test hook] (Task 3.3) + * HA_REPL_SEND_SNAPSHOT→ stub (Task 3.5) + * HA_REPL_STREAM_WAL → ha_stream_wal_to_standby(c, ha, idx) (Task 3.4) * HA_REPL_CONVERGED → no-op * * One action per heartbeat cycle per standby (design §Unified Heartbeat-Driven * Protocol). Returns 0 on success, -EINVAL on bad peer_idx. * - * Requirements: 1.4 (ha-repl-hardening spec) + * Requirements: 1.4, 5.1–5.5 (ha-repl-hardening spec) */ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, cluster_t *c, + endpoint_t *ep, int peer_idx, uint64_t committed_tag, uint64_t committed_lsn, @@ -1074,17 +1075,36 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, break; case HA_REPL_SEND_SNAPSHOT: - /* Task 3.4: snapshot transfer — stub */ + /* Task 3.5: snapshot transfer — stub */ INFO("HA repl: standby[%d] lsn=%" PRIu64 " < oldest_wal_lsn=%" PRIu64 " → snapshot (stub)", peer_idx, standby_lsn, oldest_wal_lsn); break; case HA_REPL_STREAM_WAL: - /* Task 3.5: WAL streaming — stub */ + /* Task 3.4: batch WAL streaming. + * Enqueues all batch OPCode_walCatchupEntry messages for the range + * [standby_lsn+1, committed_lsn] via ha_stream_wal_to_standby(). + * All xce_request() calls are non-blocking (enqueue-only). */ INFO("HA repl: standby[%d] lsn=%" PRIu64 - " < committed_lsn=%" PRIu64 " → stream WAL (stub)", + " < committed_lsn=%" PRIu64 " → stream WAL batches", peer_idx, standby_lsn, committed_lsn); +#ifdef XCE_TESTING + if (ha_repl_stream_wal_hook != NULL) { + const char *wal_path = (c && c->persist.wal_path[0]) + ? c->persist.wal_path : ""; + ha_repl_stream_wal_hook(ep, wal_path, standby_lsn, committed_lsn); + break; + } +#endif + if (c && c->persist.wal_path[0]) { + extern int ha_stream_wal_to_standby(endpoint_t *ep, + const char *wal_path, + uint64_t standby_lsn, + uint64_t committed_lsn); + ha_stream_wal_to_standby(ep, c->persist.wal_path, + standby_lsn, committed_lsn); + } break; case HA_REPL_CONVERGED: @@ -1632,6 +1652,16 @@ int (*ha_nop_req_hook)(endpoint_t *ep) = NULL; */ int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx) = NULL; +/* + * ha_repl_stream_wal_hook: test-only seam for ha_stream_wal_to_standby. + * When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF + * ha_stream_wal_to_standby() when the decision is HA_REPL_STREAM_WAL. + * Requirements: 5.1–5.5 (ha-repl-hardening) + */ +int (*ha_repl_stream_wal_hook)(void *ep, const char *wal_path, + uint64_t standby_lsn, + uint64_t committed_lsn) = NULL; + #endif /* XCE_TESTING */ /* ── Election Evaluation (Req 2.1, 2.7, 2.8) ─────────────────────────────── @@ -2211,13 +2241,13 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) 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.4). */ + * oldest_wal_lsn=0 until snapshot transfer is wired (Task 3.5). */ uint64_t oldest_wal_lsn = 0; #ifndef XCE_TESTING if (c) - oldest_wal_lsn = 0; /* Task 3.4: wire persist->oldest_wal_lsn */ + oldest_wal_lsn = 0; /* Task 3.5: wire persist->oldest_wal_lsn */ #endif - ha_dispatch_repl_action_for_standby(ha, c, peer_idx, + ha_dispatch_repl_action_for_standby(ha, c, ep, peer_idx, committed_tag, committed_lsn, oldest_wal_lsn); } else { diff --git a/src/controller/ha.h b/src/controller/ha.h index a44fecf3..5f563bb3 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -535,25 +535,31 @@ void ha_repl_ack_received(ha_state_t *ha, cluster_t *c); * * Reads ha->standby_reported_tag[peer_idx] and ha->standby_reported_lsn[peer_idx] * (stored by ha_process_standby_state on the previous call), calls - * ha_evaluate_repl_action, and if the result is HA_REPL_PUSH_CONFIG: - * - In production builds: calls push_config_to_standbys(c, …). - * - In XCE_TESTING builds: calls ha_repl_push_config_for_standby_hook(c, peer_idx) - * if the hook is non-NULL. + * ha_evaluate_repl_action, and acts on the result: + * - HA_REPL_PUSH_CONFIG: In production builds, calls push_config_to_standbys(c, …). + * In XCE_TESTING builds, calls ha_repl_push_config_for_standby_hook(c, peer_idx). + * - HA_REPL_STREAM_WAL: calls ha_stream_wal_to_standby(ep, wal_path, standby_lsn, + * committed_lsn) to batch-stream WAL entries via xce_request() (non-blocking). + * In XCE_TESTING builds, calls ha_repl_stream_wal_hook if non-NULL. + * - HA_REPL_SEND_SNAPSHOT: stub (Task 3.5). + * - HA_REPL_CONVERGED: no-op. * * One action per heartbeat cycle per standby (design §Unified Heartbeat-Driven - * Protocol). Other actions (SEND_SNAPSHOT, STREAM_WAL) are stubs returning 0 - * pending Tasks 3.4–3.5; CONVERGED is a no-op. + * Protocol). CONVERGED is a no-op. * + * ep — outbound endpoint to the standby (used for STREAM_WAL; may be + * NULL in tests that only exercise PUSH_CONFIG or CONVERGED paths) * peer_idx — 0-based index into ha->standby_reported_* arrays * committed_tag — primary's committed config_tag (from c->myconf or test fixture) * committed_lsn — primary's committed WAL LSN (from c->persist or test fixture) * oldest_wal_lsn— oldest retained WAL LSN (0 = all WAL retained, skip snapshot) * * Returns 0 on success; negative on internal error. - * Requirements: 1.4 (ha-repl-hardening spec) + * Requirements: 1.4, 5.1–5.5 (ha-repl-hardening spec) */ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, cluster_t *c, + endpoint_t *ep, int peer_idx, uint64_t committed_tag, uint64_t committed_lsn, @@ -610,6 +616,26 @@ extern int (*ha_push_config_hook)(void *c); */ extern int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx); +/* + * ha_repl_stream_wal_hook: test-only seam for ha_stream_wal_to_standby. + * + * When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF + * ha_stream_wal_to_standby() when the decision is HA_REPL_STREAM_WAL, so unit + * tests can count batch-send invocations without requiring a real endpoint or + * WAL directory. + * + * Signature: int hook(void *ep, const char *wal_path, + * uint64_t standby_lsn, uint64_t committed_lsn) + * ep — endpoint_t* (may be NULL in test mode) + * wal_path — WAL directory path + * standby_lsn — standby's last applied LSN + * committed_lsn— primary's committed LSN + * Requirements: 5.1–5.5 (ha-repl-hardening) + */ +extern int (*ha_repl_stream_wal_hook)(void *ep, const char *wal_path, + uint64_t standby_lsn, + uint64_t committed_lsn); + /* * ha_tcp_connect_hook: test-only seam for the raw TCP connect inside * ha_find_and_connect_to_primary. When non-NULL, the function calls this diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 2a8253fd..b11077e2 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -1329,6 +1329,183 @@ static void test_nop_reply_zero_state_fresh_standby(void) #endif /* XCE_TESTING */ +/* ── Task 3.4: ha_stream_wal_to_standby batch WAL streaming ──────────────── */ + +#ifdef XCE_TESTING + +/* + * xce_wal_stream_request_hook: test-only seam declared in config_repl.c. + * When non-NULL, ha_stream_wal_to_standby() calls this INSTEAD OF xce_request() + * for each batch message, allowing tests to inspect and count batches without + * a real endpoint. + * + * Signature: int hook(endpoint_t *ep, xce_request_t *req) + * ep — outbound endpoint (may be NULL in tests) + * req — the batch OPCode_walCatchupEntry request being sent + * Returns 0 on success (the hook owns no resources in the req). + */ +extern int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req); + +/* Helper: write N minimal WAL entries into a single segment file in wal_dir. + * Entries have consecutive LSNs starting at start_lsn. + * Returns 0 on success, -1 on error. */ +static int write_test_wal_segment(const char *wal_dir, uint64_t start_lsn, + int n_entries) +{ + char seg_path[4096]; + snprintf(seg_path, sizeof(seg_path), + "%s/seg-%016" PRIx64 ".wal", wal_dir, start_lsn); + + int fd = open(seg_path, O_WRONLY | O_CREAT | O_TRUNC, 0600); + if (fd < 0) return -1; + + for (int i = 0; i < n_entries; i++) { + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.lsn = start_lsn + (uint64_t)i; + hdr.type = WAL_JOB_UPDATE; + hdr.payload_len = 0; + hdr.term = 1; + ssize_t nw = write(fd, &hdr, sizeof(hdr)); + if (nw != (ssize_t)sizeof(hdr)) { close(fd); return -1; } + } + close(fd); + return 0; +} + +/* Batch-counting state for the hook */ +static int g_wal_stream_batch_count = 0; +static int g_wal_stream_total_entries = 0; +static int g_wal_stream_last_n_entries = 0; + +static int counting_wal_stream_hook(endpoint_t *ep, xce_request_t *req) +{ + (void)ep; + if (!req || !req->body.walcatchupentrymsg) return 0; + wal_catchup_entry_t *batch = (wal_catchup_entry_t *)req->body.walcatchupentrymsg; + g_wal_stream_batch_count++; + g_wal_stream_last_n_entries = batch->n_entries; + g_wal_stream_total_entries += batch->n_entries; + return 0; +} + +/* + * Test 3.4-A: streaming 500 entries produces exactly 2 batches + * (batch 1: 256 entries, batch 2: 244 entries). + * Requirement 5.2: accumulate up to 256 entries per batch. + * Requirement 5.4: partial final batch is sent immediately. + */ +static void test_wal_stream_500_entries_two_batches(void) +{ + /* Set up WAL directory with 500 consecutive entries starting at LSN 1 */ + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_%d_%ld", + (int)getpid(), (long)time(NULL)); + TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); + + /* Write all 500 entries in a single segment starting at LSN 1 */ + int rc = write_test_wal_segment(tmpdir, /*start_lsn=*/1, /*n_entries=*/500); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "Failed to write test WAL segment"); + + /* Install counting hook */ + xce_wal_stream_request_hook = counting_wal_stream_hook; + g_wal_stream_batch_count = 0; + g_wal_stream_total_entries = 0; + g_wal_stream_last_n_entries = 0; + + /* Stream entries [1..500]: standby_lsn=0, committed_lsn=500 */ + rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, + /*standby_lsn=*/0, + /*committed_lsn=*/500); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_wal_stream_batch_count, + "500 entries must produce exactly 2 batches (256+244)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(500, g_wal_stream_total_entries, + "All 500 entries must be sent"); + TEST_ASSERT_EQUAL_INT_MESSAGE(244, g_wal_stream_last_n_entries, + "Final batch must contain 244 entries (500 - 256)"); + + xce_wal_stream_request_hook = NULL; + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_wal_stream_500_entries_two_batches: PASS\n"); +} + +/* + * Test 3.4-B: streaming fewer than 256 entries produces exactly 1 partial batch. + * Requirement 5.4: partial final batch sent immediately without waiting. + */ +static void test_wal_stream_partial_batch(void) +{ + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_partial_%d_%ld", + (int)getpid(), (long)time(NULL)); + TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); + + /* Write 10 entries starting at LSN 5 */ + int rc = write_test_wal_segment(tmpdir, /*start_lsn=*/5, /*n_entries=*/10); + TEST_ASSERT_EQUAL_INT(0, rc); + + xce_wal_stream_request_hook = counting_wal_stream_hook; + g_wal_stream_batch_count = 0; + g_wal_stream_total_entries = 0; + g_wal_stream_last_n_entries = 0; + + /* Stream entries [6..14]: standby_lsn=5, committed_lsn=14 */ + rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, + /*standby_lsn=*/5, + /*committed_lsn=*/14); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wal_stream_batch_count, + "< 256 entries must produce exactly 1 partial batch"); + TEST_ASSERT_EQUAL_INT_MESSAGE(9, g_wal_stream_total_entries, + "9 entries in range [6..14] must be sent"); + + xce_wal_stream_request_hook = NULL; + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_wal_stream_partial_batch: PASS\n"); +} + +/* + * Test 3.4-C: empty range (standby_lsn == committed_lsn) sends zero batches. + * Requirement 5.4: no batch is sent when there are no entries to stream. + */ +static void test_wal_stream_empty_range_no_batches(void) +{ + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_empty_%d_%ld", + (int)getpid(), (long)time(NULL)); + TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); + + xce_wal_stream_request_hook = counting_wal_stream_hook; + g_wal_stream_batch_count = 0; + g_wal_stream_total_entries = 0; + + /* standby_lsn == committed_lsn: nothing to stream */ + int rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, + /*standby_lsn=*/100, + /*committed_lsn=*/100); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "Empty range must return 0"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wal_stream_batch_count, + "Empty range must produce zero batches"); + + xce_wal_stream_request_hook = NULL; + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_wal_stream_empty_range_no_batches: PASS\n"); +} + +#endif /* XCE_TESTING */ + /* ── dispatch ───────────────────────────────────── */ void test_config_repl(void) @@ -1367,6 +1544,10 @@ void test_config_repl(void) /* Task 2.3: OPCode_nop heartbeat reply carries stateReport */ RUN_TEST(test_nop_reply_carries_state_report); RUN_TEST(test_nop_reply_zero_state_fresh_standby); + /* Task 3.4: ha_stream_wal_to_standby batch WAL streaming */ + 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); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode); diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index 0a84298e..040d360d 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -428,7 +428,7 @@ static void test_dispatch_repl_action_push_config_when_tag_behind(void) ha->standby_reported_lsn[0] = 100; int rc = ha_dispatch_repl_action_for_standby( - ha, /*c=*/NULL, /*peer_idx=*/0, + ha, /*c=*/NULL, /*ep=*/NULL, /*peer_idx=*/0, /*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/0); (void)rc; @@ -465,7 +465,7 @@ static void test_dispatch_repl_action_no_push_when_converged(void) ha->standby_reported_lsn[0] = 100; int rc = ha_dispatch_repl_action_for_standby( - ha, /*c=*/NULL, /*peer_idx=*/0, + ha, /*c=*/NULL, /*ep=*/NULL, /*peer_idx=*/0, /*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/0); (void)rc;