wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

fix(ha-wal-replication): apply_wal_entry must not own the WAL CRC chain (bug-778)

Browse source 
On the standby replicated-apply path, ha_apply_single_entry wrote the entry via
persist_wal_write_replicated (which correctly set the self-computed CRC chain
link p->last_crc), then called apply_wal_entry with a SYNTHETIC header whose
crc=0. apply_wal_entry's `p->last_crc = hdr->crc` clobbered the chain link with
0, so every subsequent replicated write stamped prev_crc=0 on disk. The on-disk
CRC chain broke at the second replicated entry — invisible while live (standbys
serve in-memory state) but fatal on COLD RESTART: replay hit the break, returned
RECOVER_TRUNCATED, recovered only the config prefix, and lost all committed jobs
(joblist=0). The sticky truncated verdict then drove a promotion-fence
availability wedge (bug-777) when every survivor was truncated.

Fix — single ownership of the WAL header + CRC chain:
- apply_wal_entry is now LOGICAL entity-restore only: signature
  (p, type, lsn, payload, payload_len); no header. It advances the current_lsn
  high-water (monotonic) but NEVER touches p->last_crc.
- The physical layer is the sole owner of the CRC chain: persist_wal_write_-
  replicated on the live path, replay_wal/persist_recover on recovery.
- ha_apply_single_entry drops the fabricated header and passes entry fields
  directly. The persist_recover replay caller is updated to the new signature.

Verified: new RED->GREEN test test_replicated_apply_chain_survives_cold_restart
(drives the real ha_apply_wal_catchup_entry path, cold-replays -> RECOVER_CLEAN,
all N recovered); controller suite 0 failures; 3-node rolling-SIGKILL soak on a
real ext2/3 fs now passes 5/5 (was wedge+loss at round 3); full cold restart
recovers all 6 committed jobs (was 0) with CLEAN verdicts and a normal election.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-06-19 14:12:53 +00:00
parent 276e26c37b
commit 64ab210cbf
7 changed files with 136 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/controller/ha.c
View file

