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

feat(raft-drive-async): snapshot SlowLane offload + WAL-trim cursor guard (task 3.2)

Browse source 
Part B (LIVE, default-ON): trim_wal_segments now bounds WAL-segment deletion
by repl_trim_lsn(&p->repl, active_seg_lsn) = min(active segment LSN, every
active standby acked_lsn, every valid disconnected-slot last_disconnected_lsn),
after repl_expire_disconnect_slots ages out stale slot guards. floor <=
active_seg_lsn structurally, so the guard can only preserve more segments,
never delete one a live standby still needs — closes the bug-795/828
trim-vs-cursor race for connected/recently-disconnected standbys (bug-1147
stopgap; empty/cursor=0 join still needs entity-bearing snapshots).

Part A (DORMANT, g_drive_async_snapshot_offload default OFF): ha_snapshot_send_
dispatch routes HA_REPL_SEND_SNAPSHOT inline (flag OFF — byte-for-byte legacy,
on io_worker) or to slowlane_submit+snapshot_lane_fn (flag ON). Send body
factored into ha_snapshot_do_send. Per-peer single-flight via
ha_state_t.snapshot_inflight[] CAS + slowlane key; in-flight flag cleared on
all four terminal paths, mirroring the reconnect offload (task 2.1).

Lands dormant like task 3.1: snapshot_lane_fn borrows a standby endpoint it
does not own, so flag-ON would race io_worker RDHUP teardown (cross-thread UAF,
bug-1197 family). Flag OFF keeps send+teardown serialized on the io_worker.
Task 3.4 is the single gate that flips both this and the peerctrl flag ON after
establishing endpoint-lifetime discipline; bug-1198 stays open by design until
then. The offload flag was initially authored default-ON (would ship the UAF)
and corrected to OFF in ha.c and the ha.h accessor doc (bug-1210).

Tests (controller 196/0, 0 new leaks): test_snapshot_offload_runs_on_lane,
test_snapshot_inline_when_flag_off, test_snapshot_offload_dedups_per_peer,
test_trim_guard_preserves_standby_segments,
test_trim_without_standbys_proceeds_normally. Adversarial cpp-review: GO.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S6fDCtXkkEgupu7rUah1y2
This commit is contained in:
Rongsong Shen 2026-06-26 15:43:24 +00:00
parent e17a6a613c
commit 6c0523b69b
8 changed files with 2046 additions and 850 deletions
Download patch file Download diff file Expand all files Collapse all files

147
src/controller/testsuite/test_e2e_checkpoint.c
View file

