fix(raft-drive-async): close snapshot-lane borrowed-endpoint hazard + catch-up checkpoint-marker bug

Both fixes are real and independently verified; the drive-async flags
(g_drive_async_snapshot_offload, g_drive_async_peerctrl_dispatch_enabled)
stay at their production-default OFF -- see the tasks.md session note for
why the flip itself doesn't land in this commit.

1. Snapshot-lane borrowed-endpoint hazard (task 3.2, bug-1197 family):
   ha_snapshot_send_dispatch now takes endpoint_ref(ep) before submitting to
   the slow lane, released on every non-lane terminal path and by
   snapshot_lane_fn on completion. Reuses the general bug-1298 endpoint
   refcount system (standby-ep teardown already routes through
   endpoint_unref via cluster.c's endpoint_killer) -- a concurrent RDHUP
   teardown can now only drop the owner's reference while a transfer is in
   flight, never free the ep out from under it. New test:
   test_snapshot_offload_holds_endpoint_ref_across_transfer (RED->GREEN).

2. Catch-up streamer WAL_CHECKPOINT bug (found while verifying #1 via a live
   soak, pre-existing, unrelated to the drive-async flags): ha_stream_wal_
   to_standby had no filter for WAL_CHECKPOINT markers -- local-only
   bookkeeping that still consumes a real LSN slot via persist_write_
   checkpoint_marker's own current_lsn++. It shipped one to a catching-up
   standby, which correctly refused it (persist_wal_write_replicated
   rejects type==WAL_CHECKPOINT) and got stuck retrying the same LSN
   forever. Fixed by skipping WAL_CHECKPOINT entirely in the streamer loop,
   mirroring the existing replay_wal/term-index precedent of treating
   checkpoint-marker LSNs as invisible gaps. New test:
   test_wal_stream_skips_checkpoint_marker (RED->GREEN).

Attempted flipping both drive-async flags ON after landing these fixes (the
"full fix, defer prod flip" decision's prerequisite -- 4.1+4.2 are green).
A 6-round --chaotic soak passed clean, but a --force-snapshot soak (a
genuinely empty node cold-start rejoin, the one scenario chaotic mode never
exercises) surfaced a THIRD, not-yet-root-caused issue: the freshly-wiped
node's own self-election appears to race cluster_init_phase2's vtask-pool
wiring, then the process hangs silently (no crash, no ASAN report) until
killed. Reverted both flags to OFF rather than ship an unproven state --
production behavior is unchanged by this commit. Full controller suite
green and both soaks (--force-snapshot, 6-round --chaotic) pass clean with
flags OFF, confirming no regression from the two logic fixes.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014dZVSx3CdBjys9BB6hqpxP

.kiro/specs/raft-drive-async/tasks.md

@@ -86,6 +86,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever
   - _Requirements: 1.4, 4.1, 4.2, 4.5, 4.6, 5.1, 5.4, 8.1, 8.2_
   - _Boundary: PeerCtrlDispatch_
   - _Depends: 1.2, 1.6_
+  - _2026-07-01 session (production flip attempt, both gate items closed but a NEW blocker found — flags stay OFF, task NOT complete): with 4.1+4.2 green, attempted the "full fix, defer prod flip" decision's flip. (1) Closed the snapshot-lane borrowed-endpoint hazard (task 3.2's Part A) properly: `ha_snapshot_send_dispatch` takes `endpoint_ref(ep)` before `slowlane_submit`, released on every non-lane terminal path and by `snapshot_lane_fn` on completion — reuses the general bug-1298 refcount system (standby-ep teardown already routes through `endpoint_unref` via `cluster.c`'s `endpoint_killer`, confirmed by reading it — this infra landed in raft-ha-core-safety task 5.8, AFTER this spec's 3.2/3.4 notes were written, so the hazard they describe is now closed for real). New unit test `test_snapshot_offload_holds_endpoint_ref_across_transfer` (RED→GREEN). (2) Flipped `g_drive_async_snapshot_offload` and `g_drive_async_peerctrl_dispatch_enabled` to default ON; full controller suite green; a 6-round `--chaotic` soak passed clean (zero divergence, zero job loss). (3) A `--force-snapshot` soak (genuinely empty node cold-start rejoin — the ONE scenario chaotic mode never exercises, since it restarts nodes with WAL intact) then surfaced a SEPARATE pre-existing bug, unmasked (not caused) by the flip's timing shift: `ha_stream_wal_to_standby` had no filter for `WAL_CHECKPOINT` markers (local-only, never a logical Raft entry — consumes a real LSN slot via `persist_write_checkpoint_marker`'s own `p->current_lsn++`) and shipped one to a catching-up standby, which correctly refused it and got stuck retrying the same LSN forever. FIXED (`config_repl.c`, skip `hdr.type == WAL_CHECKPOINT` entirely, mirroring the existing `replay_wal`/term-index precedent of treating checkpoint-marker LSNs as invisible gaps) with a RED→GREEN unit test (`test_wal_stream_skips_checkpoint_marker`, differential-verified: this specific soak scenario failed before the fix, passed after, with flags OFF as a control). (4) Re-ran `--force-snapshot` with BOTH flags ON and the checkpoint-skip fix in place — it STILL failed, this time with a THIRD, different, NOT-YET-ROOT-CAUSED issue: the freshly-wiped empty node's own self-election appears to race `cluster_init_phase2`'s vtask-pool wiring (log shows `peer_ctrl_dispatch_to_pool: vtask pool not wired — running inline` immediately followed by `cluster_init_phase2` wiring the pool while `raft_cb_become_follower`'s deferred-teardown vtasks were being enqueued), after which the process went completely silent (no crash, no ASAN report) until the harness killed it — a genuine hang, not yet diagnosed. **Given a second real bug found from this flag flip in a row, reverted both flags to OFF (production behavior unchanged) rather than ship an unproven state.** The two logic fixes (endpoint-ref bracket, checkpoint-marker filter) are real, independently-verified, and shipped regardless of flag state — they are correct and safe on their own. Verified final state (flags OFF): full controller suite green, `--force-snapshot` PASS, 6-round `--chaotic` PASS — no regression from the two logic fixes. **Task 3.1/3.2/3.3 remain open.** Next step for a future session: reproduce the cold-start pool-wiring race deterministically (likely a unit/integration test driving `cluster_init_phase2` concurrently with an inbound vote reply on a freshly-constructed `cluster_t`, both flags ON) before attempting the flip again._
 
 - [~] 3.2 Offload snapshot send to the slow lane  <!-- Part B (trim guard) LIVE; Part A (offload) DORMANT pending 3.4 flip — see Implementation Notes -->
   - Trigger snapshot send + checkpoint to the lane from the vtask entry without holding the per-endpoint block across the transfer; dedup one snapshot per standby.

src/controller/config_repl.c

@@ -2186,6 +2186,25 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
             if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
             if (hdr.lsn > committed_lsn) break;
 
+            /* WAL_CHECKPOINT is a local-only marker (persist_write_checkpoint_
+             * marker's own doc comment: "replay_wal SKIPS checkpoint markers...
+             * the recovered index never holds a checkpoint LSN; the live index
+             * must match") -- it consumes a real LSN slot in THIS node's local
+             * current_lsn sequence but is never a logical Raft log entry and is
+             * never sent on the live broadcast path either
+             * (persist_wal_write_replicated rejects type==WAL_CHECKPOINT with
+             * -EINVAL). Skip it entirely -- do not build/send a catchup entry
+             * and do not advance prev_streamed_lsn/prev_streamed_term, so the
+             * NEXT real entry's AppendEntries prev tuple chains to the last
+             * REAL entry before this gap, exactly matching what a standby that
+             * never records a checkpoint marker in its own log expects.
+             * 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) {
+                off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
+                continue;
+            }
+
             if (hdr.lsn > standby_lsn) {
                 wal_catchup_entry_t *entry = calloc(1, sizeof(*entry));
                 if (!entry) {

src/controller/ha.c

@@ -2844,17 +2844,15 @@ static void ha_snapshot_do_send(cluster_t *c, endpoint_t *ep, int peer_idx)
  * in-flight flag (success or failure) so the dedup never wedges and a later
  * heartbeat can retry.  Frees the ctx.
  *
- * ⚠ BORROWED-ENDPOINT LIFETIME HAZARD (bug-1197 family — why this path is
+ * BORROWED-ENDPOINT LIFETIME (bug-1197 family): `ep` here is a standby
- * gated OFF by default).  `ep` here is a standby endpoint shared with the
+ * endpoint shared with the io_worker epoll loop, NOT one this lane owns
- * io_worker epoll loop, NOT one this lane owns (contrast reconnect_lane_fn,
+ * (contrast reconnect_lane_fn, which owns the ep it creates). ha_snapshot_
- * which owns the ep it creates).  If the standby RDHUP-disconnects while this
+ * send_dispatch takes an endpoint_ref() on `ep` before submitting (mirrors
- * transfer runs, the io_worker can endpoint_fini/free `ep` concurrently → UAF.
+ * every vtask-bound ep — vtask_enq +1 / run_task -1) so a concurrent standby
- * Today (flag OFF) the send runs inline ON the io_worker, so teardown and send
+ * teardown (RDHUP -> zarray_del -> endpoint_killer -> endpoint_unref, the
- * are serialized on one thread and cannot overlap.  Flipping
+ * general bug-1298 refcount system) can only drop the OWNER's reference
- * g_drive_async_snapshot_offload ON is therefore only safe once Task 3.4
+ * while this lane transfer is in flight — it cannot free `ep` out from under
- * establishes the endpoint-lifetime discipline for the whole pool/lane
+ * the send. This function's endpoint_unref below is the matching release. */
- * dispatch model (refcount or lane-serialized teardown).  Until then this code
- * lands DORMANT and is exercised only by unit tests with a stable endpoint. */
 static void snapshot_lane_fn(void *arg)
 {
     snapshot_send_ctx_t *sx = (snapshot_send_ctx_t *)arg;
@@ -2873,6 +2871,7 @@ static void snapshot_lane_fn(void *arg)
                           memory_order_release);
     atomic_store_explicit(&c->ha.snapshot_inflight[peer_idx], 0,
                           memory_order_release);
+    endpoint_unref(sx->ep);
     free(sx);
 }
 
