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

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

Browse source 
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
This commit is contained in:
Rongsong Shen 2026-07-02 04:54:17 +00:00
parent 1439e2655e
commit f3b6e6ae3b
5 changed files with 315 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

125
src/controller/testsuite/test_config_repl.c
View file

@ -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);

71
src/controller/testsuite/test_ha_liveness.c
View file

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