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

feat(ha-term-safety): WAL term seam + recovery read-back + legacy-0 rule (task 1.3)

Browse source 
persist_set_term sets the live election term so appends stamp hdr.term (fixes the
G1 term-0 stamping); replay_wal builds an in-memory per-LSN term index (freed in
persist_fini, no on-disk change); persist_entry_term reads it back (0=absent).
Legacy-0 -> WAL_LEGACY_TERM=1 at read-back only (raw 0 retained so present-legacy-0
is distinguishable from absent). 40-byte wal_entry_hdr_t preserved (value fix, R1.7).
5 new tests (test_term_safety_adapter.c); reviewer APPROVED. Feeds task 2.2.

All controller test groups report 0 failures; the meson full-suite timeout is the
bug-924 overlayfs /tmp fdatasync artifact, not this task.
This commit is contained in:
Rongsong Shen 2026-06-22 04:32:29 +00:00
parent a95d57f733
commit d540ebd183
6 changed files with 420 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

127
src/controller/persist.c
View file

@ -392,6 +392,90 @@ static void* fsync_worker(void* arg)
return NULL;
}
/* ------------------------------------------------------------------ *
* WAL term adapter (ha-term-safety task 1.3, R1.1/R1.5/R1.6/R1.7/R1.8) *
* ------------------------------------------------------------------ */
/* Record an entry's RAW hdr.term at its 1-based LSN in the in-memory per-LSN
* term index (R1.5). Grows the index on demand; LSNs replay/append contiguously
* so the array is dense (slot i == LSN i+1). Any gap left by a non-contiguous
* LSN (should not happen on the contiguous WAL) is zero-filled by realloc-then-
* memset, which read back as "absent" until/unless written. The raw term
* (including a legacy 0) is stored verbatim; the legacy-0 -> WAL_LEGACY_TERM
* mapping is applied only at READ-BACK (persist_entry_term), keeping "present
* legacy-0" distinguishable from "absent".
*
* Callers: the live append path (holds p->lock) and the single-threaded
* recovery path (replay_wal — no other thread touches the index before the node
* participates, so no lock is required there). Best-effort: an allocation
* failure is logged loudly and leaves the index unextended (read-back then
* reports the LSN absent) — it never aborts and never corrupts existing
* entries. */
static void record_entry_term(persist_t* p, uint64_t lsn, uint64_t term)
{
if (!p || lsn == 0)
return; /* LSN 0 is reserved — never an entry */
if (lsn > p->term_index_cap) {
/* Grow geometrically to amortize; never below the requested LSN. */
uint64_t new_cap = p->term_index_cap ? p->term_index_cap * 2 : 64;
if (new_cap < lsn)
new_cap = lsn;
uint64_t* grown = realloc(p->term_index, new_cap * sizeof(uint64_t));
if (!grown) {
ERROR("persist: term-index realloc failed (lsn=%" PRIu64
", cap=%" PRIu64 "->%" PRIu64 "); term for this LSN will read "
"back as absent", lsn, p->term_index_cap, new_cap);
return;
}
/* Zero-fill the newly grown region so any unwritten slot reads absent. */
memset(grown + p->term_index_cap, 0,
(new_cap - p->term_index_cap) * sizeof(uint64_t));
p->term_index = grown;
p->term_index_cap = new_cap;
}
p->term_index[lsn - 1] = term;
if (lsn > p->term_index_count)
p->term_index_count = lsn;
}
void persist_set_term(persist_t* p, uint64_t term)
{
if (!p)
return;
/* p->term is read on the append path under p->lock; take the lock so the
* stamped value is consistent with the append that reads it. */
pthread_mutex_lock(&p->lock);
p->term = term;
pthread_mutex_unlock(&p->lock);
}
uint64_t persist_entry_term(const persist_t* p, uint64_t lsn)
{
if (!p || lsn == 0)
return 0;
/* Take p->lock to read consistently against concurrent record_entry_term
* on the append path. Cast away const — the lock is logically a mutable
* sub-object of an otherwise-const read. */
persist_t* mp = (persist_t*)p;
pthread_mutex_lock(&mp->lock);
uint64_t result = 0;
if (lsn <= mp->term_index_count && mp->term_index != NULL) {
uint64_t raw = mp->term_index[lsn - 1];
/* Legacy-zero rule (R1.8): a present entry carrying term 0 maps to the
* single lowest-known real Raft term, applied uniformly so two legacy-0
* entries at the same LSN match. Absent LSNs are caught by the bound
* above and return 0. */
result = (raw == 0) ? WAL_LEGACY_TERM : raw;
}
pthread_mutex_unlock(&mp->lock);
return result;
}
/* ------------------------------------------------------------------ */
/* Public API */
/* ------------------------------------------------------------------ */
@ -460,6 +544,13 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
p->on_durable_advance = NULL;
p->durable_cb_ctx = NULL;
/* Per-LSN term index starts empty (ha-term-safety 1.3, R1.5): no entries
* recorded yet. Grown on demand by record_entry_term during replay/append;
* freed in persist_fini. */
p->term_index = NULL;
p->term_index_count = 0;
p->term_index_cap = 0;
/* Create WAL and snapshot directories */
if (mkdir_p(p->wal_path, 0700) != 0) return -errno;
if (mkdir_p(p->snap_path, 0700) != 0) return -errno;
@ -560,6 +651,12 @@ void persist_fini(persist_t* p)
pthread_cond_destroy(&p->checkpoint_done);
pthread_mutex_destroy(&p->fsync_mutex);
pthread_cond_destroy(&p->fsync_cond);
/* Free the in-memory per-LSN term index (ha-term-safety 1.3). */
free(p->term_index);
p->term_index = NULL;
p->term_index_count = 0;
p->term_index_cap = 0;
}
/* Forward declaration — defined after checkpoint internals */
@ -621,6 +718,10 @@ int persist_wal_async(persist_t* p, uint32_t type,
p->total_wal_bytes += entry_size;
p->last_crc = hdr.crc;
/* Record this entry's term at its LSN for recovery read-back (ha-term-safety
* 1.3, R1.5). Under p->lock; stamps the same hdr.term written to disk. */
record_entry_term(p, hdr.lsn, hdr.term);
/* Replicate to standbys — never replicate WAL_CHECKPOINT entries.
* last_replicated_crc is distinct from last_crc:
* last_crc includes checkpoint markers; last_replicated_crc skips them.
@ -757,6 +858,14 @@ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
p->total_wal_bytes += entry_size;
p->last_crc = hdr.crc;
/* Record this entry's term at its LSN for recovery read-back (ha-term-safety
* 1.3, R1.5). Stamps the same hdr.term written to disk. NOTE: the streamed
* envelope does not yet carry the leader's term, so this records the local
* best-effort p->term (the design's known hazard closed by task 2.x, which
* carries the leader's term on the envelope; this task only provides the
* read-back seam over whatever term was persisted). */
record_entry_term(p, hdr.lsn, hdr.term);
/* Advance current_lsn to one past the given lsn (max-idempotent — mirrors
* apply_wal_entry's monotonic advance). */
if (lsn + 1 > p->current_lsn)
@ -1058,6 +1167,11 @@ static int persist_write_checkpoint_marker(persist_t* p)
p->total_wal_bytes += sizeof(hdr);
p->last_crc = hdr.crc;
/* NOTE: last_replicated_crc is NOT updated — WAL_CHECKPOINT is never replicated */
/* NOTE: the term index is NOT updated for WAL_CHECKPOINT — replay_wal SKIPS
* checkpoint markers (local bookkeeping, not a logical Raft entry), so the
* recovered index never holds a checkpoint LSN; the live index must match
* (ha-term-safety 1.3). persist_entry_term is for (term,lsn) reconcile, which
* never compares against a checkpoint marker. */
pthread_mutex_unlock(&p->lock);
return 0;
@ -1583,6 +1697,19 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
/* Advance segment-local CRC */
last_crc = hdr.crc;
/* Record this entry's term at its LSN for recovery read-back
* (ha-term-safety 1.3, R1.5). Record EVERY CRC-validated logical
* entry — including snapshot-covered ones below from_lsn — so the
* index spans the full LSN range the reconcile/promotion logic may
* query. EXCLUDE WAL_CHECKPOINT markers (type == WAL_CHECKPOINT) and
* unknown future types (type > WAL_CHECKPOINT): they are skipped
* below and are never logical Raft entries, so the live append path
* does not record them either (kept symmetric). The raw hdr.term is
* stored verbatim (legacy 0 included); the legacy-0 -> WAL_LEGACY_TERM
* mapping is applied only at read-back. */
if (hdr.type < WAL_CHECKPOINT)
record_entry_term(p, hdr.lsn, hdr.term);
/* Skip entries already covered by the snapshot */
if (hdr.lsn <= from_lsn) {
free(payload);