@ -21,6 +21,24 @@
#include "config.h"
#include "persist.h"
#include "repl.h" /* repl_add_standby / repl_on_ack for trim-guard tests
* (raft-drive-async task 3.2, bug-1147) */
#include "typemap.h" /* wal_catchup_entry_t */
/* Drop hook for repl_broadcast: fixtures with dummy (int) endpoints installed
* via repl_add_standby must install this so persist_wal_async→repl_broadcast does
* not crash on the fake endpoint. Pattern from test_commit_engine.c, test_p2_integration.c. */
static int ckpt_drop_broadcast(void *ep, wal_catchup_entry_t **entries, int count)
{
(void)ep;
(void)entries;
(void)count;
return 0;
}
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int count);
/* ------------------------------------------------------------------ */
/* Helpers */
@ -178,6 +196,133 @@ static void test_snapshot_rotation_keeps_snap_keep_n(void)
printf("test_snapshot_rotation_keeps_snap_keep_n: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Helper: check if a WAL segment file at a given start LSN exists. */
/* Builds the name "seg-<lsn16>.wal" inside dir. */
/* ------------------------------------------------------------------ */
static int seg_file_exists(const char* wal_dir, uint64_t seg_start_lsn)
{
char path[4096];
snprintf(path, sizeof(path), "%s/seg-%016" PRIx64 ".wal",
wal_dir, seg_start_lsn);
return access(path, F_OK) == 0 ? 1 : 0;
}
/* ------------------------------------------------------------------ */
/* Test 4: standby cursor-bound trim guard — Req 1.1 (task 3.2, */
/* bug-1147). With a standby whose acked_lsn is below the */
/* active segment, segments the standby still needs for */
/* catch-up are NOT deleted. */
/* */
/* RED proof: revert the repl_trim_lsn floor call in trim_wal_segments */
/* → the segment at the standby's acked LSN is deleted because it's */
/* below the active segment → re-creates the bug-795/828 gap where */
/* catch-up hits a missing segment → permanent committed-job loss. */
/* ------------------------------------------------------------------ */
static void test_trim_guard_preserves_standby_segments(void)
{
char tmpdir[256];
persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2);
TEST_ASSERT_NOT_NULL(p);
/* Install drop hook so repl_broadcast doesn't crash on the dummy ep */
repl_broadcast_send_hook = ckpt_drop_broadcast;
/* Add a standby with acked_lsn = 2 (simulating a standby that has
* caught up through LSN 2 but not beyond) */
int dummy_ep = 0;
int idx = repl_add_standby(&p->repl, (void*)&dummy_ep, "standby1", 7001);
TEST_ASSERT_TRUE(idx >= 0);
repl_on_ack(&p->repl, (void*)&dummy_ep, 2);
/* Write entries to create segments.
* init: current_lsn = 1 (LSN 0 reserved, fresh store)
* write "x" @ LSN 1 → current_lsn = 2
* force → close seg-at-1, open seg-at-2
* write "y" @ LSN 2 → current_lsn = 3
* force → close seg-at-2, open seg-at-3
*
* Now we have 3 segments: seg-...001.wal, seg-...002.wal, seg-...003.wal.
* Active segment = seg-at-3. Standby acked_lsn = 2. */
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1));
TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p));
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "y", 1));
TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p));
/* Trigger checkpoint → trim_wal_segments runs */
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p));
/* floor = repl_trim_lsn(active_seg_lsn=3, ...) = min(3, acked=2) = 2.
* So seg_start_lsn < 2 means only seg-at-1 is deleted (1 < 2).
* seg-at-2 (the standby still needs it) is NOT deleted (2 >= 2). */
char wal_path[512];
snprintf(wal_path, sizeof(wal_path), "%s/e2e-ckpt/data/wal", tmpdir);
/* The old segment below the standby's acked LSN IS trimmed */
TEST_ASSERT_FALSE_MESSAGE(
seg_file_exists(wal_path, 1),
"seg-at-1: below standby acked_lsn=2, MUST be trimmed");
/* The segment at the standby's acked LSN is PRESERVED (bug-1147 guard) */
TEST_ASSERT_TRUE_MESSAGE(
seg_file_exists(wal_path, 2),
"seg-at-2: at standby acked_lsn=2, MUST be preserved (bug-1147 guard)");
/* The active segment is never trimmed */
TEST_ASSERT_TRUE_MESSAGE(
seg_file_exists(wal_path, 3),
"seg-at-3: active segment, MUST always be preserved");
repl_broadcast_send_hook = NULL;
teardown_ckpt_e2e(p);
printf("test_trim_guard_preserves_standby_segments: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 5: trim with no standbys proceeds as before — Req 1.1 */
/* (task 3.2). With no standbys registered, the floor */
/* defaults to active_seg_lsn, so all old segments are */
/* trimmed. This confirms the guard does not prevent normal */
/* cleanup when no standbys need protection. */
/* ------------------------------------------------------------------ */
static void test_trim_without_standbys_proceeds_normally(void)
{
char tmpdir[256];
persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2);
TEST_ASSERT_NOT_NULL(p);
/* No standbys — just create segments and checkpoint */
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p));
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1));
TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p));
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p));
char wal_path[512];
snprintf(wal_path, sizeof(wal_path), "%s/e2e-ckpt/data/wal", tmpdir);
/* Without standbys, floor = active_seg_lsn = 3.
* seg_at_1 < 3 → deleted. seg_at_2 < 3 → deleted too. */
TEST_ASSERT_FALSE_MESSAGE(
seg_file_exists(wal_path, 1),
"seg-at-1: no standbys, MUST be trimmed");
TEST_ASSERT_FALSE_MESSAGE(
seg_file_exists(wal_path, 2),
"seg-at-2: no standbys, MUST be trimmed");
/* Active segment still there */
TEST_ASSERT_TRUE_MESSAGE(
seg_file_exists(wal_path, 3),
"seg-at-3: active segment, MUST always be preserved");
teardown_ckpt_e2e(p);
printf("test_trim_without_standbys_proceeds_normally: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -189,4 +334,6 @@ void test_e2e_checkpoint(void)
RUN_TEST(test_checkpoint_recovery_lsn_continuity);
RUN_TEST(test_checkpoint_wal_trim_after_checkpoint);
RUN_TEST(test_snapshot_rotation_keeps_snap_keep_n);
RUN_TEST(test_trim_guard_preserves_standby_segments);
RUN_TEST(test_trim_without_standbys_proceeds_normally);
}

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

