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

feat(ha-repl-hardening): add batch WAL catch-up on standby side (task 2.2)

Browse source 
Extend ha_apply_wal_catchup_entry with batch branch:
- Check n_entries > 0, iterate entries[] list in order
- Apply each via apply_wal_entry(); skip NULL/errors without crash
- Advance applied_lsn once after full batch (not per-entry)
- n_entries == 0 falls through to task 2.1 single-entry path
4 new unit tests covering full batch, partial batch, error skip, fallback
This commit is contained in:
Rongsong Shen 2026-06-04 15:38:20 +00:00
parent 647b88ce57
commit 6bbcff62ca
3 changed files with 295 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/ha-repl-hardening/tasks.md
View file

@ -51,7 +51,7 @@
- _Requirements: 3.1, 3.3, 3.4, 3.5_
- _Boundary: Standby Dispatch (cluster.c / ha.c)_
- [ ] 2.2 Handle batch WAL catch-up entries on standby
- [x] 2.2 Handle batch WAL catch-up entries on standby
- Extend the `OPCode_walCatchupEntry` handler from task 2.1 to decode the `entries` list for batch messages
- Apply each entry in order via `apply_wal_entry()`; advance `applied_lsn` only after the full batch is durably written
- Standby remains in `CTRL_STANDBY` during batch apply; `applied_lsn` not persisted to disk

136
src/controller/ha.c
View file

@ -803,53 +803,38 @@ void ha_fini(ha_state_t *ha)
}
}
/* ── Task 2.1: WAL catch-up entry application (standby side) ────────────────
/* ── Task 2.1/2.2: WAL catch-up entry application (standby side) ─────────────
*
* ha_apply_wal_catchup_entry: apply a single WAL catch-up entry received from
* the primary to local persistent storage and advance applied_lsn on success.
* ha_apply_wal_catchup_entry: apply WAL catch-up entries received from the
* primary to local persistent storage and advance applied_lsn on success.
*
* On success (rc == 0): atomically advances ha->applied_lsn to entry->lsn+1
* so the next heartbeat reply's stateReport reflects the latest state.
* Two modes (Req 5.3, 5.5):
*
* On error (rc < 0): logs ERROR and returns the error code. The dispatch
* handler in cluster.c ignores the error and continues, satisfying Req 3.3
* (corrupt/disk-error entries are skipped, not crashed).
* Batch mode (entry->n_entries > 0):
* Iterate entry->entries[0..n_entries-1] in order. Apply each sub-entry
* individually via apply_wal_entry(). NULL sub-entries are treated as
* corrupt and skipped (Req 3.3). After ALL sub-entries are processed,
* advance applied_lsn to last-successfully-applied lsn+1 so the next
* heartbeat reply reflects post-batch state (Req 5.3).
*
* applied_lsn is in-memory only (not persisted to disk). On restart the
* standby reports lsn=0 and re-bootstraps via the heartbeat decision loop
* (design §Durability).
* Single-entry mode (entry->n_entries == 0):
* Backward-compatible path from task 2.1. Apply entry->lsn/entry_type
* directly and advance applied_lsn to lsn+1 on success.
*
* Exported under XCE_TESTING so unit tests can call it directly without
* constructing a full cluster_t or endpoint_t.
* In both modes:
* - applied_lsn is advanced atomically and monotonically.
* - On error for any individual entry: log ERROR, skip, continue (Req 3.3).
* - applied_lsn is in-memory only; not persisted to disk (design §Durability).
* - Standby stays in CTRL_STANDBY throughout (Req 3.4).
*
* Returns 0 on success (even if some batch entries were skipped due to errors).
* Returns -EINVAL if ha, p, or entry is NULL.
*/
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
wal_catchup_entry_t *entry)
/* ha_advance_applied_lsn: atomically advance ha->applied_lsn to new_lsn
* if new_lsn > current value (monotonic). */
static void ha_advance_applied_lsn(ha_state_t *ha, uint64_t new_lsn)
{
if (!ha || !p || !entry)
return -EINVAL;
/* Build a wal_entry_hdr_t from the decoded catch-up entry fields.
* payload_len is 0 when entry->payload is NULL (no payload for this
* entry type, or single-entry path before task 3.4 encodes real data). */
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = entry->lsn;
hdr.type = (uint32_t)entry->entry_type;
hdr.payload_len = 0;
hdr.crc = 0;
hdr.prev_crc = 0;
int rc = apply_wal_entry(p, &hdr, NULL);
if (rc != 0) {
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for "
"lsn=%" PRIu64 " type=%u rc=%d — skipping entry",
entry->lsn, (unsigned)entry->entry_type, rc);
return rc;
}
/* Advance applied_lsn atomically and monotonically (lsn+1 mirrors
* persist->current_lsn after a successful apply). */
uint64_t new_lsn = entry->lsn + 1;
uint64_t old_lsn = (uint64_t)atomic_load_explicit(&ha->applied_lsn,
memory_order_acquire);
while (new_lsn > old_lsn) {
@ -858,7 +843,78 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
memory_order_acq_rel, memory_order_acquire))
break;
}
}
/* ha_apply_single_entry: apply one decoded wal_catchup_entry_t sub-entry.
* Returns 0 on success, non-zero on error (caller logs and skips). */
static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
{
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = e->lsn;
hdr.type = (uint32_t)e->entry_type;
hdr.payload_len = 0;
hdr.crc = 0;
hdr.prev_crc = 0;
return apply_wal_entry(p, &hdr, NULL);
}
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
wal_catchup_entry_t *entry)
{
if (!ha || !p || !entry)
return -EINVAL;
/* ── Batch mode: entry->n_entries > 0 ─────────────────────────────────── */
if (entry->n_entries > 0) {
/* Apply each sub-entry in LSN order. Track the last successfully
* applied LSN so we can advance applied_lsn after the full batch.
* Req 5.5: applied_lsn advances only after the full batch is written.
* Req 3.3: NULL or failed entries are logged and skipped; no crash. */
uint64_t last_good_lsn = 0;
int had_good = 0;
for (int i = 0; i < entry->n_entries; i++) {
wal_catchup_entry_t *sub = entry->entries[i];
if (!sub) {
ERROR("ha_apply_wal_catchup_entry: NULL sub-entry at "
"batch index %d — skipping", i);
continue;
}
int rc = ha_apply_single_entry(p, sub);
if (rc != 0) {
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed "
"for batch index %d lsn=%" PRIu64 " type=%u rc=%d "
"— skipping entry",
i, sub->lsn, (unsigned)sub->entry_type, rc);
continue;
}
last_good_lsn = sub->lsn;
had_good = 1;
}
/* Advance applied_lsn to last_good_lsn+1 after the full batch.
* Only advance if at least one entry was applied successfully. */
if (had_good)
ha_advance_applied_lsn(ha, last_good_lsn + 1);
return 0;
}
/* ── Single-entry mode (task 2.1 backward-compatible path) ────────────── */
int rc = ha_apply_single_entry(p, entry);
if (rc != 0) {
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for "
"lsn=%" PRIu64 " type=%u rc=%d — skipping entry",
entry->lsn, (unsigned)entry->entry_type, rc);
return rc;
}
/* Advance applied_lsn atomically (lsn+1 mirrors persist->current_lsn
* after a successful apply). */
ha_advance_applied_lsn(ha, entry->lsn + 1);
return 0;
}

