lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-repl-hardening): batch WAL streaming from decision loop (task 3.4)

Browse source 
ha_stream_wal_to_standby() reads WAL from committed_lsn range, accumulates
entries to 256/64KB per batch, enqueues all via xce_request() (non-blocking).
Partial final batch sent immediately. Wired into STREAM_WAL case.
3 tests: 500-entries/2-batches, partial batch, empty range.
Fixed stale task-number comments per review.
This commit is contained in:
Rongsong Shen 2026-06-04 17:57:55 +00:00
parent d1c7872d73
commit 6b69a47c50
7 changed files with 469 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

184
src/controller/config_repl.c
View file

@ -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;
}