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

feat(ha-term-safety): on_send_append_entries callback body (task 2.3)

Browse source 
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:
Rongsong Shen 2026-06-22 06:01:30 +00:00
parent ed2443533f
commit 2c5d298cb5
6 changed files with 452 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

164
src/controller/ha.c
View file

@ -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() /