@ -1578,14 +1578,12 @@ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
return prc;
}
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = e->lsn;
hdr.type = (uint32_t)e->entry_type;
hdr.payload_len = (uint32_t)e->raw_payload_len;
hdr.crc = 0;
hdr.prev_crc = 0;
return apply_wal_entry(p, &hdr, e->raw_payload);
/* Logical entity restore only. persist_wal_write_replicated (above) already
* wrote the entry and owns the on-disk CRC chain + current_lsn — so we pass
* the entry fields directly rather than fabricating a header. (bug-778:
* a synthetic crc=0 header here used to clobber p->last_crc.) */
return apply_wal_entry(p, (uint32_t)e->entry_type, e->lsn,
e->raw_payload, (uint32_t)e->raw_payload_len);
}
/* ha_send_wal_ack_reply: send WalAckReply {term, lsn, success=1} back to

57
src/controller/persist.c
View file

@ -1302,49 +1302,60 @@ int persist_load_snapshot(persist_t *p, const char *path, uint64_t lsn)
*
* Returns 0 on success (including skip for checkpoint / unknown types),
* negative errno on error (NULL args, oversized payload). */
int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload)
int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
const void* payload, uint32_t payload_len)
{
if (!p || !hdr)
if (!p)
return -EINVAL;
/* Reject oversized payloads BEFORE any allocation */
if (hdr->payload_len > MAX_WAL_PAYLOAD_BYTES) {
if (payload_len > MAX_WAL_PAYLOAD_BYTES) {
ERROR("apply_wal_entry: payload_len %u exceeds MAX_WAL_PAYLOAD_BYTES %u",
hdr->payload_len, MAX_WAL_PAYLOAD_BYTES);
payload_len, MAX_WAL_PAYLOAD_BYTES);
return -EMSGSIZE;
}
/* LOGICAL entity restore ONLY. The on-disk WAL header, the CRC chain
* (p->last_crc), and current_lsn are owned exclusively by the physical
* layer — persist_wal_write_replicated on the live replicated path, and
* replay_wal/persist_recover on recovery. This function MUST NOT touch
* them. (bug-778: it previously set p->last_crc from a synthetic crc=0
* header, breaking the on-disk CRC chain and cold-restart recovery.) */
/* WAL_CHECKPOINT (100) is a local-only marker — standbys never receive it
* via replication. Skip silently; do NOT advance LSN. */
if (hdr->type == WAL_CHECKPOINT)
* via replication. Skip silently. */
if (type == WAL_CHECKPOINT)
return 0;
/* Unknown types above WAL_ENTRY_TYPE_MAX: log WARN and skip for forward
* compatibility — an older controller can boot on WAL written by a newer
* one without hard-failing on unrecognised entry types. */
if (hdr->type > WAL_ENTRY_TYPE_MAX) {
if (type > WAL_ENTRY_TYPE_MAX) {
WARN("apply_wal_entry: unknown type %u > WAL_ENTRY_TYPE_MAX %u"
" (lsn=%" PRIu64 ") — skipping for forward compatibility",
hdr->type, WAL_ENTRY_TYPE_MAX, hdr->lsn);
type, WAL_ENTRY_TYPE_MAX, lsn);
return 0;
}
/* Advance LSN chain: track the highest replicated LSN seen. */
/* Advance the current_lsn high-water (monotonic, idempotent). This is a
* write-cursor high-water that is safe to bump from multiple paths. The
* CRC chain link (p->last_crc) is deliberately NOT touched here — it has a
* single owner: the physical writer (persist_wal_write_replicated) and
* replay (persist_recover). (bug-778) */
pthread_mutex_lock(&p->lock);
if (hdr->lsn + 1 > p->current_lsn)
p->current_lsn = hdr->lsn + 1;
p->last_crc = hdr->crc;
if (lsn + 1 > p->current_lsn)
p->current_lsn = lsn + 1;
pthread_mutex_unlock(&p->lock);
if (!p->cluster)
return 0; /* unit tests without a full cluster — skip entity restore */
switch (hdr->type) {
switch (type) {
case WAL_JOB_SUBMIT:
/* Payload: Cap'n Proto serialized xce_job_t. Thread the entry LSN
* through so the restored job is tagged for the egress watermark
* (ha-convergence-recovery 3.3). */
restore_job(p->cluster, 0, payload, hdr->payload_len, hdr->lsn);
restore_job(p->cluster, 0, payload, payload_len, lsn);
break;
case WAL_JOB_START:
@ -1357,7 +1368,7 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa
case WAL_JOB_PREASON:
case WAL_JOB_RUSAGE:
/* Payload: wal_jobstate_t {jobid, state} */
if (hdr->payload_len >= sizeof(wal_jobstate_t)) {
if (payload_len >= sizeof(wal_jobstate_t)) {
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
restore_job_state(p->cluster, js->jobid, js->state);
}
@ -1365,13 +1376,13 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa
case WAL_WORKER_CTRL:
/* Payload: [uint32_t detail][capnp worker] */
if (hdr->payload_len > WORKER_WAL_HDR_SIZE)
worker_from_capnp(p->cluster, payload, hdr->payload_len);
if (payload_len > WORKER_WAL_HDR_SIZE)
worker_from_capnp(p->cluster, payload, payload_len);
break;
case WAL_QUEUE_CTRL:
/* Payload: [wal_queuehdr_t][name NUL-terminated] */
if (hdr->payload_len > sizeof(wal_queuehdr_t)) {
if (payload_len > sizeof(wal_queuehdr_t)) {
const wal_queuehdr_t* qh = (const wal_queuehdr_t*)payload;
const char* name = (const char*)(payload + sizeof(wal_queuehdr_t));
restore_queue_state(p->cluster, name, qh->status);
@ -1380,8 +1391,8 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa
case WAL_NOOP:
/* Internal promotion no-op: empty payload, no entity restore, no
* hooks. The common-path current_lsn advance (above) is its only
* effect — recognised explicitly here rather than via default. */
* hooks. current_lsn is advanced by the physical layer (writer /
* replay), not here — recognised explicitly rather than via default. */
break;
default:
@ -1581,8 +1592,10 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
continue;
}
/* Apply the entry (P0 stub: no-op) */
apply_wal_entry(p, &hdr, payload);
/* Logical entity restore. The CRC chain + current_lsn are owned by
* this replay loop (current_lsn advanced below; last_crc at line
* ~1601), not by apply_wal_entry. */
apply_wal_entry(p, hdr.type, hdr.lsn, payload, hdr.payload_len);
free(payload);
/* Advance current_lsn to one past this entry */

70
src/controller/testsuite/test_commit_durability.c
View file

@ -1305,9 +1305,77 @@ static void test_recover_verdict_crc_break_first_entry_truncated(void)
printf("test_recover_verdict_crc_break_first_entry_truncated: PASS\n");
}
/* ------------------------------------------------------------------ *
* bug-778 regression: the standby replicated-apply path must leave an *
* intact on-disk CRC chain that cold-restart replay can recover. *
* *
* Drives the REAL standby path (ha_apply_wal_catchup_entry -> *
* ha_apply_single_entry -> persist_wal_write_replicated + apply_wal_ *
* entry) for N entries, then cold-recovers. Before the fix, *
* apply_wal_entry clobbered p->last_crc with the synthetic header's *
* crc=0, so every entry after the first wrote prev_crc=0 -> CRC chain *
* break -> RECOVER_TRUNCATED and the committed entries were lost on *
* cold restart. cluster==NULL here (entity restore is skipped); the *
* physical WAL write + CRC chain is what is under test. *
* ------------------------------------------------------------------ */
static void test_replicated_apply_chain_survives_cold_restart(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_replchain_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const int N = 6;
/* Writer: apply N replicated entries via the real standby path, then die. */
{
persist_t* p = make_durability_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
ha_state_t ha;
memset(&ha, 0, sizeof(ha));
for (int i = 1; i <= N; i++) {
wal_jobstate_t js = { .jobid = (uint64_t)i, .state = 0x10 };
wal_catchup_entry_t e;
memset(&e, 0, sizeof(e));
e.lsn = (uint64_t)i;
e.entry_type = WAL_JOB_UPDATE;
e.n_entries = 0; /* single-entry mode */
e.raw_payload = (uint8_t*)&js;
e.raw_payload_len = sizeof(js);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, ha_apply_wal_catchup_entry(&ha, p, &e, NULL, NULL),
"replicated apply of a streamed entry must succeed");
}
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
teardown_durability_persist(p);
}
/* Cold restart: the standby's own WAL must form an intact CRC chain. */
{
persist_t* p2 = make_durability_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_INT_MESSAGE(
RECOVER_CLEAN, persist_recovery_verdict(p2),
"replicated-apply WAL must form an intact CRC chain on cold restart "
"(bug-778: apply_wal_entry must not clobber the chain link)");
TEST_ASSERT_FALSE_MESSAGE(
persist_recovery_truncated(p2),
"no truncation: every replicated entry must be recoverable");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
(uint64_t)N + 1, p2->current_lsn,
"all N replicated entries must replay (current_lsn == N+1)");
teardown_durability_persist(p2);
}
printf("test_replicated_apply_chain_survives_cold_restart: PASS\n");
}
/* ------------------------------------------------------------------ *
* Test 3.1d (snapshot-sentinel edge): a member that recovered from a *
* 16-byte LSN-sentinel snapshot has a committed position at the *
* 16-byte LSN-sentinel snapshot has a committed position at the *
* sentinel LSN but NO log entries below it. A clean replay over that *
* sentinel snapshot must classify CLEAN, never TRUNCATED — the absence *
* of sub-sentinel entries is normal, not corruption (design Open *
@ -2277,6 +2345,8 @@ void test_commit_durability(void)
RUN_TEST(test_recover_verdict_empty_first_boot_clean);
RUN_TEST(test_recover_verdict_crc_break_first_entry_truncated);
RUN_TEST(test_recover_verdict_snapshot_sentinel_clean);
/* bug-778 — standby replicated-apply must leave an intact CRC chain. */
RUN_TEST(test_replicated_apply_chain_survives_cold_restart);
/* task 3.4 — re-derive the durable floor from the recovered WAL tip. */
RUN_TEST(test_durable_floor_truncating_restart_is_intact_tip);
RUN_TEST(test_durable_floor_intact_recovery_full_tip);

3
src/controller/testsuite/test_egress_filter.c
View file

@ -380,7 +380,8 @@ static void test_replay_restore_tags_entry_lsn(void)
hdr.lsn = replay_lsn;
hdr.payload_len = (uint32_t)plen;
int rc = apply_wal_entry(&g_c.persist, &hdr, payload);
int rc = apply_wal_entry(&g_c.persist, hdr.type, hdr.lsn, payload,
hdr.payload_len);
TEST_ASSERT_EQUAL_INT(0, rc);
free(payload);

4
src/controller/testsuite/test_ha_role_lifecycle.c
View file

@ -279,7 +279,7 @@ static void test_noop_apply_advances_lsn_no_entity_effect(void)
hdr.prev_crc = 0;
/* Act: apply the no-op with a NULL payload (header-only) */
int rc = apply_wal_entry(p, &hdr, NULL);
int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
/* Assert: clean apply, LSN advanced past the entry, nothing else */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
@ -288,7 +288,7 @@ static void test_noop_apply_advances_lsn_no_entity_effect(void)
"applying a WAL_NOOP must advance current_lsn to entry LSN + 1");
/* No cluster → no entity state to inspect; the absence of effect is the
* point. Re-applying must be idempotent on LSN (already advanced). */
int rc2 = apply_wal_entry(p, &hdr, NULL);
int rc2 = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc2,
"re-applying a WAL_NOOP must remain a clean no-op");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(noop_lsn + 1, p->current_lsn,

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

@ -190,14 +190,16 @@ static void test_apply_wal_entry_advances_lsn(void)
uint8_t payload[4] = {1, 2, 3, 4};
int rc = apply_wal_entry(p, &hdr, payload);
int rc = apply_wal_entry(p, hdr.type, hdr.lsn, payload, hdr.payload_len);
TEST_ASSERT_EQUAL_INT(0, rc);
/* current_lsn must have advanced to at least lsn+1 */
/* current_lsn high-water must have advanced to at least lsn+1 */
TEST_ASSERT_TRUE(p->current_lsn > 42);
/* last_crc must reflect the entry's crc */
TEST_ASSERT_EQUAL_UINT32(0xDEADBEEF, p->last_crc);
/* bug-778: apply_wal_entry is LOGICAL restore only — it must NOT touch the
* on-disk CRC chain link. last_crc stays at its initial value (the physical
* writer / replay owns it); it must NOT adopt the header's crc. */
TEST_ASSERT_EQUAL_UINT32(WAL_CRC_NONE, p->last_crc);
destroy_test_persist(p);
printf("test_apply_wal_entry_advances_lsn: PASS\n");
@ -215,7 +217,7 @@ static void test_apply_wal_entry_skips_checkpoint(void)
hdr.term = 1;
hdr.payload_len = 0;
int rc = apply_wal_entry(p, &hdr, NULL);
int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
TEST_ASSERT_EQUAL_INT(0, rc); /* skip = success, not error */
/* Checkpoint entries are local-only markers — standbys never receive them.
@ -237,7 +239,7 @@ static void test_apply_wal_entry_rejects_oversized(void)
hdr.term = 1;
hdr.payload_len = MAX_WAL_PAYLOAD_BYTES + 1; /* too large */
int rc = apply_wal_entry(p, &hdr, NULL);
int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
TEST_ASSERT_NOT_EQUAL(0, rc); /* must be rejected */
destroy_test_persist(p);
@ -260,7 +262,7 @@ static void test_apply_wal_entry_skips_unknown_type(void)
hdr.term = 1;
hdr.payload_len = 0;
int rc = apply_wal_entry(p, &hdr, NULL);
int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
TEST_ASSERT_EQUAL_INT(0, rc); /* unknown type: skip not error */
/* LSN must NOT advance for a skipped-unknown type */
@ -272,8 +274,8 @@ static void test_apply_wal_entry_skips_unknown_type(void)
static void test_apply_wal_entry_null_guard(void)
{
/* Both NULL inputs must return non-zero (EINVAL) */
int rc = apply_wal_entry(NULL, NULL, NULL);
/* NULL persist must return non-zero (EINVAL) */
int rc = apply_wal_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0);
TEST_ASSERT_NOT_EQUAL(0, rc);
printf("test_apply_wal_entry_null_guard: PASS\n");
}

13
src/include/persist.h
View file

@ -532,11 +532,18 @@ int persist_queue_hook_fun(int type, int situation, int detail,
int persist_worker_hook_fun(int type, int situation, int detail,
void* cls, va_list ap);
/* Apply a single WAL entry to in-memory state (standby replay path).
/* Apply a single WAL entry to in-memory state (LOGICAL entity restore only).
* No hooks fire — hook system is empty on standbys until promotion.
*
* This restores entity state from (type, lsn, payload) ONLY. It does NOT touch
* the on-disk WAL header, the CRC chain (p->last_crc), or current_lsn — those
* are owned exclusively by the physical layer: persist_wal_write_replicated on
* the live replicated path, and replay_wal/persist_recover on recovery. (bug-778)
*
* Returns 0 on success (including skip for checkpoint/unknown types),
* negative errno on error (oversized payload, NULL arguments). */
int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload);
* negative errno on error (oversized payload, NULL persist). */
int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
const void* payload, uint32_t payload_len);
/* Load a snapshot file from path, set p->checkpoint_lsn and p->current_lsn.
* lsn hint is informational — the actual LSN is read from the file header.