From 647b88ce573f5c24d4fd042276209be254b67a3e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 4 Jun 2026 15:05:52 +0000 Subject: [PATCH] feat(ha-repl-hardening): wire OPCode_walCatchupEntry standby handler (task 2.1) Replace stub with ha_apply_wal_catchup_entry() calling apply_wal_entry(). CAS-advance applied_lsn on success; log ERROR and skip on failure. Standby stays in CTRL_STANDBY throughout. 3 new unit tests. --- .kiro/specs/ha-repl-hardening/tasks.md | 2 +- src/controller/cluster.c | 25 ++++ src/controller/ha.c | 59 ++++++++ src/controller/ha.h | 15 ++ src/controller/testsuite/test_config_repl.c | 151 ++++++++++++++++++++ 5 files changed, 251 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 2f83050..0380a1b 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -42,7 +42,7 @@ --- - [ ] 2. Standby-side: Wire WAL catch-up and heartbeat state reporting -- [ ] 2.1 (P) Wire `OPCode_walCatchupEntry` single-entry handler on standby +- [x] 2.1 (P) Wire `OPCode_walCatchupEntry` single-entry handler on standby - Replace the stub in the dispatch switch (returns 0) with a call to `apply_wal_entry()` for single entries - On success, advance `applied_lsn` atomically; on error, log ERROR and skip — do not crash - Standby remains in `CTRL_STANDBY` throughout; does not accept job dispatch during catch-up diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 44f3464..2e897c4 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1,6 +1,7 @@ #include "config.h" #include +#include #include #include #include @@ -154,6 +155,23 @@ static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req) } return 0; } +/* handle_wal_catchup_entry: dispatch handler for OPCode_walCatchupEntry. + * Delegates to ha_apply_wal_catchup_entry (ha.c); on error logs and returns 0 + * (skip, no crash). Standby remains in CTRL_STANDBY throughout (Req 3.4). */ +static int handle_wal_catchup_entry(cluster_t *c, xce_request_t *req) +{ + wal_catchup_entry_t *entry = + (wal_catchup_entry_t *)req->body.walcatchupentrymsg; + + /* body may be NULL if the sender used Request_body_nomsg; skip gracefully */ + if (!entry) + return 0; + + ha_apply_wal_catchup_entry(&c->ha, &c->persist, entry); + /* Return 0 regardless of per-entry errors — Req 3.3: skip, don't crash */ + return 0; +} + static int myreq_handler(vtask_t* task) { vtask_data_t* data = (vtask_data_t*)task->data; @@ -351,6 +369,13 @@ static int myreq_handler(vtask_t* task) /* Primary pushes config to this standby — save and reload */ rc = handle_repl_config_push(c, ep, &request); break; + case OPCode_walCatchupEntry: + /* Task 2.1: standby applies a single WAL catch-up entry from the + * primary. applied_lsn advances on success; errors are logged and + * skipped (Req 3.1, 3.3, 3.4, 3.5). No reply is sent — primary + * learns of progress via the next heartbeat reply's stateReport. */ + rc = handle_wal_catchup_entry(c, &request); + break; case OPCode_upgradeConfig: #if CONF_ENABLE_UPGRADE_CONFIG /* OPCode_upgradeConfig is sent BY the primary TO standbys to push config. diff --git a/src/controller/ha.c b/src/controller/ha.c index ec4ca2b..5691f4d 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -803,6 +803,65 @@ void ha_fini(ha_state_t *ha) } } +/* ── Task 2.1: 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. + * + * On success (rc == 0): atomically advances ha->applied_lsn to entry->lsn+1 + * so the next heartbeat reply's stateReport reflects the latest state. + * + * 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). + * + * 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). + * + * Exported under XCE_TESTING so unit tests can call it directly without + * constructing a full cluster_t or endpoint_t. + */ +int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, + wal_catchup_entry_t *entry) +{ + 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) { + if (atomic_compare_exchange_weak_explicit( + &ha->applied_lsn, &old_lsn, new_lsn, + memory_order_acq_rel, memory_order_acquire)) + break; + } + + return 0; +} + /* ── Dispatch gate (req 1.3, 1.4) ────────────────────────────────────────── * * Called from myreq_handler() in cluster.c before the opcode switch. diff --git a/src/controller/ha.h b/src/controller/ha.h index e849269..b25b3bf 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -26,6 +26,8 @@ #include #include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */ +#include "persist.h" /* persist_t, apply_wal_entry */ +#include "typemap.h" /* wal_catchup_entry_t */ #include "xce/cluster.h" /* xce_candidate_t */ #include "zarray.h" #include "raft/raft.h" /* raft_state_t, raft_init */ @@ -265,6 +267,19 @@ int ha_is_primary(const ha_state_t *ha); */ int ha_dispatch_gate(ha_state_t *ha, int opcode); +/* + * ha_apply_wal_catchup_entry: apply a single WAL catch-up entry to local + * persistent storage and advance ha->applied_lsn on success. + * + * Called from cluster.c OPCode_walCatchupEntry dispatch handler (Task 2.1). + * On success (rc == 0): applied_lsn advances to entry->lsn+1. + * On error (rc < 0): logs ERROR, returns error; caller skips (Req 3.3). + * applied_lsn is in-memory only — not persisted to disk. + * Requirements: 3.1, 3.2, 3.3, 3.4, 3.5 + */ +int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, + wal_catchup_entry_t *entry); + /* ── HA Config Bootstrap & Term Durability (ha-config-bootstrap spec) ── */ /* diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index a0b62a1..44673d6 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -23,11 +23,23 @@ #include "config_repl.h" #include "controller.capnp.h" +#include "ha.h" #include "myconf.h" +#include "persist.h" #include "request.h" #include "typemap.h" +#include "wal_types.h" #include "xce/cluster_def.h" +#ifdef XCE_TESTING +/* Declared in cluster.c under #ifdef XCE_TESTING — test seam for + * ha_apply_wal_catchup_entry: applies a single WAL catch-up entry to + * persistent storage and advances ha->applied_lsn on success. + * On error: logs ERROR, returns non-zero, does not crash. */ +int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, + wal_catchup_entry_t *entry); +#endif + /* Per-test directory helpers — not Unity setUp/tearDown (those are in * controller_test.c). Each test that needs a temp dir calls cr_setup() * at the top and cr_teardown() via the label before returning. */ @@ -900,6 +912,141 @@ static void test_upgrade_config_req_extract_raw(void) printf("test_upgrade_config_req_extract_raw: PASS\n"); } +/* ── Task 2.1: OPCode_walCatchupEntry handler ────────────────────────────── */ + +#ifdef XCE_TESTING + +/* Helpers shared by Task 2.1 tests: create/destroy a minimal persist_t. */ + +static persist_t *make_test_persist_catchup(char *tmpdir_out, size_t sz) +{ + persist_t *p = calloc(1, sizeof(persist_t)); + if (!p) { fprintf(stderr, "calloc failed\n"); exit(1); } + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; /* no real fsync in unit tests */ + + snprintf(tmpdir_out, sz, "/tmp/xce_catchup_XXXXXX"); + if (!mkdtemp(tmpdir_out)) { + fprintf(stderr, "mkdtemp failed\n"); + free(p); + exit(1); + } + + persist_set_workdir(p, tmpdir_out, "ctrl-catchup"); + persist_init(p, &cfg, NULL); + persist_start(p); + return p; +} + +static void destroy_test_persist_catchup(persist_t *p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Test 2.1-A: successful single WAL entry advances applied_lsn */ +static void test_wal_catchup_entry_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 a single-entry WAL catch-up with LSN=42, no payload */ + wal_catchup_entry_t entry; + memset(&entry, 0, sizeof(entry)); + entry.lsn = 42; + entry.entry_type = (uint8_t)WAL_JOB_UPDATE; + entry.payload = NULL; /* payload_len=0; apply_wal_entry handles NULL */ + entry.n_entries = 0; + entry.entries = NULL; + + /* GREEN will implement ha_apply_wal_catchup_entry() */ + int rc = ha_apply_wal_catchup_entry(&ha, p, &entry); + TEST_ASSERT_EQUAL_INT(0, rc); + /* applied_lsn must advance to lsn+1 = 43 (mirrors persist->current_lsn) */ + TEST_ASSERT_EQUAL_UINT64(43, + (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_entry_advances_applied_lsn: PASS\n"); +} + +/* Test 2.1-B: apply_wal_entry failure (NULL persist = -EINVAL) is logged and + * skipped — applied_lsn must NOT advance, handler must not crash. + * Req 3.3: corrupt/error entries are logged and skipped without crashing. */ +static void test_wal_catchup_entry_error_skipped_no_crash(void) +{ + 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); + atomic_store_explicit(&ha.applied_lsn, 10, memory_order_release); + + /* Pass NULL persist — apply_wal_entry returns -EINVAL for NULL p */ + wal_catchup_entry_t entry; + memset(&entry, 0, sizeof(entry)); + entry.lsn = 99; + entry.entry_type = (uint8_t)WAL_JOB_UPDATE; + entry.payload = NULL; + entry.n_entries = 0; + entry.entries = NULL; + + /* Handler must NOT crash; must log ERROR; applied_lsn must remain at 10 */ + int rc = ha_apply_wal_catchup_entry(&ha, /*persist=*/NULL, &entry); + /* rc is non-zero (error propagated from apply_wal_entry(-EINVAL)) */ + TEST_ASSERT_TRUE(rc != 0); + /* applied_lsn must not advance on failure */ + TEST_ASSERT_EQUAL_UINT64(10, + (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); + + ha_fini(&ha); + printf("test_wal_catchup_entry_error_skipped_no_crash: PASS\n"); +} + +/* Test 2.1-C: NULL entry is handled gracefully (null guard) */ +static void test_wal_catchup_entry_null_entry_no_crash(void) +{ + 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); + + /* NULL entry — must not crash, must return error */ + int rc = ha_apply_wal_catchup_entry(&ha, NULL, NULL); + TEST_ASSERT_TRUE(rc != 0); + + ha_fini(&ha); + printf("test_wal_catchup_entry_null_entry_no_crash: PASS\n"); +} + +#endif /* XCE_TESTING */ + /* ── dispatch ───────────────────────────────────── */ void test_config_repl(void) @@ -926,6 +1073,10 @@ void test_config_repl(void) RUN_TEST(test_load_primary_config_missing_returns_error); RUN_TEST(test_wait_all_with_timeout_slow_threads); RUN_TEST(test_wait_all_with_timeout_fast_threads); + /* Task 2.1: OPCode_walCatchupEntry handler */ + 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); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode);