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

feat(ha-convergence-recovery): clamped rejoin reconciliation + intake trigger (task 3.2)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-14 04:53:08 +00:00
parent 5521720b0c
commit 63c12461a5
3 changed files with 168 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

152
src/controller/ha_convergence.c
View file

@ -212,23 +212,137 @@ int ha_schedule_demotion_convergence(cluster_t *c)
}
/*
* ha_rejoin_reconcile — RED scaffold (ha-convergence-recovery task 3.1).
* ha_rejoin_reconcile_ctx_t — heap context for the one-shot rejoin delaymgr task.
* Carries the clamped cutoff computed at trigger time (under the same lock that
* read the tip) plus the prior tip for the WARN range, so the deferred body does
* not re-read state that may have advanced between schedule and run.
*/
typedef struct {
cluster_t *c;
uint64_t cutoff; /* truncate above this LSN (clamped, Req 2.2) */
uint64_t prior_tip; /* local tip at trigger time (WARN range, Req 2.1) */
} ha_rejoin_reconcile_ctx_t;
/*
* ha_rejoin_reconcile_fn — delaymgr task body (runs OFF the CtrlHelloReply intake
* thread, which may be iomgr; D5/D1). Truncate above the clamped cutoff (Req 2.1,
* 2.2), WARN with the discarded range, then ha_self_restart (Req 2.1). On
* truncation failure ERROR then restart anyway — boot recovery + a subsequent
* rejoin converge the untruncated case (crash-safe by design, mirrors demotion).
*/
int ha_rejoin_reconcile_fn(void *arg)
{
ha_rejoin_reconcile_ctx_t *ctx = (ha_rejoin_reconcile_ctx_t *)arg;
if (ctx == NULL || ctx->c == NULL) {
free(ctx);
return -EINVAL;
}
cluster_t *c = ctx->c;
uint64_t cutoff = ctx->cutoff;
uint64_t prior_tip = ctx->prior_tip;
int trc = persist_truncate_wal_above(&c->persist, cutoff);
if (trc < 0) {
ERROR("HA rejoin reconcile: truncate above cutoff=%" PRIu64
" failed: %d — restarting anyway (boot recovery + next rejoin will "
"converge)", cutoff, trc);
} else {
/* Req 2.1: log the discarded range — (cutoff, prior tip]. */
WARN("HA rejoin reconcile: local tip=%" PRIu64 " exceeded the cluster "
"committed boundary; truncated WAL above cutoff=%" PRIu64
" (clamped to the locally committed prefix), restarting once to "
"rejoin from converged state", prior_tip, cutoff);
}
/* Free the task ctx before restart (production exec never returns). */
free(ctx);
/* Req 2.1: rebuild in-memory state through boot recovery. */
return ha_self_restart(c);
}
/*
* ha_rejoin_reconcile — converge a rejoining node whose local WAL log runs ahead
* of the cluster's committed boundary (ha-convergence-recovery Req 2.1–2.3, D5).
*
* TASK 3.2 REPLACES THIS with the real reconciliation: when the rejoining node's
* local WAL tip exceeds the primary-reported committed boundary, truncate above
* max(boundary, local committed_lsn) (clamp, Req 2.2), WARN the range, set the
* rejoin_reconciled once-guard (Req 2.1), and ha_self_restart (Req 2.1) —
* deferred to a delaymgr task if the intake runs on iomgr. Otherwise return 0
* with no side effects (Req 2.3).
* Called from the CtrlHelloReply intake (cluster_apply_ctrlhello_reply) after the
* primary-reported committed boundary is stored — the moment the rejoining node
* learns where the cluster has committed.
*
* This stub does NOTHING (no truncation, no restart, no once-guard set) so the
* task 3.1 RED tests FAIL until 3.2 lands the contract above.
* Decision (design "Rejoin Reconciliation", research D5):
* - local_tip = persist.current_lsn - 1.
* - If local_tip <= primary_committed_lsn → return 0, NO side effects (Req 2.3).
* - If the rejoin_reconciled once-guard is already set → return 0; never restart
* a second time in one process lifetime (Req 2.1 defensive guard; "once" is
* also structural — after restart the tip equals the cutoff, so the condition
* cannot re-fire for the same divergence).
* - Else: cutoff = max(primary_committed_lsn, local committed_lsn) — the CLAMP
* (Req 2.2): never truncate at or below the locally known committed prefix.
* Set the once-guard, then defer truncation + restart to a delaymgr one-shot
* (the intake may run on iomgr, so file I/O / exec must leave it — D5/D1).
*
* Returns 0 on the no-action paths and on a successfully scheduled reconcile;
* negative on a scheduling/allocation failure (the once-guard is rolled back so a
* later rejoin can retry).
*/
int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn)
{
(void)c;
(void)primary_committed_lsn;
return 0; /* RED scaffold — task 3.2 replaces */
if (c == NULL)
return -EINVAL;
/* Read the local tip + committed prefix together under the persist lock so
* the clamp decision is consistent. */
pthread_mutex_lock(&c->persist.lock);
uint64_t current_lsn = c->persist.current_lsn;
pthread_mutex_unlock(&c->persist.lock);
uint64_t local_tip = current_lsn > 0 ? current_lsn - 1 : 0;
/* Req 2.3: tip does not exceed the boundary → join with no truncation/restart. */
if (local_tip <= primary_committed_lsn)
return 0;
/* Req 2.1 (defensive once-guard): reconcile at most once per process lifetime.
* CAS 0→1 so a concurrent/duplicate CtrlHelloReply cannot schedule twice. */
int expected = 0;
if (!atomic_compare_exchange_strong_explicit(
&c->ha.rejoin_reconciled, &expected, 1,
memory_order_acq_rel, memory_order_acquire))
return 0; /* already reconciled — no second restart */
/* Req 2.2 (clamp): never truncate at/below the locally known committed prefix.
* cutoff = max(primary_committed_lsn, local committed_lsn). */
uint64_t local_committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
uint64_t cutoff = primary_committed_lsn > local_committed
? primary_committed_lsn
: local_committed;
ha_rejoin_reconcile_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL) {
ERROR("HA rejoin reconcile: ctx alloc failed — divergent suffix will be "
"reconciled at the next rejoin");
atomic_store_explicit(&c->ha.rejoin_reconciled, 0,
memory_order_release); /* roll back: allow retry */
return -ENOMEM;
}
ctx->c = c;
ctx->cutoff = cutoff;
ctx->prior_tip = local_tip;
/* D5/D1: defer truncation + restart off the (possibly iomgr) intake thread.
* One-shot, minimal delay — mirrors ha_schedule_demotion_convergence. Under
* XCE_TESTING the fixture drives the queued task synchronously. */
int rc = delay_task_ms(&c->delaymgr, 1, 1,
(delay_task_entry_fn_t)ha_rejoin_reconcile_fn, ctx);
if (rc < 0) {
ERROR("HA rejoin reconcile: delay_task_ms failed: %d", rc);
free(ctx);
atomic_store_explicit(&c->ha.rejoin_reconciled, 0,
memory_order_release); /* roll back: allow retry */
return rc;
}
return 0;
}
/*
@ -241,10 +355,12 @@ int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn)
* CtrlHelloReply; the function stores the primary-reported committed boundary +
* config tag (existing behavior), then is the rejoin trigger point.
*
* TASK 3.2 ADDS the ha_rejoin_reconcile(c, committed_lsn) call AFTER the store
* below (design "Rejoin Reconciliation (ha.c + cluster.c intake)", research D5).
* Task 3.1 ships only the extracted store, so the intake-integration RED test —
* which asserts a restart fired — FAILS until 3.2 wires the trigger.
* Rejoin trigger (Req 2.1–2.3, design "Rejoin Reconciliation", research D5):
* AFTER the store, ha_rejoin_reconcile(c, committed_lsn) evaluates the local tip
* against the primary-reported boundary and (when the tip runs ahead) defers a
* clamped truncate + one-time restart. ha_rejoin_reconcile is internally a
* no-op when the tip is at/below the boundary or already reconciled, so it is
* safe to call on every CtrlHelloReply.
*/
void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn,
uint64_t config_tag)
@ -254,5 +370,7 @@ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn,
c->ha.last_known_committed_config_tag = config_tag;
INFO("HA: CtrlHelloReply from primary: committed_lsn=%" PRIu64
" config_tag=%" PRIu64, committed_lsn, config_tag);
/* TASK 3.2: ha_rejoin_reconcile(c, committed_lsn); ← rejoin trigger */
/* Req 2.1–2.3: reconcile a log that runs ahead of the cluster boundary. */
(void)ha_rejoin_reconcile(c, committed_lsn);
}

40
src/controller/testsuite/test_demotion_convergence.c
View file

@ -211,21 +211,39 @@ static int dc_task_count(const delaymgr_t *mgr)
return count;
}
/* Run the single pending delaymgr task synchronously by invoking its stored
* entry(args), then return the entry's rc. Returns INT_MIN sentinel if there
* is no pending task (RED: nothing was scheduled). The task is NOT removed
* from the queue (delaymgr_fini frees it at teardown) — we only need to run
* its body once to observe truncation + restart. */
/* Run the single pending delaymgr one-shot task synchronously by invoking its
* stored entry(args), then return the entry's rc. Returns DC_NO_TASK sentinel if
* there is no pending task (RED: nothing was scheduled).
*
* The run task is DEQUEUED and its node freed, mirroring the real delaymgr worker
* (delay.c: ordq_get → runner_fn → runcount-- → FREEUP for a runcount==1 task).
* This is required for correctness, not just fidelity: the convergence task bodies
* free their own heap ctx (the args), exactly as production does (the delaymgr
* killer frees only the delay_task_t node, never args; production execv never
* re-runs a one-shot). Peeking without dequeuing would let a SECOND
* dc_run_pending_task call re-run an already-run task on a freed ctx
* (heap-use-after-free) — e.g. the rejoin once-guard test, whose second reconcile
* correctly schedules NO new task so the head still points at the consumed one.
* Dequeuing makes that second call return DC_NO_TASK, as production would. */
#define DC_NO_TASK 0x7fffffff
static int dc_run_pending_task(const delaymgr_t *mgr)
static int dc_run_pending_task(delaymgr_t *mgr)
{
const ordq_item_t *it = mgr->tasks.head;
if (it == NULL || it->data == NULL)
ordq_item_t *it = ordq_get(&mgr->tasks, NULL);
if (it == NULL || it->data == NULL) {
if (it != NULL)
free(it);
return DC_NO_TASK;
const delay_task_t *t = (const delay_task_t *)it->data;
if (t->entry == NULL)
}
delay_task_t *t = (delay_task_t *)it->data;
if (t->entry == NULL) {
free(t);
free(it);
return DC_NO_TASK;
return t->entry(t->args);
}
int rc = t->entry(t->args); /* task body frees its own args ctx */
free(t); /* mirror the worker's FREEUP(t) for runcount==1 */
free(it); /* mirror the worker's FREEUP(it) */
return rc;
}
/* ── Tests ──────────────────────────────────────────────────────────────── */