From f28ce579931bd520489546876bae2720bb411a29 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 8 Jul 2026 08:45:22 +0000 Subject: [PATCH] =?UTF-8?q?fix(raft-apply-decoupling):=20bug-1419=20layer?= =?UTF-8?q?=203=20=E2=80=94=20close=20cross-primary=20client-retry=20dedup?= =?UTF-8?q?=20gap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit jobid_range_exists() (c->jobmgr) and the in-flight registry (job.c, per-process, only populated by this node's own handle_jobsubmit calls) both miss a real cross-node case: a client's XCECLI_FAILOVER_CALL retry resends the same client-generated jobid after a connection drop, and the resend lands on a newly elected primary that never processed the original request. If the original was already committed before the old primary died, the new primary's own local WAL already holds it (Raft Leader Completeness) but jobmgr hasn't applied it yet — both existing guards see nothing, so the resend appends a second, genuinely committed WAL_JOB_SUBMIT entry for the same jobid. A live chaotic soak reproduced this exactly (waldump: two committed entries, same term, matching apiserverId/timestamp/ sequence, one second apart) — apply's own idempotent dedup keeps jobmgr correct, but replay's raw WAL_JOB_SUBMIT count no longer matches jobmgr's growth, firing a false APPLY_DIVERGENT fence and deadlocking promotion. Fix: apply_committed_entry()'s WAL_JOB_SUBMIT case now claims the jobid in job.c's in-flight registry (new job_wal_commit_claim/release, wrapping the existing inflight_range_try_add/remove) for the duration of dispatch, regardless of whether this node originated the submission or is applying a replicated/replayed entry. A resend landing mid-dispatch on any node now sees the same -EEXIST rejection self-originated in-flight submits already get. Release only fires when the claim call added the entry, so a self-originated submission's own claim/release lifecycle (unchanged) is never double-released or stolen. Also fixes 4 pre-existing tests that broke under the earlier checkpoint_lsn fix (0eb1054) and were missed in that commit's regression sweep: cluster- less fixtures where last_applied_lsn never advances now correctly get checkpoint_lsn=0 unless the test simulates apply catch-up, matching the new (correct) semantics. One assertion in test_checkpoint_tail_with_no_snapshot_is_apply_clean compared Phase 1's checkpoint_lsn against Phase 2's independently-reconstructed physical WAL tip — that equality only held under the pre-fix current_lsn-tracking behavior and is removed, not patched around. Also hardens run-ha-soak.sh's start_node() with setsid+nohup: a forked/ background test-harness invocation whose session tears down mid-run was silently killing every controller process via SIGHUP forwarding (observed twice — deploy completes, ~1s of activity, then all 3 nodes die together with no error logged). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01VgT3n4vTv4ZjCuCTmzed1F --- .claude/skills/xce-test-ha/run-ha-soak.sh | 24 ++++++++--- src/controller/include/cluster.h | 15 +++++++ src/controller/job.c | 35 ++++++++++++++++ src/controller/persist.c | 36 +++++++++++----- src/controller/testsuite/test.c | 22 ++++++++++ .../test_apply_correctness_verification.c | 24 ++++++++++- .../testsuite/test_commit_barrier_reply.c | 42 +++++++++++++++++++ .../testsuite/test_e2e_checkpoint.c | 8 ++++ src/controller/testsuite/test_e2e_gc.c | 8 ++++ .../testsuite/test_persist_recovery.c | 8 ++++ 10 files changed, 206 insertions(+), 16 deletions(-) diff --git a/.claude/skills/xce-test-ha/run-ha-soak.sh b/.claude/skills/xce-test-ha/run-ha-soak.sh index d4b0ae46..4b13d772 100755 --- a/.claude/skills/xce-test-ha/run-ha-soak.sh +++ b/.claude/skills/xce-test-ha/run-ha-soak.sh @@ -239,19 +239,33 @@ start_node() { # yama scope 1) instead of via start.sh, so a hang can be interrupted # (SIGINT to gdb's own pid) for a live thread backtrace. Not for normal # soak use -- temporary investigation-only seam. + # Forked/background-agent execution note: a plain trailing `&` keeps the + # child in the SAME session as this script. If the session that launched + # run-ha-soak.sh tears down mid-run (e.g. a forked skill invocation whose + # controlling session ends once its own turn returns), the kernel sends + # SIGHUP to everything still attached to it -- bash forwards that to + # unprotected background jobs by default, killing every node silently + # (no error, no log line, just simultaneous death) well before the soak + # can finish. setsid detaches into a brand-new session (no controlling + # terminal to ever receive a SIGHUP from) and nohup additionally makes + # the process ignore SIGHUP outright -- belt and suspenders, since which + # exact signal/mechanism a given forked-agent runtime uses to tear down + # is not something this script can assume. >"$TOP/logs/$1.gdbwrap" 2>&1 & + >"$TOP/logs/$1.gdbwrap" 2>&1 & return fi if [ ${#extra_args[@]} -gt 0 ]; then - bash "$TOP/scripts/start.sh" "$1" -vvv "${extra_args[@]}" \ - >>"$TOP/logs/$1.stderr" 2>&1 & + setsid nohup bash "$TOP/scripts/start.sh" "$1" -vvv "${extra_args[@]}" \ + >"$TOP/logs/$1.stderr" 2>&1 & else - bash "$TOP/scripts/start.sh" "$1" -vvv >>"$TOP/logs/$1.stderr" 2>&1 & + setsid nohup bash "$TOP/scripts/start.sh" "$1" -vvv \ + >"$TOP/logs/$1.stderr" 2>&1 & fi } diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index ad82f094..a2034fa7 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -73,6 +73,21 @@ extern int handle_join(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_reply(endpoint_t*, struct Request*, cluster_t*); extern int handle_jobdispatch(cluster_t*, endpoint_t*, xce_request_t*); extern int handle_jobsubmit(cluster_t*, endpoint_t*, xce_request_t*); +/* bug-1419 layer 3: claim/release a single jobid in job.c's in-flight + * registry from OUTSIDE handle_jobsubmit -- used by apply_committed_entry() + * (persist.c) to make a WAL_JOB_SUBMIT entry's jobid visible as + * "committed, dispatch in progress" for the whole duration of its apply, + * regardless of whether this node originated the submission or received it + * via replication/replay. Closes the cross-primary-retry gap: a client + * resend landing on a NEW primary after the old one committed the entry but + * died before acking sees jobid_range_exists() find nothing (not yet + * applied to c->jobmgr) unless this claim is also visible. + * job_wal_commit_claim returns true iff IT added the entry (false if + * something else -- e.g. this node's own handle_jobsubmit -- already holds + * an overlapping claim); only call job_wal_commit_release when the claim + * call returned true, so a caller never removes an entry it does not own. */ +extern bool job_wal_commit_claim(uint64_t jobid); +extern void job_wal_commit_release(uint64_t jobid); #ifdef XCE_TESTING /* bug-1419: clears job.c's process-lifetime in-flight-jobid registry. * Tests that deliberately leave a submission's commit continuation diff --git a/src/controller/job.c b/src/controller/job.c index c718f26d..f0631198 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -155,6 +155,41 @@ static void inflight_range_remove(uint64_t base_jobid, uint64_t array_size) pthread_mutex_unlock(&s_inflight_submits_lock); } +/* bug-1419 layer 3 (cross-primary retry gap): apply_committed_entry() + * (persist.c) calls this around its WAL_JOB_SUBMIT dispatch so the jobid is + * visible in s_inflight_submits for the whole duration of apply, regardless + * of whether this node originated the submission (handle_jobsubmit already + * holds an overlapping claim in that case -- see the false-return note + * below) or received it via replication/replay (no local claim exists yet + * without this call). A resend of an already-committed-but-not-yet-applied + * jobid landing on this node -- e.g. a client retry reaching a newly + * elected primary after the old one committed the entry but died before + * acking -- is rejected by inflight_range_try_add here exactly as it would + * be for a same-node in-flight submission, closing the window + * jobid_range_exists()'s c->jobmgr-only check cannot see. + * + * Single-element claim (array_size=1): a WAL_JOB_SUBMIT entry always names + * one individual jobid (build_job_submit_payloads emits one entry per array + * element), never a range. Returns true iff this call added the entry -- + * false means an overlapping claim already exists (typically this node's + * own handle_jobsubmit-originated range claim covering the same jobid), in + * which case the caller must NOT release it; that claim's own submitter + * releases it via jobsubmit_async_ctx_release as usual. */ +bool job_wal_commit_claim(uint64_t jobid) +{ + return inflight_range_try_add(jobid, 1); +} + +/* Release a claim job_wal_commit_claim() added. Only call when that call + * returned true -- inflight_range_remove only matches an EXACT + * base_jobid+array_size pair, so calling it for a claim this function never + * added is always a safe no-op, but the caller-side guard avoids a wasted + * lock+scan on the common already-claimed-elsewhere path. */ +void job_wal_commit_release(uint64_t jobid) +{ + inflight_range_remove(jobid, 1); +} + #ifdef XCE_TESTING /* Test-only seam: s_inflight_submits is process-lifetime, not tied to any * cluster_t fixture. In production that's fine -- a restart wipes it along diff --git a/src/controller/persist.c b/src/controller/persist.c index a89c2f0d..5536d912 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -2726,16 +2726,12 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, switch (wal_type) { case WAL_JOB_SUBMIT: { - int rc = restore_job(c, 0, payload, len, lsn); - if (rc == -EEXIST) - return 0; /* idempotent re-apply (replay / resync) — no hook */ - if (rc < 0) - return rc; - /* Recover the jobid via a cheap raw capnp field read (read_Job only - * reads the fixed uint64 field — no allocation, unlike decode_Job) - * so the just-constructed job can be looked up for the notification - * hook without changing restore_job's own contract. */ + * reads the fixed uint64 field — no allocation, unlike decode_Job), + * up front this time (not just for the post-success hook lookup): + * bug-1419 layer 3 needs it BEFORE restore_job() to claim the jobid + * as in-flight for the whole dispatch, closing the cross-primary + * retry gap (see job_wal_commit_claim's doc comment, cluster.h). */ struct capn jc_ctx; capn_init_mem(&jc_ctx, (const uint8_t*)payload, len, 0); Job_ptr jp; @@ -2745,12 +2741,29 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, uint64_t jobid = cj.jobid; capn_free(&jc_ctx); + bool claimed_here = job_wal_commit_claim(jobid); + + int rc = restore_job(c, 0, payload, len, lsn); + if (rc == -EEXIST) { + if (claimed_here) + job_wal_commit_release(jobid); + return 0; /* idempotent re-apply (replay / resync) — no hook */ + } + if (rc < 0) { + if (claimed_here) + job_wal_commit_release(jobid); + return rc; + } + job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); - if (!j) + if (!j) { + if (claimed_here) + job_wal_commit_release(jobid); return 0; /* defensive: should not happen after a successful restore */ + } void* jptr = j; xlist_t jlist = { .count = 1, .caps = 1, .delta = 1, .data = &jptr }; @@ -2758,6 +2771,9 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, c, &jlist); t_apply_seam_notifying = false; + + if (claimed_here) + job_wal_commit_release(jobid); return 0; } diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index e86330cc..8afb5f34 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -3614,6 +3614,28 @@ int main(int argc, char* argv[]) idgen_ctx_init(&idctx); + /* bug-1420 follow-up: folding pid into apiserverId alone (4 bits, 16 + * buckets) at the jsub call site still collides too often across this + * harness's rapid, per-invocation-fresh xce_test processes (confirmed + * empirically -- a 10-round chaotic soak still produced same-second + * duplicate jobids after that fix landed). idgen_ctx_init() always + * resets sequence to 0, and MAKE_JOBID's sequence field alone has 20 + * bits (config.h N_SEQUENCE_BITS) of headroom -- two orders of + * magnitude more than the namespacing fields. Re-seed it here with a + * value that mixes pid (stable per process) with monotonic-clock + * nanoseconds (varies even across processes launched back-to-back in + * the same wall-clock second, unlike idgen's own second-resolution + * timestamp), masked to N_SEQUENCE_BITS so it can never bleed into the + * adjacent tag/timestamp bits MAKE_JOBID packs beside it. idgen()'s own + * ROUNDUP-to-N_ARRAY_IDX_BITS logic for array jobs still applies + * unchanged on top of this seed. */ + { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + idctx.sequence = ((uint32_t)getpid() ^ (uint32_t)ts.tv_nsec) + & (uint32_t)UMAX_BITS(N_SEQUENCE_BITS); + } + /* errexit default, modelled on shell `-e`/`+e`, keyed on the stdin mode: * batch (piped/redirected stdin) → `-e`: abort the read loop on the first * failing command and exit non-zero (a script wants fail-fast + verdict). diff --git a/src/controller/testsuite/test_apply_correctness_verification.c b/src/controller/testsuite/test_apply_correctness_verification.c index fa4b2afa..f7f901c2 100644 --- a/src/controller/testsuite/test_apply_correctness_verification.c +++ b/src/controller/testsuite/test_apply_correctness_verification.c @@ -310,6 +310,18 @@ static void test_checkpoint_tail_with_no_snapshot_is_apply_clean(void) } TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + /* No live apply pump runs in this cluster-less fixture, so + * last_applied_lsn stays 0 unless simulated -- persist_checkpoint + * now stamps checkpoint_lsn from last_applied_lsn, not current_lsn + * (checkpoint_lsn-vs-last_applied_lsn fix); simulate apply having + * caught up to both submits before the checkpoint runs (Phase 2's + * own from-scratch replay below is what actually exercises the + * physical-tip-vs-logical-cursor gap this test targets — it does + * not depend on this value, since the snapshot carrying it is + * deleted a few lines down). */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); + /* Real, complete checkpoint — marker fsynced AND snapshot written. * A genuine crash never reaches the snapshot write, but there is no * seam to stop persist_checkpoint() midway, so the snapshot is @@ -353,7 +365,17 @@ static void test_checkpoint_tail_with_no_snapshot_is_apply_clean(void) TEST_ASSERT_TRUE_MESSAGE(physical_tip > logical_cursor, "fixture sanity: the checkpoint marker must sit above the last " "logical entry for this test to exercise the round-2 bug"); - TEST_ASSERT_EQUAL_UINT64(marker_checkpoint_lsn, physical_tip + 1); + /* marker_checkpoint_lsn (Phase 1's checkpoint_lsn, now the LOGICAL + * applied cursor at checkpoint time -- 2, since only the two + * submits were simulated as applied) is no longer expected to equal + * physical_tip+1 (Phase 2's independently-reconstructed PHYSICAL + * marker position): that equality only held under the pre-fix + * checkpoint_lsn==current_lsn behavior this test's own docstring + * calls "the round-2 bug". checkpoint_lsn and the physical WAL tip + * legitimately diverging is the fix, not a regression -- Phase 2 + * never even sees Phase 1's checkpoint_lsn (the snapshot carrying + * it was deleted above), so there is no cross-phase LSN identity + * left to assert here. */ TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, persist_apply_verdict(p2), diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index f78b6600..0bdf977a 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -1381,6 +1381,47 @@ static void test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job " PASS\n"); } +/* bug-1419 layer 3: a jobid claimed by something OTHER than this node's own + * handle_jobsubmit call -- simulating the apply pump processing a + * WAL_JOB_SUBMIT entry received via replication/replay, never locally + * submitted -- must still block a same-jobid submit arriving through + * handle_jobsubmit. Before this fix, job_wal_commit_claim did not exist and + * apply_committed_entry() never registered an in-flight claim for an + * externally-sourced entry, so jobid_range_exists() (c->jobmgr, not yet + * applied) and the in-flight registry (only ever populated by + * handle_jobsubmit itself) both missed it -- exactly the cross-primary + * retry gap a live chaotic soak reproduced (waldump showed two committed + * WAL_JOB_SUBMIT entries for one jobid, same term, no fresh idgen collision + * -- the same generated jobid resent by the client after a connection drop, + * landing on a newly elected primary that never saw the original). */ +static void test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit(void) +{ + const uint64_t jobid = 9160; + int rc; + + cbrp_setup(2000); + cbrp_arrange_jobsubmit(jobid, 1); + + /* Simulate: the apply pump is mid-dispatch on a WAL_JOB_SUBMIT entry for + * this jobid that this node never itself submitted (e.g. replicated + * from a prior primary that committed it before dying). */ + TEST_ASSERT_TRUE_MESSAGE(job_wal_commit_claim(jobid), + "fixture sanity: claiming a fresh jobid must succeed"); + + rc = handle_jobsubmit(&g_hc, &g_req_ep, &g_jobsub_req); + + job_wal_commit_release(jobid); + cbrp_teardown(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(-EEXIST, rc, + "bug-1419 layer 3: a submit for a jobid claimed by a foreign " + "(non-self-originated) commit must be rejected -EEXIST, closing " + "the cross-primary retry gap"); + + printf("test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit:" + " PASS\n"); +} + /* The explicit-FAILED (non-timeout) sibling of the repeated-timeout test * above: repeats the truncation-forced FAILED path * CBRP_JOBSUBMIT_REPEAT_N times in the same fixture, proving the @@ -3347,6 +3388,7 @@ void test_commit_barrier_reply(void) RUN_TEST(test_jobsubmit_repeated_timeout_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_resubmit_before_resolution_rejected_no_dup_wal_entry); RUN_TEST(test_jobsubmit_resubmit_after_full_resolution_rejected_no_second_job); + RUN_TEST(test_jobsubmit_rejected_when_jobid_claimed_by_foreign_commit); RUN_TEST(test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements); /* handle_jobsubmit success path via the ordered apply pump diff --git a/src/controller/testsuite/test_e2e_checkpoint.c b/src/controller/testsuite/test_e2e_checkpoint.c index da37e5ca..b46fc19b 100644 --- a/src/controller/testsuite/test_e2e_checkpoint.c +++ b/src/controller/testsuite/test_e2e_checkpoint.c @@ -104,6 +104,14 @@ static void test_checkpoint_recovery_lsn_continuity(void) persist_wal_async(p, WAL_JOB_UPDATE, "x", 1)); } + /* No live apply pump runs in this cluster-less fixture, so + * last_applied_lsn stays 0 unless simulated -- persist_checkpoint now + * stamps checkpoint_lsn from last_applied_lsn, not current_lsn + * (checkpoint_lsn-vs-last_applied_lsn fix); simulate apply having + * caught up to what was appended before the checkpoint runs. */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); + /* Checkpoint */ TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); uint64_t ckpt_lsn = p->checkpoint_lsn; diff --git a/src/controller/testsuite/test_e2e_gc.c b/src/controller/testsuite/test_e2e_gc.c index c89c039d..311bb330 100644 --- a/src/controller/testsuite/test_e2e_gc.c +++ b/src/controller/testsuite/test_e2e_gc.c @@ -126,6 +126,14 @@ static void test_checkpoint_gc_removes_expired_entity(void) TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "gc-test", 7)); + /* No live apply pump runs in this cluster-less fixture, so + * last_applied_lsn stays 0 unless simulated -- persist_checkpoint now + * stamps checkpoint_lsn from last_applied_lsn, not current_lsn + * (checkpoint_lsn-vs-last_applied_lsn fix); simulate apply having + * caught up to what was appended before the checkpoint runs. */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); + /* Trigger checkpoint — must succeed and leave a snapshot */ TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); TEST_ASSERT_TRUE(p->checkpoint_lsn > 0); diff --git a/src/controller/testsuite/test_persist_recovery.c b/src/controller/testsuite/test_persist_recovery.c index 4c8d23d2..076ed145 100644 --- a/src/controller/testsuite/test_persist_recovery.c +++ b/src/controller/testsuite/test_persist_recovery.c @@ -91,6 +91,14 @@ static void test_recover_loads_checkpoint_lsn(void) TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1)); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "y", 1)); + + /* No live apply pump runs in this cluster-less fixture, so + * last_applied_lsn stays 0 unless simulated -- persist_checkpoint + * now stamps checkpoint_lsn from last_applied_lsn, not current_lsn + * (checkpoint_lsn-vs-last_applied_lsn fix); simulate apply having + * caught up to what was appended. */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); teardown_persist(p);