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.
This commit is contained in:
Rongsong Shen 2026-06-04 15:05:52 +00:00
parent e3ca775f0d
commit 647b88ce57
5 changed files with 251 additions and 1 deletions

View file

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

View file

@ -1,6 +1,7 @@
#include "config.h"
#include <errno.h>
#include <inttypes.h>
#include <rtti.h>
#include <stdatomic.h>
#include <string.h>
@ -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.

View file

@ -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.

View file

@ -26,6 +26,8 @@
#include <time.h>
#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) ── */
/*

View file

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