feat(ha-term-safety): retire term-0 repl_broadcast live send; carry leader term on replication (task 3.4)
Closes bug-990: the live replicator repl_broadcast wire-encoded term=0 (entry.term was never set), so standbys persisted hdr.term=0 and took the legacy direct-apply path instead of the library (term,lsn) intake. Now the live broadcast and catch-up streamer carry the leader's real term + AE prev tuple; ha_apply_wal_catchup_entry routes every term-carrying entry through raft_on_append_entries_request -> on_persist_entry (leader term stamped), retiring the term-0 direct-apply. Standbys now persist the leader's real term and reconcile by (term,lsn); the live commit chain is end-to-end. 848/849/850 + persist-before-ACK + commit-barrier + bug-960 preserved. Proven at integration level; the definitive live STRICT_TERM failover run is env-blocked here (bug-924 overlayfs) and recommended on a real fs.
This commit is contained in:
parent
87242aa8e9
commit
efe621c124
11 changed files with 307 additions and 45 deletions
|
|
@ -2055,6 +2055,16 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
int batch_n = 0;
|
||||
size_t batch_bytes = 0;
|
||||
|
||||
/* ha-term-safety task 3.4: track the previous streamed entry's
|
||||
* (lsn, term) so each catch-up entry carries the AppendEntries prev tuple
|
||||
* (prev_log_index, prev_log_term) + its own term. The standby intake routes
|
||||
* every term-carrying sub-entry through raft_on_append_entries_request so it
|
||||
* persists the LEADER's term (not a local substitute). Entries are streamed
|
||||
* in strict ascending LSN order, so the prev tuple of entry N is exactly the
|
||||
* (lsn, term) of the entry emitted just before it (0/0 before the first). */
|
||||
uint64_t prev_streamed_lsn = 0;
|
||||
uint64_t prev_streamed_term = 0;
|
||||
|
||||
for (int si = 0; si < nseg; si++) {
|
||||
/* Skip segments whose entire range is at or below standby_lsn */
|
||||
if (si + 1 < nseg && seg_lsns[si + 1] <= standby_lsn)
|
||||
|
|
@ -2091,6 +2101,10 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
batch_n = 0;
|
||||
batch_bytes = 0;
|
||||
}
|
||||
/* Advance the prev tracker even on the OOM-skip so a later
|
||||
* entry's prev tuple stays correct. */
|
||||
prev_streamed_lsn = hdr.lsn;
|
||||
prev_streamed_term = hdr.term;
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -2099,6 +2113,15 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
entry->n_entries = 0;
|
||||
entry->entries = NULL;
|
||||
entry->payload = NULL;
|
||||
/* ha-term-safety task 3.4: carry the leader's term + prev tuple
|
||||
* so the standby intake routes this through the library and
|
||||
* persists the LEADER's term (R1.3/R1.4). prev_streamed_* holds
|
||||
* the (lsn,term) of the immediately preceding WAL entry (0/0
|
||||
* before the first), which IS the predecessor in contiguous LSN
|
||||
* order. */
|
||||
entry->term = hdr.term;
|
||||
entry->prev_log_index = prev_streamed_lsn;
|
||||
entry->prev_log_term = prev_streamed_term;
|
||||
|
||||
if (hdr.payload_len > 0) {
|
||||
uint8_t *pbuf = malloc(hdr.payload_len);
|
||||
|
|
@ -2130,6 +2153,13 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
batch_bytes = 0;
|
||||
}
|
||||
}
|
||||
/* Advance the prev tracker for EVERY scanned entry — including those
|
||||
* at/below standby_lsn that are not sent — so the first streamed
|
||||
* entry's prev tuple references the entry the standby already holds
|
||||
* (prev_log_index == standby_lsn). (Skips on the OOM path above set
|
||||
* it inline before continuing.) */
|
||||
prev_streamed_lsn = hdr.lsn;
|
||||
prev_streamed_term = hdr.term;
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
}
|
||||
close(fd);
|
||||
|
|
|
|||
|
|
@ -2073,6 +2073,54 @@ static void ha_send_wal_ack_reply(ha_state_t *ha, struct endpoint_t *ep,
|
|||
xce_reply(ep, c->myhost, &reply);
|
||||
}
|
||||
|
||||
/* ha_apply_term_entry_via_library: route ONE term-carrying replicated entry
|
||||
* (a live broadcast or a catch-up sub-entry) through the raft library
|
||||
* (raft_on_append_entries_request via ha_consume_append_entries) so the LIBRARY
|
||||
* does logOk + (term,lsn) reconcile and on_persist_entry stamps the LEADER's
|
||||
* term — never the local term-0 direct-apply path (ha-term-safety task 3.4).
|
||||
*
|
||||
* `e` is the single entry (its term/prev_log_index/prev_log_term carry the AE
|
||||
* metadata, set by the leader's broadcast/stream path). On accept (the entry's
|
||||
* term is now recorded at its LSN) the caller ACKs and advances applied. Returns
|
||||
* 1 if the entry was accepted (recorded), 0 otherwise (reject / not yet
|
||||
* applicable — e.g. a logOk-failing forward gap that catch-up will backfill). */
|
||||
static int ha_apply_term_entry_via_library(ha_state_t *ha, persist_t *p,
|
||||
cluster_t *c, struct endpoint_t *ep,
|
||||
wal_catchup_entry_t *e)
|
||||
{
|
||||
/* Wrap the single entry in an AE container whose TOP-LEVEL fields carry the
|
||||
* prev tuple + term that ha_consume_append_entries reads, with entries[0]
|
||||
* pointing at the entry itself (the same shape repl_send_append_entries /
|
||||
* repl_broadcast produce on the wire). */
|
||||
wal_catchup_entry_t *sub_arr[1] = { e };
|
||||
wal_catchup_entry_t ae;
|
||||
memset(&ae, 0, sizeof(ae));
|
||||
ae.n_entries = 1;
|
||||
ae.entries = sub_arr;
|
||||
ae.term = e->term;
|
||||
ae.prev_log_index = e->prev_log_index;
|
||||
ae.prev_log_term = e->prev_log_term;
|
||||
ae.leader_commit = e->leader_commit;
|
||||
|
||||
uint64_t before_term = persist_entry_term(p, e->lsn);
|
||||
int rc = ha_consume_append_entries(c, &ae, /*leader_idx*/ -1);
|
||||
if (rc != 0)
|
||||
return 0;
|
||||
|
||||
uint64_t after_term = persist_entry_term(p, e->lsn);
|
||||
if (after_term == e->term) {
|
||||
/* Accepted (appended / conflict-truncate-then-persist) or already-have
|
||||
* (idempotent re-stream): ACK + advance applied on that signal. */
|
||||
(void)before_term;
|
||||
ha_advance_applied_lsn(ha, e->lsn + 1);
|
||||
ha_send_wal_ack_reply(ha, ep, c, e->lsn);
|
||||
return 1;
|
||||
}
|
||||
/* logOk reject (e.g. forward gap before the prefix arrives): no ACK; the
|
||||
* ACK-driven catch-up continuation re-streams from current_lsn in order. */
|
||||
return 0;
|
||||
}
|
||||
|
||||
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
||||
wal_catchup_entry_t *entry,
|
||||
struct endpoint_t *ep, cluster_t *c)
|
||||
|
|
@ -2094,23 +2142,7 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
|||
* computes the target index from the prev tuple, then ACK on accept. */
|
||||
if (c && c->ha.raft_c == c && entry->n_entries == 1 &&
|
||||
entry->entries && entry->entries[0] && entry->entries[0]->term != 0) {
|
||||
wal_catchup_entry_t *e = entry->entries[0];
|
||||
uint64_t before_term = persist_entry_term(p, e->lsn);
|
||||
int rc = ha_consume_append_entries(c, entry, /*leader_idx*/-1);
|
||||
if (rc == 0) {
|
||||
/* Accept (append or conflict-truncate-then-persist) leaves the entry
|
||||
* term recorded at its LSN — ACK + advance applied on that signal.
|
||||
* A reject (logOk fail) leaves the prior term unchanged → no ACK. */
|
||||
uint64_t after_term = persist_entry_term(p, e->lsn);
|
||||
if (after_term == e->term && after_term != before_term) {
|
||||
ha_advance_applied_lsn(ha, e->lsn + 1);
|
||||
ha_send_wal_ack_reply(ha, ep, c, e->lsn);
|
||||
} else if (after_term == e->term) {
|
||||
/* already-have (idempotent re-stream): ACK without a rewrite. */
|
||||
ha_advance_applied_lsn(ha, e->lsn + 1);
|
||||
ha_send_wal_ack_reply(ha, ep, c, e->lsn);
|
||||
}
|
||||
}
|
||||
ha_apply_term_entry_via_library(ha, p, c, ep, entry->entries[0]);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -2131,6 +2163,22 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
|||
continue;
|
||||
}
|
||||
|
||||
/* ha-term-safety task 3.4: a term-carrying catch-up sub-entry routes
|
||||
* through the LIBRARY (logOk + (term,lsn) reconcile + on_persist_entry
|
||||
* stamps the LEADER's term). Each sub-entry carries its own prev tuple
|
||||
* (wal_entry_to_wire encodes prevLog{Index,Term} per element). This
|
||||
* retires the term-0 direct-apply path for live/catch-up replication.
|
||||
* Entries stream in ascending LSN order, so the library's contiguity
|
||||
* (prev tuple) is satisfied in sequence; a not-yet-applicable entry
|
||||
* (logOk reject) is left for the catch-up continuation. */
|
||||
if (c && c->ha.raft_c == c && sub->term != 0) {
|
||||
if (ha_apply_term_entry_via_library(ha, p, c, ep, sub)) {
|
||||
last_good_lsn = sub->lsn;
|
||||
had_good = 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
int rc = ha_apply_single_entry(p, sub);
|
||||
if (rc != 0) {
|
||||
/* -EAGAIN = contiguity gap (bug-850): entry ahead of our tail,
|
||||
|
|
|
|||
|
|
@ -440,6 +440,24 @@ static void record_entry_term(persist_t* p, uint64_t lsn, uint64_t term)
|
|||
p->term_index_count = lsn;
|
||||
}
|
||||
|
||||
/* entry_term_locked: the lock-free core of persist_entry_term — read the
|
||||
* recorded Raft term for `lsn` from the in-memory term index, applying the
|
||||
* legacy-0 rule (R1.8: a present term-0 entry maps to WAL_LEGACY_TERM; an absent
|
||||
* LSN returns 0). Caller MUST hold p->lock. Used by the append path
|
||||
* (persist_wal_async) to source the AppendEntries prev_log_term while already
|
||||
* under the lock — calling persist_entry_term there would re-acquire p->lock
|
||||
* and deadlock. */
|
||||
static uint64_t entry_term_locked(const persist_t* p, uint64_t lsn)
|
||||
{
|
||||
if (!p || lsn == 0)
|
||||
return 0;
|
||||
if (lsn <= p->term_index_count && p->term_index != NULL) {
|
||||
uint64_t raw = p->term_index[lsn - 1];
|
||||
return (raw == 0) ? WAL_LEGACY_TERM : raw;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Shrink the in-memory per-LSN term index so that every LSN > `keep_lsn` reads
|
||||
* back as absent (ha-term-safety task 2.2). Called from the truncate path after
|
||||
* the on-disk entries above keep_lsn are removed, keeping the term index
|
||||
|
|
@ -477,15 +495,11 @@ uint64_t persist_entry_term(const persist_t* p, uint64_t lsn)
|
|||
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;
|
||||
}
|
||||
/* Legacy-zero rule (R1.8) lives in entry_term_locked: 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
|
||||
* return 0. */
|
||||
uint64_t result = entry_term_locked(mp, lsn);
|
||||
|
||||
pthread_mutex_unlock(&mp->lock);
|
||||
return result;
|
||||
|
|
@ -744,7 +758,13 @@ int persist_wal_async(persist_t* p, uint32_t type,
|
|||
* Note: WAL_CHECKPOINT is blocked by the assert at function entry, so
|
||||
* every entry that reaches here is a replicable entry. */
|
||||
p->last_replicated_crc = hdr.crc;
|
||||
repl_broadcast(&p->repl, &hdr, payload, len);
|
||||
/* ha-term-safety task 3.4: source the AppendEntries prev_log_term (term of
|
||||
* the entry at hdr.lsn-1) under p->lock via the lock-free accessor, so the
|
||||
* live broadcast carries the leader's term + prev tuple and the standby
|
||||
* routes it through the library (raft_on_append_entries_request). 0 when
|
||||
* hdr.lsn == 1 (no predecessor). */
|
||||
uint64_t prev_log_term = entry_term_locked(p, hdr.lsn - 1);
|
||||
repl_broadcast(&p->repl, &hdr, payload, len, prev_log_term);
|
||||
|
||||
/* Rotate segment when full */
|
||||
if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
|
||||
|
|
|
|||
|
|
@ -237,7 +237,8 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
|
|||
* ============================================================ */
|
||||
|
||||
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
|
||||
const void *payload, uint32_t payload_len)
|
||||
const void *payload, uint32_t payload_len,
|
||||
uint64_t prev_log_term)
|
||||
{
|
||||
/* WAL_CHECKPOINT entries must never be replicated */
|
||||
assert(hdr != NULL);
|
||||
|
|
@ -276,13 +277,25 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
|
|||
continue;
|
||||
}
|
||||
|
||||
/* Build a single-entry batch and enqueue it for this standby. */
|
||||
/* Build a single-entry AppendEntries and enqueue it for this standby.
|
||||
*
|
||||
* ha-term-safety task 3.4 (Req 1.3/1.4): carry the leader's per-entry
|
||||
* term + the AppendEntries prev tuple so the standby intake
|
||||
* (ha_apply_wal_catchup_entry, entry->term != 0) routes this through
|
||||
* raft_on_append_entries_request — the library does UpdateTerm/logOk/
|
||||
* (term,lsn) reconcile and on_persist_entry stamps the LEADER's term.
|
||||
* The term-0 legacy direct-apply path is retired for live entries.
|
||||
* prev_log_index == hdr->lsn - 1 (0 for lsn 1); prev_log_term is the
|
||||
* term recorded for that LSN, supplied by the caller (term index). */
|
||||
wal_catchup_entry_t entry;
|
||||
memset(&entry, 0, sizeof(entry));
|
||||
entry.lsn = hdr->lsn;
|
||||
entry.entry_type = (uint8_t)hdr->type;
|
||||
entry.raw_payload = (uint8_t *)(uintptr_t)payload;
|
||||
entry.raw_payload_len = payload_len;
|
||||
entry.term = hdr->term;
|
||||
entry.prev_log_index = (hdr->lsn > 0) ? hdr->lsn - 1 : 0;
|
||||
entry.prev_log_term = prev_log_term;
|
||||
|
||||
wal_catchup_entry_t *ep_arr[1] = { &entry };
|
||||
|
||||
|
|
|
|||
|
|
@ -754,7 +754,7 @@ static int cv_broadcast_one(replicator_t *r, uint64_t lsn)
|
|||
hdr.lsn = lsn;
|
||||
hdr.type = WAL_NOOP;
|
||||
hdr.payload_len = 0;
|
||||
return repl_broadcast(r, &hdr, NULL, 0);
|
||||
return repl_broadcast(r, &hdr, NULL, 0, 0 /* prev_log_term */);
|
||||
}
|
||||
|
||||
/* (a) umbrella 6.1: repl_broadcast enqueues to EVERY registered active standby
|
||||
|
|
|
|||
|
|
@ -279,7 +279,7 @@ static void test_broadcast_over_limit_removes_standby(void)
|
|||
|
||||
/* Broadcast a wal entry — the over-limit standby should be evicted */
|
||||
wal_entry_hdr_t hdr = { .type = WAL_JOB_SUBMIT, .lsn = 1 };
|
||||
int rc = repl_broadcast(&r, &hdr, NULL, 0);
|
||||
int rc = repl_broadcast(&r, &hdr, NULL, 0, 0 /* prev_log_term */);
|
||||
/* repl_broadcast itself may return 0 even after eviction */
|
||||
(void)rc;
|
||||
|
||||
|
|
|
|||
|
|
@ -186,6 +186,11 @@ static uint8_t g_hook_last_entry_type;
|
|||
static const void *g_hook_last_payload;
|
||||
static size_t g_hook_last_payload_len;
|
||||
static int g_hook_return_val; /* 0 = success, negative = error */
|
||||
/* ha-term-safety task 3.4: capture the AppendEntries term + prev tuple the live
|
||||
* broadcast carries, so a test can assert the leader's term is on the wire. */
|
||||
static uint64_t g_hook_last_term;
|
||||
static uint64_t g_hook_last_prev_log_index;
|
||||
static uint64_t g_hook_last_prev_log_term;
|
||||
|
||||
static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries,
|
||||
int n_entries)
|
||||
|
|
@ -193,10 +198,13 @@ static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries,
|
|||
g_hook_call_count++;
|
||||
g_hook_last_ep = ep;
|
||||
if (n_entries >= 1 && entries && entries[0]) {
|
||||
g_hook_last_lsn = entries[0]->lsn;
|
||||
g_hook_last_entry_type = entries[0]->entry_type;
|
||||
g_hook_last_payload = entries[0]->raw_payload;
|
||||
g_hook_last_payload_len = entries[0]->raw_payload_len;
|
||||
g_hook_last_lsn = entries[0]->lsn;
|
||||
g_hook_last_entry_type = entries[0]->entry_type;
|
||||
g_hook_last_payload = entries[0]->raw_payload;
|
||||
g_hook_last_payload_len = entries[0]->raw_payload_len;
|
||||
g_hook_last_term = entries[0]->term;
|
||||
g_hook_last_prev_log_index = entries[0]->prev_log_index;
|
||||
g_hook_last_prev_log_term = entries[0]->prev_log_term;
|
||||
}
|
||||
return g_hook_return_val;
|
||||
}
|
||||
|
|
@ -210,6 +218,9 @@ static void reset_hook_state(void)
|
|||
g_hook_last_payload = NULL;
|
||||
g_hook_last_payload_len = 0;
|
||||
g_hook_return_val = 0;
|
||||
g_hook_last_term = 0;
|
||||
g_hook_last_prev_log_index = 0;
|
||||
g_hook_last_prev_log_term = 0;
|
||||
}
|
||||
|
||||
/* Test: no standbys — broadcast returns 0, hook not called */
|
||||
|
|
@ -223,7 +234,7 @@ static void test_broadcast_no_standbys_returns_zero(void)
|
|||
|
||||
wal_entry_hdr_t hdr = { .lsn = 10, .type = WAL_JOB_SUBMIT };
|
||||
uint8_t payload[] = { 0xAA, 0xBB };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(0, g_hook_call_count);
|
||||
|
|
@ -249,7 +260,7 @@ static void test_broadcast_one_standby_sends_entry(void)
|
|||
|
||||
wal_entry_hdr_t hdr = { .lsn = 42, .type = WAL_JOB_SUBMIT };
|
||||
uint8_t payload[] = { 1, 2, 3, 4 };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(1, g_hook_call_count);
|
||||
|
|
@ -267,6 +278,46 @@ static void test_broadcast_one_standby_sends_entry(void)
|
|||
#endif
|
||||
}
|
||||
|
||||
/* ha-term-safety task 3.4 (Req 1.3/1.4): the LIVE broadcast must carry the
|
||||
* leader's per-entry term and the AppendEntries prev tuple, so the standby
|
||||
* intake routes it through raft_on_append_entries_request and persists the
|
||||
* LEADER's term — not the term-0 legacy direct-apply path. Before the cutover
|
||||
* repl_broadcast sent term=0 (RED): the wire entry's term was always 0. */
|
||||
static void test_broadcast_carries_leader_term_and_prev_tuple(void)
|
||||
{
|
||||
#ifdef XCE_TESTING
|
||||
replicator_t r;
|
||||
int dummy_ep = 0;
|
||||
repl_init(&r);
|
||||
repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001);
|
||||
reset_hook_state();
|
||||
repl_broadcast_send_hook = test_broadcast_hook;
|
||||
|
||||
/* Leader appends LSN 5 at term 7; predecessor (LSN 4) was term 7. */
|
||||
wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT, .term = 7 };
|
||||
uint8_t payload[] = { 9, 8, 7 };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload),
|
||||
7 /* prev_log_term of LSN 4 */);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(1, g_hook_call_count);
|
||||
TEST_ASSERT_EQUAL_UINT64(5, g_hook_last_lsn);
|
||||
/* The load-bearing assertion: the leader's real term is on the wire entry
|
||||
* (NOT 0). This FAILS before the task 3.4 cutover. */
|
||||
TEST_ASSERT_EQUAL_UINT64(7, g_hook_last_term);
|
||||
/* prev tuple: prev_log_index == lsn - 1; prev_log_term == supplied. */
|
||||
TEST_ASSERT_EQUAL_UINT64(4, g_hook_last_prev_log_index);
|
||||
TEST_ASSERT_EQUAL_UINT64(7, g_hook_last_prev_log_term);
|
||||
|
||||
repl_broadcast_send_hook = NULL;
|
||||
repl_fini(&r);
|
||||
printf("test_broadcast_carries_leader_term_and_prev_tuple: PASS\n");
|
||||
#else
|
||||
printf("test_broadcast_carries_leader_term_and_prev_tuple: "
|
||||
"SKIPPED (no XCE_TESTING)\n");
|
||||
#endif
|
||||
}
|
||||
|
||||
/* Test: standby in catch-up is skipped — hook not called for it */
|
||||
static void test_broadcast_skips_catchup_standby(void)
|
||||
{
|
||||
|
|
@ -283,7 +334,7 @@ static void test_broadcast_skips_catchup_standby(void)
|
|||
|
||||
wal_entry_hdr_t hdr = { .lsn = 99, .type = WAL_JOB_SUBMIT };
|
||||
uint8_t payload[] = { 0xFF };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
/* Only sb2 should be sent to */
|
||||
|
|
@ -313,7 +364,7 @@ static void test_broadcast_evicts_overloaded_standby(void)
|
|||
|
||||
wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT };
|
||||
uint8_t payload[] = { 0x01 };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
/* Evicted — hook must NOT be called */
|
||||
|
|
@ -477,7 +528,7 @@ static void test_broadcast_send_failure_continues(void)
|
|||
|
||||
wal_entry_hdr_t hdr = { .lsn = 77, .type = WAL_JOB_SUBMIT };
|
||||
uint8_t payload[] = { 0xDE, 0xAD };
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
|
||||
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */);
|
||||
|
||||
/* broadcast should still return 0 — errors are WARN+continue */
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
|
@ -548,6 +599,7 @@ void test_repl_unit(void) {
|
|||
/* Task 1.1: broadcast send path */
|
||||
RUN_TEST(test_broadcast_no_standbys_returns_zero);
|
||||
RUN_TEST(test_broadcast_one_standby_sends_entry);
|
||||
RUN_TEST(test_broadcast_carries_leader_term_and_prev_tuple);
|
||||
RUN_TEST(test_broadcast_skips_catchup_standby);
|
||||
RUN_TEST(test_broadcast_evicts_overloaded_standby);
|
||||
RUN_TEST(test_broadcast_send_failure_continues);
|
||||
|
|
|
|||
|
|
@ -1278,6 +1278,83 @@ static void test_intake_logmismatch_rejected_no_false_convergence(void)
|
|||
printf("test_intake_logmismatch_rejected_no_false_convergence: PASS\n");
|
||||
}
|
||||
|
||||
/* ================================================================== */
|
||||
/* Task 3.4 — live broadcast/catch-up receive cutover (closes bug-990) */
|
||||
/* ================================================================== */
|
||||
|
||||
/* The live cutover's observable: a standby applying a LIVE-broadcast-shaped
|
||||
* AppendEntries (a single entry in the `entries` LIST carrying the leader's term
|
||||
* + prev tuple — exactly what repl_broadcast → xce_wal_batch_req_raw produces and
|
||||
* xce_decode yields) through the PRODUCTION receive entry point
|
||||
* ha_apply_wal_catchup_entry persists the LEADER's term (hdr.term == leader_term),
|
||||
* NOT 0 (the retired term-0 legacy direct-apply path) and NOT its local term.
|
||||
*
|
||||
* Before the cutover repl_broadcast sent entry->term == 0, so this entry took the
|
||||
* term-0 legacy direct-apply branch and persisted hdr.term = local p->term — the
|
||||
* bug-990 term-identity gap. With the cutover the term-carrying entry routes
|
||||
* through raft_on_append_entries_request → on_persist_entry → leader term stamped. */
|
||||
static void test_live_broadcast_receive_persists_leader_term(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_ts_live_bcast_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
cluster_t* c = make_cluster(tmp);
|
||||
TEST_ASSERT_NOT_NULL(c);
|
||||
/* The standby is a follower at a LOW local term (3) so we can prove the
|
||||
* persisted term is the LEADER's (7), not the local term. */
|
||||
make_follower(c, /*term*/3, /*n_peers*/3);
|
||||
|
||||
/* Seed LSN 1 at the leader's term 7 so the broadcast of LSN 2 satisfies the
|
||||
* library's logOk (prev=(1,7)). This mirrors the contiguous live stream. */
|
||||
uint8_t blob[64];
|
||||
size_t n = pack_entry(blob, WAL_JOB_SUBMIT, "a", 1);
|
||||
TEST_ASSERT_EQUAL_INT(0, ha_cb_persist_entry_for_test(&c->ha, 1, 7, blob, n));
|
||||
c->ha.raft.log_len = 1;
|
||||
TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 1));
|
||||
|
||||
/* Build the LIVE receive shape: a single entry in the `entries` list, each
|
||||
* sub-entry carrying its term + prev tuple (what wal_entry_to_wire now encodes
|
||||
* per element and xce_decode reproduces). repl_broadcast at term 7 of LSN 2
|
||||
* with prev=(1,7). */
|
||||
uint8_t lpay[2] = { 'b', 'c' };
|
||||
wal_catchup_entry_t sub;
|
||||
memset(&sub, 0, sizeof(sub));
|
||||
sub.lsn = 2;
|
||||
sub.entry_type = WAL_JOB_SUBMIT;
|
||||
sub.term = 7; /* the LEADER's term on the wire */
|
||||
sub.prev_log_index = 1;
|
||||
sub.prev_log_term = 7;
|
||||
sub.raw_payload = lpay;
|
||||
sub.raw_payload_len = sizeof(lpay);
|
||||
|
||||
wal_catchup_entry_t* sub_arr[1] = { &sub };
|
||||
wal_catchup_entry_t batch;
|
||||
memset(&batch, 0, sizeof(batch));
|
||||
batch.n_entries = 1;
|
||||
batch.entries = sub_arr;
|
||||
/* Top-level container also carries the AE metadata (the encoder mirrors it). */
|
||||
batch.term = 7;
|
||||
batch.prev_log_index = 1;
|
||||
batch.prev_log_term = 7;
|
||||
|
||||
int rc = ha_apply_wal_catchup_entry(&c->ha, &c->persist, &batch,
|
||||
/*ep*/NULL, c);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
/* The load-bearing assertion (bug-990): LSN 2 persists the LEADER's term 7,
|
||||
* NOT 0 and NOT the local term 3. Before the cutover this read back as the
|
||||
* standby's local term via the term-0 direct-apply path. */
|
||||
TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 2));
|
||||
/* The agreed prefix (LSN 1@term7) is untouched. */
|
||||
TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 1));
|
||||
/* The library adopted the leader's term on the RPC (G5). */
|
||||
TEST_ASSERT_EQUAL_UINT64(7, c->ha.raft.term);
|
||||
|
||||
teardown_cluster(c);
|
||||
printf("test_live_broadcast_receive_persists_leader_term: PASS\n");
|
||||
}
|
||||
|
||||
/* ================================================================== */
|
||||
/* Task 3.3 — drive raft_on_append_entries_reply + raft_advance_commit */
|
||||
/* on the replication ACK path */
|
||||
|
|
@ -1800,6 +1877,7 @@ void test_term_safety_adapter(void)
|
|||
RUN_TEST(test_intake_higher_term_heartbeat_adopts_term);
|
||||
RUN_TEST(test_intake_divergent_tail_truncated_and_reconverged);
|
||||
RUN_TEST(test_intake_logmismatch_rejected_no_false_convergence);
|
||||
RUN_TEST(test_live_broadcast_receive_persists_leader_term);
|
||||
RUN_TEST(test_repl_ack_currentterm_quorum_advances_commit);
|
||||
RUN_TEST(test_repl_ack_priorterm_quorum_holds_commit);
|
||||
RUN_TEST(test_repl_ack_staleterm_reply_dropped);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue