From 6bbcff62ca4b342e45886f524bd48d03abca4cbb Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 4 Jun 2026 15:38:20 +0000 Subject: [PATCH] feat(ha-repl-hardening): add batch WAL catch-up on standby side (task 2.2) 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 --- .kiro/specs/ha-repl-hardening/tasks.md | 2 +- src/controller/ha.c | 136 ++++++++++---- src/controller/testsuite/test_config_repl.c | 198 ++++++++++++++++++++ 3 files changed, 295 insertions(+), 41 deletions(-) diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 0380a1b..1abf073 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -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 diff --git a/src/controller/ha.c b/src/controller/ha.c index 5691f4d..c5e8a4f 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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; } diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 44673d6..337659d 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -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);