diff --git a/.kiro/specs/ha-term-safety/tasks.md b/.kiro/specs/ha-term-safety/tasks.md index e187c44c..8e39ad3c 100644 --- a/.kiro/specs/ha-term-safety/tasks.md +++ b/.kiro/specs/ha-term-safety/tasks.md @@ -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. diff --git a/src/controller/ha.c b/src/controller/ha.c index eceb5616..bd5dcd68 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2,6 +2,7 @@ #include "ha.h" #include +#include #include #include #include @@ -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() / diff --git a/src/controller/ha.h b/src/controller/ha.h index 7b7d45cd..5827da7c 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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 /* diff --git a/src/controller/repl.c b/src/controller/repl.c index 8e6e8a6a..16155c42 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -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 * ============================================================ */ diff --git a/src/controller/testsuite/test_term_safety_adapter.c b/src/controller/testsuite/test_term_safety_adapter.c index 2184f7ba..56b8eea0 100644 --- a/src/controller/testsuite/test_term_safety_adapter.c +++ b/src/controller/testsuite/test_term_safety_adapter.c @@ -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); } diff --git a/src/include/repl.h b/src/include/repl.h index f15c320f..e1f085bb 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -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