198
src/controller/testsuite/test_config_repl.c
View file

@ -1045,6 +1045,199 @@ static void test_wal_catchup_entry_null_entry_no_crash(void)
printf("test_wal_catchup_entry_null_entry_no_crash: PASS\n");
}
/* ── Task 2.2: batch WAL catch-up entries on standby ─────────────────────── */
/* Test 2.2-A: batch of 3 entries — applied_lsn advances to last+1.
* Requirements 5.3, 5.5: apply each in order, advance after full batch. */
static void test_wal_catchup_batch_advances_applied_lsn(void)
{
char tmpdir[256];
persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir));
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);
/* applied_lsn starts at 0 */
TEST_ASSERT_EQUAL_UINT64(0,
(uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire));
/* Build 3 sub-entries at LSNs 10, 11, 12 */
wal_catchup_entry_t e0, e1, e2;
memset(&e0, 0, sizeof(e0));
memset(&e1, 0, sizeof(e1));
memset(&e2, 0, sizeof(e2));
e0.lsn = 10; e0.entry_type = (uint8_t)WAL_JOB_UPDATE;
e1.lsn = 11; e1.entry_type = (uint8_t)WAL_JOB_UPDATE;
e2.lsn = 12; e2.entry_type = (uint8_t)WAL_JOB_UPDATE;
wal_catchup_entry_t *batch_entries[3] = { &e0, &e1, &e2 };
/* Outer entry carries the batch; lsn/entry_type ignored when n_entries > 0 */
wal_catchup_entry_t outer;
memset(&outer, 0, sizeof(outer));
outer.n_entries = 3;
outer.entries = batch_entries;
int rc = ha_apply_wal_catchup_entry(&ha, p, &outer);
TEST_ASSERT_EQUAL_INT(0, rc);
/* applied_lsn must advance to last_entry->lsn + 1 = 13 */
TEST_ASSERT_EQUAL_UINT64(13,
(uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire));
ha_fini(&ha);
destroy_test_persist_catchup(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_wal_catchup_batch_advances_applied_lsn: PASS\n");
}
/* Test 2.2-B: partial batch (2 of max 3) — applied_lsn advances correctly.
* Requirement 5.4: partial batch (fewer than max) sent and applied without wait. */
static void test_wal_catchup_partial_batch(void)
{
char tmpdir[256];
persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir));
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);
/* Pre-set applied_lsn to 5 to verify monotonic advance */
atomic_store_explicit(&ha.applied_lsn, 5, memory_order_release);
wal_catchup_entry_t e0, e1;
memset(&e0, 0, sizeof(e0));
memset(&e1, 0, sizeof(e1));
e0.lsn = 20; e0.entry_type = (uint8_t)WAL_JOB_UPDATE;
e1.lsn = 21; e1.entry_type = (uint8_t)WAL_JOB_UPDATE;
wal_catchup_entry_t *batch_entries[2] = { &e0, &e1 };
wal_catchup_entry_t outer;
memset(&outer, 0, sizeof(outer));
outer.n_entries = 2;
outer.entries = batch_entries;
int rc = ha_apply_wal_catchup_entry(&ha, p, &outer);
TEST_ASSERT_EQUAL_INT(0, rc);
/* applied_lsn must advance to 21+1=22 */
TEST_ASSERT_EQUAL_UINT64(22,
(uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire));
ha_fini(&ha);
destroy_test_persist_catchup(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_wal_catchup_partial_batch: PASS\n");
}
/* Test 2.2-C: batch with one error entry — error is logged and skipped,
* remaining entries are still applied, applied_lsn advances for the good ones.
* Requirement 3.3/5.5: error in batch → skip that entry, continue with rest. */
static void test_wal_catchup_batch_error_entry_skipped(void)
{
char tmpdir[256];
persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir));
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);
/* Good entry at LSN 30, then a NULL pointer (simulates corrupt entry),
* then good entry at LSN 32. */
wal_catchup_entry_t e0, e2;
memset(&e0, 0, sizeof(e0));
memset(&e2, 0, sizeof(e2));
e0.lsn = 30; e0.entry_type = (uint8_t)WAL_JOB_UPDATE;
e2.lsn = 32; e2.entry_type = (uint8_t)WAL_JOB_UPDATE;
/* NULL in the middle represents a missing/corrupt sub-entry */
wal_catchup_entry_t *batch_entries[3] = { &e0, NULL, &e2 };
wal_catchup_entry_t outer;
memset(&outer, 0, sizeof(outer));
outer.n_entries = 3;
outer.entries = batch_entries;
/* Handler must not crash; applied_lsn advances for successful entries */
int rc = ha_apply_wal_catchup_entry(&ha, p, &outer);
/* Overall rc == 0: outer batch call succeeds even with inner skips */
TEST_ASSERT_EQUAL_INT(0, rc);
/* applied_lsn must reach at least 31 (e0 applied); ideally 33 (e2 applied) */
uint64_t final_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn,
memory_order_acquire);
TEST_ASSERT_TRUE(final_lsn >= 31);
ha_fini(&ha);
destroy_test_persist_catchup(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_wal_catchup_batch_error_entry_skipped: PASS\n");
}
/* Test 2.2-D: n_entries==0 falls through to single-entry path (backward compat) */
static void test_wal_catchup_batch_zero_entries_fallback(void)
{
char tmpdir[256];
persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir));
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);
/* n_entries == 0 → single-entry path from task 2.1 */
wal_catchup_entry_t entry;
memset(&entry, 0, sizeof(entry));
entry.lsn = 50;
entry.entry_type = (uint8_t)WAL_JOB_UPDATE;
entry.n_entries = 0;
entry.entries = NULL;
int rc = ha_apply_wal_catchup_entry(&ha, p, &entry);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Single-entry path: applied_lsn = 50+1 = 51 */
TEST_ASSERT_EQUAL_UINT64(51,
(uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire));
ha_fini(&ha);
destroy_test_persist_catchup(p);
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
system(cmd);
printf("test_wal_catchup_batch_zero_entries_fallback: PASS\n");
}
#endif /* XCE_TESTING */
/* ── dispatch ───────────────────────────────────── */
@ -1077,6 +1270,11 @@ void test_config_repl(void)
RUN_TEST(test_wal_catchup_entry_advances_applied_lsn);
RUN_TEST(test_wal_catchup_entry_error_skipped_no_crash);
RUN_TEST(test_wal_catchup_entry_null_entry_no_crash);
/* Task 2.2: batch WAL catch-up entries */
RUN_TEST(test_wal_catchup_batch_advances_applied_lsn);
RUN_TEST(test_wal_catchup_partial_batch);
RUN_TEST(test_wal_catchup_batch_error_entry_skipped);
RUN_TEST(test_wal_catchup_batch_zero_entries_fallback);
#endif
/* Task A: raw capnp shims */
RUN_TEST(test_replconfig_req_ssl_encode);