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

feat(raft-ha-core-safety): establish snapshot (term,lsn) boundary on install (task 3.2)

Browse source 
Adds checkpointTerm (@6, append-only) to the SnapshotTransfer wire
message. The leader stamps it via persist_entry_term(p, checkpoint_lsn)
when sending a snapshot; the standby reconciles it into its own
per-LSN term index (persist_reconcile_checkpoint_term) right after
persist_load_snapshot succeeds, before applied_lsn advances. This
authoritatively overwrites any stale local term a divergent
pre-snapshot entry left at the checkpoint LSN, closing the gap task
3.1's RED test proved (Expected 7 Was 3 -> now passes). Transport
stays persist_t-free; the reconcile lives in persist.c per the
existing dependency direction. Decode is auto-generated via capnp
mapname; only the hand-rolled encode path (request.c/typemap.h, needed
for the raw snapshot payload) required a matching manual update.

test_ha_core_safety 25/0, test_config_repl 66/0, no ASAN reports.
This commit is contained in:
Rongsong Shen 2026-07-01 13:08:10 +00:00
parent 21858faba4
commit a53b9a7b6c
11 changed files with 132 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/raft-ha-core-safety/tasks.md
View file

@ -73,7 +73,7 @@
- _Requirements: 3.1, 5.1_
- _Boundary: Snapshot Boundary Reconcile_
- [ ] 3.2 Establish the snapshot `(term, lsn)` boundary
- [x] 3.2 Establish the snapshot `(term, lsn)` boundary
- Make snapshot install establish a verifiable `(term, lsn)` at the checkpoint so post-install AE intake reconciles correctly and no committed entry below the checkpoint can diverge
- Observable completion: 3.1 passes; a member that installed a snapshot matches the leader's `(term, payload-crc)` at and below the checkpoint boundary
- _Requirements: 3.1, 5.1_

9
src/controller/config_repl.c
View file

@ -2535,7 +2535,8 @@ int (*xce_snapshot_chunk_hook)(endpoint_t *ep, xce_request_t *req) = NULL;
#define SNAPSHOT_CHUNK_SIZE 65536
int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir,
uint64_t checkpoint_lsn)
uint64_t checkpoint_lsn,
uint64_t checkpoint_term)
{
if (!snap_dir)
return -EINVAL;
@ -2591,6 +2592,12 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir,
chunk.chunk_offset = offset;
chunk.total_size = total_size;
chunk.is_last = is_last;
/* raft-ha-core-safety task 3.2: stamp every chunk with the
* checkpoint's own term so the standby can reconcile its per-LSN
* term index at the boundary regardless of which chunk it last
* processes (is_last framing already exists per-chunk; cheap to
* repeat). */
chunk.checkpoint_term = checkpoint_term;
chunk.raw_payload = buf; /* borrowed for hook/encode only */
chunk.raw_payload_len = chunk_len;

10
src/controller/config_repl.h
View file

@ -397,8 +397,16 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
#define SNAPSHOT_CHUNK_SIZE 65536
/* raft-ha-core-safety task 3.2 (Req 3.1/5.1): checkpoint_term is the leader's
* own term for checkpoint_lsn (caller sources it via
* persist_entry_term(p, checkpoint_lsn) before calling — this function stays
* persist_t-free, per the layering invariant that the transport layer never
* reaches into persistence internals). Stamped onto every chunk so the
* standby can reconcile its per-LSN term index at the boundary after
* install, regardless of which chunk it processes last. */
int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir,
uint64_t checkpoint_lsn);
uint64_t checkpoint_lsn,
uint64_t checkpoint_term);
uint64_t ha_compute_oldest_wal_lsn(const char *wal_path);

25
src/controller/ha.c
View file

@ -2606,6 +2606,21 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
return rc;
}
/* raft-ha-core-safety task 3.2 (Req 3.1/5.1) — Snapshot Boundary
* Reconcile: authoritatively overwrite the per-LSN term index at the
* checkpoint LSN with the checkpoint's own term (as carried on the
* wire). Establishes a verifiable (term, lsn) boundary so a divergent
* pre-snapshot entry's STALE local term at this exact LSN is reconciled
* rather than silently surviving the install. Applies identically to
* both a legacy (position-only) and an entity-bearing checkpoint — the
* term reconcile is orthogonal to entity content. chunk->checkpoint_term
* == 0 means a legacy sender never stamped the wire field; persist's
* existing legacy-0 -> WAL_LEGACY_TERM convention (applied uniformly at
* read-back by persist_entry_term) already covers that case, so no
* special-casing is needed here. */
persist_reconcile_checkpoint_term(p, chunk->snapshot_lsn,
chunk->checkpoint_term);
/* Advance applied_lsn to checkpoint_lsn (in-memory only — Req 4.2) */
atomic_store_explicit(&ha->applied_lsn, p->checkpoint_lsn,
memory_order_release);
@ -2753,7 +2768,7 @@ static void ha_snapshot_do_send(cluster_t *c, endpoint_t *ep, int peer_idx)
#endif
if (c && c->persist.snap_path[0]) {
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *,
uint64_t);
uint64_t, uint64_t);
/* Req 4.1: create a FRESH checkpoint so the transferred snapshot
* reflects current state (also trims WAL + rotates old snaps).
* -EBUSY means another checkpoint is in flight — fall back to the
@ -2763,8 +2778,14 @@ static void ha_snapshot_do_send(cluster_t *c, endpoint_t *ep, int peer_idx)
if (ckpt_rc != 0 && ckpt_rc != -EBUSY)
WARN("HA repl: persist_checkpoint before snapshot send "
"failed rc=%d — sending existing checkpoint", ckpt_rc);
/* raft-ha-core-safety task 3.2: source the checkpoint's own term so
* the standby can reconcile its per-LSN term index at the boundary
* after install (Snapshot Boundary Reconcile). */
uint64_t checkpoint_term =
persist_entry_term(&c->persist, c->persist.checkpoint_lsn);
ha_send_snapshot_to_standby(ep, c->persist.snap_path,
c->persist.checkpoint_lsn);
c->persist.checkpoint_lsn,
checkpoint_term);
}
}