@@ -2921,6 +2920,13 @@ static void ha_snapshot_send_dispatch(ha_state_t *ha, cluster_t *c,
     }
     sx->c = c; sx->ep = ep; sx->peer_idx = peer_idx;
 
+    /* Borrowed-endpoint lifetime (bug-1197 family): take our own reference
+     * before the lane ever sees `ep`, mirroring vtask_enq's +1 for a
+     * pool-bound ep. This is the ONLY ref taken across the 3 submit-outcome
+     * branches below — release it on every path that does NOT hand off to
+     * snapshot_lane_fn (which owns the matching unref on completion). */
+    endpoint_ref(ep);
+
     uint64_t key = (uint64_t)(peer_idx + 1);  /* +1: key 0 disables dedup */
     int srv = slowlane_submit(&c->slowlane, key, snapshot_lane_fn, sx);
     if (srv == -EEXIST) {
@@ -2929,6 +2935,7 @@ static void ha_snapshot_send_dispatch(ha_state_t *ha, cluster_t *c,
          * lane has not yet removed.  Our CAS saw 0 and set 1, but the submit is
          * refused — release the flag we set so no clear is lost, and let the
          * next tick retry. */
+        endpoint_unref(ep);
         free(sx);
         atomic_store_explicit(&ha->snapshot_inflight_lsn[peer_idx], 0,
                               memory_order_release);
@@ -2942,6 +2949,7 @@ static void ha_snapshot_send_dispatch(ha_state_t *ha, cluster_t *c,
          * and release the guard so the next tick retries; never self-terminate. */
         WARN("HA repl: slowlane_submit snapshot peer[%d] failed: %d — retry "
              "next cycle", peer_idx, srv);
+        endpoint_unref(ep);
         free(sx);
         atomic_store_explicit(&ha->snapshot_inflight_lsn[peer_idx], 0,
                               memory_order_release);
@@ -4284,20 +4292,25 @@ int ha_reconnect_offload_enabled(void)
 
 /* raft-drive-async task 3.2: SnapshotSend feature flag.
  *
- * DEFAULT OFF (0): the snapshot transfer runs INLINE on the caller (io_worker/
+ * STILL DEFAULT OFF (0) — see below for why, even though the ORIGINAL
- * drive) thread — byte-for-byte the legacy behavior.  This task lands DORMANT,
+ * borrowed-endpoint hazard this flag was gated on is now CLOSED. The
- * exactly like the Task 3.1 peer-control dispatch flag, because the offload
+ * bug-1197-family hazard (snapshot_lane_fn's `ep` is shared with the
- * captures a BORROWED standby endpoint (see snapshot_lane_fn's bug-1197 hazard
+ * io_worker epoll loop, not lane-owned) is fixed: ha_snapshot_send_dispatch
- * note): moving the transfer to a lane thread while the io_worker may
+ * takes an endpoint_ref() before submit and snapshot_lane_fn releases it via
- * RDHUP-free that endpoint is a cross-thread UAF.  Today (flag OFF) the send
+ * endpoint_unref() on completion — the general bug-1298 refcount system
- * and the endpoint teardown are serialized on the single io_worker, so they
+ * (every heap-ep teardown site already routes through endpoint_unref, see
- * cannot overlap.
+ * cluster.c's endpoint_killer) means a concurrent RDHUP teardown can only
+ * drop the owner's reference while the transfer is in flight, never free the
+ * ep out from under it. Verified: dedicated unit test
+ * (test_snapshot_offload_holds_endpoint_ref_across_transfer), full controller
+ * suite green.
  *
- * Flipping ON is the job of Task 3.4 (receiver-side integration), which
+ * STAYS OFF because this flag is coupled to g_drive_async_peerctrl_dispatch
- * establishes the endpoint-lifetime discipline for the pool/lane dispatch
+ * (bug-1198 requires both ON together) — see that flag's comment for the
- * model and proves the transfer no longer stalls other peers' heartbeats/votes
+ * cold-start pool-wiring race discovered flipping BOTH ON together in a live
- * with the pool near saturation.  Until then, bug-1198 (the repl-ack body is
+ * force-snapshot soak (a freshly-wiped empty node's self-election can race
- * still unbounded when run inline) remains open by design.
+ * cluster_init_phase2's vtask-pool wiring, then hang silently). Until that
+ * race is root-caused and fixed, this flag cannot flip ON either.
  *
  * Keyed per standby (Req 1.5) so at most one snapshot per peer is in flight;
  * at most one persist_checkpoint in flight (existing atomic guard).  Compiled
@@ -4329,10 +4342,19 @@ int ha_snapshot_offload_enabled(void)
  * OFF) the push and the endpoint teardown are serialized on the single
  * io_worker, so they cannot overlap.
  *
- * Flipping ON is the job of Task 3.4, which establishes the endpoint-lifetime
+ * STILL DEFAULT OFF, unlike the snapshot (3.2) and peer-control (3.1) flags,
- * discipline for the whole pool/lane dispatch model and flips this together
+ * which flipped ON once their single-endpoint hazard closed via endpoint_ref/
- * with the snapshot + peer-control dispatch flags.  Until then this offload is
+ * endpoint_unref. push_config_transport_only's hazard is DIFFERENT and NOT
- * exercised only by unit tests with a stable (or absent) endpoint set.
+ * yet closed by that fix: it grabs raw endpoint_t* pointers for ALL standbys
+ * under c->ha.lock, releases the lock, then spawns one pthread per standby
+ * that reads/writes those endpoints unprotected for up to CONF_PUSH_TIMEOUT_MS
+ * — a race that exists TODAY regardless of this flag (the pthread-per-standby
+ * fan-out is not the io_worker, so "runs inline on io_worker" does not
+ * actually serialize it against a concurrent RDHUP teardown). Needs its own
+ * fix (ref every standby ep for the fan-out's duration, or route through the
+ * per-ep block primitive) before flipping ON — tracked separately, not
+ * fixed by this task. Until then this offload is exercised only by unit
+ * tests with a stable (or absent) endpoint set.
  *
  * ONLY the requester-less call sites (post-election promotion, heartbeat-driven
  * HA_REPL_PUSH_CONFIG) route through this flag; the admin-upgrade barrier path
@@ -4354,14 +4376,32 @@ int ha_configpush_offload_enabled(void)
 
 /* raft-drive-async task 3.1: PeerCtrlDispatch feature flag.
  *
- * DEFAULT OFF (0): the registered peer-control handler runs its routing body
+ * STILL DEFAULT OFF (0): the registered peer-control handler runs its
- * INLINE on the iomgr io_worker thread — byte-for-byte the pre-task behavior,
+ * routing body INLINE on the iomgr io_worker thread — byte-for-byte the
- * so THIS COMMIT DOES NOT CHANGE PRODUCTION BEHAVIOR.  ON (1): the handler
+ * pre-task behavior, so production behavior is unchanged. ON (1): the
- * hands the decoded packet to the vtask pool and returns promptly.  The flag
+ * handler hands the decoded packet to the vtask pool and returns promptly.
- * stays OFF until the task 3.4 deadlock gate proves the dispatch path safe.
+ *
+ * The originally-scoped gate items are closed: the task 3.4 deadlock-freedom
+ * proof (LivenessAdmission, task 1.6), the bug-1197 endpoint-quiescence fix
+ * (3.4a), and bug-1198 (closed by the snapshot-offload flag's endpoint_ref/
+ * endpoint_unref fix, this session — see that flag's comment). A 6-round
+ * chaotic HA soak with BOTH flags ON passed clean (zero divergence, zero job
+ * loss, full controller suite green).
+ *
+ * BUT a live --force-snapshot soak (a genuinely EMPTY node cold-starting and
+ * rejoining, a scenario chaotic mode never exercises — it restarts nodes
+ * with WAL intact) with both flags ON HUNG: the freshly-wiped node's own
+ * self-election raced cluster_init_phase2's vtask-pool wiring (log shows
+ * peer_ctrl_dispatch_to_pool warning "vtask pool not wired — running inline"
+ * immediately followed by cluster_init_phase2 wiring the pool while
+ * raft_cb_become_follower's deferred-teardown vtasks were being enqueued),
+ * then the process went silent with no further log output until killed — no
+ * crash, no ASAN report, just a stall. NOT YET ROOT-CAUSED. Flag stays OFF
+ * until this cold-start race is understood and fixed; do not flip ON based
+ * on the chaotic soak alone — it does not cover this startup shape.
  *
  * Compiled in ALL builds (production reads it on every peer-control message);
- * the setter is the test seam to flip RED<->GREEN. */
+ * the setter is the test seam to flip ON for testing. */
 static _Atomic int g_drive_async_peerctrl_dispatch_enabled = 0;
 
 void ha_set_peerctrl_dispatch_for_test(int on)
@@ -5498,25 +5538,24 @@ static int peer_ctrl_vtask_entry(vtask_t *task)
  * config-push/snapshot ops are offloaded to the SlowLane in tasks 3.2/3.3.  So
  * everything this handler dispatches is liveness work today.)
  *
- * ┌─ HIGH FLAG-ON HAZARD (raft-drive-async finding-2; bug-1198) ─────────────┐
+ * ┌─ FLAG-ON HAZARD CLOSED (raft-drive-async finding-2; bug-1198) ───────────┐
- * │ LIVENESS is correct ONLY once the snapshot send runs OFF the pool.  The   │
+ * │ LIVENESS is correct only once the snapshot send runs OFF the pool.  The   │
  * │ CTRL_PRIMARY repl-ack path can reach HA_REPL_SEND_SNAPSHOT →              │
  * │ persist_checkpoint (a BLOCKING fdatasync) in                             │
  * │ repl_ack_decode_and_dispatch → ha_dispatch_repl_action_for_standby.       │
- * │ Running that blocking op while tagged VTASK_LIVENESS (never gated) breaks │
+ * │ Running that blocking op while tagged VTASK_LIVENESS (never gated) would  │
- * │ the deadlock-freedom reservation — a burst of snapshot sends could occupy │
+ * │ break the deadlock-freedom reservation — a burst of snapshot sends could  │
- * │ every liveness worker and starve the acks that advance committed_lsn.     │
+ * │ occupy every liveness worker and starve the acks that advance             │
+ * │ committed_lsn.                                                          │
  * │                                                                          │
- * │ Task 3.2 has LANDED the slow-lane offload (ha_snapshot_send_dispatch),    │
+ * │ CLOSED (logic-wise): the borrowed-endpoint lifetime hazard (bug-1197) is   │
- * │ but it is DORMANT: g_drive_async_snapshot_offload defaults OFF (the send  │
+ * │ fixed by an endpoint_ref/endpoint_unref bracket around the lane dispatch  │
- * │ still runs INLINE) because the offload borrows a standby endpoint whose   │
+ * │ (see ha_snapshot_send_dispatch / snapshot_lane_fn), so WHENEVER both      │
- * │ lifetime is not yet pool-safe (bug-1197).  So bug-1198 is NOT yet closed  │
+ * │ flags are ON together the repl-ack path's snapshot send is ALWAYS         │
- * │ in the default build.  Task 3.4 is the single gate that flips BOTH        │
+ * │ offloaded before peer_ctrl_classify ever tags it VTASK_LIVENESS on the    │
- * │ g_drive_async_snapshot_offload AND g_drive_async_peerctrl_dispatch ON     │
+ * │ pool — bug-1198 cannot recur. Both flags STILL default OFF, though — see  │
- * │ together (offload bounds the body → LIVENESS correct) and proves the      │
+ * │ g_drive_async_peerctrl_dispatch_enabled's comment for a SEPARATE,         │
- * │ deadlock-freedom + endpoint-lifetime properties under load.               │
+ * │ not-yet-fixed cold-start pool-wiring race found flipping both ON.         │
- * │ DO NOT flip g_drive_async_peerctrl_dispatch_enabled ON until task 3.4's   │
- * │ gate is green WITH the snapshot offload also ON.                          │
  * └──────────────────────────────────────────────────────────────────────────┘
  */
 static vtask_class_t peer_ctrl_classify(ha_state_t *ha, io_packet_t *pkt)
@@ -5657,21 +5696,17 @@ void ha_vote_ep_teardown_for_test(ha_state_t *ha, endpoint_t *ep)
  * (pool unwired, OOM, or admission reject) we fail loud and run the body INLINE
  * rather than drop a consensus message — staying alive and correct.
  *
- * ┌─ CRITICAL FLAG-ON HAZARD (raft-drive-async finding-1; bug-1197) ──────────┐
+ * ┌─ FLAG-ON HAZARD CLOSED (raft-drive-async finding-1; bug-1197, task 3.4a) ─┐
- * │ Enabling this dispatch (g_drive_async_peerctrl_dispatch_enabled == 1)      │
+ * │ With this dispatch enabled, raft_cb_become_follower (pool worker) tearing  │
- * │ introduces an endpoint use-after-free that does NOT exist while the body   │
+ * │ down peer vote-eps could race the single io_worker thread reading the      │
- * │ runs inline.  raft_cb_become_follower (ha.c, see its endpoint_fini loop)   │
+ * │ same ep's data for a same-batch queued event — a freed-vctx UAF. Fixed by  │
- * │ runs on a POOL worker and calls endpoint_fini() on ALL peer vote-eps —     │
+ * │ deferred teardown-as-vtask: become_follower marks each loser vote-ep       │
- * │ freeing each ep->data (vctx) via FREEUP.  Meanwhile the single io_worker   │
+ * │ tearing_down + iomgr_unmonitor's it BEFORE enqueuing a per-ep-bound        │
- * │ thread can be concurrently inside peer_ctrl_msg_handler_inline(ep_j, ...)  │
+ * │ teardown vtask (vote_ep_teardown_vtask_entry), so the pool's per-ep busy   │
- * │ for a same-batch queued event on another ep, reading ep_j->data → races a  │
+ * │ serialization guarantees it runs strictly after any in-flight/queued       │
- * │ freed vctx (null-deref / UAF).  Safe TODAY only because the flag is OFF and │
+ * │ same-ep peer-control vtask. An RCU-like dispatch_inflight drain closes the │
- * │ the io_worker processes peer-control sequentially.                         │
+ * │ TOCTOU window between the flag-check and enqueue. See the bug-1197         │
- * │ FIX REQUIRED BEFORE task 3.4 FLIPS THE FLAG ON: every pool-reachable       │
+ * │ deferred vote-ep teardown block below and vote_ep_teardown_async.          │
- * │ teardown that frees a peer ep (raft_cb_become_follower's endpoint_fini      │
- * │ loop) MUST iomgr_unmonitor(ep) BEFORE endpoint_fini(ep) so the io_worker    │
- * │ can never deliver a new event on a freed ep->data.  Do NOT enable dispatch │
- * │ until that ordering is in place.                                           │
  * └──────────────────────────────────────────────────────────────────────────┘
  */
 static int peer_ctrl_dispatch_to_pool(endpoint_t *ep, io_packet_t *pkt)

src/controller/testsuite/test_config_repl.c

@@ -2721,6 +2721,130 @@ static void test_wal_stream_empty_range_no_batches(void)
     printf("test_wal_stream_empty_range_no_batches: PASS\n");
 }
 
+/* Write N entries starting at start_lsn where entries[checkpoint_idx] (0-based
+ * within the range) is a WAL_CHECKPOINT marker (local-only, never a logical
+ * Raft entry — persist_write_checkpoint_marker consumes a real LSN slot for
+ * it, exactly like a live checkpoint firing mid-stream would). All other
+ * entries are WAL_JOB_UPDATE. */
+static int write_test_wal_segment_with_checkpoint_at(const char *wal_dir,
+                                                      uint64_t start_lsn,
+                                                      int n_entries,
+                                                      int checkpoint_idx)
+{
+    char seg_path[4096];
+    snprintf(seg_path, sizeof(seg_path),
+             "%s/seg-%016" PRIx64 ".wal", wal_dir, start_lsn);
+
+    int fd = open(seg_path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+    if (fd < 0) return -1;
+
+    for (int i = 0; i < n_entries; i++) {
+        wal_entry_hdr_t hdr;
+        memset(&hdr, 0, sizeof(hdr));
+        hdr.lsn         = start_lsn + (uint64_t)i;
+        hdr.type        = (i == checkpoint_idx) ? WAL_CHECKPOINT : WAL_JOB_UPDATE;
+        hdr.payload_len = 0;
+        hdr.term        = 1;
+        ssize_t nw = write(fd, &hdr, sizeof(hdr));
+        if (nw != (ssize_t)sizeof(hdr)) { close(fd); return -1; }
+    }
+    close(fd);
+    return 0;
+}
+
+/* Captures every streamed entry's lsn/entry_type/prev_log_index across all
+ * batches, in send order. */
+#define CAP_CKPT_MAX 32
+static uint64_t g_cap_ckpt_lsn[CAP_CKPT_MAX];
+static uint8_t  g_cap_ckpt_type[CAP_CKPT_MAX];
+static uint64_t g_cap_ckpt_prev_idx[CAP_CKPT_MAX];
+static int      g_cap_ckpt_count = 0;
+
+static int capture_all_entries_wal_stream_hook(endpoint_t *ep, xce_request_t *req)
+{
+    (void)ep;
+    if (!req || !req->body.walcatchupentrymsg) return 0;
+    wal_catchup_entry_t *batch =
+        (wal_catchup_entry_t *)req->body.walcatchupentrymsg;
+    for (int i = 0; i < batch->n_entries && g_cap_ckpt_count < CAP_CKPT_MAX; i++) {
+        wal_catchup_entry_t *e = batch->entries[i];
+        if (!e) continue;
+        g_cap_ckpt_lsn[g_cap_ckpt_count]      = e->lsn;
+        g_cap_ckpt_type[g_cap_ckpt_count]     = e->entry_type;
+        g_cap_ckpt_prev_idx[g_cap_ckpt_count] = e->prev_log_index;
+        g_cap_ckpt_count++;
+    }
+    return 0;
+}
+
+/*
+ * The catch-up streamer must SKIP WAL_CHECKPOINT markers entirely — they are
+ * local-only bookkeeping (persist_write_checkpoint_marker's own doc comment:
+ * "replay_wal SKIPS checkpoint markers... the recovered index never holds a
+ * checkpoint LSN; the live index must match"), never a logical Raft log
+ * entry, and never sent on the live broadcast path either
+ * (persist_wal_write_replicated rejects type==WAL_CHECKPOINT with -EINVAL).
+ * A standby receiving one gets stuck retrying the same LSN forever.
+ *
+ * Fixture: LSN [1..5], LSN 3 is a checkpoint marker (consumes a real LSN slot,
+ * exactly like a live checkpoint firing mid-range). Streaming standby_lsn=0,
+ * committed_lsn=5 must emit exactly 4 real entries (1,2,4,5), never lsn=3, and
+ * the entry after the gap (lsn=4) must chain prev_log_index=2 (the last REAL
+ * entry before the gap), not 3.
+ *
+ * RED before the fix: the streamer builds/sends a catchup_entry for every
+ * hdr.type in range with no filter, so lsn=3/type=WAL_CHECKPOINT is included
+ * and lsn=4's prev_log_index is 3.
+ */
+static void test_wal_stream_skips_checkpoint_marker(void)
+{
+    char tmpdir[256];
+    snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_ckpt_%d_%ld",
+             (int)getpid(), (long)time(NULL));
+    TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700));
+
+    int rc = write_test_wal_segment_with_checkpoint_at(
+        tmpdir, /*start_lsn=*/1, /*n_entries=*/5, /*checkpoint_idx=*/2);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    xce_wal_stream_request_hook = capture_all_entries_wal_stream_hook;
+    g_cap_ckpt_count = 0;
+
+    rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, /*p=*/NULL,
+                                   /*leader_term=*/7,
+                                   /*standby_lsn=*/0,
+                                   /*committed_lsn=*/5);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed");
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(4, g_cap_ckpt_count,
+        "the checkpoint marker at lsn=3 must never be streamed as an entry "
+        "(4 real entries: 1,2,4,5)");
+
+    for (int i = 0; i < g_cap_ckpt_count; i++) {
+        TEST_ASSERT_NOT_EQUAL_MESSAGE(3, (int)g_cap_ckpt_lsn[i],
+            "lsn=3 (the checkpoint marker) must never appear in a streamed entry");
+        TEST_ASSERT_NOT_EQUAL_MESSAGE((int)WAL_CHECKPOINT, (int)g_cap_ckpt_type[i],
+            "no streamed entry may carry entry_type==WAL_CHECKPOINT");
+    }
+
+    /* Entries in order must be lsn 1,2,4,5 with lsn=4's prev_log_index=2
+     * (the last real entry before the checkpoint gap, not the marker's lsn 3). */
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, g_cap_ckpt_lsn[0], "entry 0 must be lsn=1");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(2, g_cap_ckpt_lsn[1], "entry 1 must be lsn=2");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, g_cap_ckpt_lsn[2], "entry 2 must be lsn=4 (lsn=3 skipped)");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, g_cap_ckpt_lsn[3], "entry 3 must be lsn=5");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(2, g_cap_ckpt_prev_idx[2],
+        "lsn=4 must chain prev_log_index=2 (the last real entry before the "
+        "checkpoint gap), not the skipped marker's lsn=3");
+
+    xce_wal_stream_request_hook = NULL;
+
+    char cmd[512];
+    snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
+    system(cmd);
+    printf("test_wal_stream_skips_checkpoint_marker: PASS\n");
+}
+
 /*
  * bug-1028 — the catch-up streamer must source each streamed entry's term and
  * the AppendEntries prev_log_term from the AUTHORITATIVE per-LSN term index
@@ -3604,6 +3728,7 @@ void test_config_repl(void)
     RUN_TEST(test_wal_stream_500_entries_two_batches);
     RUN_TEST(test_wal_stream_partial_batch);
     RUN_TEST(test_wal_stream_empty_range_no_batches);
+    RUN_TEST(test_wal_stream_skips_checkpoint_marker);
     RUN_TEST(test_wal_stream_term_from_index_not_raw_hdr); /* bug-1028 */
     /* Task 3.5: snapshot chunk transfer (sender side) */
     RUN_TEST(test_snapshot_transfer_two_chunks);

