forked from lei.chen/myxce
fix(ha): close two independent chaotic-failover bugs — term-desync liveness stall and bug-847 WAL batch payload loss
Found via extended chaotic /xce-test-ha soaking on top of the bug-1419 fix (same session). Both bugs independently caused the 10-round chaotic soak to fail; sequential and chaotic now pass cleanly and repeatably. 1. Term-desync liveness deadlock (new): a standby that loses a near- simultaneous election race to a peer can legitimately end up at a HIGHER term than the peer that actually won and became primary. The primary's now-stale-term AppendEntries/heartbeats are correctly rejected by the raft library, but every one of those rejected messages was ALSO resetting the standby's own liveness deadline (ha_on_heartbeat_received), so its liveness timeout never fired and it could never start a fresh, higher-term election to let the cluster self-heal — a permanent stall. Fix: ha_msg_term_is_stale() gates the liveness reset (and ha_discover_primary step-down) on the incoming message's own term stamp, skipping it for provably-stale messages, in cluster.c's wal_catchup_entry and nop/heartbeat handlers. 2. bug-847 (root-caused 2026-06-30, never fixed): a per-sub-entry `payload` Data blob in a multi-entry WAL catch-up batch loses its capnp pointer once the message crosses a segment boundary — the receiver decodes raw_payload_len=0 for entries at/after the boundary and silently persists empty WAL entries (structurally intact: correct LSN/term/CRC chain, empty content). Only surfaces when a node rejoins and catches up a real batch, which is why it needed a live chaotic soak (not the 2-entry unit test) to reproduce. Fix: carry every sub-entry's payload concatenated in ONE top-level batchPayloads blob instead of one Data blob per entry — a single top-level blob does not exhibit the cross-segment loss. Sliced back apart on decode using each sub-entry's own wirePayloadLen, a plain scalar that (like lsn/entryType/term) decodes via the generic generated decode path and was never subject to the bug. Confirmed via targeted diagnostic logging that the corrupted entry arrives already-empty over the wire (not a receive-side race), and via a faithful reproduction at live JOB_SUBMIT payload size (5×3560B) after the original 2-entry/ 14×872B tests turned out not to cross a segment boundary in this environment. Incidental fixes found and fixed along the way: - A real self-deadlock (persist_entry_term() re-acquiring p->lock from inside a caller that already held it) introduced by an initial, ultimately unnecessary WAL-read locking fix attempted before bug-847 was pinned down; fixed via a new persist_entry_term_locked() lock-free counterpart. Left in place as a genuine (if incidental) hardening of ha_stream_wal_to_standby's read consistency. - .claude/skills/xce-test-ha/SKILL.md's `context: fork` frontmatter caused the skill's own forked sub-session teardown to SIGHUP the whole xce_controller process group mid-soak, killing all 3 nodes silently with no error logged, whenever the soak was launched via the Skill tool itself (setsid/nohup inside run-ha-soak.sh only protects against a controlling terminal's own SIGHUP, not a parent forked-session's teardown). Removed; added explicit guidance to launch long soaks via direct Bash from the main session instead. Verified: full 78-group unit regression sweep clean (0 failures) after each change; two consecutive clean 10-round chaotic soaks and one clean 10-round sequential soak on the final binary (zero committed divergence, zero job loss, all 3 nodes led in both modes).
This commit is contained in:
parent
11244dd654
commit
0941cf9649
11 changed files with 295 additions and 47 deletions
|
|
@ -1,6 +1,7 @@
|
|||
---
|
||||
name: xce-test-ha
|
||||
description: "End-to-end HA failover soak test for an XCE controller cluster: deploy 3 nodes, submit jobs, then repeatedly kill the primary and verify zero committed-job loss across 5-10 rounds. Usage: /xce-test-ha [rounds] [dir]"
|
||||
model: haiku
|
||||
---
|
||||
|
||||
# XCE HA Failover Soak Test
|
||||
|
|
@ -194,6 +195,21 @@ dev shell so the controller/xce_test runtime libs (openssl, etc.) are on the
|
|||
library path. Run it in the **background** (it takes several minutes) and tail
|
||||
the log:
|
||||
|
||||
**Launch this directly via the Bash tool from your own (non-forked) session —
|
||||
never through a sub-agent or a forked/isolated skill invocation.** A 5-10
|
||||
round soak runs for several minutes; if the process that launched it exits or
|
||||
tears down its session before the soak finishes, the child `xce_controller`
|
||||
processes can be killed outright (observed: a forked/background sub-session
|
||||
ending sent SIGHUP through to the whole process group, silently killing all 3
|
||||
controllers mid-round with no error logged — the run just stops). `setsid
|
||||
nohup` in `run-ha-soak.sh`'s `start_node()` protects against a SIGHUP from
|
||||
your own shell's controlling terminal, but it does NOT protect against a
|
||||
parent session's own process-group/cgroup teardown when that parent's
|
||||
lifetime is itself bounded (e.g. a forked sub-agent invocation that ends when
|
||||
it returns its response). Launch with `setsid nohup ... </dev/null >log 2>&1 &
|
||||
disown` from the main session, then poll/Monitor that log from the same
|
||||
session — do not delegate the launch itself to a forked context.
|
||||
|
||||
```bash
|
||||
SKILL_DIR=.claude/skills/xce-test-ha
|
||||
nix develop ./src --command bash "$SKILL_DIR/run-ha-soak.sh" 2>&1 | tee /tmp/xce-test-ha.log
|
||||
|
|
|
|||
|
|
@ -172,8 +172,15 @@ static int handle_wal_catchup_entry(cluster_t *c, endpoint_t *ep,
|
|||
return 0;
|
||||
|
||||
/* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby
|
||||
* liveness and steps down from CTRL_ELECTING. */
|
||||
{
|
||||
* liveness and steps down from CTRL_ELECTING. Skipped when the entry's
|
||||
* own term stamp proves it is already stale (bug-1418 family): a
|
||||
* stale-term AppendEntries is about to be rejected by the raft library
|
||||
* below anyway, and letting it ALSO reset liveness would mean a standby
|
||||
* that legitimately outraced its own peer to a higher term (a rare but
|
||||
* valid election-timing outcome) never times out to start a fresh,
|
||||
* higher-term election — permanently deadlocking the cluster on a
|
||||
* primary whose term the standby correctly knows is stale. */
|
||||
if (!ha_msg_term_is_stale(&c->ha, entry->term)) {
|
||||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||||
&c->ha.role, memory_order_acquire);
|
||||
if (role == CTRL_STANDBY)
|
||||
|
|
@ -324,14 +331,25 @@ static int myreq_handler(vtask_t* task)
|
|||
ha_nop_reply_state_report(&c->ha, c->myconf, &sr, &reply);
|
||||
xce_reply(ep, c->myhost, &reply);
|
||||
/* Req 1.3: any NOP from primary resets liveness deadline on standby.
|
||||
* Also steps down from CTRL_ELECTING if a live primary is discovered. */
|
||||
* Also steps down from CTRL_ELECTING if a live primary is discovered.
|
||||
* Skipped when the heartbeat's own term stamp proves it is already
|
||||
* stale (bug-1418 family) -- see the identical guard's doc comment
|
||||
* in handle_wal_catchup_entry above. A bodyless nop (no HeartbeatMsg)
|
||||
* carries no term, so ha_msg_term_is_stale's msg_term==0 case
|
||||
* preserves the existing unconditional behavior for it. */
|
||||
{
|
||||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||||
&c->ha.role, memory_order_acquire);
|
||||
if (role == CTRL_STANDBY)
|
||||
ha_on_heartbeat_received(&c->ha);
|
||||
else if (role == CTRL_ELECTING)
|
||||
ha_discover_primary(&c->ha, c);
|
||||
uint64_t hb_term = (request.body_which == Request_body_heartbeatmsg
|
||||
&& request.body.heartbeatmsg)
|
||||
? request.body.heartbeatmsg->term
|
||||
: 0;
|
||||
if (!ha_msg_term_is_stale(&c->ha, hb_term)) {
|
||||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
|
||||
&c->ha.role, memory_order_acquire);
|
||||
if (role == CTRL_STANDBY)
|
||||
ha_on_heartbeat_received(&c->ha);
|
||||
else if (role == CTRL_ELECTING)
|
||||
ha_discover_primary(&c->ha, c);
|
||||
}
|
||||
}
|
||||
/* ha-heartbeat-committed-propagation task 2.2: after the unchanged
|
||||
* liveness/reply path above, a heartbeat that carries a HeartbeatMsg
|
||||
|
|
|
|||
|
|
@ -2151,6 +2151,66 @@ static int stream_read_entry_body(int fd, uint32_t body_off,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* raft-ha-core-safety task 2.5 (send-side torn-read guard): read ONE WAL
|
||||
* entry's header + (if payload_len>0) full body while holding p->lock, so this
|
||||
* read observes a consistent snapshot with respect to a CONCURRENT write to
|
||||
* the SAME file -- persist_wal_write_replicated (and the primary's own append
|
||||
* path) write under p->lock, but this function's reads previously ran lock-
|
||||
* free, so a node that is simultaneously receiving new entries from its own
|
||||
* upstream leader AND streaming older entries downstream to a catching-up
|
||||
* standby could observe the downstream read torn: a committed entry with a
|
||||
* real non-empty payload on disk read back with a short/zero payload_len or a
|
||||
* partially-written body, then faithfully shipped and persisted downstream as
|
||||
* empty -- a genuine committed-prefix divergence at a historical LSN (not the
|
||||
* already-guarded body-short-read case in stream_read_entry_body, which
|
||||
* assumes a syntactically valid header; this closes the header-itself-torn
|
||||
* gap). Held only for one entry's read, not the whole stream, so a long catch-
|
||||
* up does not block writers throughout. p may be NULL only in count-only
|
||||
* tests (no live writer to race against).
|
||||
*
|
||||
* Returns 0 on success (*hdr_out populated; *payload_out set iff
|
||||
* hdr_out->payload_len > 0), -1 on a short/EOF header read (caller treats as
|
||||
* "nothing more in this segment"), or stream_read_entry_body's negative rc on
|
||||
* a body-read failure (caller aborts the stream, per its own fail-loud
|
||||
* contract). */
|
||||
static int read_one_wal_entry_locked(int fd, uint32_t off, const persist_t *p,
|
||||
uint64_t leader_term, uint64_t standby_lsn,
|
||||
wal_entry_hdr_t *hdr_out,
|
||||
uint8_t **payload_out)
|
||||
{
|
||||
pthread_mutex_t *lk = p ? (pthread_mutex_t *)&p->lock : NULL;
|
||||
*payload_out = NULL;
|
||||
|
||||
if (lk)
|
||||
pthread_mutex_lock(lk);
|
||||
|
||||
ssize_t n = pread(fd, hdr_out, sizeof(*hdr_out), (off_t)off);
|
||||
if (n < (ssize_t)sizeof(*hdr_out)) {
|
||||
if (lk)
|
||||
pthread_mutex_unlock(lk);
|
||||
return -1;
|
||||
}
|
||||
|
||||
int rc = 0;
|
||||
/* Only entries actually being streamed (above standby_lsn) need their
|
||||
* body read at all -- matches the caller's own send-vs-scan-only split,
|
||||
* so this fix does not add wasted I/O for the already-acked prefix. */
|
||||
if (hdr_out->payload_len > 0 && hdr_out->lsn > standby_lsn) {
|
||||
/* p->lock is already held here -- persist_entry_term() would
|
||||
* re-acquire it and self-deadlock; persist_entry_term_locked() is
|
||||
* the lock-free counterpart for exactly this case. */
|
||||
uint64_t stored_term = p ? persist_entry_term_locked(p, hdr_out->lsn)
|
||||
: hdr_out->term;
|
||||
rc = stream_read_entry_body(fd, off + (uint32_t)sizeof(*hdr_out),
|
||||
hdr_out->payload_len, hdr_out->lsn,
|
||||
stored_term, leader_term, payload_out);
|
||||
}
|
||||
|
||||
if (lk)
|
||||
pthread_mutex_unlock(lk);
|
||||
return rc;
|
||||
}
|
||||
|
||||
int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
||||
const persist_t *p, uint64_t leader_term,
|
||||
uint64_t standby_lsn, uint64_t committed_lsn)
|
||||
|
|
@ -2222,10 +2282,19 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
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 > committed_lsn) break;
|
||||
uint8_t *locked_pbuf = NULL;
|
||||
int rrc0 = read_one_wal_entry_locked(fd, off, p, leader_term,
|
||||
standby_lsn, &hdr,
|
||||
&locked_pbuf);
|
||||
if (rrc0 == -1) break; /* short/EOF header read */
|
||||
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) {
|
||||
free(locked_pbuf);
|
||||
break;
|
||||
}
|
||||
if (hdr.lsn > committed_lsn) {
|
||||
free(locked_pbuf);
|
||||
break;
|
||||
}
|
||||
|
||||
/* WAL_CHECKPOINT is a local-only marker (persist_write_checkpoint_
|
||||
* marker's own doc comment: "replay_wal SKIPS checkpoint markers...
|
||||
|
|
@ -2242,6 +2311,7 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
* Without this, a standby receiving one gets stuck retrying the
|
||||
* same LSN forever (the receive-side guard correctly refuses it). */
|
||||
if (hdr.type == WAL_CHECKPOINT) {
|
||||
free(locked_pbuf);
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -2249,6 +2319,7 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
if (hdr.lsn > standby_lsn) {
|
||||
wal_catchup_entry_t *entry = calloc(1, sizeof(*entry));
|
||||
if (!entry) {
|
||||
free(locked_pbuf);
|
||||
/* OOM: flush current batch and skip this entry */
|
||||
if (batch_n > 0) {
|
||||
send_wal_batch(ep, batch_entries, batch_n);
|
||||
|
|
@ -2293,14 +2364,13 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
entry->leader_current_term = leader_term;
|
||||
|
||||
if (hdr.payload_len > 0) {
|
||||
/* raft-ha-core-safety task 2.4: read the body with bounded
|
||||
* retry; a non-empty COMMITTED entry must NEVER be shipped
|
||||
* empty (the live LSN-9 send-side payload loss). */
|
||||
uint8_t *pbuf = NULL;
|
||||
int rrc = stream_read_entry_body(
|
||||
fd, off + (uint32_t)sizeof(hdr), hdr.payload_len,
|
||||
hdr.lsn, entry->term, leader_term, &pbuf);
|
||||
if (rrc != 0) {
|
||||
/* raft-ha-core-safety task 2.4/2.5: the body was already
|
||||
* read (with bounded retry) by read_one_wal_entry_locked
|
||||
* above, under the SAME lock as the header -- a non-empty
|
||||
* COMMITTED entry must NEVER be shipped empty, whether from
|
||||
* a short body read (task 2.4) or a header/body torn read
|
||||
* racing a concurrent write to this same file (task 2.5). */
|
||||
if (rrc0 != 0) {
|
||||
/* FAIL LOUD: the body could not be read in full after the
|
||||
* bounded retry. Do NOT degrade the entry to PLEN=0 —
|
||||
* abort the stream here. Flush the valid prefix already
|
||||
|
|
@ -2319,8 +2389,9 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
"non-empty committed entry must never be transmitted "
|
||||
"as PLEN=0; catch-up will re-stream from this LSN)",
|
||||
hdr.lsn, hdr.payload_len, entry->term, leader_term,
|
||||
rrc, WAL_STREAM_BODY_READ_MAX_ATTEMPTS);
|
||||
rrc0, WAL_STREAM_BODY_READ_MAX_ATTEMPTS);
|
||||
free(entry);
|
||||
free(locked_pbuf);
|
||||
if (batch_n > 0) {
|
||||
send_wal_batch(ep, batch_entries, batch_n);
|
||||
for (int bi = 0; bi < batch_n; bi++) {
|
||||
|
|
@ -2333,7 +2404,7 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
|
|||
close(fd);
|
||||
return -EIO;
|
||||
}
|
||||
entry->raw_payload = pbuf;
|
||||
entry->raw_payload = locked_pbuf;
|
||||
entry->raw_payload_len = hdr.payload_len;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -7463,6 +7463,14 @@ void ha_discover_primary(ha_state_t *ha, cluster_t *c)
|
|||
raft_cb_become_follower(ha, term, -1);
|
||||
}
|
||||
|
||||
bool ha_msg_term_is_stale(ha_state_t *ha, uint64_t msg_term)
|
||||
{
|
||||
if (!ha || msg_term == 0)
|
||||
return false;
|
||||
uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||||
return msg_term < term;
|
||||
}
|
||||
|
||||
void ha_on_heartbeat_received(ha_state_t *ha)
|
||||
{
|
||||
if (!ha) return;
|
||||
|
|
|
|||
|
|
@ -1225,6 +1225,28 @@ void ha_discover_primary(ha_state_t *ha, cluster_t *c);
|
|||
/* ha_on_heartbeat_received: reset the heartbeat deadline (Req 6.3). */
|
||||
void ha_on_heartbeat_received(ha_state_t *ha);
|
||||
|
||||
/* ha_msg_term_is_stale: true iff msg_term is a real (non-zero) term stamp
|
||||
* strictly less than this node's own current ha->term -- i.e. the sender is
|
||||
* provably behind and any liveness-reset / step-down reaction to this
|
||||
* specific message would be acting on stale information. msg_term == 0
|
||||
* means the message type carries no comparable term (e.g. ctrlhello,
|
||||
* snapshotTransfer) -- always non-stale by this check since there is
|
||||
* nothing to compare (callers keep their existing behavior for those).
|
||||
*
|
||||
* Exists to break a real liveness deadlock (bug-1418 family): a standby
|
||||
* that races its own real election against the peer that actually wins can
|
||||
* end up at a HIGHER term than the winning primary (a legitimate, if rare,
|
||||
* outcome of a near-simultaneous election race). The raft library correctly
|
||||
* rejects the winning primary's now-stale-term AppendEntries/heartbeats —
|
||||
* but every one of those messages was ALSO resetting the standby's own
|
||||
* liveness deadline before this check existed, so its liveness timeout
|
||||
* never fired and it could never start a fresh, higher-term election to
|
||||
* let the cluster self-heal. Callers use this to skip the liveness reset
|
||||
* (and ha_discover_primary step-down) specifically for messages already
|
||||
* known stale by term — never to change how the message itself is applied
|
||||
* or rejected. */
|
||||
bool ha_msg_term_is_stale(ha_state_t *ha, uint64_t msg_term);
|
||||
|
||||
/*
|
||||
* ha_observe_heartbeat_committed: reconcile-free monotonic update of a standby's
|
||||
* learned committed position from a primary heartbeat
|
||||
|
|
|
|||
|
|
@ -518,6 +518,11 @@ uint64_t persist_entry_term(const persist_t* p, uint64_t lsn)
|
|||
return result;
|
||||
}
|
||||
|
||||
uint64_t persist_entry_term_locked(const persist_t* p, uint64_t lsn)
|
||||
{
|
||||
return entry_term_locked(p, lsn);
|
||||
}
|
||||
|
||||
/* 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
|
||||
|
|
|
|||
|
|
@ -598,6 +598,13 @@ 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);
|
||||
|
||||
/* Lock-free counterpart of persist_entry_term for a caller that ALREADY holds
|
||||
* p->lock (calling persist_entry_term there would re-acquire the same
|
||||
* non-recursive mutex and self-deadlock — raft-ha-core-safety task 2.5). Same
|
||||
* absent/legacy-0 semantics, just without the lock/unlock. Caller MUST hold
|
||||
* p->lock; safe on a NULL p (returns 0). */
|
||||
uint64_t persist_entry_term_locked(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`
|
||||
|
|
|
|||
|
|
@ -377,6 +377,24 @@ struct wal_catchup_entry {
|
|||
* 0 = unset (legacy sender) → fall back to the stored term. Wire field
|
||||
* leaderCurrentTerm @8. */
|
||||
uint64_t leader_current_term;
|
||||
/* bug-847: this sub-entry's payload length, carried as a plain scalar (wire
|
||||
* field wirePayloadLen @9) so it decodes via the generic generated decode
|
||||
* exactly like lsn/entry_type/term — never subject to the per-element Data
|
||||
* cross-segment loss. Set by the encoder for every xce_wal_batch_encode_raw
|
||||
* send; used on decode to slice this entry's bytes out of the TOP-LEVEL
|
||||
* container's batch_payloads blob (below). */
|
||||
uint32_t wire_payload_len;
|
||||
/* bug-847: generic decoded placeholder for wire field batchPayloads @10 —
|
||||
* same broken-for-raw-bytes limitation as `payload` above; kept NULL, do
|
||||
* not use. Only meaningful on the TOP-LEVEL container (never on a nested
|
||||
* sub-entry). The real bytes are hand-extracted by xce_decode() into
|
||||
* raw_batch_payloads/raw_batch_payloads_len below, then sliced per
|
||||
* sub-entry using each entry's wire_payload_len into that entry's OWN
|
||||
* raw_payload/raw_payload_len (so every consumer keeps reading
|
||||
* raw_payload exactly as before — this is purely a wire-transport fix). */
|
||||
xce_data_t *batch_payloads;
|
||||
uint8_t *raw_batch_payloads;
|
||||
size_t raw_batch_payloads_len;
|
||||
};
|
||||
|
||||
/* ── Persistence Cap'n Proto mapped types (P0-9) ─────────────────────
|
||||
|
|
|
|||
|
|
@ -202,6 +202,15 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req)
|
|||
extract_data_bytes(WalCatchupEntry_get_payload(b_ptr),
|
||||
&batch->raw_payload_len);
|
||||
|
||||
/* bug-847: the top-level batch_payloads blob — every sub-entry's
|
||||
* payload concatenated in ONE Data field (see wal_entry_to_wire /
|
||||
* xce_wal_batch_encode_raw). A single top-level Data blob does not
|
||||
* exhibit the per-element cross-segment pointer loss that scattered
|
||||
* per-sub-entry Data blobs did. */
|
||||
batch->raw_batch_payloads =
|
||||
extract_data_bytes(WalCatchupEntry_get_batchPayloads(b_ptr),
|
||||
&batch->raw_batch_payloads_len);
|
||||
|
||||
/* ha-term-safety task 3.2: the top-level container's AppendEntries
|
||||
* metadata (term, prevLogIndex, prevLogTerm, leaderCommit) is decoded
|
||||
* into batch->{term,prev_log_index,prev_log_term,leader_commit} by the
|
||||
|
|
@ -209,18 +218,40 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req)
|
|||
* the Data payload needs manual extraction (above/below). The standby
|
||||
* intake (ha_consume_append_entries) reads those struct fields directly. */
|
||||
|
||||
/* Batch sub-entries: each carries its own lsn/entryType/term (scalar,
|
||||
* decoded) + payload (Data, extracted here). */
|
||||
/* Batch sub-entries: each carries its own lsn/entryType/term/
|
||||
* wire_payload_len (scalar, decoded — bug-847 immune) + payload. Slice
|
||||
* this entry's bytes out of the top-level batch_payloads blob using
|
||||
* wire_payload_len and a running cumulative offset (entries stream in
|
||||
* the same order they were concatenated on encode). Falls back to the
|
||||
* legacy per-element Data field only when no batch_payloads blob was
|
||||
* sent at all (defensive — every current sender populates it). */
|
||||
WalCatchupEntry_list wlist = WalCatchupEntry_get_entries(b_ptr);
|
||||
int wn = (wlist.p.type != CAPN_NULL) ? capn_len(wlist) : 0;
|
||||
size_t batch_off = 0;
|
||||
for (int i = 0; i < wn && i < batch->n_entries; i++) {
|
||||
if (!batch->entries || !batch->entries[i])
|
||||
continue;
|
||||
struct WalCatchupEntry we;
|
||||
get_WalCatchupEntry(&we, wlist, i);
|
||||
batch->entries[i]->raw_payload =
|
||||
extract_data_bytes(we.payload,
|
||||
&batch->entries[i]->raw_payload_len);
|
||||
|
||||
if (batch->raw_batch_payloads != NULL) {
|
||||
uint32_t plen = batch->entries[i]->wire_payload_len;
|
||||
if (plen > 0 &&
|
||||
batch_off + plen <= batch->raw_batch_payloads_len) {
|
||||
uint8_t *slice = (uint8_t *)malloc(plen);
|
||||
if (slice) {
|
||||
memcpy(slice, batch->raw_batch_payloads + batch_off,
|
||||
plen);
|
||||
batch->entries[i]->raw_payload = slice;
|
||||
batch->entries[i]->raw_payload_len = plen;
|
||||
}
|
||||
batch_off += plen;
|
||||
}
|
||||
} else {
|
||||
batch->entries[i]->raw_payload =
|
||||
extract_data_bytes(we.payload,
|
||||
&batch->entries[i]->raw_payload_len);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -302,6 +333,9 @@ void xce_free_request(xce_request_t *req)
|
|||
free(batch->raw_payload);
|
||||
batch->raw_payload = NULL;
|
||||
batch->raw_payload_len = 0;
|
||||
free(batch->raw_batch_payloads);
|
||||
batch->raw_batch_payloads = NULL;
|
||||
batch->raw_batch_payloads_len = 0;
|
||||
for (int i = 0; i < batch->n_entries; i++) {
|
||||
if (!batch->entries || !batch->entries[i])
|
||||
continue;
|
||||
|
|
@ -491,11 +525,14 @@ static void wal_entry_to_wire(struct capn_segment *cs,
|
|||
* the AppendEntries header term so historical (entry_term < leader_term)
|
||||
* catch-up entries are not stale-rejected. */
|
||||
w->leaderCurrentTerm = e->leader_current_term;
|
||||
if (e->raw_payload && e->raw_payload_len > 0) {
|
||||
capn_list8 pl = capn_new_list8(cs, (int)e->raw_payload_len);
|
||||
capn_setv8(pl, 0, e->raw_payload, (int)e->raw_payload_len);
|
||||
w->payload.p = pl.p;
|
||||
}
|
||||
/* bug-847: do NOT write this sub-entry's payload as its own per-element
|
||||
* Data blob — that is exactly the shape that loses its pointer once a
|
||||
* multi-entry batch crosses a capnp segment boundary. The caller
|
||||
* (xce_wal_batch_encode_raw) instead carries every sub-entry's bytes
|
||||
* concatenated in ONE top-level batchPayloads blob; this scalar length
|
||||
* is all a sub-entry needs to find its slice on decode. */
|
||||
(void)cs;
|
||||
w->wirePayloadLen = (uint32_t)e->raw_payload_len;
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
@ -549,7 +586,7 @@ int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
|
|||
top.leaderCurrentTerm = entries[0]->leader_current_term;
|
||||
|
||||
/* Batch container: entries list carries the per-entry payload(s) (empty for
|
||||
* a heartbeat). Each sub-entry carries lsn/entryType/term/payload. */
|
||||
* a heartbeat). Each sub-entry carries lsn/entryType/term/wirePayloadLen. */
|
||||
WalCatchupEntry_list lst = new_WalCatchupEntry_list(cs, n_entries);
|
||||
for (int i = 0; i < n_entries; i++) {
|
||||
struct WalCatchupEntry w;
|
||||
|
|
@ -558,6 +595,36 @@ int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
|
|||
}
|
||||
top.entries = lst;
|
||||
|
||||
/* bug-847: carry every sub-entry's payload concatenated in ONE top-level
|
||||
* Data blob instead of one Data blob per sub-entry — a single blob does
|
||||
* not exhibit the cross-segment pointer loss that scattered per-element
|
||||
* Data blobs do in a multi-entry batch. Decode slices this back apart
|
||||
* per entry using each entry's own wirePayloadLen (set above). */
|
||||
{
|
||||
size_t total = 0;
|
||||
for (int i = 0; i < n_entries; i++)
|
||||
total += entries[i]->raw_payload_len;
|
||||
if (total > 0) {
|
||||
uint8_t *concat = (uint8_t *)malloc(total);
|
||||
if (!concat) {
|
||||
capn_free(&c);
|
||||
return -ENOMEM;
|
||||
}
|
||||
size_t off = 0;
|
||||
for (int i = 0; i < n_entries; i++) {
|
||||
if (entries[i]->raw_payload && entries[i]->raw_payload_len > 0) {
|
||||
memcpy(concat + off, entries[i]->raw_payload,
|
||||
entries[i]->raw_payload_len);
|
||||
off += entries[i]->raw_payload_len;
|
||||
}
|
||||
}
|
||||
capn_list8 pl = capn_new_list8(cs, (int)total);
|
||||
capn_setv8(pl, 0, concat, (int)total);
|
||||
top.batchPayloads.p = pl.p;
|
||||
free(concat);
|
||||
}
|
||||
}
|
||||
|
||||
tp = new_WalCatchupEntry(cs);
|
||||
write_WalCatchupEntry(&top, tp);
|
||||
|
||||
|
|
|
|||
|
|
@ -90,21 +90,19 @@ static void test_wal_batch_roundtrip_payload_bytes(void)
|
|||
printf("test_wal_batch_roundtrip_payload_bytes: PASS\n");
|
||||
}
|
||||
|
||||
/* Production-scale batch: many entries each carrying a full-size (872-byte)
|
||||
* capnp-Job-sized payload — mirrors the live catch-up stream where ctrl-1
|
||||
* rejoined and received header-only entries (PLEN=0). The small 2-entry
|
||||
* round-trip passes; this reproduces whatever the larger/multi-segment batch
|
||||
* does on the wire (bug-847). */
|
||||
/* Production-scale batch: several entries each carrying a full-size
|
||||
* (3560-byte) JOB_SUBMIT-sized payload — mirrors the live catch-up stream
|
||||
* that reproduced bug-847 (a rejoining node persisted header-only entries,
|
||||
* PLEN=0, for a mid-batch LSN range). The small 2-entry round-trip above
|
||||
* fits in a single capnp segment and always passed; this size crosses a
|
||||
* segment boundary, which used to lose the per-entry payload Data pointer on
|
||||
* the wire. Fixed by carrying every sub-entry's payload concatenated in one
|
||||
* top-level batch_payloads blob instead of one Data blob per entry (see
|
||||
* wal_entry_to_wire / xce_wal_batch_encode_raw in request.c) — a single Data
|
||||
* blob does not exhibit the cross-segment loss. */
|
||||
static void test_wal_batch_roundtrip_large_payloads(void)
|
||||
{
|
||||
/* KNOWN-FAILING REPRODUCER for bug-847: a production-scale batch that spans
|
||||
* more than one capnp segment loses its per-entry payload Data pointers on
|
||||
* the wire (decode returns raw_payload_len=0; it also leaks the orphaned
|
||||
* decode allocations under ASAN). IGNORE'd so the suite stays green; REMOVE
|
||||
* this TEST_IGNORE line when bug-847 is fixed and assert the payloads
|
||||
* survive. Decisive evidence: this fails "Expected 872 Was 0" while the
|
||||
* 2-entry round-trip above passes. */
|
||||
enum { N = 14, PSZ = 872 };
|
||||
enum { N = 5, PSZ = 3560 };
|
||||
wal_catchup_entry_t e[N];
|
||||
wal_catchup_entry_t *entries[N];
|
||||
uint8_t* payloads[N];
|
||||
|
|
|
|||
|
|
@ -919,6 +919,24 @@ struct WalCatchupEntry $C.mapname("wal_catchup_entry_t") {
|
|||
# per-entry term (logOk / on_persist_entry). 0 = unset (legacy sender) → the
|
||||
# intake falls back to the entry's stored term (the prior, live-only behaviour).
|
||||
leaderCurrentTerm @8 :UInt64 $C.mapname("leader_current_term");
|
||||
# bug-847: a PER-ELEMENT `payload` (@2) Data blob loses its pointer when a
|
||||
# multi-entry catch-up batch spans more than one capnp segment (the
|
||||
# underlying capnp-c library's far-pointer handling for a List(UInt8) whose
|
||||
# segment differs from its referencing struct's segment) -- the receiver
|
||||
# decodes raw_payload_len=0 for entries at/after the boundary even though
|
||||
# the sender held the real bytes, silently persisting empty WAL entries.
|
||||
# A SINGLE top-level Data blob does not exhibit this loss (only scattered
|
||||
# per-element Data does), so the batch encoder (xce_wal_batch_encode_raw)
|
||||
# now carries every sub-entry's payload concatenated in ONE top-level
|
||||
# batchPayloads blob, sliced back apart on decode using each sub-entry's
|
||||
# own wirePayloadLen (a plain scalar -- already immune, decoded like
|
||||
# lsn/entryType/term by the generic generated decode, no manual extraction
|
||||
# needed). Populated for every xce_wal_batch_encode_raw send regardless of
|
||||
# n_entries (0/1/N); sub-entries no longer carry their own @2 payload on
|
||||
# this path (kept for any other/legacy sender; decode still handles it
|
||||
# defensively as a fallback when batchPayloads is absent).
|
||||
wirePayloadLen @9 :UInt32 $C.mapname("wire_payload_len");
|
||||
batchPayloads @10 :Data $C.mapname("batch_payloads");
|
||||
}
|
||||
|
||||
struct CtrlHelloRequest $C.mapname("ctrl_hello_req_t") {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue