From ffd08c6472c1b7b739d1d7523769c90e40530e6d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 14 Jun 2026 05:45:20 +0000 Subject: [PATCH] feat(ha-convergence-recovery): ACK-driven catch-up continuation + parity re-inclusion (task 4.2) --- .kiro/specs/ha-convergence-recovery/tasks.md | 4 +- src/controller/config_repl.c | 195 +++++++++++++++++++ src/controller/ha.c | 7 + src/controller/ha.h | 28 +++ src/controller/ha_convergence.c | 31 ++- 5 files changed, 243 insertions(+), 22 deletions(-) diff --git a/.kiro/specs/ha-convergence-recovery/tasks.md b/.kiro/specs/ha-convergence-recovery/tasks.md index 9bb11ea9..67a7ae87 100644 --- a/.kiro/specs/ha-convergence-recovery/tasks.md +++ b/.kiro/specs/ha-convergence-recovery/tasks.md @@ -45,14 +45,14 @@ - _Depends: 2.2_ - _Requirements: 2.1, 2.2, 2.3_ -- [ ] 4. ACK-driven catch-up continuation and re-inclusion trigger +- [x] 4. ACK-driven catch-up continuation and re-inclusion trigger - [x] 4.1 Write RED tests for continuation, single-flight, and parity re-inclusion - Create `testsuite/test_convergence_verification.c` and register it in meson - Tests (use the existing `ha_repl_stream_wal_hook` / `xce_wal_stream_request_hook` seams to count streaming invocations): a walAck with `acked_lsn < max(committed_lsn, last_written)` schedules exactly one continuation that streams `[acked_lsn+1, target]` WITHOUT any intervening heartbeat/stateReport dispatch; a second walAck while the continuation is in flight schedules nothing (single-flight); after the continuation completes, the next lagging walAck schedules again; a walAck with `acked_lsn >= target` calls `repl_complete_catchup` and the standby's `in_catch_up` clears; the heartbeat decision path (`ha_dispatch_repl_action_for_standby`) still produces HA_REPL_SEND_SNAPSHOT / HA_REPL_STREAM_WAL decisions unchanged (fallback + snapshot ownership pin) - Observable completion: new cases FAIL (red) — current walAck branch (ha.c:2737-2754 anchor) only records the ACK; `repl_complete_catchup` (repl.c:521) has no production caller - _Requirements: 5.1, 5.2, 5.3, 7.1_ -- [ ] 4.2 Implement the continuation trigger and parity re-inclusion to green +- [x] 4.2 Implement the continuation trigger and parity re-inclusion to green - ha.c walAck branch: after `repl_on_ack`, derive target as at ha.c:2727-2732; behind target → schedule the continuation via delaymgr unless the standby's `catchup_inflight` flag is set; at parity → `repl_complete_catchup` (ha-commit-engine wiring propagates the membership re-evaluation) - config_repl.c: continuation task body — re-validate the endpoint against the registry, re-derive target, call `ha_stream_wal_to_standby(ep, wal_path, acked_lsn, target)` (internals untouched), clear `catchup_inflight` on all exit paths - ha.h: per-standby `_Atomic int catchup_inflight` storage with slot identity stable across schedule/run diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index ffdcd8e4..bfd833cb 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1651,6 +1651,201 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, return 0; } +/* ── ACK-driven catch-up continuation (ha-convergence-recovery task 4.2) ────── + * + * Design D6 (Req 5.1, 5.2, 7.1). The walAck branch of + * repl_ack_decode_and_dispatch (ha.c, iomgr) calls ha_walack_on_ack after + * repl_on_ack; the unit test drives the same logic through + * ha_walack_continuation_for_test. Streaming reads WAL files, so it must never + * run inline on iomgr — the decision schedules a one-shot delaymgr task that + * does the streaming and clears the per-standby single-flight flag. + * + * This lives in config_repl.c (per design) alongside ha_stream_wal_to_standby: + * the continuation wraps the existing streaming primitive with no change to its + * internals. + * ──────────────────────────────────────────────────────────────────────────── */ + +/* repl_ep_still_registered: is `ep` still an active standby slot? The ep may + * have RDHUPed between schedule and run; on its disappearance the continuation + * is a no-op (design Error Handling: clear the flag and return, no error). */ +static int repl_ep_still_registered(replicator_t *r, repl_ep_t ep) +{ + int found = 0; + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep == ep) { + found = 1; + break; + } + } + mylock_unlock(&r->lock); + return found; +} + +/* re-derive the replication target exactly as ha.c:2771-2776 / D6: + * target = max(committed_lsn, last_written), last_written = current_lsn - 1. */ +static uint64_t catchup_derive_target(cluster_t *c) +{ + uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn, + memory_order_acquire); + pthread_mutex_lock(&c->persist.lock); + uint64_t last_written = c->persist.current_lsn > 0 + ? c->persist.current_lsn - 1 + : 0; + pthread_mutex_unlock(&c->persist.lock); + return last_written > committed ? last_written : committed; +} + +/* Invoke the catch-up streaming path for [standby_lsn+1, target]. Mirrors the + * HA_REPL_STREAM_WAL dispatch in ha.c: under XCE_TESTING the + * ha_repl_stream_wal_hook seam intercepts so unit tests can count the stream + * without a real endpoint or WAL directory; otherwise ha_stream_wal_to_standby + * runs unchanged. */ +static void catchup_stream_range(cluster_t *c, repl_ep_t ep, + uint64_t standby_lsn, uint64_t target) +{ +#ifdef XCE_TESTING + if (ha_repl_stream_wal_hook != NULL) { + const char *wal_path = (c && c->persist.wal_path[0]) + ? c->persist.wal_path : ""; + ha_repl_stream_wal_hook(ep, wal_path, standby_lsn, target); + return; + } +#endif + if (c && c->persist.wal_path[0]) + ha_stream_wal_to_standby((endpoint_t *)ep, c->persist.wal_path, + standby_lsn, target); +} + +/* ha_catchup_continuation_ctx_t — heap context for the one-shot delaymgr task. + * Carries the standby slot identity (peer_idx — stable schedule→run, indexes the + * single-flight flag) and the ACKed LSN; target is re-derived at run time so the + * stream extends to entries appended meanwhile (Req 5.1). */ +typedef struct { + cluster_t *c; + repl_ep_t ep; + int peer_idx; + uint64_t acked_lsn; +} ha_catchup_continuation_ctx_t; + +/* + * ha_catchup_continuation_fn — delaymgr task body (runs OFF the iomgr walAck + * branch, D6/D1). Re-validate the endpoint against the registry, re-derive + * target, stream [acked_lsn+1, target], and clear catchup_inflight on EVERY exit + * path (success, ep-gone, nothing-to-stream) so a stuck flag never blocks future + * continuations (Req 5.2). Frees its own ctx (body-frees-own-ctx pattern, + * mirroring ha_demotion_convergence_fn). + */ +int ha_catchup_continuation_fn(void *arg) +{ + ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg; + if (ctx == NULL || ctx->c == NULL) { + free(ctx); + return -EINVAL; + } + cluster_t *c = ctx->c; + repl_ep_t ep = ctx->ep; + int peer_idx = ctx->peer_idx; + uint64_t acked = ctx->acked_lsn; + + /* Endpoint may have RDHUPed between schedule and run — no-op if gone. */ + if (!repl_ep_still_registered(&c->persist.repl, ep)) { + INFO("HA catch-up continuation: standby[%d] endpoint gone before run " + "— skipping (heartbeat fallback re-evaluates)", peer_idx); + goto clear_and_done; + } + + uint64_t target = catchup_derive_target(c); + if (acked < target) { + INFO("HA catch-up continuation: standby[%d] acked=%" PRIu64 + " < target=%" PRIu64 " → streaming [%" PRIu64 "+1..%" PRIu64 "]", + peer_idx, acked, target, acked, target); + catchup_stream_range(c, ep, acked, target); + } + /* If acked >= target the standby reached parity meanwhile — nothing to + * stream; parity re-inclusion is owned by the decision path, not here. */ + +clear_and_done: + /* Req 5.2: release the single-flight slot so the next lagging ACK schedules + * again. release order so the decision path's acquire load observes it. */ + if (peer_idx >= 0 && peer_idx < MAX_STANDBYS) + atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, + memory_order_release); + free(ctx); + return 0; +} + +/* + * ha_schedule_catchup_continuation — enqueue one one-shot delaymgr task running + * ha_catchup_continuation_fn. The single-flight flag is set by the caller + * (ha_walack_on_ack) before this call; on a scheduling failure this clears it so + * the slot does not stay stuck. Returns 0 on success, negative on failure. + */ +static int ha_schedule_catchup_continuation(cluster_t *c, repl_ep_t ep, + int peer_idx, uint64_t acked_lsn) +{ + ha_catchup_continuation_ctx_t *ctx = calloc(1, sizeof(*ctx)); + if (ctx == NULL) { + ERROR("HA catch-up continuation: ctx alloc failed for standby[%d]", + peer_idx); + atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, + memory_order_release); + return -ENOMEM; + } + ctx->c = c; + ctx->ep = ep; + ctx->peer_idx = peer_idx; + ctx->acked_lsn = acked_lsn; + + /* One-shot (runcount=1), minimal delay: streaming reads WAL files and must + * run off the iomgr walAck thread. delay_task_ms only enqueues (ordq_put); + * the delaymgr worker runs it. In XCE_TESTING the fixture drives the queued + * task synchronously via its stored entry/args. */ + int rc = delay_task_ms(&c->delaymgr, 1, 1, + (delay_task_entry_fn_t)ha_catchup_continuation_fn, + ctx); + if (rc < 0) { + ERROR("HA catch-up continuation: delay_task_ms failed for standby[%d]:" + " %d", peer_idx, rc); + free(ctx); + atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, + memory_order_release); + return rc; + } + return 0; +} + +int ha_walack_on_ack(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn) +{ + if (c == NULL || ep == NULL) + return -EINVAL; + if (peer_idx < 0 || peer_idx >= MAX_STANDBYS) + return -EINVAL; + + uint64_t target = catchup_derive_target(c); + + /* Parity (Req 7.1): the standby reached the boundary — re-include it in live + * replication via the existing-but-untriggered re-inclusion call. No + * continuation; ha-commit-engine wiring propagates commit_on_membership_change. */ + if (acked_lsn >= target) { + repl_complete_catchup(&c->persist.repl, (repl_ep_t)ep); + return 0; + } + + /* Behind target (Req 5.1/5.2): single-flight — claim the per-standby slot + * 0→1; if already in flight, schedule nothing. */ + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &c->ha.catchup_inflight[peer_idx], &expected, 1, + memory_order_acq_rel, memory_order_acquire)) { + /* A continuation is already in flight for this standby (Req 5.2). */ + return 0; + } + + return ha_schedule_catchup_continuation(c, (repl_ep_t)ep, peer_idx, + acked_lsn); +} + /* ── Snapshot chunk transfer ───────────────────────────────────────────── */ #ifdef XCE_TESTING diff --git a/src/controller/ha.c b/src/controller/ha.c index 8a5798a2..5d0dfe7c 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2792,6 +2792,13 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) repl_on_ack(&c->persist.repl, ep, acked_lsn); TRACE("repl_ack_decode_and_dispatch: WAL ACK from standby" " lsn=%lu success=1", (unsigned long)acked_lsn); + /* ha-convergence-recovery task 4.2 (D6, Req 5.1/5.2/7.1): + * ACK-driven catch-up — continue streaming to a standby still + * behind target, or re-include it at parity. Shares the exact + * decision the unit test drives via ha_walack_continuation_for_test. + * Scheduling is O(1) on iomgr; all WAL file I/O runs in the + * deferred continuation task. */ + ha_walack_on_ack(c, ep, vctx->peer_idx, acked_lsn); } else if (!success) { WARN("repl_ack_decode_and_dispatch: WAL NACK from standby" " lsn=%lu success=0 — ignoring", (unsigned long)acked_lsn); diff --git a/src/controller/ha.h b/src/controller/ha.h index 8d2d01fe..5ba7bf73 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -349,6 +349,34 @@ int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn); void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn, uint64_t config_tag); +/* + * ha_walack_on_ack: the post-ACK catch-up decision (ha-convergence-recovery + * task 4.2, Req 5.1, 5.2, 7.1, design D6). Called from the walAck branch of + * repl_ack_decode_and_dispatch (ha.c) AFTER repl_on_ack, and from the + * ha_walack_continuation_for_test shim so production and tests exercise the + * same logic. Defined in config_repl.c (alongside ha_stream_wal_to_standby and + * the continuation task body). + * + * target = max(committed_lsn, last_written) (same derivation as ha.c:2771-2776) + * if acked_lsn < target → schedule ONE catch-up continuation that streams + * [acked_lsn+1, target] on delaymgr (never inline on + * iomgr), UNLESS c->ha.catchup_inflight[peer_idx] is + * already set (per-standby single-flight, Req 5.2). + * if acked_lsn >= target → repl_complete_catchup(&c->persist.repl, ep) to + * re-include the standby in live replication (Req 7.1). + * + * c — cluster_t* (real persist + repl registry; never NULL). + * ep — repl registry endpoint key for the standby (endpoint_t* / + * repl_ep_t); identifies the standby slot for streaming + parity. + * peer_idx — 0-based standby slot index (into c->ha.catchup_inflight[]). + * acked_lsn— the LSN the standby just acknowledged. + * + * Returns 0 on the parity / single-flight-suppressed paths and on a successfully + * scheduled continuation; negative on a scheduling/allocation failure. Never + * performs file I/O itself (streaming runs in the deferred continuation task). + */ +int ha_walack_on_ack(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn); + /* ── Redirect helpers used by join.c ──────────────────────── */ diff --git a/src/controller/ha_convergence.c b/src/controller/ha_convergence.c index 0672ab32..379faf48 100644 --- a/src/controller/ha_convergence.c +++ b/src/controller/ha_convergence.c @@ -377,30 +377,21 @@ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn, #ifdef XCE_TESTING /* ─────────────────────────────────────────────────────────────────────────── - * RED scaffold — ha-convergence-recovery task 4.1 (replaced by task 4.2 GREEN). + * ha_walack_continuation_for_test — task 4.2 GREEN. * - * ha_walack_continuation_for_test stands in for the post-ACK decision that the - * walAck branch of repl_ack_decode_and_dispatch (ha.c) must perform after - * repl_on_ack (D6): derive target = max(committed_lsn, last_written); when the - * standby is still behind target, schedule exactly one catch-up continuation - * (single-flight via ha.catchup_inflight[peer_idx]); at/above target call - * repl_complete_catchup to re-include the standby (Req 7.1). - * - * THIS SCAFFOLD INTENTIONALLY DOES NOTHING beyond recording that the ACK was - * observed (it mirrors the current production walAck branch, which only calls - * repl_on_ack). It does NOT schedule a continuation and does NOT call - * repl_complete_catchup — so the continuation / single-flight / parity tests in - * test_convergence_verification.c FAIL (RED) until task 4.2 implements the real - * decision here and wires it into the production walAck branch. + * Test-only shim onto the REAL post-ACK catch-up decision ha_walack_on_ack + * (config_repl.c, D6). The production walAck branch of + * repl_ack_decode_and_dispatch (ha.c) calls the SAME ha_walack_on_ack after + * repl_on_ack, so production and the unit tests exercise identical logic + * (derive target = max(committed_lsn, last_written); behind target → schedule + * one single-flight continuation; at parity → repl_complete_catchup, Req + * 5.1/5.2/7.1). This shim exists only because the production caller decodes a + * real Cap'n Proto WalAckReply off the wire, which is too heavy to drive from a + * unit fixture. * ─────────────────────────────────────────────────────────────────────────── */ int ha_walack_continuation_for_test(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn) { - /* RED scaffold — task 4.2 replaces this body with the D6 decision. */ - (void)c; - (void)ep; - (void)peer_idx; - (void)acked_lsn; - return 0; + return ha_walack_on_ack(c, ep, peer_idx, acked_lsn); } #endif /* XCE_TESTING */