23
src/controller/persist.c
View file

@ -505,6 +505,29 @@ uint64_t persist_entry_term(const persist_t* p, uint64_t lsn)
return result;
}
/* persist_reconcile_checkpoint_term — raft-ha-core-safety task 3.2 (Req
* 3.1/5.1), Snapshot Boundary Reconcile. See persist.h for the full contract.
* Reuses record_entry_term (same translation unit) to authoritatively
* overwrite whatever term (if any) the local term index held at
* checkpoint_lsn — this is a genuine reconcile-overwrite, not an idempotent
* skip, because a pre-snapshot divergent entry may have left a stale local
* term at this exact LSN. Only the checkpoint LSN's own term is reconciled:
* post-install Raft semantics mean no AppendEntries prevLogIndex/prevLogTerm
* check against this member ever references an LSN below the checkpoint
* again (the checkpoint is the new log-start boundary), and the WAL bytes
* below it were compacted away, so there is no data to reconstruct a full
* per-LSN history from even if it were needed. */
void persist_reconcile_checkpoint_term(persist_t* p, uint64_t checkpoint_lsn,
uint64_t checkpoint_term)
{
if (!p || checkpoint_lsn == 0)
return;
pthread_mutex_lock(&p->lock);
record_entry_term(p, checkpoint_lsn, checkpoint_term);
pthread_mutex_unlock(&p->lock);
}
/* ------------------------------------------------------------------ */
/* Public API */
/* ------------------------------------------------------------------ */

10
src/controller/testsuite/test_config_repl.c
View file

@ -2848,8 +2848,9 @@ static void test_snapshot_transfer_two_chunks(void)
xce_snapshot_chunk_hook = snaptest_hook;
snaptest_reset();
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t);
int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn);
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *,
uint64_t, uint64_t);
int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_snaptest_chunk_count);
@ -2891,8 +2892,9 @@ static void test_snapshot_transfer_small_less_than_one_chunk(void)
xce_snapshot_chunk_hook = snaptest_hook;
snaptest_reset();
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t);
int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn);
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *,
uint64_t, uint64_t);
int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunk_count);

23
src/controller/testsuite/test_ha_core_safety.c
View file

