forked from lei.chen/myxce
feat(raft-apply-decoupling): resolve pending continuations on demotion/truncation (task 1.4)
commit_resolve_pending_above(ce, lsn_floor, rc) fails every registered continuation for lsn > lsn_floor, reusing commit_resolve()'s existing exactly-once/idempotent guarantee (no parallel bookkeeping path). Truncation: a new on_truncate/truncate_cb_ctx callback on persist_t (same inversion-of-control shape as the existing on_durable_advance hook) fires from persist_truncate_wal_above after releasing p->lock, only when entries were actually removed. This one wrapper covers all three real truncation call sites: raft_cb_become_leader's post-election truncate, the library's on_truncate_from conflict-reconcile, and ha_demotion_convergence_fn's post-demotion truncate. Demotion: raft_cb_become_follower calls commit_resolve_pending_above(&c->commit, 0, -XCE_ERR_STANDBY) -- lsn_floor=0 fails everything, since a demoted node can never authoritatively confirm anything still pending (CP posture). Without this, a demoted or truncated node would leave any pending async command permanently unresolved -- the requester would simply never get a reply, worse than today's bounded-timeout behavior. Also closes bug-1407 (pre-existing test_config_repl flake, unrelated to this feature) -- confirmed fixed as a side effect of prior tasks in this series, 70/70 passing across 3 independent runs.
This commit is contained in:
parent
530a02bda3
commit
9b4b805a27
9 changed files with 748 additions and 301 deletions
|
|
@ -79,6 +79,12 @@ static void commit_resolve_pending(commit_engine_t* ce);
|
|||
* commit_sweep_scan_locked, the locked-scan helper it drives). */
|
||||
static void* commit_sweep_worker(void* arg);
|
||||
|
||||
/* Forward declaration: raft-apply-decoupling task 1.4. commit_engine_init
|
||||
* (below) registers this as persist's truncation-wake callback; it is
|
||||
* defined in the Continuation Resolution section further down this file
|
||||
* (alongside commit_resolve_pending_above, the helper it delegates to). */
|
||||
static void commit_on_truncate(void* ctx, uint64_t cutoff_lsn);
|
||||
|
||||
/* ============================================================
|
||||
* Internal helpers
|
||||
* ============================================================ */
|
||||
|
|
@ -387,6 +393,14 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp
|
|||
* the reverse) dependency direction. */
|
||||
persist_set_durable_cb(p, commit_on_durable_advance, ce);
|
||||
|
||||
/* raft-apply-decoupling task 1.4: register the truncation wake alongside
|
||||
* the durable-advance wake -- same dependency direction (persist stores
|
||||
* an opaque callback; commit.c is never a dependency of persist.c). Fired
|
||||
* by persist_truncate_wal_above after it actually removes entries above
|
||||
* a new tail, so any continuation registered above that tail (which can
|
||||
* never commit -- its entry no longer exists) resolves FAILED promptly. */
|
||||
persist_set_truncate_cb(p, commit_on_truncate, ce);
|
||||
|
||||
/* raft-apply-decoupling task 1.3: start the dedicated timeout-sweep
|
||||
* thread (see file header for why this is a private thread rather than
|
||||
* the shared controller delaymgr). A creation failure is treated as a
|
||||
|
|
@ -402,6 +416,7 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp
|
|||
int prc = pthread_create(&ce->sweep_thread, NULL, commit_sweep_worker, ce);
|
||||
if (prc != 0) {
|
||||
persist_set_durable_cb(p, NULL, NULL);
|
||||
persist_set_truncate_cb(p, NULL, NULL);
|
||||
free(ce->continuations);
|
||||
ce->continuations = NULL;
|
||||
ce->n_continuations = 0;
|
||||
|
|
@ -453,6 +468,7 @@ void commit_engine_fini(commit_engine_t* ce)
|
|||
* cheap. After this, persist holds no pointer to ce. */
|
||||
if (ce->p != NULL) {
|
||||
persist_set_durable_cb(ce->p, NULL, NULL);
|
||||
persist_set_truncate_cb(ce->p, NULL, NULL);
|
||||
}
|
||||
free(ce->continuations);
|
||||
ce->continuations = NULL;
|
||||
|
|
@ -801,6 +817,65 @@ static void commit_resolve_pending(commit_engine_t* ce)
|
|||
}
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.4: the FAILED-outcome sibling of
|
||||
* commit_resolve_pending -- instead of scanning for majority-satisfied
|
||||
* entries, resolves every registered continuation whose lsn > lsn_floor as
|
||||
* FAILED with rc. See commit.h for the full contract and the two intended
|
||||
* callers (log-tail truncation, demotion). Caller must NOT hold ce->lock. */
|
||||
void commit_resolve_pending_above(commit_engine_t* ce, uint64_t lsn_floor,
|
||||
int rc)
|
||||
{
|
||||
int n = 0;
|
||||
|
||||
if (ce == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
uint64_t failed[ce->n_continuations]; /* VLA: bounded by n_vcpus -- see
|
||||
* the commit.h field-comment
|
||||
* sizing invariant */
|
||||
|
||||
pthread_mutex_lock(&ce->lock);
|
||||
for (int i = 0; i < ce->n_continuations; i++) {
|
||||
uint64_t lsn = ce->continuations[i].lsn;
|
||||
|
||||
if (lsn != 0 && lsn > lsn_floor) {
|
||||
failed[n++] = lsn;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&ce->lock);
|
||||
|
||||
for (int i = 0; i < n; i++) {
|
||||
commit_resolve(ce, failed[i], COMMIT_OUTCOME_FAILED, rc);
|
||||
}
|
||||
}
|
||||
|
||||
/* Truncation wake (raft-apply-decoupling task 1.4): registered with persist
|
||||
* via persist_set_truncate_cb (commit_engine_init) and invoked by
|
||||
* persist_truncate_wal_above AFTER it actually removes entries above the new
|
||||
* tail (cutoff_lsn) and AFTER it releases p->lock -- persist only fires this
|
||||
* when a truncation genuinely happened (current_lsn strictly decreased); a
|
||||
* no-op truncate never calls this. Every continuation registered above the
|
||||
* new tail can never commit (its entry no longer exists in the log) --
|
||||
* resolve it FAILED now rather than leaving it for the timeout-sweep thread
|
||||
* to eventually time out. -ENOENT mirrors the truncation-outcome vocabulary
|
||||
* already exercised in test_commit_resolve.c (task 1.2).
|
||||
*
|
||||
* LOCK ORDERING: mirrors commit_on_durable_advance -- runs with NO persist
|
||||
* lock held, takes only ce->lock (via commit_resolve_pending_above /
|
||||
* commit_resolve). Same cycle-free p->lock -> ce->lock edge; the commit side
|
||||
* never takes p->lock while holding ce->lock. ctx is the commit_engine_t
|
||||
* registered in commit_engine_init. */
|
||||
static void commit_on_truncate(void* ctx, uint64_t cutoff_lsn)
|
||||
{
|
||||
commit_engine_t* ce = ctx;
|
||||
|
||||
if (ce == NULL || ce->p == NULL) {
|
||||
return;
|
||||
}
|
||||
commit_resolve_pending_above(ce, cutoff_lsn, -ENOENT);
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
* Timeout sweep (raft-apply-decoupling task 1.3)
|
||||
* ============================================================ */
|
||||
|
|
|
|||
|
|
@ -1378,6 +1378,21 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
|
|||
(hook_fn_t)persist_worker_hook_fun);
|
||||
ha->hooks_registered = 0;
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.4 (Req 1.3, 1.4, 9.3, 9.4): a demoted
|
||||
* PRIMARY can never advance committed_lsn again for anything it had
|
||||
* pending -- resolve every currently registered commit continuation
|
||||
* as FAILED now, synchronously with the rest of this "stop being
|
||||
* primary" sequence, rather than leaving a requester parked until
|
||||
* the commit engine's own timeout-sweep thread (task 1.3) eventually
|
||||
* times it out. lsn_floor=0 resolves ALL pending continuations (lsn
|
||||
* 0 is never a real registration -- the free-slot sentinel). Uses
|
||||
* the same -XCE_ERR_STANDBY vocabulary the HA dispatch gate already
|
||||
* uses to reject writes to a non-primary. Reachable only when this
|
||||
* node was already PRIMARY, which means cluster_init_phase2 (and
|
||||
* therefore commit_engine_init) already ran, so c->commit is always
|
||||
* initialized here. */
|
||||
commit_resolve_pending_above(&c->commit, 0, -XCE_ERR_STANDBY);
|
||||
}
|
||||
|
||||
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY,
|
||||
|
|
|
|||
|
|
@ -179,6 +179,32 @@ int commit_submit_async(commit_engine_t* ce, uint64_t lsn,
|
|||
void commit_resolve(commit_engine_t* ce, uint64_t lsn, commit_outcome_t outcome,
|
||||
int rc);
|
||||
|
||||
/* raft-apply-decoupling task 1.4: resolve every currently pending registered
|
||||
* continuation whose lsn > lsn_floor as FAILED with rc, in one bounded pass --
|
||||
* the FAILED-outcome sibling of commit_resolve's internal majority-ack scan.
|
||||
* Two intended callers:
|
||||
* - log-tail truncation: lsn_floor = the new tail (cutoff_lsn) AFTER a
|
||||
* truncation actually removed entries above it. Those entries no longer
|
||||
* exist in the log and can never commit. A continuation registered AT OR
|
||||
* BELOW the new tail is UNAFFECTED and stays pending -- its entry is
|
||||
* still in the log and may still reach majority.
|
||||
* - demotion (loss of leadership): lsn_floor = 0. A demoted node can never
|
||||
* advance committed_lsn again for anything it had pending, so every
|
||||
* currently registered continuation fails, regardless of its lsn. lsn ==
|
||||
* 0 is never a real registration (the free-slot sentinel -- see
|
||||
* commit_continuation_t), so lsn_floor = 0 correctly means "all".
|
||||
* Snapshot-then-resolve, mirroring commit_resolve_pending's own pattern:
|
||||
* takes a bounded snapshot of matching lsns under ce->lock, then resolves
|
||||
* each from an unlocked context via the self-locking commit_resolve()
|
||||
* (dispatch-after-unlock ordering rule, design §Commit Engine Resolve).
|
||||
* Idempotent: a continuation already resolved by a concurrent caller (e.g.
|
||||
* the timeout-sweep thread winning a race) is a silent no-op via
|
||||
* commit_resolve's own exactly-once guarantee -- no continuation is ever
|
||||
* double-resolved. Safe to call from any thread that does NOT already hold
|
||||
* ce->lock. */
|
||||
void commit_resolve_pending_above(commit_engine_t* ce, uint64_t lsn_floor,
|
||||
int rc);
|
||||
|
||||
/* iomgr path: record standby ACK, evaluate majority, advance committed_lsn
|
||||
* contiguously, wake satisfied waiters. Never blocks. */
|
||||
void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn);
|
||||
|
|
|
|||
|
|
@ -604,6 +604,12 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
|
|||
p->on_durable_advance = NULL;
|
||||
p->durable_cb_ctx = NULL;
|
||||
|
||||
/* Truncation wake hook starts unregistered (raft-apply-decoupling task
|
||||
* 1.4): the commit engine registers it in commit_engine_init. Until
|
||||
* then every truncation is a no-op wake. */
|
||||
p->on_truncate = NULL;
|
||||
p->truncate_cb_ctx = NULL;
|
||||
|
||||
/* Per-LSN term index starts empty (ha-term-safety 1.3, R1.5): no entries
|
||||
* recorded yet. Grown on demand by record_entry_term during replay/append;
|
||||
* freed in persist_fini. */
|
||||
|
|
@ -1279,6 +1285,17 @@ void persist_set_durable_cb(persist_t* p, void (*fn)(void* ctx), void* ctx)
|
|||
p->on_durable_advance = fn;
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.4: registration counterpart of
|
||||
* persist_set_durable_cb -- see persist.h for the full contract. */
|
||||
void persist_set_truncate_cb(persist_t* p, void (*fn)(void* ctx, uint64_t cutoff_lsn),
|
||||
void* ctx)
|
||||
{
|
||||
if (!p)
|
||||
return;
|
||||
p->truncate_cb_ctx = ctx;
|
||||
p->on_truncate = fn;
|
||||
}
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
void persist_test_set_force_fsync_fail(persist_t* p, int fail)
|
||||
{
|
||||
|
|
@ -3681,8 +3698,25 @@ static void clamp_durable_lsn_locked(persist_t* p, uint64_t tip)
|
|||
int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
|
||||
{
|
||||
pthread_mutex_lock(&p->lock);
|
||||
int rc = truncate_wal_above_locked(p, cutoff_lsn);
|
||||
uint64_t before = p->current_lsn;
|
||||
int rc = truncate_wal_above_locked(p, cutoff_lsn);
|
||||
uint64_t after = p->current_lsn;
|
||||
pthread_mutex_unlock(&p->lock);
|
||||
|
||||
/* raft-apply-decoupling task 1.4: fire the truncation wake only when a
|
||||
* truncation actually removed entries above cutoff_lsn (current_lsn
|
||||
* strictly decreased -- both truncate_wal_above_locked branches set
|
||||
* current_lsn = cutoff_lsn + 1 on a real truncation) -- mirrors the
|
||||
* durable-advance wake's own before/after guard (see fsync_worker
|
||||
* above) so a no-op truncate (nothing above cutoff_lsn, or an I/O
|
||||
* error before current_lsn is touched) never spuriously fails a
|
||||
* continuation whose entry is still safely in the log. Called AFTER
|
||||
* releasing p->lock: the callback (the commit engine) takes only
|
||||
* ce->lock, preserving the cycle-free p->lock -> ce->lock edge already
|
||||
* established by the durable-advance wake. */
|
||||
if (rc == 0 && after < before && p->on_truncate != NULL) {
|
||||
p->on_truncate(p->truncate_cb_ctx, cutoff_lsn);
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@
|
|||
#include "persist.h" /* persist_t, persist_truncate_wal_above */
|
||||
#include "repl.h" /* repl_add_standby / repl_set_commit_engine */
|
||||
#include "xce/cluster_def.h" /* full cluster_t layout */
|
||||
#include "xce/errnox.h" /* XCE_ERR_STANDBY (task 1.4 resolve rc) */
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
|
||||
|
|
@ -271,6 +272,27 @@ static int dc_run_pending_task(delaymgr_t *mgr)
|
|||
return rc;
|
||||
}
|
||||
|
||||
/* ── Continuation recorder (raft-apply-decoupling task 1.4) ───────────────
|
||||
* Minimal local recorder for commit_submit_async continuations, mirroring
|
||||
* test_commit_resolve.c's continuation_recorder_t/recorder_fn pattern (kept
|
||||
* as a separate file-static copy -- no cross-TU sharing of test statics). */
|
||||
typedef struct {
|
||||
_Atomic int calls;
|
||||
uint64_t last_lsn;
|
||||
commit_outcome_t last_outcome;
|
||||
int last_rc;
|
||||
} dc_continuation_recorder_t;
|
||||
|
||||
static void dc_recorder_fn(void *ctx, uint64_t lsn, commit_outcome_t outcome,
|
||||
int rc)
|
||||
{
|
||||
dc_continuation_recorder_t *r = (dc_continuation_recorder_t *)ctx;
|
||||
r->last_lsn = lsn;
|
||||
r->last_outcome = outcome;
|
||||
r->last_rc = rc;
|
||||
atomic_fetch_add(&r->calls, 1);
|
||||
}
|
||||
|
||||
/* ── Tests ──────────────────────────────────────────────────────────────── */
|
||||
|
||||
/* 1.1: demotion from PRIMARY disables the persistence hooks synchronously in
|
||||
|
|
@ -376,6 +398,66 @@ static void test_self_restart_failure_invokes_fatal_hook(void)
|
|||
printf("test_self_restart_failure_invokes_fatal_hook: checked\n");
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.4 (Req 1.3/1.4, 9.3/9.4): demotion from
|
||||
* PRIMARY must resolve every currently registered commit continuation as
|
||||
* FAILED -- otherwise a client/fact waiting on a pending commit is left
|
||||
* permanently unresolved once this node can never commit it. Registers TWO
|
||||
* continuations at different uncommitted LSNs to prove the resolve loop
|
||||
* handles the whole pending set, not just a single slot.
|
||||
* RED: before task 1.4, raft_cb_become_follower's was-PRIMARY branch never
|
||||
* calls into the commit engine, so both continuations stay pending forever
|
||||
* (calls == 0). */
|
||||
static void test_demotion_from_primary_resolves_pending_continuations_failed(void)
|
||||
{
|
||||
dc_setup("resolve", /*n_peers=*/3, /*n_standbys=*/2,
|
||||
/*appended_lsn=*/9, /*n_vcpus=*/8);
|
||||
atomic_store(&g_dc.persist.committed_lsn, 4);
|
||||
|
||||
dc_continuation_recorder_t low = {0};
|
||||
dc_continuation_recorder_t high = {0};
|
||||
|
||||
/* Both lsns sit above committed_lsn=4 (majority is unreachable in this
|
||||
* fixture -- standbys never ACK -- so both stay genuinely pending, not
|
||||
* already resolved COMMITTED by the time we demote). */
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
commit_submit_async(&g_dc.commit, 5, dc_recorder_fn, &low));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
commit_submit_async(&g_dc.commit, 9, dc_recorder_fn, &high));
|
||||
|
||||
/* Act: demotion edge from a LIVE PRIMARY. This fixture never wires a
|
||||
* vtaskmgr pool (commit_set_pool is never called), so commit_resolve's
|
||||
* dispatch falls back to the inline fallback -- the recorder runs
|
||||
* synchronously on this thread, no polling required. */
|
||||
raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&low.calls),
|
||||
"demotion must resolve every pending continuation, including one "
|
||||
"for an lsn that could still reach majority under normal operation "
|
||||
"(Req 1.3/1.4/9.4)");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
|
||||
(int)low.last_outcome, "demotion must resolve as FAILED, not COMMITTED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_STANDBY, low.last_rc,
|
||||
"demotion must use the existing -XCE_ERR_STANDBY vocabulary (the "
|
||||
"same code the HA dispatch gate already uses to reject writes to a "
|
||||
"non-primary)");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, low.last_lsn,
|
||||
"the recorder must observe the lsn it was registered for");
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&high.calls),
|
||||
"demotion must resolve the uncommitted-tail continuation as FAILED "
|
||||
"exactly once");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
|
||||
(int)high.last_outcome, "demotion must resolve as FAILED, not COMMITTED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_STANDBY, high.last_rc,
|
||||
"demotion must use the existing -XCE_ERR_STANDBY vocabulary");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(9, high.last_lsn,
|
||||
"the recorder must observe the lsn it was registered for");
|
||||
|
||||
dc_teardown();
|
||||
printf("test_demotion_from_primary_resolves_pending_continuations_failed: "
|
||||
"checked\n");
|
||||
}
|
||||
|
||||
/* Regression: demotion from STANDBY must NOT truncate, NOT restart, NOT disable
|
||||
* hooks beyond today's behavior — byte-equivalent to the pre-spec path. A
|
||||
* STANDBY has no registered hooks and no convergence work. */
|
||||
|
|
@ -814,6 +896,7 @@ void test_demotion_convergence(void)
|
|||
printf("\n>> demotion convergence + self-restart tests (RED, task 2.1)\n\n");
|
||||
|
||||
RUN_TEST(test_demotion_from_primary_disables_hooks);
|
||||
RUN_TEST(test_demotion_from_primary_resolves_pending_continuations_failed);
|
||||
RUN_TEST(test_demotion_schedules_convergence_truncate_and_restart);
|
||||
RUN_TEST(test_self_restart_failure_invokes_fatal_hook);
|
||||
RUN_TEST(test_demotion_from_standby_no_truncation_no_restart);
|
||||
|
|
|
|||
|
|
@ -21,6 +21,8 @@
|
|||
#include "persist.h"
|
||||
#include "wal_types.h"
|
||||
#include "config.h"
|
||||
#include "commit.h" /* raft-apply-decoupling task 1.4: commit_engine_t,
|
||||
* commit_submit_async, commit_outcome_t */
|
||||
|
||||
/* ── Forward declaration of function-under-test (not yet implemented) ─────── */
|
||||
extern int persist_truncate_wal_above(persist_t *p, uint64_t cutoff_lsn);
|
||||
|
|
@ -230,6 +232,149 @@ static void test_truncate_noop_at_current_lsn(void)
|
|||
printf("test_truncate_noop_at_current_lsn: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test: truncation resolves invalidated commit continuations (task 1.4) ── */
|
||||
|
||||
/* Minimal local recorder for commit_submit_async continuations -- mirrors
|
||||
* test_commit_resolve.c's continuation_recorder_t/recorder_fn (kept as a
|
||||
* separate file-static copy; no cross-TU sharing of test statics). */
|
||||
typedef struct {
|
||||
_Atomic int calls;
|
||||
uint64_t last_lsn;
|
||||
commit_outcome_t last_outcome;
|
||||
int last_rc;
|
||||
} trunc_continuation_recorder_t;
|
||||
|
||||
static void trunc_recorder_fn(void *ctx, uint64_t lsn, commit_outcome_t outcome,
|
||||
int rc)
|
||||
{
|
||||
trunc_continuation_recorder_t *r = (trunc_continuation_recorder_t *)ctx;
|
||||
r->last_lsn = lsn;
|
||||
r->last_outcome = outcome;
|
||||
r->last_rc = rc;
|
||||
atomic_fetch_add(&r->calls, 1);
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.4 (Req 1.3/1.4, 9.3/9.4): a real truncation
|
||||
* through persist_truncate_wal_above must resolve every commit continuation
|
||||
* registered ABOVE the new tail as FAILED -- those entries no longer exist
|
||||
* in the log and can never commit -- while a continuation registered AT OR
|
||||
* BELOW the new tail is left untouched (its entry is still safely in the
|
||||
* log and may still reach majority).
|
||||
*
|
||||
* Uses n_peers=3 with NO standbys registered so majority is permanently
|
||||
* unreachable (majority_candidate_locked's zero-fill keeps committed_lsn at
|
||||
* 0 regardless of local append/durable progress) -- every appended entry
|
||||
* stays genuinely pending, giving a stable LSN range to register
|
||||
* continuations against without racing the commit engine's own majority
|
||||
* evaluation.
|
||||
*
|
||||
* RED: before task 1.4, persist_truncate_wal_above never calls into the
|
||||
* commit engine, so all three continuations stay pending forever (calls ==
|
||||
* 0 for all). */
|
||||
static void test_truncate_above_resolves_pending_continuations_as_failed(void)
|
||||
{
|
||||
char tmpdir[] = "/tmp/xce-trunc-resolve-XXXXXX";
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
|
||||
|
||||
persist_t *p = make_fresh_persist(tmpdir);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
|
||||
commit_engine_t ce;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8));
|
||||
|
||||
int dummy = 0;
|
||||
for (int i = 0; i < 5; i++)
|
||||
persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(6, p->current_lsn,
|
||||
"fixture: 5 entries appended, current_lsn must be 6");
|
||||
|
||||
trunc_continuation_recorder_t below = {0};
|
||||
trunc_continuation_recorder_t above1 = {0};
|
||||
trunc_continuation_recorder_t above2 = {0};
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &below));
|
||||
TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 4, trunc_recorder_fn, &above1));
|
||||
TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 5, trunc_recorder_fn, &above2));
|
||||
|
||||
/* Act: truncate above lsn=3 -- entries 4 and 5 are removed. */
|
||||
int rc = persist_truncate_wal_above(p, 3);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, p->current_lsn,
|
||||
"current_lsn must collapse to cutoff_lsn + 1");
|
||||
|
||||
/* Assert: continuations ABOVE the new tail resolve FAILED exactly once.
|
||||
* No vtaskmgr pool is wired on this engine (commit_set_pool never
|
||||
* called), so commit_resolve's dispatch falls back to the inline
|
||||
* fallback -- synchronous with the persist_truncate_wal_above call
|
||||
* above, no polling required. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above1.calls),
|
||||
"a continuation above the new tail must resolve on truncation");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
|
||||
(int)above1.last_outcome, "truncation must resolve as FAILED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above1.last_rc,
|
||||
"truncation must use the existing -ENOENT vocabulary already "
|
||||
"exercised in test_commit_resolve.c (task 1.2)");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, above1.last_lsn,
|
||||
"the recorder must observe the lsn it was registered for");
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above2.calls),
|
||||
"a second continuation above the new tail must also resolve");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
|
||||
(int)above2.last_outcome, "truncation must resolve as FAILED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above2.last_rc,
|
||||
"truncation must use the existing -ENOENT vocabulary");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, above2.last_lsn,
|
||||
"the recorder must observe the lsn it was registered for");
|
||||
|
||||
/* Assert: a continuation AT OR BELOW the new tail is left untouched --
|
||||
* its entry is still in the log and may still reach majority. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&below.calls),
|
||||
"a continuation at/below the new tail must NOT be resolved by "
|
||||
"truncation -- only entries strictly above the new tail were "
|
||||
"removed from the log");
|
||||
|
||||
commit_engine_fini(&ce);
|
||||
teardown_persist(p);
|
||||
printf("test_truncate_above_resolves_pending_continuations_as_failed: PASS\n");
|
||||
}
|
||||
|
||||
/* Regression: a no-op truncate (cutoff at/above the current tail) must NOT
|
||||
* fire the truncation wake -- a pending continuation for any lsn must stay
|
||||
* untouched, since nothing was actually removed from the log. */
|
||||
static void test_truncate_noop_does_not_resolve_pending_continuations(void)
|
||||
{
|
||||
char tmpdir[] = "/tmp/xce-trunc-resolve-noop-XXXXXX";
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
|
||||
|
||||
persist_t *p = make_fresh_persist(tmpdir);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
|
||||
commit_engine_t ce;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8));
|
||||
|
||||
int dummy = 0;
|
||||
persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
|
||||
persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
|
||||
|
||||
trunc_continuation_recorder_t rec = {0};
|
||||
TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &rec));
|
||||
|
||||
/* Truncate AT the current boundary -- a no-op (nothing above cutoff). */
|
||||
uint64_t cutoff = p->current_lsn - 1;
|
||||
int rc = persist_truncate_wal_above(p, cutoff);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&rec.calls),
|
||||
"a no-op truncate (nothing above cutoff_lsn) must not resolve any "
|
||||
"pending continuation");
|
||||
|
||||
commit_engine_fini(&ce);
|
||||
teardown_persist(p);
|
||||
printf("test_truncate_noop_does_not_resolve_pending_continuations: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ────────────────────────────────────────────────── */
|
||||
|
||||
void test_persist_truncate(void)
|
||||
|
|
@ -240,5 +385,7 @@ void test_persist_truncate(void)
|
|||
RUN_TEST(test_truncate_idempotent);
|
||||
RUN_TEST(test_truncate_updates_applied_lsn);
|
||||
RUN_TEST(test_truncate_noop_at_current_lsn);
|
||||
RUN_TEST(test_truncate_above_resolves_pending_continuations_as_failed);
|
||||
RUN_TEST(test_truncate_noop_does_not_resolve_pending_continuations);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -340,6 +340,22 @@ struct persist_s {
|
|||
void (*on_durable_advance)(void* ctx);
|
||||
void* durable_cb_ctx;
|
||||
|
||||
/* Truncation wake hook (raft-apply-decoupling task 1.4). Persistence owns
|
||||
* this slot; the commit engine REGISTERS a callback (ctx = the
|
||||
* commit_engine_t) via persist_set_truncate_cb, mirroring
|
||||
* on_durable_advance's registration pattern and dependency direction
|
||||
* (persist only stores+calls an opaque function pointer; it never
|
||||
* depends on commit.c). persist_truncate_wal_above invokes it AFTER it
|
||||
* actually removes entries above the new tail and AFTER releasing
|
||||
* p->lock, so any commit continuation registered for an LSN above the
|
||||
* new tail (which can never commit -- its entry no longer exists) is
|
||||
* resolved FAILED promptly instead of waiting out the commit engine's
|
||||
* timeout sweep. NOT invoked on a no-op truncate (nothing above
|
||||
* cutoff_lsn -- current_lsn unchanged). NULL until registered (a
|
||||
* truncate is then a silent no-op wake). */
|
||||
void (*on_truncate)(void* ctx, uint64_t cutoff_lsn);
|
||||
void* truncate_cb_ctx;
|
||||
|
||||
/* Checkpoint (stubs for now — wired in Task 10) */
|
||||
atomic_int checkpoint_requested;
|
||||
pthread_cond_t checkpoint_done;
|
||||
|
|
@ -590,6 +606,16 @@ void persist_clear_recovery_unrecoverable(persist_t* p);
|
|||
* unregister. Safe to call before persist_start (no flusher running yet). */
|
||||
void persist_set_durable_cb(persist_t* p, void (*fn)(void* ctx), void* ctx);
|
||||
|
||||
/* Register a truncation wake callback (raft-apply-decoupling task 1.4).
|
||||
* persist_truncate_wal_above calls fn(ctx, cutoff_lsn) after it actually
|
||||
* truncates entries above cutoff_lsn (the new log tail) and after releasing
|
||||
* p->lock -- fn MUST NOT take p->lock (same rule as persist_set_durable_cb;
|
||||
* the commit engine's callback takes only ce->lock, keeping the lock order
|
||||
* p->lock → ce->lock cycle-free). Not called for a no-op truncate (nothing
|
||||
* above cutoff_lsn). Pass fn = NULL to unregister. */
|
||||
void persist_set_truncate_cb(persist_t* p, void (*fn)(void* ctx, uint64_t cutoff_lsn),
|
||||
void* ctx);
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
/* ============================================================
|
||||
* Durability-verification test seam (ha-commit-durability Req 3.2, 3.3)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue