feat(ha-term-safety): on_send_append_entries callback body (task 2.3)
Replaces the stub: builds a WalCatchupEntry AppendEntries to standbys[peer_idx].ep carrying prev_log_index/prev_log_term; for n_entries==1 sources the entry at prev_log_index+1 from the local WAL (term via persist_entry_term, type+payload via ha_read_wal_entry_at); n_entries==0 emits a zero-entry heartbeat. Async send (xce_wal_batch_req_raw), payload freed on all paths. 2 tests via a capture seam. Review-applied: corrected the misleading repl.c comment — leader_commit is accepted but NOT yet wire-encoded (deferred to task 3.2: WalCatchupEntry leaderCommit field + standby intake). Reviewer APPROVED; per-group 0 failures.
This commit is contained in:
parent
ed2443533f
commit
2c5d298cb5
6 changed files with 452 additions and 4 deletions
|
|
@ -44,7 +44,7 @@
|
|||
- _Requirements: 1.5, 1.8, 3.1, 3.2_
|
||||
- _Depends: 1.3_
|
||||
|
||||
- [ ] 2.3 Implement `on_send_append_entries` over the repl envelope
|
||||
- [x] 2.3 Implement `on_send_append_entries` over the repl envelope
|
||||
- Build and enqueue an AppendEntries on the repl transport carrying `prev_log_index`, `prev_log_term`, the entry's `term`, and `leader_commit = commit_index`; a heartbeat carries zero entries
|
||||
- Observable completion: emitting to a peer produces an envelope with the correct prev tuple and entry term; a 0-entry heartbeat is emitted when there is nothing to send
|
||||
- _Requirements: 1.3, 3.1_
|
||||
|
|
@ -155,3 +155,7 @@
|
|||
- **3.1/3.2 producer format:** must pack `raft_entry_blob_t {type,payload_len}+payload` to match the on_persist_entry unpack (single shared def in wal_types.h).
|
||||
|
||||
- **2.2 done.** on_get_entry_term → persist_entry_term (legacy-0 caveat documented per 1.3 review); on_truncate_from → persist_truncate_wal_above(p,index-1) (delete LSN>=index), returns rc, non-promotable fence preserved. NEW `truncate_term_index_locked` (persist.c) shrinks the in-memory term index on BOTH truncate return paths (under p->lock, cutoff-aligned) so on_get_entry_term returns 0 for truncated LSNs — no stale-term hazard. 3 tests. Reviewer APPROVED.
|
||||
|
||||
- **2.3 done.** on_send_append_entries → repl_send_append_entries: builds a WalCatchupEntry AE to standbys[peer_idx].ep carrying prev_log_index/prev_log_term; for n_entries==1 sources the entry at prev_log_index+1 from the local WAL (term via persist_entry_term, type+payload via ha_read_wal_entry_at); heartbeat = n_entries==0 (zero entries). Async (xce_wal_batch_req_raw), payload freed on all paths. 2 tests via repl_send_ae_capture_hook seam. Reviewer APPROVED.
|
||||
- **DEFERRED to task 3.2:** leader_commit is accepted+captured but NOT yet wire-encoded — WalCatchupEntry needs a `leaderCommit` capnp field + standby intake of it (3.2). Comment at repl.c corrected to state this.
|
||||
- **LOW (future refactor):** ha_read_wal_entry_at duplicates WAL segment-file parsing already in persist.c/config_repl.c/waldump.c — a persist_read_entry_at() accessor would consolidate; correct as written.
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
#include "ha.h"
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <dirent.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <inttypes.h>
|
||||
|
|
@ -1437,14 +1438,156 @@ static uint64_t ha_cb_get_entry_term_stub(void *ctx, uint64_t index)
|
|||
return persist_entry_term(&ha->raft_c->persist, index);
|
||||
}
|
||||
|
||||
/* ha_read_wal_entry_at: source ONE local WAL entry at LSN `lsn` for the
|
||||
* AppendEntries send path (ha-term-safety task 2.3). The library holds no entry
|
||||
* bytes, so the leader reads its own WAL segment files (seg-%016llx.wal under
|
||||
* wal_path) and returns the entry's type + payload. The per-LSN term is NOT read
|
||||
* from the on-disk header here — the caller takes it from persist_entry_term
|
||||
* (the authoritative in-memory term index), per the library contract that the
|
||||
* wire entry's term == the term recorded for that LSN. `payload` is malloc'd and
|
||||
* owned by the caller (freed after the send). Returns 0 on success, negative on
|
||||
* absence/read error. */
|
||||
static int ha_read_wal_entry_at(const char *wal_path, uint64_t lsn,
|
||||
uint8_t *type_out, uint8_t **payload_out,
|
||||
uint32_t *plen_out)
|
||||
{
|
||||
*type_out = 0;
|
||||
*payload_out = NULL;
|
||||
*plen_out = 0;
|
||||
|
||||
DIR *d = opendir(wal_path);
|
||||
if (!d)
|
||||
return -errno ? -errno : -EIO;
|
||||
|
||||
/* Enumerate segment start-LSNs, descending, so we open the latest segment
|
||||
* whose start LSN is <= the target first. */
|
||||
uint64_t seg_lsns[4096];
|
||||
int nseg = 0;
|
||||
struct dirent *de;
|
||||
while ((de = readdir(d)) != NULL &&
|
||||
nseg < (int)(sizeof(seg_lsns) / sizeof(*seg_lsns))) {
|
||||
uint64_t sl = 0;
|
||||
if (sscanf(de->d_name, "seg-%016" SCNx64 ".wal", &sl) == 1)
|
||||
seg_lsns[nseg++] = sl;
|
||||
}
|
||||
closedir(d);
|
||||
if (nseg == 0)
|
||||
return -ENOENT;
|
||||
|
||||
/* insertion sort ascending */
|
||||
for (int i = 1; i < nseg; i++) {
|
||||
uint64_t key = seg_lsns[i];
|
||||
int j = i - 1;
|
||||
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j + 1] = seg_lsns[j]; j--; }
|
||||
seg_lsns[j + 1] = key;
|
||||
}
|
||||
|
||||
int rc = -ENOENT;
|
||||
for (int si = 0; si < nseg; si++) {
|
||||
/* Skip a segment whose successor already starts past the target. */
|
||||
if (si + 1 < nseg && seg_lsns[si + 1] <= lsn)
|
||||
continue;
|
||||
if (seg_lsns[si] > lsn)
|
||||
break;
|
||||
|
||||
char seg_path[4096];
|
||||
snprintf(seg_path, sizeof(seg_path),
|
||||
"%s/seg-%016" PRIx64 ".wal", wal_path, seg_lsns[si]);
|
||||
int fd = open(seg_path, O_RDONLY);
|
||||
if (fd < 0)
|
||||
continue;
|
||||
|
||||
uint32_t off = 0;
|
||||
for (;;) {
|
||||
wal_entry_hdr_t hdr;
|
||||
ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off);
|
||||
if (n < (ssize_t)sizeof(hdr)) break;
|
||||
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
|
||||
|
||||
if (hdr.lsn == lsn) {
|
||||
*type_out = (uint8_t)hdr.type;
|
||||
if (hdr.payload_len > 0) {
|
||||
uint8_t *pbuf = malloc(hdr.payload_len);
|
||||
if (!pbuf) { rc = -ENOMEM; }
|
||||
else {
|
||||
ssize_t pr = pread(fd, pbuf, hdr.payload_len,
|
||||
(off_t)(off + sizeof(hdr)));
|
||||
if (pr == (ssize_t)hdr.payload_len) {
|
||||
*payload_out = pbuf;
|
||||
*plen_out = hdr.payload_len;
|
||||
rc = 0;
|
||||
} else {
|
||||
free(pbuf);
|
||||
rc = -EIO;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
rc = 0; /* zero-payload entry (e.g. WAL_NOOP) */
|
||||
}
|
||||
close(fd);
|
||||
return rc;
|
||||
}
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
}
|
||||
close(fd);
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
/* on_send_append_entries (ha-term-safety task 2.3, R1.3/R3.1): transmit ONE
|
||||
* AppendEntries to the standby at slot `peer_idx`, carrying the prev tuple
|
||||
* (prev_log_index, prev_log_term) + leader_commit, and — for n_entries==1 — the
|
||||
* single entry at LSN prev_log_index+1 sourced from this leader's OWN WAL (the
|
||||
* library holds no entry bytes). The entry's term is taken from
|
||||
* persist_entry_term (the authoritative per-LSN term index), so the standby
|
||||
* persists the LEADER's recorded term, not a substitute. n_entries==0 emits a
|
||||
* heartbeat (zero entries) carrying only the prev tuple + leader_commit.
|
||||
*
|
||||
* Non-blocking (ha-async-io): delegates to repl_send_append_entries which
|
||||
* enqueues via the async repl transport — never blocks in the callback.
|
||||
* peer_idx is the library's raft peer index, mapped 1:1 to the repl standby
|
||||
* registry slot (the same index match_index/next_index use). */
|
||||
static int ha_cb_send_append_entries_stub(void *ctx, int peer_idx,
|
||||
uint64_t prev_log_index,
|
||||
uint64_t prev_log_term,
|
||||
uint64_t leader_commit, int n_entries)
|
||||
{
|
||||
(void)ctx; (void)peer_idx; (void)prev_log_index; (void)prev_log_term;
|
||||
(void)leader_commit; (void)n_entries;
|
||||
return 0; /* TODO task 2.3 */
|
||||
ha_state_t *ha = (ha_state_t *)ctx;
|
||||
if (!ha || !ha->raft_c)
|
||||
return -EINVAL; /* test mode without a cluster: cannot send */
|
||||
|
||||
persist_t *p = &ha->raft_c->persist;
|
||||
replicator_t *r = ha->repl ? ha->repl : &p->repl;
|
||||
|
||||
if (n_entries == 0) {
|
||||
/* Heartbeat: prev tuple + leader_commit, no entry. */
|
||||
return repl_send_append_entries(r, peer_idx, prev_log_index,
|
||||
prev_log_term, leader_commit,
|
||||
/*n_entries*/0,
|
||||
/*entry_lsn*/0, /*type*/0, /*term*/0,
|
||||
NULL, 0);
|
||||
}
|
||||
|
||||
/* n_entries == 1: source the entry at prev_log_index+1 from the local WAL. */
|
||||
uint64_t entry_lsn = prev_log_index + 1;
|
||||
uint64_t entry_term = persist_entry_term(p, entry_lsn);
|
||||
|
||||
uint8_t type = 0;
|
||||
uint8_t *payload = NULL;
|
||||
uint32_t plen = 0;
|
||||
int rc = ha_read_wal_entry_at(p->wal_path, entry_lsn, &type, &payload, &plen);
|
||||
if (rc != 0) {
|
||||
WARN("on_send_append_entries: cannot source local WAL entry at lsn=%lu "
|
||||
"(rc=%d) — not sending to peer %d",
|
||||
(unsigned long)entry_lsn, rc, peer_idx);
|
||||
return rc;
|
||||
}
|
||||
|
||||
rc = repl_send_append_entries(r, peer_idx, prev_log_index, prev_log_term,
|
||||
leader_commit, /*n_entries*/1,
|
||||
entry_lsn, type, entry_term, payload, plen);
|
||||
free(payload);
|
||||
return rc;
|
||||
}
|
||||
|
||||
/* on_truncate_from (ha-term-safety task 2.2, R1.8/R3.2): delete the local log
|
||||
|
|
@ -1680,6 +1823,21 @@ int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index)
|
|||
{
|
||||
return s_real_cbs.on_truncate_from(ha, index);
|
||||
}
|
||||
|
||||
/* ha-term-safety task 2.3: drive the WIRED on_send_append_entries callback so a
|
||||
* unit test exercises the real AppendEntries send body (peer→ep map, prev tuple
|
||||
* + leader_commit carry, local-WAL entry sourcing). Pair with
|
||||
* repl_send_ae_capture_hook to assert envelope content. Exposed only under
|
||||
* XCE_TESTING. */
|
||||
int ha_cb_send_append_entries_for_test(ha_state_t *ha, int peer_idx,
|
||||
uint64_t prev_log_index,
|
||||
uint64_t prev_log_term,
|
||||
uint64_t leader_commit, int n_entries)
|
||||
{
|
||||
return s_real_cbs.on_send_append_entries(ha, peer_idx, prev_log_index,
|
||||
prev_log_term, leader_commit,
|
||||
n_entries);
|
||||
}
|
||||
#endif
|
||||
|
||||
/* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() /
|
||||
|
|
|
|||
|
|
@ -789,6 +789,20 @@ uint64_t ha_cb_get_entry_term_for_test(ha_state_t *ha, uint64_t index);
|
|||
* member stays non-promotable). TEST-ONLY.
|
||||
*/
|
||||
int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index);
|
||||
|
||||
/*
|
||||
* ha_cb_send_append_entries_for_test (ha-term-safety task 2.3): invoke the WIRED
|
||||
* on_send_append_entries callback directly so a unit test exercises the real
|
||||
* AppendEntries send body — peer→endpoint mapping, prev-tuple + leader_commit
|
||||
* carry, and (for n_entries==1) sourcing the single entry at prev_log_index+1
|
||||
* from the leader's own WAL. Combine with repl_send_ae_capture_hook to assert
|
||||
* the built envelope content without network I/O. Returns the callback rc.
|
||||
* TEST-ONLY.
|
||||
*/
|
||||
int ha_cb_send_append_entries_for_test(ha_state_t *ha, int peer_idx,
|
||||
uint64_t prev_log_index,
|
||||
uint64_t prev_log_term,
|
||||
uint64_t leader_commit, int n_entries);
|
||||
#endif
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -310,6 +310,102 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* AppendEntries send (ha-term-safety task 2.3)
|
||||
*
|
||||
* The library drives on_send_append_entries(peer_idx, prev tuple,
|
||||
* leader_commit, n_entries); the ha.c adapter sources the single entry (for
|
||||
* n_entries==1) from its own WAL and calls this to build + enqueue the
|
||||
* AppendEntries envelope to the standby at slot peer_idx. Non-blocking: reuses
|
||||
* the async xce_wal_batch_req_raw transport.
|
||||
* ============================================================ */
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
repl_ae_capture_t *repl_send_ae_capture_hook = NULL;
|
||||
#endif
|
||||
|
||||
int repl_send_append_entries(replicator_t *r, int peer_idx,
|
||||
uint64_t prev_log_index, uint64_t prev_log_term,
|
||||
uint64_t leader_commit, int n_entries,
|
||||
uint64_t entry_lsn, uint8_t entry_type,
|
||||
uint64_t entry_term,
|
||||
const void *payload, uint32_t payload_len)
|
||||
{
|
||||
if (r == NULL || peer_idx < 0 || peer_idx >= MAX_STANDBYS)
|
||||
return -EINVAL;
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
/* Capture seam: record the built envelope without network I/O. */
|
||||
if (repl_send_ae_capture_hook) {
|
||||
repl_ae_capture_t *cap = repl_send_ae_capture_hook;
|
||||
cap->called = 1;
|
||||
cap->peer_idx = peer_idx;
|
||||
cap->prev_log_index = prev_log_index;
|
||||
cap->prev_log_term = prev_log_term;
|
||||
cap->leader_commit = leader_commit;
|
||||
cap->n_entries = n_entries;
|
||||
cap->entry_lsn = (n_entries == 1) ? entry_lsn : 0;
|
||||
cap->entry_type = (n_entries == 1) ? entry_type : 0;
|
||||
cap->entry_term = (n_entries == 1) ? entry_term : 0;
|
||||
if (n_entries == 1 && payload && payload_len > 0) {
|
||||
uint32_t cpy = payload_len;
|
||||
if (cpy > sizeof(cap->payload))
|
||||
cpy = sizeof(cap->payload);
|
||||
memcpy(cap->payload, payload, cpy);
|
||||
cap->payload_len = cpy;
|
||||
} else {
|
||||
cap->payload_len = 0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
#endif
|
||||
|
||||
mylock_lock(&r->lock);
|
||||
standby_conn_t *s = &r->standbys[peer_idx];
|
||||
repl_ep_t ep = s->ep;
|
||||
char host[XCE_MAXNAME_LEN];
|
||||
uint32_t port = s->port;
|
||||
strncpy(host, s->host, sizeof(host) - 1);
|
||||
host[sizeof(host) - 1] = '\0';
|
||||
mylock_unlock(&r->lock);
|
||||
|
||||
if (ep == NULL)
|
||||
return -ENOENT; /* no live endpoint at this peer slot */
|
||||
|
||||
/* Build a single AppendEntries envelope carrying the prev tuple +
|
||||
* leader_commit and, for n_entries==1, the entry sourced from the leader's
|
||||
* WAL. The carrier struct already holds the Raft term fields (task 1.1). */
|
||||
wal_catchup_entry_t entry;
|
||||
memset(&entry, 0, sizeof(entry));
|
||||
entry.prev_log_index = prev_log_index;
|
||||
entry.prev_log_term = prev_log_term;
|
||||
/* NOTE: leader_commit is accepted as a parameter (and captured by the test
|
||||
* seam) but is NOT yet wire-encoded — WalCatchupEntry has no leaderCommit
|
||||
* field. Adding that capnp field and the standby-side intake of leader_commit
|
||||
* is deferred to task 3.2; no standby acts on it until then.
|
||||
* A heartbeat is marked by n_entries==0 (zero entries; lsn left 0). */
|
||||
if (n_entries == 1) {
|
||||
entry.lsn = entry_lsn;
|
||||
entry.entry_type = entry_type;
|
||||
entry.term = entry_term;
|
||||
entry.raw_payload = (uint8_t *)(uintptr_t)payload;
|
||||
entry.raw_payload_len = payload_len;
|
||||
}
|
||||
|
||||
wal_catchup_entry_t *ep_arr[1] = { &entry };
|
||||
int rc = xce_wal_batch_req_raw(ep, ep_arr, n_entries == 1 ? 1 : 0);
|
||||
if (rc != 0) {
|
||||
WARN("repl_send_append_entries: send to standby %s:%u failed (rc=%d)",
|
||||
host, port, rc);
|
||||
return rc;
|
||||
}
|
||||
TRACE("repl_send_append_entries: peer=%d prevIdx=%lu prevTerm=%lu "
|
||||
"leaderCommit=%lu n=%d to %s:%u",
|
||||
peer_idx, (unsigned long)prev_log_index, (unsigned long)prev_log_term,
|
||||
(unsigned long)leader_commit, n_entries, host, port);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* Sync replication — stub in P2-1; wired in P2-3
|
||||
* ============================================================ */
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@
|
|||
#include "config.h"
|
||||
#include "persist.h"
|
||||
#include "wal_types.h"
|
||||
#include "repl.h"
|
||||
#include "xce/cluster_def.h"
|
||||
#include "ha.h"
|
||||
|
||||
|
|
@ -494,6 +495,132 @@ static void test_on_truncate_from_removes_all(void)
|
|||
printf("test_on_truncate_from_removes_all: PASS\n");
|
||||
}
|
||||
|
||||
/* ================================================================== */
|
||||
/* Task 2.3 — on_send_append_entries callback body */
|
||||
/* ================================================================== */
|
||||
|
||||
/* A dummy non-NULL endpoint pointer for a registered standby slot. The send
|
||||
* is diverted to the capture hook, so the pointer is never dereferenced. */
|
||||
static int g_ae_dummy_ep;
|
||||
|
||||
/* Register one standby in the cluster's embedded replicator at slot 0, wiring
|
||||
* ha.repl so the callback maps peer_idx 0 → this endpoint. */
|
||||
static void register_one_standby(cluster_t *c)
|
||||
{
|
||||
c->ha.repl = &c->persist.repl;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
repl_add_standby(&c->persist.repl, &g_ae_dummy_ep, "sb0", 7000));
|
||||
}
|
||||
|
||||
/* Test: on_send_append_entries for n_entries==1 builds an envelope carrying the
|
||||
* correct (prev_log_index, prev_log_term, leader_commit, n_entries) and, for the
|
||||
* single entry sourced from the leader's own WAL at prev_log_index+1, its term
|
||||
* (== persist_entry_term(prev_log_index+1)) and payload (R1.3, R3.1). */
|
||||
static void test_send_append_entries_one_entry_from_wal(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_ts_ae1_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
cluster_t *c = make_cluster(tmp);
|
||||
TEST_ASSERT_NOT_NULL(c);
|
||||
atomic_store(&c->ha.role, CTRL_PRIMARY);
|
||||
|
||||
/* Populate the leader's own WAL BEFORE registering a standby — otherwise the
|
||||
* persist append fans out via repl_broadcast to the dummy int ep (ASAN
|
||||
* global-buffer-overflow reading ep->peer). With no standby, the append is a
|
||||
* no-op broadcast. The entry to be sent is at prev_log_index+1 == LSN 2. */
|
||||
uint8_t blob[64];
|
||||
size_t n1 = pack_entry(blob, WAL_JOB_UPDATE, "p", 1);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
ha_cb_persist_entry_for_test(&c->ha, /*index*/1, /*term*/4, blob, n1));
|
||||
size_t n2 = pack_entry(blob, WAL_JOB_UPDATE, "QQ", 2);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
ha_cb_persist_entry_for_test(&c->ha, /*index*/2, /*term*/6, blob, n2));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
|
||||
|
||||
/* Now register the standby so peer_idx 0 maps to a slot; the AE send below
|
||||
* is diverted to the capture hook, so the dummy ep is never dereferenced. */
|
||||
register_one_standby(c);
|
||||
|
||||
/* The entry-term oracle the send must carry for LSN 2. */
|
||||
uint64_t expect_term = persist_entry_term(&c->persist, 2);
|
||||
TEST_ASSERT_EQUAL_UINT64(6, expect_term);
|
||||
|
||||
/* Drive the callback: prev_log_index=1, prev_log_term=4 (term of LSN 1),
|
||||
* leader_commit=1, n_entries=1 → sends the entry at LSN 2. */
|
||||
repl_ae_capture_t cap;
|
||||
memset(&cap, 0, sizeof(cap));
|
||||
repl_send_ae_capture_hook = ∩
|
||||
|
||||
int rc = ha_cb_send_append_entries_for_test(&c->ha, /*peer*/0,
|
||||
/*prev_idx*/1, /*prev_term*/4,
|
||||
/*leader_commit*/1, /*n*/1);
|
||||
repl_send_ae_capture_hook = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(1, cap.called);
|
||||
TEST_ASSERT_EQUAL_INT(0, cap.peer_idx);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, cap.prev_log_index);
|
||||
TEST_ASSERT_EQUAL_UINT64(4, cap.prev_log_term);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, cap.leader_commit);
|
||||
TEST_ASSERT_EQUAL_INT(1, cap.n_entries);
|
||||
|
||||
/* The single entry is sourced from the local WAL at prev_log_index+1 == 2. */
|
||||
TEST_ASSERT_EQUAL_UINT64(2, cap.entry_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT64(expect_term, cap.entry_term);
|
||||
TEST_ASSERT_EQUAL_UINT8(WAL_JOB_UPDATE, cap.entry_type);
|
||||
TEST_ASSERT_EQUAL_UINT32(2, cap.payload_len);
|
||||
TEST_ASSERT_EQUAL_MEMORY("QQ", cap.payload, 2);
|
||||
|
||||
teardown_cluster(c);
|
||||
printf("test_send_append_entries_one_entry_from_wal: PASS\n");
|
||||
}
|
||||
|
||||
/* Test: on_send_append_entries for n_entries==0 emits a heartbeat carrying the
|
||||
* prev tuple + leader_commit and ZERO entries (R1.3). */
|
||||
static void test_send_append_entries_heartbeat_zero_entries(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_ts_ae0_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
cluster_t *c = make_cluster(tmp);
|
||||
TEST_ASSERT_NOT_NULL(c);
|
||||
atomic_store(&c->ha.role, CTRL_PRIMARY);
|
||||
|
||||
/* One entry on disk so prev_log_index=1 is a real tip; heartbeat sends none.
|
||||
* Persist BEFORE registering the standby (see one-entry test note). */
|
||||
uint8_t blob[64];
|
||||
size_t n1 = pack_entry(blob, WAL_JOB_UPDATE, "h", 1);
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
ha_cb_persist_entry_for_test(&c->ha, /*index*/1, /*term*/3, blob, n1));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
|
||||
|
||||
register_one_standby(c);
|
||||
|
||||
repl_ae_capture_t cap;
|
||||
memset(&cap, 0, sizeof(cap));
|
||||
repl_send_ae_capture_hook = ∩
|
||||
|
||||
int rc = ha_cb_send_append_entries_for_test(&c->ha, /*peer*/0,
|
||||
/*prev_idx*/1, /*prev_term*/3,
|
||||
/*leader_commit*/1, /*n*/0);
|
||||
repl_send_ae_capture_hook = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_INT(1, cap.called);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, cap.prev_log_index);
|
||||
TEST_ASSERT_EQUAL_UINT64(3, cap.prev_log_term);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, cap.leader_commit);
|
||||
TEST_ASSERT_EQUAL_INT(0, cap.n_entries);
|
||||
/* No entry sourced for a heartbeat. */
|
||||
TEST_ASSERT_EQUAL_UINT32(0, cap.payload_len);
|
||||
|
||||
teardown_cluster(c);
|
||||
printf("test_send_append_entries_heartbeat_zero_entries: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Suite entry point */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
|
@ -511,4 +638,6 @@ void test_term_safety_adapter(void)
|
|||
RUN_TEST(test_on_get_entry_term_present_legacy_absent);
|
||||
RUN_TEST(test_on_truncate_from_drops_suffix);
|
||||
RUN_TEST(test_on_truncate_from_removes_all);
|
||||
RUN_TEST(test_send_append_entries_one_entry_from_wal);
|
||||
RUN_TEST(test_send_append_entries_heartbeat_zero_entries);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -71,6 +71,28 @@ int repl_active_standby_count(replicator_t *r);
|
|||
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
|
||||
const void *payload, uint32_t payload_len);
|
||||
|
||||
/* AppendEntries send (ha-term-safety task 2.3, Req 1.3/3.1).
|
||||
*
|
||||
* Build and enqueue ONE AppendEntries envelope to the standby at slot
|
||||
* `peer_idx` (== the library's raft peer index, which maps 1:1 to the repl
|
||||
* standby registry slot — the same index used by match_index/next_index).
|
||||
* The envelope carries the AppendEntries prev tuple (`prev_log_index`,
|
||||
* `prev_log_term`) + `leader_commit`; when `n_entries == 1` it also carries the
|
||||
* single entry at LSN `prev_log_index + 1` whose `entry_term` is the term the
|
||||
* leader recorded for that LSN (== persist_entry_term) and whose
|
||||
* `entry_type`/`payload` are read from the leader's own WAL. `n_entries == 0`
|
||||
* is a heartbeat (zero entries) carrying only the prev tuple + leader_commit.
|
||||
*
|
||||
* Non-blocking: enqueues via the existing async repl transport
|
||||
* (xce_wal_batch_req_raw), never blocks. Returns 0 on enqueue, or a negative
|
||||
* errno (e.g. -ENOENT when peer_idx has no live endpoint). */
|
||||
int repl_send_append_entries(replicator_t *r, int peer_idx,
|
||||
uint64_t prev_log_index, uint64_t prev_log_term,
|
||||
uint64_t leader_commit, int n_entries,
|
||||
uint64_t entry_lsn, uint8_t entry_type,
|
||||
uint64_t entry_term,
|
||||
const void *payload, uint32_t payload_len);
|
||||
|
||||
/* WAL ACK handling — call from io_worker read callback ONLY */
|
||||
void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn);
|
||||
|
||||
|
|
@ -108,3 +130,28 @@ void repl_complete_catchup(replicator_t *r, repl_ep_t ep);
|
|||
/* Returns 1 if standby needs a full snapshot transfer (its LSN predates oldest WAL).
|
||||
* oldest_wal_lsn == 0 means no WAL exists — no snapshot needed for LSN 0 standbys. */
|
||||
int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn, uint64_t oldest_wal_lsn);
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
/* AppendEntries send capture seam (ha-term-safety task 2.3).
|
||||
*
|
||||
* When `repl_send_ae_capture_hook` is non-NULL, repl_send_append_entries
|
||||
* records the FULL built envelope into this struct instead of enqueuing it on a
|
||||
* real endpoint — so a unit test asserts the prev tuple / leader_commit /
|
||||
* n_entries / entry term + payload without network I/O. The payload bytes are
|
||||
* copied into a fixed buffer (capped at sizeof(payload)). */
|
||||
typedef struct {
|
||||
int called;
|
||||
int peer_idx;
|
||||
uint64_t prev_log_index;
|
||||
uint64_t prev_log_term;
|
||||
uint64_t leader_commit;
|
||||
int n_entries;
|
||||
uint64_t entry_lsn;
|
||||
uint8_t entry_type;
|
||||
uint64_t entry_term;
|
||||
uint32_t payload_len;
|
||||
uint8_t payload[256];
|
||||
} repl_ae_capture_t;
|
||||
|
||||
extern repl_ae_capture_t *repl_send_ae_capture_hook;
|
||||
#endif
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue