diff --git a/.claude/skills/xce-test-ha/SKILL.md b/.claude/skills/xce-test-ha/SKILL.md index e2b0e8e..09e6675 100644 --- a/.claude/skills/xce-test-ha/SKILL.md +++ b/.claude/skills/xce-test-ha/SKILL.md @@ -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 ... 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 diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 7ccf7a0..59a9402 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -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 diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 72976c1..31f4811 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -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; } diff --git a/src/controller/ha.c b/src/controller/ha.c index f3e8413..7c339e2 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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; diff --git a/src/controller/ha.h b/src/controller/ha.h index 8c3e5c7..08bd75b 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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 diff --git a/src/controller/persist.c b/src/controller/persist.c index 1b40888..2bb6d86 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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 diff --git a/src/include/persist.h b/src/include/persist.h index 14d02e7..41a17e6 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -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` diff --git a/src/include/typemap.h b/src/include/typemap.h index a211a4b..b939986 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -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) ───────────────────── diff --git a/src/lib/request.c b/src/lib/request.c index f41e6b8..0ba2976 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -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); diff --git a/src/lib/testsuite/test_wal_wire.c b/src/lib/testsuite/test_wal_wire.c index 718697b..f2288dd 100644 --- a/src/lib/testsuite/test_wal_wire.c +++ b/src/lib/testsuite/test_wal_wire.c @@ -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]; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 4254df2..beeba1e 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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") {