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

fix(ha-wal-replication): standby durably persists replicated WAL before ACK

Browse source 
A standby applied replicated WAL entries via apply_wal_entry (the replay/
in-memory path) which restores entities + advances applied_lsn but NEVER writes
the entry to the standby's WAL segment. The standby then ACKed (feeding the
primary's commit majority) without persisting — a Raft durability violation:
entries counted toward a commit could be lost on standby restart, and standby
WAL files stayed empty (contradicting steering: log[i] = WAL segments for every
node). This is why ctrl-2/ctrl-3 received + acked the streamed own-term NOOP yet
kept empty seg-*.wal.

Fix: new persist_wal_write_replicated(p, lsn, type, payload, len) writes the
entry to the local WAL at the PRIMARY's lsn (no lsn-reassignment), with a
self-consistent intra-segment CRC chain (the streamed entry carries no term/crc;
each node chains independently and recovery validates intra-segment), fdatasyncs
for durability, advances current_lsn = max(current_lsn, lsn+1), rotates segments,
and does NOT re-broadcast. Wired into ha_apply_single_entry as
persist(durable) -> apply(in-memory) -> ACK; on persist failure: no apply, no ACK
(a standby never ACKs an entry it didn't persist). Covers single + batch
catch-up; OPCode_replEntry has no separate apply path (allow-list only), so all
live WAL apply now persists. Dedup guard (lsn+1<=current_lsn) makes re-streams
idempotent while still ACKing.

TDD: standby write -> drop persist_t -> reopen + persist_recover replays it
(single, 3-entry crc chain, post-dedup); checkpoint/oversized rejected. Full
controller suite green under ASAN; persist_wal_async / recovery / truncation /
streamer / wire untouched.
This commit is contained in:
Rongsong Shen 2026-06-14 13:10:02 +00:00
parent 007bf0c403
commit 708f2279e3
4 changed files with 359 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/controller/ha.c
View file

@ -1030,6 +1030,27 @@ static void ha_advance_applied_lsn(ha_state_t *ha, uint64_t new_lsn)
* 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)
{
/* Persist-before-apply-before-ACK: durably write the replicated entry to
* this standby's own WAL BEFORE applying it in memory (and before the
* caller ACKs). The ACK feeds the primary's commit majority, so it must
* imply durability — a standby must not ACK an entry it could lose on
* restart (Raft log[i] includes standbys; ha-async-io.md).
*
* persist_wal_write_replicated does NOT assign a new lsn (uses e->lsn) and
* does NOT re-broadcast; it fsyncs before returning and is idempotent on
* re-stream. WAL_CHECKPOINT is never streamed, but it would be rejected
* by both this primitive and apply_wal_entry. */
int prc = persist_wal_write_replicated(p, e->lsn, (uint32_t)e->entry_type,
e->raw_payload,
(uint32_t)e->raw_payload_len);
if (prc != 0) {
ERROR("ha_apply_single_entry: persist_wal_write_replicated failed for "
"lsn=%" PRIu64 " type=%u rc=%d — not applying/acking",
e->lsn, (unsigned)e->entry_type, prc);
/* No apply, no ACK on persist failure (mirrors no-ACK-on-apply-fail). */
return prc;
}
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = e->lsn;

104
src/controller/persist.c
View file

@ -397,6 +397,110 @@ int persist_wal_async(persist_t* p, uint32_t type,
return 0;
}
int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
const void* payload, uint32_t len)
{
if (!p)
return -EINVAL;
/* WAL_CHECKPOINT is a local-only marker, never streamed to standbys. */
if (type == WAL_CHECKPOINT)
return -EINVAL;
/* Reject oversized payloads BEFORE touching the segment. */
if (len > MAX_WAL_PAYLOAD_BYTES)
return -EMSGSIZE;
pthread_mutex_lock(&p->lock);
/* Idempotency / dedup guard: if this lsn is already persisted (current_lsn
* already advanced past it) treat the re-stream as a no-op. This keeps the
* write idempotent and avoids a duplicate entry / CRC-chain corruption when
* the primary re-sends an lsn the standby already wrote. */
if (lsn + 1 <= p->current_lsn) {
pthread_mutex_unlock(&p->lock);
return 0;
}
/* Open first segment lazily. open_new_segment names the file from
* current_lsn — for a fresh standby receiving lsn 1 first, current_lsn is
* 1 (reserve-lsn-0), so the segment is seg-...001.wal. */
if (p->wal_fd < 0) {
if (open_new_segment(p) != 0) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
/* Build entry header. Unlike persist_wal_async, lsn is GIVEN (not
* current_lsn++). term isn't streamed, so use our best-effort p->term.
* The CRC chain is SELF-computed (the streamed entry carries no crc):
* recovery validates an intra-segment chain (prev_crc == last_crc), so the
* standby's own writes must chain to each other — cross-node CRC equality
* is neither required nor possible. */
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.term = p->term;
hdr.lsn = lsn;
hdr.timestamp = now_us();
hdr.type = type;
hdr.payload_len = len;
hdr.prev_crc = (p->seg_offset == 0) ? WAL_CRC_NONE : p->last_crc;
uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
if (payload && len > 0)
c = crc32(c, (const Bytef*)payload, len);
hdr.crc = c;
/* Write header */
ssize_t hw = pwrite(p->wal_fd, &hdr, sizeof(hdr), (off_t)p->seg_offset);
if (hw != (ssize_t)sizeof(hdr)) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
/* Write payload (if any) */
if (payload && len > 0) {
ssize_t pw = pwrite(p->wal_fd, payload, len,
(off_t)(p->seg_offset + sizeof(hdr)));
if (pw != (ssize_t)len) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
uint32_t entry_size = (uint32_t)(sizeof(hdr) + len);
p->seg_offset += entry_size;
p->total_wal_bytes += entry_size;
p->last_crc = hdr.crc;
/* Advance current_lsn to one past the given lsn (max-idempotent — mirrors
* apply_wal_entry's monotonic advance). */
if (lsn + 1 > p->current_lsn)
p->current_lsn = lsn + 1;
/* Durability: fsync the segment BEFORE returning so the caller's ACK
* implies the entry is on stable storage (persist-before-ACK).
* persist_wal_async defers fsync to the background thread / persist_flush;
* here we sync inline because the entry feeds the primary's commit
* majority and must survive a standby restart. fdatasync is sufficient
* (entry bytes + segment metadata are pre-allocated by open_new_segment). */
fdatasync(p->wal_fd);
/* Rotate segment when full (same threshold as the primary append). */
if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
if (rotate_segment(p) != 0) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
/* NOTE: no repl_broadcast — standbys never re-replicate. */
pthread_mutex_unlock(&p->lock);
return 0;
}
int persist_flush(persist_t* p)
{
pthread_mutex_lock(&p->lock);

219
src/controller/testsuite/test_repl_apply.c
View file

@ -8,9 +8,11 @@
* - Oversized payloads are rejected (rc != 0)
* - Unknown types (> WAL_ENTRY_TYPE_MAX) are skipped with rc=0
*/
#include <dirent.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unity/unity.h>
@ -83,6 +85,85 @@ static persist_t* make_test_persist_for_apply(void)
return p;
}
/* Like make_test_persist_for_apply but reports the chosen tmpdir so a test
* can re-open the same WAL via a FRESH persist_t (restart/recover proof). */
static persist_t* make_test_persist_in(char* tmpdir_out, size_t cap)
{
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;
snprintf(tmpdir_out, cap, "/tmp/xce_replwr_XXXXXX");
if (!mkdtemp(tmpdir_out)) {
fprintf(stderr, "mkdtemp failed\n");
free(p);
exit(1);
}
persist_set_workdir(p, tmpdir_out, "ctrl-replwr");
persist_init(p, &cfg, NULL);
persist_start(p);
return p;
}
/* Re-open an existing workdir as a fresh persist_t (simulates a restart). */
static persist_t* reopen_test_persist(const char* tmpdir)
{
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;
persist_set_workdir(p, tmpdir, "ctrl-replwr");
persist_init(p, &cfg, NULL);
persist_start(p);
return p;
}
/* Count *.wal files under <tmpdir>/ctrl-replwr/data/wal */
static int count_replwr_wal_files(const char* tmpdir)
{
char wal_path[512];
snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir);
DIR* dir = opendir(wal_path);
if (!dir) return 0;
int count = 0;
struct dirent* ent;
while ((ent = readdir(dir)) != NULL) {
size_t len = strlen(ent->d_name);
if (len > 4 && strcmp(ent->d_name + len - 4, ".wal") == 0)
count++;
}
closedir(dir);
return count;
}
/* Total bytes across all *.wal segment files (proxy for "entry on disk"). */
static off_t total_replwr_wal_bytes(const char* tmpdir)
{
char wal_path[512];
snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir);
DIR* dir = opendir(wal_path);
if (!dir) return 0;
off_t total = 0;
struct dirent* ent;
while ((ent = readdir(dir)) != NULL) {
size_t len = strlen(ent->d_name);
if (len <= 4 || strcmp(ent->d_name + len - 4, ".wal") != 0)
continue;
char full[1024];
snprintf(full, sizeof(full), "%s/%s", wal_path, ent->d_name);
struct stat st;
if (stat(full, &st) == 0)
total += st.st_size;
}
closedir(dir);
return total;
}
static void destroy_test_persist(persist_t* p)
{
persist_stop(p);
@ -197,6 +278,138 @@ static void test_apply_wal_entry_null_guard(void)
printf("test_apply_wal_entry_null_guard: PASS\n");
}
/* ================================================================== */
/* persist_wal_write_replicated — standby persists replicated entries */
/* (durability: persist-before-ACK). RED before the fix. */
/* ================================================================== */
/* A streamed entry written to the standby's WAL must survive a restart:
* a FRESH persist_recover over the same WAL replays it (crc-valid) and
* advances current_lsn. Before the fix the segment is empty / nothing
* is recovered. */
static void test_repl_write_persists_and_recovers(void)
{
char tmpdir[256];
persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));
/* Fresh standby: first replicated entry is lsn 1 (1-based; reserve-lsn-0). */
const char payload[] = "replicated_job_update_payload";
int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE,
payload, (uint32_t)strlen(payload));
TEST_ASSERT_EQUAL_INT(0, rc);
/* current_lsn advanced to lsn+1 */
TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn);
/* A .wal segment must now exist on disk. */
TEST_ASSERT_TRUE(count_replwr_wal_files(tmpdir) >= 1);
TEST_ASSERT_TRUE(total_replwr_wal_bytes(tmpdir) > 0);
/* Simulate a restart: drop this persist_t, re-open the same workdir,
* and run recovery. The replicated entry must replay (crc-chain valid)
* and current_lsn must advance past it. */
destroy_test_persist(p);
persist_t* p2 = reopen_test_persist(tmpdir);
int rec = persist_recover(p2);
TEST_ASSERT_TRUE(rec >= 0);
/* Recovery must have advanced current_lsn to 2 (replayed lsn 1). */
TEST_ASSERT_EQUAL_UINT64(2, p2->current_lsn);
destroy_test_persist(p2);
printf("test_repl_write_persists_and_recovers: PASS\n");
}
/* Multiple replicated entries in sequence form a self-consistent crc chain
* that the standby's OWN recovery validates with no chain break (all replay). */
static void test_repl_write_crc_chain_recovers(void)
{
char tmpdir[256];
persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));
const char* payloads[] = { "entry-one", "entry-two", "entry-three" };
for (int i = 0; i < 3; i++) {
uint64_t lsn = (uint64_t)(i + 1); /* 1,2,3 */
int rc = persist_wal_write_replicated(p, lsn, WAL_JOB_UPDATE,
payloads[i],
(uint32_t)strlen(payloads[i]));
TEST_ASSERT_EQUAL_INT(0, rc);
}
TEST_ASSERT_EQUAL_UINT64(4, p->current_lsn);
destroy_test_persist(p);
/* Fresh recovery must replay ALL three (no crc-chain break) → current_lsn=4. */
persist_t* p2 = reopen_test_persist(tmpdir);
int rec = persist_recover(p2);
TEST_ASSERT_TRUE(rec >= 0);
TEST_ASSERT_EQUAL_UINT64(4, p2->current_lsn);
destroy_test_persist(p2);
printf("test_repl_write_crc_chain_recovers: PASS\n");
}
/* Re-streaming the same lsn must be idempotent: the dedup guard skips the
* second write (no double-write, no chain corruption). After a dup write,
* recovery still yields exactly the single entry chain. */
static void test_repl_write_dedup_idempotent(void)
{
char tmpdir[256];
persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));
const char payload[] = "dup-entry";
int rc1 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE,
payload, (uint32_t)strlen(payload));
TEST_ASSERT_EQUAL_INT(0, rc1);
uint64_t lsn_after_first = p->current_lsn; /* expect 2 */
off_t bytes_after_first = total_replwr_wal_bytes(tmpdir);
/* Re-stream lsn 1 (already persisted): must be a no-op success. */
int rc2 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE,
payload, (uint32_t)strlen(payload));
TEST_ASSERT_EQUAL_INT(0, rc2);
/* current_lsn must NOT advance further; no extra bytes written. */
TEST_ASSERT_EQUAL_UINT64(lsn_after_first, p->current_lsn);
TEST_ASSERT_EQUAL_INT64((long long)bytes_after_first,
(long long)total_replwr_wal_bytes(tmpdir));
/* Now legitimately stream lsn 2 — should append fine after the dup. */
int rc3 = persist_wal_write_replicated(p, 2, WAL_JOB_UPDATE,
payload, (uint32_t)strlen(payload));
TEST_ASSERT_EQUAL_INT(0, rc3);
TEST_ASSERT_EQUAL_UINT64(3, p->current_lsn);
destroy_test_persist(p);
/* Recovery must replay lsn 1 and lsn 2 (no duplicate / no break). */
persist_t* p2 = reopen_test_persist(tmpdir);
int rec = persist_recover(p2);
TEST_ASSERT_TRUE(rec >= 0);
TEST_ASSERT_EQUAL_UINT64(3, p2->current_lsn);
destroy_test_persist(p2);
printf("test_repl_write_dedup_idempotent: PASS\n");
}
/* WAL_CHECKPOINT must never be persisted via the replicated-write path
* (standbys never receive it; it must be rejected, not written). */
static void test_repl_write_rejects_checkpoint(void)
{
char tmpdir[256];
persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));
int rc = persist_wal_write_replicated(p, 1, WAL_CHECKPOINT, NULL, 0);
TEST_ASSERT_NOT_EQUAL(0, rc);
/* current_lsn must not advance for a rejected checkpoint. */
TEST_ASSERT_EQUAL_UINT64(1, p->current_lsn);
destroy_test_persist(p);
printf("test_repl_write_rejects_checkpoint: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point (registered in controller_test.c) */
/* ------------------------------------------------------------------ */
@ -208,4 +421,10 @@ void test_repl_apply(void)
RUN_TEST(test_apply_wal_entry_rejects_oversized);
RUN_TEST(test_apply_wal_entry_skips_unknown_type);
RUN_TEST(test_apply_wal_entry_null_guard);
/* persist-before-ACK durability (standby persists replicated entries) */
RUN_TEST(test_repl_write_persists_and_recovers);
RUN_TEST(test_repl_write_crc_chain_recovers);
RUN_TEST(test_repl_write_dedup_idempotent);
RUN_TEST(test_repl_write_rejects_checkpoint);
}

15
src/include/persist.h
View file

@ -116,6 +116,21 @@ int persist_wal_sync_append(persist_t* p, uint32_t type,
const void* payload, uint32_t len,
int* sync_rc_out);
/* Write a REPLICATED entry (received from the primary) into the local WAL
* at its GIVEN lsn — used by standbys to durably persist streamed entries
* BEFORE acking them (persist-before-ACK; an ACK must imply durability).
*
* Unlike persist_wal_async it does NOT assign a new lsn (the lsn is supplied
* by the primary) and does NOT repl_broadcast (standbys never re-replicate).
* It builds a SELF-consistent CRC chain (the streamed entry carries no crc),
* fsyncs the segment before returning, and is idempotent on re-stream: if the
* lsn is already persisted (lsn+1 <= current_lsn) it is a no-op returning 0.
*
* WAL_CHECKPOINT is rejected with -EINVAL (never streamed to standbys).
* Returns 0 on success (incl. dedup no-op), negative errno on failure. */
int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
const void* payload, uint32_t len);
/* ============================================================
* Write-path commit classification (ha-commit-barrier-rollout)
* ============================================================ */