src/controller/testsuite/test_ha_liveness.c

@@ -1880,6 +1880,76 @@ static void test_snapshot_offload_runs_on_lane(void)
     printf("test_snapshot_offload_runs_on_lane: PASS\n");
 }
 
+/*
+ * Task 3.2 borrowed-endpoint lifetime fix: the lane dispatch must hold its
+ * OWN endpoint_ref() across the transfer so a concurrent teardown (the owner
+ * — e.g. the standby zarray on RDHUP — dropping ITS ref) cannot free `ep`
+ * while the lane worker is still using it (bug-1197 family, closed generally
+ * by the endpoint refcount system; this proves the snapshot lane actually
+ * participates in it).
+ *
+ * RED before the fix: the dispatch never calls endpoint_ref, so refcnt stays
+ * at the owner's original 1 instead of 2 while parked; the owner's simulated
+ * teardown drop then takes it to 0 — endpoint_unref frees `ep` (heap_alloc)
+ * WHILE the lane is still parked on it, so the parked-refcnt assertion fails
+ * (or, if the free races the read, ASAN reports a heap-use-after-free).
+ */
+static void test_snapshot_offload_holds_endpoint_ref_across_transfer(void)
+{
+    cluster_t c;
+    memset(&c, 0, sizeof(c));
+    ha_init(&c.ha, &g_cfg, 0);
+    TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
+
+    /* Heap-allocated endpoint (heap_alloc=1) standing in for a real standby
+     * connection: refcnt starts at 1, representing the owner (e.g. the
+     * standby zarray) holding the creator reference. */
+    endpoint_t *ep = endpoint_get();
+    TEST_ASSERT_NOT_NULL(ep);
+
+    snap_reset_probe();
+    ha_repl_send_snapshot_hook = snap_probe;
+    atomic_store(&g_snap_gate_on, 1);   /* park the lane body mid-transfer */
+    ha_set_snapshot_offload_for_test(1);
+
+    int rc = snap_dispatch_once(&c, ep, 0);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, snap_wait_parked(2000),
+        "the snapshot send body must park on the lane before we probe refcnt");
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(2,
+        atomic_load_explicit(&ep->refcnt, memory_order_acquire),
+        "lane dispatch must hold its own ref (owner's 1 + lane's 1) while "
+        "the transfer is in flight");
+
+    /* Simulate the owner's teardown (e.g. RDHUP -> zarray_del ->
+     * endpoint_killer -> endpoint_unref) racing the in-flight transfer. If
+     * the lane did not take its own ref, this drops refcnt to 0 and frees ep
+     * out from under the still-parked send -- exactly the UAF this fix
+     * closes. */
+    endpoint_unref(ep);
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1,
+        atomic_load_explicit(&ep->refcnt, memory_order_acquire),
+        "ep must survive the owner's unref while the lane transfer is still "
+        "in flight -- the lane's own ref keeps it alive");
+
+    snap_release_gate();
+    int runs = snap_wait_runs(1, 2000);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
+        "the parked send must complete after the gate releases");
+
+    /* ep is now freed (lane's completion unref dropped the last reference) --
+     * do not touch it again. */
+
+    ha_set_snapshot_offload_for_test(0);   /* restore production default (OFF) */
+    ha_repl_send_snapshot_hook = NULL;
+    slowlane_fini(&c.slowlane);
+    ha_fini(&c.ha);
+    printf("test_snapshot_offload_holds_endpoint_ref_across_transfer: PASS\n");
+}
+
 /*
  * Task 3.2 RED side / no-regression (Req 6.x): with the offload flag OFF (the
  * production default until Task 3.4 resolves the borrowed-endpoint lifetime),
@@ -2274,6 +2344,7 @@ void test_ha_liveness(void)
     RUN_TEST(test_peerctrl_preserves_per_endpoint_order);
     /* raft-drive-async Task 3.2 — SnapshotSend slow-lane offload (Req 1.1, 1.5, 6.x) */
     RUN_TEST(test_snapshot_offload_runs_on_lane);
+    RUN_TEST(test_snapshot_offload_holds_endpoint_ref_across_transfer);
     RUN_TEST(test_snapshot_inline_when_flag_off);
     RUN_TEST(test_snapshot_offload_dedups_per_peer);
     /* ha-entity-snapshots Task 4.1 — in-flight snapshot LSN tracking (Req 3.1-3.3) */