@ -1745,6 +1745,230 @@ static void test_peerctrl_preserves_per_endpoint_order(void)
printf("test_peerctrl_preserves_per_endpoint_order: PASS\n");
}
/* ───────────────────────────────────────────────────────────────────────
* Task 3.2 — SnapshotSend slow-lane offload (Req 1.1, 1.5; bug-1198).
*
* Probe: a snapshot-send hook (ha_repl_send_snapshot_hook) that records the
* thread it ran on and supports an optional gate so a test can hold the lane
* worker in flight while it submits a second snapshot for the same peer
* (per-peer dedup proof, Req 1.5). Mirrors the peerctrl pc_probe gate.
* ─────────────────────────────────────────────────────────────────────── */
extern int (*ha_repl_send_snapshot_hook)(int peer_idx);
static _Atomic int g_snap_runs = 0;
static _Atomic pthread_t g_snap_last_thread;
static _Atomic int g_snap_gate_on = 0;
static pthread_mutex_t g_snap_mu = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t g_snap_parked_cv = PTHREAD_COND_INITIALIZER;
static pthread_cond_t g_snap_release_cv = PTHREAD_COND_INITIALIZER;
static int g_snap_parked = 0;
static int g_snap_release = 0;
static int snap_probe(int peer_idx)
{
(void)peer_idx;
atomic_store_explicit(&g_snap_last_thread, pthread_self(),
memory_order_relaxed);
if (atomic_load(&g_snap_gate_on)) {
pthread_mutex_lock(&g_snap_mu);
g_snap_parked = 1;
pthread_cond_broadcast(&g_snap_parked_cv);
while (!g_snap_release)
pthread_cond_wait(&g_snap_release_cv, &g_snap_mu);
pthread_mutex_unlock(&g_snap_mu);
}
atomic_fetch_add(&g_snap_runs, 1);
return 0;
}
/* Bounded poll until the gated body parks; 0 on park, -1 on timeout. */
static int snap_wait_parked(int timeout_ms)
{
pthread_mutex_lock(&g_snap_mu);
int parked = 0;
for (int i = 0; i < timeout_ms; i++) {
if (g_snap_parked) { parked = 1; break; }
pthread_mutex_unlock(&g_snap_mu);
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
nanosleep(&ts, NULL);
pthread_mutex_lock(&g_snap_mu);
}
pthread_mutex_unlock(&g_snap_mu);
return parked ? 0 : -1;
}
static int snap_wait_runs(int want, int timeout_ms)
{
for (int i = 0; i < timeout_ms; i++) {
if (atomic_load(&g_snap_runs) >= want) break;
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
nanosleep(&ts, NULL);
}
return atomic_load(&g_snap_runs);
}
static void snap_release_gate(void)
{
pthread_mutex_lock(&g_snap_mu);
g_snap_release = 1;
pthread_cond_broadcast(&g_snap_release_cv);
pthread_mutex_unlock(&g_snap_mu);
}
static void snap_reset_probe(void)
{
atomic_store(&g_snap_runs, 0);
atomic_store(&g_snap_gate_on, 0);
pthread_mutex_lock(&g_snap_mu);
g_snap_parked = 0;
g_snap_release = 0;
pthread_mutex_unlock(&g_snap_mu);
}
/* Drive one SEND_SNAPSHOT decision: standby tag current but lsn behind the
* oldest WAL window → ha_evaluate_repl_action yields HA_REPL_SEND_SNAPSHOT. */
static int snap_dispatch_once(cluster_t *c, endpoint_t *ep, int peer_idx)
{
c->ha.standby_reported_tag[peer_idx] = 5;
c->ha.standby_reported_lsn[peer_idx] = 10; /* < oldest_wal_lsn below */
return ha_dispatch_repl_action_for_standby(
&c->ha, c, ep, peer_idx,
/*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/50);
}
/*
* Task 3.2 GREEN (Req 1.1, 8.1/8.2): with the offload flag ON, SEND_SNAPSHOT
* is submitted to the slow lane — the send body runs on a LANE worker (a
* DIFFERENT thread from the caller) and the dispatch returns promptly. This
* proves the snapshot transfer no longer blocks the io_worker/drive thread.
*
* RED before Task 3.2 lands: the snapshot ran inline on the caller thread, so
* the different-thread assertion fails.
*/
static void test_snapshot_offload_runs_on_lane(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));
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
snap_reset_probe();
ha_repl_send_snapshot_hook = snap_probe;
ha_set_snapshot_offload_for_test(1);
pthread_t caller = pthread_self();
int rc = snap_dispatch_once(&c, &ep, 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"Req 8.1/8.2: dispatch must return promptly (0) after lane submit");
int runs = snap_wait_runs(1, 2000);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
"Req 1.1: the snapshot send body must run on the lane exactly once");
TEST_ASSERT_FALSE_MESSAGE(
pthread_equal(atomic_load_explicit(&g_snap_last_thread,
memory_order_relaxed), caller),
"Req 1.1: the snapshot send must run on a LANE worker, NOT the "
"caller's (io_worker/drive) thread");
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_runs_on_lane: 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),
* SEND_SNAPSHOT runs INLINE on the calling thread — byte-for-byte the
* pre-task behavior.
*/
static void test_snapshot_inline_when_flag_off(void)
{
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
snap_reset_probe();
ha_repl_send_snapshot_hook = snap_probe;
ha_set_snapshot_offload_for_test(0); /* explicit OFF */
pthread_t caller = pthread_self();
int rc = snap_dispatch_once(&c, &ep, 0);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Synchronous: the body already ran on THIS thread before returning. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_snap_runs),
"Req 6.x: with the flag OFF the snapshot send must run inline");
TEST_ASSERT_TRUE_MESSAGE(
pthread_equal(atomic_load_explicit(&g_snap_last_thread,
memory_order_relaxed), caller),
"Req 6.x: with the flag OFF the send must run on the caller's thread");
ha_repl_send_snapshot_hook = NULL;
ha_fini(&c.ha);
printf("test_snapshot_inline_when_flag_off: PASS\n");
}
/*
* Task 3.2 (Req 1.5): per-peer single-flight. While a snapshot to peer 0 is
* in flight on the lane (parked at the gate), a second SEND_SNAPSHOT decision
* for the SAME peer enqueues nothing — the CAS on snapshot_inflight[0] fails.
* Proof: with the first body parked, the run count stays at 0; after release,
* exactly ONE body has run (the duplicate was deduped, not queued).
*/
static void test_snapshot_offload_dedups_per_peer(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));
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
snap_reset_probe();
atomic_store(&g_snap_gate_on, 1); /* lane body parks at the gate */
ha_repl_send_snapshot_hook = snap_probe;
ha_set_snapshot_offload_for_test(1);
/* First snapshot — submitted; its body parks in the lane. */
TEST_ASSERT_EQUAL_INT(0, snap_dispatch_once(&c, &ep, 0));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, snap_wait_parked(2000),
"the first snapshot body must reach the gate (running on the lane)");
/* Second snapshot for the SAME peer while the first is in flight. */
TEST_ASSERT_EQUAL_INT(0, snap_dispatch_once(&c, &ep, 0));
/* Give a generous window: the duplicate must NOT run. */
struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */
nanosleep(&ts, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_snap_runs),
"Req 1.5: the duplicate snapshot must be deduped while one is in flight");
/* Release; exactly the first body completes. */
snap_release_gate();
int runs = snap_wait_runs(1, 2000);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
"Req 1.5: exactly one snapshot ran for the peer (duplicate deduped)");
/* And no second body sneaks in after release. */
nanosleep(&ts, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_snap_runs),
"Req 1.5: the deduped duplicate was dropped, not queued behind the first");
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_dedups_per_peer: PASS\n");
}
/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_ha_liveness(void)
@ -1782,5 +2006,9 @@ void test_ha_liveness(void)
RUN_TEST(test_peerctrl_dispatch_runs_on_pool);
RUN_TEST(test_peerctrl_inline_when_flag_off);
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_inline_when_flag_off);
RUN_TEST(test_snapshot_offload_dedups_per_peer);
UNITY_END();
}