@ -977,10 +977,14 @@ static void test_3_1_snapshot_install_does_not_reconcile_boundary_term(void)
"at the STALE local term");
/* A leader at a DIFFERENT (higher) term ships a checkpoint whose position
* IS this exact boundary LSN. Today's on-disk/wire format carries no
* term field at all, so the legacy (position-only) shape is exactly what
* a real checkpoint transfer installs via ha_handle_snapshot_chunk's real
* chunk-assembly + rename + load path (not a stub). */
* IS this exact boundary LSN. The on-disk snapshot payload itself stays
* legacy (position-only: magic + lsn, no term) -- exactly what a real
* checkpoint transfer installs via ha_handle_snapshot_chunk's real
* chunk-assembly + rename + load path (not a stub). The wire chunk's
* checkpoint_term field (task 3.2) is what the leader now stamps with its
* own term for this checkpoint (via persist_entry_term at send time,
* ha_send_snapshot_to_standby) -- set here to simulate that real sender
* behavior, independent of the (still-legacy) snapshot file format. */
char snap_body[16];
memcpy(snap_body, "XCESNAP\0", 8);
uint64_t boundary_lsn_u64 = (uint64_t)BOUNDARY_LSN;
@ -988,11 +992,12 @@ static void test_3_1_snapshot_install_does_not_reconcile_boundary_term(void)
snapshot_transfer_t chunk;
memset(&chunk, 0, sizeof(chunk));
chunk.snapshot_lsn = boundary_lsn_u64;
chunk.chunk_index = 0;
chunk.chunk_offset = 0;
chunk.total_size = sizeof(snap_body);
chunk.is_last = 1;
chunk.snapshot_lsn = boundary_lsn_u64;
chunk.chunk_index = 0;
chunk.chunk_offset = 0;
chunk.total_size = sizeof(snap_body);
chunk.is_last = 1;
chunk.checkpoint_term = (uint64_t)CHECKPOINT_TERM;
int rc = ha_handle_snapshot_chunk(&ha, m.p, m.p->snap_path, &chunk,
(const uint8_t*)snap_body,

25
src/include/persist.h
View file

@ -354,6 +354,31 @@ void persist_set_term(persist_t* p, uint64_t term);
* Safe on a NULL p (returns 0). */
uint64_t persist_entry_term(const persist_t* p, uint64_t lsn);
/* raft-ha-core-safety task 3.2 (Req 3.1/5.1) — Snapshot Boundary Reconcile.
*
* Authoritatively overwrite the per-LSN term index entry at `checkpoint_lsn`
* with `checkpoint_term` (the checkpoint's own true term, as carried on the
* wire by the leader that shipped the snapshot). Called from the standby
* snapshot-install receive path (ha_handle_snapshot_chunk) immediately after
* persist_load_snapshot() succeeds, so a divergent pre-snapshot entry's STALE
* local term at the checkpoint LSN is reconciled to the checkpoint's own
* term. Establishes a verifiable (term, lsn) boundary: post-install AE intake
* (prevLogIndex/prevLogTerm checks) never references an LSN below the
* checkpoint against this member again, so only the checkpoint LSN's own
* term needs reconciling — not the full compacted history below it.
*
* `checkpoint_term == 0` means a legacy sender (pre-task-3.2 peer) never
* stamped the wire field; treat that as WAL_LEGACY_TERM per the existing
* legacy-0 convention (record_entry_term stores the raw value verbatim; the
* legacy-0 -> WAL_LEGACY_TERM mapping is applied uniformly at read-back by
* persist_entry_term/entry_term_locked, so this call is consistent with that
* convention without needing to special-case it here).
*
* Takes p->lock internally (callers must NOT already hold it). Safe on a
* NULL p or checkpoint_lsn == 0 (no-op). */
void persist_reconcile_checkpoint_term(persist_t* p, uint64_t checkpoint_lsn,
uint64_t checkpoint_term);
/* Lifecycle */
int persist_init(persist_t* p, persist_config_t* config, cluster_t* c);
int persist_start(persist_t* p);

5
src/include/typemap.h
View file

@ -338,6 +338,11 @@ struct snapshot_transfer_s {
* owned by the struct, freed by xce_free_request(). */
uint8_t *raw_payload;
size_t raw_payload_len;
/* raft-ha-core-safety task 3.2 (Req 3.1/5.1): the leader's term for the
* checkpoint LSN above. 0 = unset (legacy sender) -> receiver treats as
* WAL_LEGACY_TERM, same convention as persist's existing legacy-0 term
* mapping (see entry_term_locked in persist.c). */
uint64_t checkpoint_term;
};
struct wal_catchup_entry {

13
src/lib/request.c
View file

@ -632,11 +632,14 @@ int xce_snapshot_chunk_encode_raw(const snapshot_transfer_t *chunk,
memset(&rq, 0, sizeof(rq));
memset(&st, 0, sizeof(st));
st.snapshotLsn = chunk->snapshot_lsn;
st.chunkIndex = chunk->chunk_index;
st.chunkOffset = chunk->chunk_offset;
st.totalSize = chunk->total_size;
st.isLast = chunk->is_last ? 1 : 0;
st.snapshotLsn = chunk->snapshot_lsn;
st.chunkIndex = chunk->chunk_index;
st.chunkOffset = chunk->chunk_offset;
st.totalSize = chunk->total_size;
st.isLast = chunk->is_last ? 1 : 0;
/* raft-ha-core-safety task 3.2: checkpoint's term at snapshotLsn, so the
* receiver can reconcile its per-LSN term index at the boundary. */
st.checkpointTerm = chunk->checkpoint_term;
if (payload && payload_len > 0) {
capn_list8 pl = capn_new_list8(cs, (int)payload_len);
capn_setv8(pl, 0, payload, (int)payload_len);

10
src/schema/controller.capnp
View file

@ -878,6 +878,16 @@ struct SnapshotTransfer $C.mapname("snapshot_transfer_t") {
chunkOffset @3 :UInt64 $C.mapname("chunk_offset"); # byte offset in the full snapshot
totalSize @4 :UInt64 $C.mapname("total_size"); # total snapshot size in bytes
isLast @5 :Bool $C.mapname("is_last"); # true for the final chunk
# raft-ha-core-safety task 3.2 (Req 3.1/5.1): the leader's term for the
# checkpoint LSN (persist_entry_term(p, snapshotLsn) at send time). The
# receiver reconciles this into its own per-LSN term index after install so
# the snapshot boundary establishes a verifiable (term, lsn) pair and any
# divergent stale-term pre-snapshot entry at this LSN is authoritatively
# overwritten. Append-only field number (@6) — no collision with @0-@5
# above. 0 = unset (legacy sender that never stamped this field) — the
# receiver treats 0 as WAL_LEGACY_TERM, the same convention persist's
# existing legacy-0 term mapping already uses.
checkpointTerm @6 :UInt64 $C.mapname("checkpoint_term");
}
struct WalCatchupEntry $C.mapname("wal_catchup_entry_t") {