diff --git a/.kiro/specs/raft-ha-core-safety/tasks.md b/.kiro/specs/raft-ha-core-safety/tasks.md index 99592f3..834f4fa 100644 --- a/.kiro/specs/raft-ha-core-safety/tasks.md +++ b/.kiro/specs/raft-ha-core-safety/tasks.md @@ -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_ diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index c64532c..c437549 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -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; diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 3e32027..f4f1609 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -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); diff --git a/src/controller/ha.c b/src/controller/ha.c index c9f8ea9..9675ae0 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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); } } diff --git a/src/controller/persist.c b/src/controller/persist.c index 455d9cb..f3cfcbb 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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 */ /* ------------------------------------------------------------------ */ diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index bdf1062..25da2c7 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -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); diff --git a/src/controller/testsuite/test_ha_core_safety.c b/src/controller/testsuite/test_ha_core_safety.c index 220e7a5..72e0781 100644 --- a/src/controller/testsuite/test_ha_core_safety.c +++ b/src/controller/testsuite/test_ha_core_safety.c @@ -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, diff --git a/src/include/persist.h b/src/include/persist.h index f4f524b..ad9a75f 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -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); diff --git a/src/include/typemap.h b/src/include/typemap.h index c41e09e..a211a4b 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -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 { diff --git a/src/lib/request.c b/src/lib/request.c index f8fbe43..f41e6b8 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -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); diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index e753b73..4254df2 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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") {