From c2221077fffd9dd3abd594eac584617ac6d8859a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 14 Jun 2026 05:16:54 +0000 Subject: [PATCH] test(ha-convergence-recovery): RED ACK-driven catch-up continuation tests + seam (task 4.1) --- .kiro/specs/ha-convergence-recovery/tasks.md | 2 +- src/controller/ha.h | 49 ++ src/controller/ha_convergence.c | 30 ++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_convergence_verification.c | 483 ++++++++++++++++++ 6 files changed, 566 insertions(+), 1 deletion(-) create mode 100644 src/controller/testsuite/test_convergence_verification.c diff --git a/.kiro/specs/ha-convergence-recovery/tasks.md b/.kiro/specs/ha-convergence-recovery/tasks.md index f77f4cb..9bb11ea 100644 --- a/.kiro/specs/ha-convergence-recovery/tasks.md +++ b/.kiro/specs/ha-convergence-recovery/tasks.md @@ -46,7 +46,7 @@ - _Requirements: 2.1, 2.2, 2.3_ - [ ] 4. ACK-driven catch-up continuation and re-inclusion trigger -- [ ] 4.1 Write RED tests for continuation, single-flight, and parity re-inclusion +- [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 diff --git a/src/controller/ha.h b/src/controller/ha.h index 76a4aeb..8d2d01f 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -200,6 +200,17 @@ typedef struct { * CtrlHelloReply in the same process * lifetime cannot re-fire the restart. * 0 at init (cleared by ha_init memset). */ + /* ── New field added by ha-convergence-recovery spec (Task 4.x) ──────── */ + _Atomic int catchup_inflight[MAX_STANDBYS]; /* Per-standby single-flight + * guard for ACK-driven catch-up + * continuation (Req 5.2): set when a + * continuation is scheduled for a + * standby slot, cleared when it + * completes; a duplicate walAck while + * the flag is set schedules nothing. + * Slot identity (the repl registry slot + * index) is stable across schedule/run. + * 0 at init (cleared by ha_init memset). */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */ @@ -909,6 +920,44 @@ extern void (*ha_fatal_hook)(const char *msg); * used by the promotion-side lifecycle tests. */ void raft_cb_become_follower_with_cluster_for_test(cluster_t *c, uint64_t term); + +/* + * ha_walack_continuation_for_test: test-only entry point for the ACK-driven + * catch-up continuation logic (ha-convergence-recovery task 4.x, Req 5.1, 5.2, + * 7.1). + * + * The production trigger lives in the walAck branch of + * repl_ack_decode_and_dispatch (ha.c), which decodes a real Cap'n Proto + * WalAckReply off the wire — too heavy to drive from a unit test. This shim + * exposes the SAME post-ACK decision the walAck branch performs after + * repl_on_ack, so the continuation / single-flight / parity behaviors can be + * exercised directly against a unit fixture: + * + * target = max(committed_lsn, last_written) (same derivation as ha.c:2771-2776) + * if acked_lsn < target → schedule ONE continuation that streams + * [acked_lsn+1, target] via the catch-up streaming + * path, UNLESS ha.catchup_inflight[peer_idx] is set + * (single-flight, Req 5.2); the continuation must not + * dispatch any heartbeat/stateReport action (Req 5.1). + * if acked_lsn >= target → repl_complete_catchup(&c->persist.repl, ep), which + * clears the standby's in_catch_up (Req 7.1). + * + * c — cluster_t* (real persist + repl registry; never NULL in tests) + * ep — repl registry endpoint key for the standby (the same pointer + * passed to repl_add_standby); identifies the standby slot. + * peer_idx — 0-based standby slot index (into ha.catchup_inflight[]). + * acked_lsn— the LSN the standby just acknowledged. + * + * Returns 0 on the no-op / parity paths and on a successfully scheduled + * continuation; negative on a scheduling failure. + * + * RED scaffold (task 4.1): ha_convergence.c currently records the ACK only and + * neither schedules a continuation nor calls repl_complete_catchup — task 4.2 + * implements the real decision. The continuation / single-flight / parity tests + * therefore FAIL until 4.2 lands. + */ +int ha_walack_continuation_for_test(cluster_t *c, void *ep, int peer_idx, + uint64_t acked_lsn); #endif #endif /* HA_H */ diff --git a/src/controller/ha_convergence.c b/src/controller/ha_convergence.c index 0f8e80d..0672ab3 100644 --- a/src/controller/ha_convergence.c +++ b/src/controller/ha_convergence.c @@ -374,3 +374,33 @@ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn, /* Req 2.1–2.3: reconcile a log that runs ahead of the cluster boundary. */ (void)ha_rejoin_reconcile(c, committed_lsn); } + +#ifdef XCE_TESTING +/* ─────────────────────────────────────────────────────────────────────────── + * RED scaffold — ha-convergence-recovery task 4.1 (replaced by 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. + * ─────────────────────────────────────────────────────────────────────────── */ +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; +} +#endif /* XCE_TESTING */ diff --git a/src/controller/meson.build b/src/controller/meson.build index 77491cb..054ddbc 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -126,6 +126,7 @@ controller_unit_test_src = [ 'testsuite/test_commit_barrier_reply.c', 'testsuite/test_egress_filter.c', 'testsuite/test_demotion_convergence.c', + 'testsuite/test_convergence_verification.c', 'commit.c', 'job.c', 'egress.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index f4a6ede..59f651e 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -54,6 +54,7 @@ extern void test_commit_barrier_rollout(void); extern void test_commit_barrier_reply(void); extern void test_egress_filter(void); extern void test_demotion_convergence(void); +extern void test_convergence_verification(void); void setUp(void) { return; } @@ -177,6 +178,7 @@ int main(int argc, char* argv[]) TEST(test_commit_barrier_reply); TEST(test_egress_filter); TEST(test_demotion_convergence); + TEST(test_convergence_verification); return UNITY_END(); } diff --git a/src/controller/testsuite/test_convergence_verification.c b/src/controller/testsuite/test_convergence_verification.c new file mode 100644 index 0000000..5c2fb49 --- /dev/null +++ b/src/controller/testsuite/test_convergence_verification.c @@ -0,0 +1,483 @@ +#define _GNU_SOURCE +/* + * test_convergence_verification.c — RED-phase tests for ACK-driven catch-up + * continuation, single-flight, and parity re-inclusion + * (ha-convergence-recovery task 4.1). + * + * Requirements (this spec's IDs — see requirements.md): + * 5.1 — when the primary receives a WAL ACK from a standby that remains BEHIND + * the replication target, it continues streaming the missing entries to + * that standby WITHOUT waiting for the next heartbeat exchange (the + * continuation streams [acked_lsn+1, target] and dispatches no + * heartbeat/stateReport replication action of its own). + * 5.2 — while a catch-up continuation for a standby is already in flight, the + * primary does NOT start a duplicate continuation for that standby + * (per-standby single-flight guard); once the in-flight continuation + * completes, the next lagging ACK schedules again. + * 5.3 — the heartbeat-driven replication evaluation remains in place as the + * fallback initiator of catch-up AND the initiator of snapshot-transfer + * decisions: ha_evaluate_repl_action still returns HA_REPL_STREAM_WAL / + * HA_REPL_SEND_SNAPSHOT for the same inputs (decision boundary pin). + * 7.1 — when a lagging standby catches up to the replication boundary (a WAL + * ACK at/above target), the primary re-includes it in live replication + * via repl_complete_catchup, which clears the standby's in_catch_up. + * + * RED status against the current tree: the production walAck branch + * (repl_ack_decode_and_dispatch, ha.c — the OPCode_reply / + * RequestReply_data_walackreplymsg case) only calls repl_on_ack; it derives no + * target, schedules no continuation, and never calls repl_complete_catchup + * (which has NO production caller, repl.c). The task-4.1 test seam + * ha_walack_continuation_for_test is a RED scaffold (ha_convergence.c) that + * records the ACK only. The continuation / single-flight / parity cases below + * therefore FAIL until task 4.2 lands the real D6 decision; the 5.3 decision-pin + * case exercises existing behavior and PASSES now (pinning the boundary so 4.2 + * cannot disturb the heartbeat path). + * + * Harness: a partial-but-real cluster_t (real persist over a private tmpdir, + * a wired commit engine, stub standby registry, un-started delaymgr) mirroring + * test_demotion_convergence.c's dc_setup, plus the catch-up streaming count seam + * ha_repl_stream_wal_hook and a dispatch-observer seam to prove the continuation + * takes no heartbeat path. + * + * AAA structure throughout. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "commit.h" /* commit_engine_t, commit_engine_init */ +#include "delay.h" /* delaymgr_t, delay_task_t */ +#include "ha.h" /* ha_walack_continuation_for_test, ha_evaluate_repl_action, + * ha_repl_stream_wal_hook, HA_REPL_* */ +#include "ordq.h" /* ordq_item_t (drive the scheduled task) */ +#include "persist.h" /* persist_t */ +#include "repl.h" /* repl_add_standby / repl_set_commit_engine / standby slots */ +#include "xce/cluster_def.h" /* full cluster_t layout */ + +#ifdef XCE_TESTING + +/* ── Broadcast drop-hook (precedent: test_demotion_convergence.c) — the stub int + * endpoints registered via repl_add_standby are not real endpoint_t, so any + * repl_broadcast triggered by priming appends must not dereference ep->peer. */ +extern int (*repl_broadcast_send_hook)(void *ep, + wal_catchup_entry_t **entries, + int n_entries); +static int cv_drop_send_hook(void *ep, wal_catchup_entry_t **entries, + int n_entries) +{ + (void)ep; (void)entries; (void)n_entries; + return 0; +} + +/* ── Catch-up streaming count seam (Req 5.1) ───────────────────────────────── + * The ACK-driven continuation must stream the missing range through the existing + * catch-up streaming path. ha_repl_stream_wal_hook is the established XCE_TESTING + * seam for that path (config_repl.c ha_stream_wal_to_standby / ha.c + * HA_REPL_STREAM_WAL). We count invocations and capture the [start, target] + * range so the continuation's bounds can be asserted. */ +static atomic_int g_stream_calls; +static atomic_uint_least64_t g_stream_last_standby_lsn; +static atomic_uint_least64_t g_stream_last_target_lsn; + +static int cv_stream_hook(void *ep, const char *wal_path, + uint64_t standby_lsn, uint64_t committed_lsn) +{ + (void)ep; (void)wal_path; + atomic_fetch_add(&g_stream_calls, 1); + atomic_store(&g_stream_last_standby_lsn, standby_lsn); + atomic_store(&g_stream_last_target_lsn, committed_lsn); + return 0; +} + +/* ── Cluster fixture ─────────────────────────────────────────────────────── */ +static cluster_t g_cv; /* large — must not live on the test stack */ +static int g_cv_eps[MAX_STANDBYS]; +#define CV_PERSIST_NAME "ctrl-test" + +static ha_config_t cv_ha_cfg(void) +{ + ha_config_t cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + }; + return cfg; +} + +/* Build g_cv: real persist over a fresh tmpdir, n_standbys stub standbys, a + * wired commit engine, an initialised-but-UNSTARTED delaymgr (so any task in + * tasks.head was armed by the code under test, not by a background runner). + * Role left at INIT (the walAck continuation is a primary-side concern but the + * decision under test needs only the persist watermark + repl registry, not the + * full promotion sequence). Leaves committed_lsn at 0; callers set it. */ +static void cv_setup(const char *suffix, int n_peers, int n_standbys, + uint64_t appended_lsn, int n_vcpus) +{ + char name[32]; + char tmpdir[256]; + + memset(&g_cv, 0, sizeof(g_cv)); + atomic_store(&g_stream_calls, 0); + atomic_store(&g_stream_last_standby_lsn, 0); + atomic_store(&g_stream_last_target_lsn, 0); + + ha_config_t cfg = cv_ha_cfg(); + TEST_ASSERT_EQUAL_INT(0, ha_init(&g_cv.ha, &cfg, 0)); + g_cv.ha.phase2_done = 1; + g_cv.ha.n_peers = n_peers; + + persist_config_t pcfg = persist_config_defaults(); + pcfg.wal_sync_ms = 5000; /* slow background fsync — no races */ + pcfg.sync_timeout_ms = 50; /* no real standby ACKs here */ + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cv_%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + persist_set_workdir(&g_cv.persist, tmpdir, CV_PERSIST_NAME); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_cv.persist, &pcfg, &g_cv)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_cv.persist)); + + /* Prime the appended watermark: current_lsn = appended_lsn + 1, so + * last_written = appended_lsn (the replication target's upper bound). */ + pthread_mutex_lock(&g_cv.persist.lock); + g_cv.persist.current_lsn = appended_lsn + 1; + pthread_mutex_unlock(&g_cv.persist.lock); + atomic_store(&g_cv.persist.committed_lsn, 0); + + for (int i = 0; i < n_standbys; i++) { + snprintf(name, sizeof(name), "cvsb%d", i); + TEST_ASSERT_TRUE_MESSAGE( + repl_add_standby(&g_cv.persist.repl, &g_cv_eps[i], name, + 7600 + (uint32_t)i) >= 0, + "repl_add_standby must return a slot index >= 0"); + } + TEST_ASSERT_EQUAL_INT(0, + commit_engine_init(&g_cv.commit, &g_cv.persist, n_peers, n_vcpus)); + repl_set_commit_engine(&g_cv.persist.repl, &g_cv.commit); + + repl_broadcast_send_hook = cv_drop_send_hook; + ha_repl_stream_wal_hook = cv_stream_hook; + + TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_cv.delaymgr, 100, NULL)); +} + +static void cv_teardown(void) +{ + repl_broadcast_send_hook = NULL; + ha_repl_stream_wal_hook = NULL; + commit_engine_fini(&g_cv.commit); + delaymgr_fini(&g_cv.delaymgr); + persist_stop(&g_cv.persist); + persist_fini(&g_cv.persist); + repl_fini(&g_cv.persist.repl); + ha_fini(&g_cv.ha); + /* tmpdir left on disk — /tmp is ephemeral */ +} + +/* Count pending delaymgr tasks by walking the ordq head chain (read-only; + * single-threaded test, delaymgr never started). */ +static int cv_task_count(const delaymgr_t *mgr) +{ + int count = 0; + const ordq_item_t *it = mgr->tasks.head; + while (it != NULL) { + count++; + it = it->next; + } + return count; +} + +/* Run the single pending delaymgr one-shot task synchronously by invoking its + * stored entry(args), then return the entry's rc. Returns CV_NO_TASK if there + * is no pending task (RED: nothing was scheduled). DEQUEUES + frees, mirroring + * the real delaymgr worker (delay.c) and test_demotion_convergence.c's + * dc_run_pending_task — required so a second call after a body that frees its own + * ctx does not re-run a freed ctx (heap-UAF). */ +#define CV_NO_TASK 0x7fffffff +static int cv_run_pending_task(delaymgr_t *mgr) +{ + ordq_item_t *it = ordq_get(&mgr->tasks, NULL); + if (it == NULL || it->data == NULL) { + if (it != NULL) + free(it); + return CV_NO_TASK; + } + delay_task_t *t = (delay_task_t *)it->data; + if (t->entry == NULL) { + free(t); + free(it); + return CV_NO_TASK; + } + int rc = t->entry(t->args); /* task body frees its own args ctx */ + free(t); + free(it); + return rc; +} + +/* Read the per-standby in_catch_up flag (repl registry slot identified by ep). */ +static int cv_standby_in_catch_up(repl_ep_t ep) +{ + replicator_t *r = &g_cv.persist.repl; + int result = -1; + mylock_lock(&r->lock); + for (int i = 0; i < MAX_STANDBYS; i++) { + if (r->standbys[i].ep == ep) { + result = atomic_load_explicit(&r->standbys[i].in_catch_up, + memory_order_acquire); + break; + } + } + mylock_unlock(&r->lock); + return result; +} + +/* ── Tests ──────────────────────────────────────────────────────────────── */ + +/* 5.1: a walAck with acked_lsn < target = max(committed_lsn, last_written) + * schedules EXACTLY ONE continuation that streams [acked_lsn+1, target] without + * any intervening heartbeat/stateReport dispatch. + * RED: the walAck continuation scaffold schedules nothing → no task, the stream + * hook is never called. */ +static void test_walack_behind_target_schedules_one_continuation(void) +{ + /* appended_lsn=9 → last_written=9; committed_lsn=4 → target=max(4,9)=9. */ + cv_setup("behind", /*n_peers=*/3, /*n_standbys=*/1, + /*appended_lsn=*/9, /*n_vcpus=*/8); + atomic_store(&g_cv.persist.committed_lsn, 4); + + const int peer_idx = 0; + const uint64_t acked_lsn = 3; /* behind target 9 */ + const uint64_t target = 9; + + int tasks_before = cv_task_count(&g_cv.delaymgr); + + /* Act: standby acknowledges lsn=3 while target=9. */ + int rc = ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], + peer_idx, acked_lsn); + (void)rc; + + /* A continuation must have been scheduled OFF the ACK path (exactly one). */ + int tasks_after = cv_task_count(&g_cv.delaymgr); + TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before + 1, tasks_after, + "a walAck behind target must schedule exactly one catch-up continuation " + "(Req 5.1)"); + + /* No streaming may have happened on the ACK path itself (it defers). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_stream_calls), + "the walAck branch must NOT stream inline — streaming runs in the " + "deferred continuation (Req 5.1, iomgr-safe)"); + + /* Run the scheduled continuation; it must stream [acked_lsn+1, target]. */ + int trc = cv_run_pending_task(&g_cv.delaymgr); + TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc, + "no continuation task was scheduled on the walAck-behind-target edge " + "(Req 5.1)"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_stream_calls), + "the continuation must stream exactly once (Req 5.1)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(acked_lsn, + atomic_load(&g_stream_last_standby_lsn), + "the continuation must stream starting just above the ACKed LSN " + "(ha_stream_wal_to_standby streams (standby_lsn, target] = " + "[acked_lsn+1, target]) (Req 5.1)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(target, + atomic_load(&g_stream_last_target_lsn), + "the continuation must stream up to target = max(committed_lsn, " + "last_written) (Req 5.1)"); + + cv_teardown(); + printf("test_walack_behind_target_schedules_one_continuation: checked\n"); +} + +/* 5.2 single-flight: a SECOND walAck while a continuation is already in flight + * for the same standby schedules NOTHING. + * RED: the scaffold schedules nothing for the first ACK either, so the + * "exactly one continuation across two ACKs" assertion fails (0 != 1). */ +static void test_walack_single_flight_no_duplicate_continuation(void) +{ + cv_setup("single", /*n_peers=*/3, /*n_standbys=*/1, + /*appended_lsn=*/9, /*n_vcpus=*/8); + atomic_store(&g_cv.persist.committed_lsn, 4); + + const int peer_idx = 0; + + int tasks_before = cv_task_count(&g_cv.delaymgr); + + /* Act: first walAck (behind target) → schedules one continuation and marks + * the standby's catchup_inflight. */ + (void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx, + /*acked_lsn=*/3); + int tasks_after_first = cv_task_count(&g_cv.delaymgr); + TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before + 1, tasks_after_first, + "the first walAck behind target must schedule one continuation (Req 5.2 " + "setup)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + atomic_load(&g_cv.ha.catchup_inflight[peer_idx]), + "scheduling a continuation must set the per-standby single-flight flag " + "(Req 5.2)"); + + /* Act: a SECOND walAck while the continuation is still in flight (the task + * has NOT been drained). */ + (void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx, + /*acked_lsn=*/5); + + /* Assert (Req 5.2): no duplicate continuation — task count unchanged. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_after_first, + cv_task_count(&g_cv.delaymgr), + "a second walAck while a continuation is in flight must schedule NOTHING " + "(single-flight, Req 5.2)"); + + /* Drain the single in-flight continuation (clears the flag). */ + (void)cv_run_pending_task(&g_cv.delaymgr); + + cv_teardown(); + printf("test_walack_single_flight_no_duplicate_continuation: checked\n"); +} + +/* 5.2 release: after the in-flight continuation completes (flag cleared), the + * NEXT lagging walAck schedules a fresh continuation again. + * RED: the scaffold never schedules/clears, so neither continuation appears. */ +static void test_walack_reschedules_after_continuation_completes(void) +{ + cv_setup("respawn", /*n_peers=*/3, /*n_standbys=*/1, + /*appended_lsn=*/9, /*n_vcpus=*/8); + atomic_store(&g_cv.persist.committed_lsn, 4); + + const int peer_idx = 0; + + /* Act 1: first walAck → continuation #1. */ + (void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx, + /*acked_lsn=*/3); + int trc1 = cv_run_pending_task(&g_cv.delaymgr); + TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc1, + "first walAck must schedule a continuation (Req 5.2 release setup)"); + + /* After the continuation completes the single-flight flag must be clear. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + atomic_load(&g_cv.ha.catchup_inflight[peer_idx]), + "the continuation must clear the single-flight flag on completion " + "(Req 5.2 release)"); + + int streams_after_first = atomic_load(&g_stream_calls); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, streams_after_first, + "the first continuation must have streamed once"); + + /* Act 2: a later lagging walAck (still behind target 9) → continuation #2. */ + (void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx, + /*acked_lsn=*/6); + int trc2 = cv_run_pending_task(&g_cv.delaymgr); + TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc2, + "a lagging walAck after the prior continuation completed must schedule " + "AGAIN (Req 5.2 release)"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(2, atomic_load(&g_stream_calls), + "the second continuation must stream once more after the first released " + "(Req 5.2 release)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(6, + atomic_load(&g_stream_last_standby_lsn), + "the second continuation streams from the newly ACKed LSN (Req 5.2 " + "release)"); + + cv_teardown(); + printf("test_walack_reschedules_after_continuation_completes: checked\n"); +} + +/* 7.1 parity: a walAck with acked_lsn >= target calls repl_complete_catchup, + * which clears the standby's in_catch_up (re-inclusion in live replication). + * RED: the scaffold never calls repl_complete_catchup, so in_catch_up — set here + * via repl_start_catchup — stays 1. */ +static void test_walack_at_parity_completes_catchup_and_clears_in_catch_up(void) +{ + /* appended_lsn=9 → last_written=9; committed_lsn=9 → target=9. */ + cv_setup("parity", /*n_peers=*/3, /*n_standbys=*/1, + /*appended_lsn=*/9, /*n_vcpus=*/8); + atomic_store(&g_cv.persist.committed_lsn, 9); + + const int peer_idx = 0; + repl_ep_t ep = &g_cv_eps[peer_idx]; + const uint64_t acked_lsn = 9; /* at parity with target 9 */ + + /* Put the standby into catch-up so re-inclusion is observable. */ + repl_start_catchup(&g_cv.persist.repl, ep, /*join_applied_lsn=*/0); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, cv_standby_in_catch_up(ep), + "fixture precondition: the standby must be in_catch_up before parity ACK"); + + int tasks_before = cv_task_count(&g_cv.delaymgr); + + /* Act: the standby acknowledges the target → parity. */ + (void)ha_walack_continuation_for_test(&g_cv, ep, peer_idx, acked_lsn); + + /* Assert (Req 7.1): no continuation scheduled at parity, and the standby is + * re-included — in_catch_up cleared by repl_complete_catchup. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before, cv_task_count(&g_cv.delaymgr), + "a parity walAck must NOT schedule a streaming continuation (Req 7.1)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_stream_calls), + "a parity walAck must NOT stream (nothing left to send) (Req 7.1)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_standby_in_catch_up(ep), + "a parity walAck must call repl_complete_catchup, clearing the standby's " + "in_catch_up so it re-enters live replication (Req 7.1)"); + + cv_teardown(); + printf("test_walack_at_parity_completes_catchup_and_clears_in_catch_up: checked\n"); +} + +/* 5.3 decision pin: the heartbeat decision function ha_evaluate_repl_action is + * the fallback initiator of catch-up and the SOLE owner of the snapshot + * decision; the ACK-driven continuation must not change its verdicts. This pins + * the existing boundary so task 4.2 cannot disturb the heartbeat path. + * + * This case exercises only existing pure-function behavior, so it PASSES against + * the current tree (intended — it locks the boundary; the RED signal for task + * 4.1 comes from the continuation/single-flight/parity cases above). */ +static void test_heartbeat_decision_path_unchanged(void) +{ + /* HA_REPL_STREAM_WAL: tag current, lsn within the WAL window + * (oldest_wal_lsn <= standby_lsn < committed_lsn). */ + ha_repl_action_t a_stream = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/3, + /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/1); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)a_stream, + "heartbeat decision: standby_lsn within the WAL window must remain " + "HA_REPL_STREAM_WAL — the heartbeat path stays the fallback catch-up " + "initiator (Req 5.3)"); + + /* HA_REPL_SEND_SNAPSHOT: standby_lsn predates the oldest retained WAL. */ + ha_repl_action_t a_snap = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/2, + /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)a_snap, + "heartbeat decision: standby_lsn below oldest_wal_lsn must remain " + "HA_REPL_SEND_SNAPSHOT — the heartbeat path remains the SOLE snapshot " + "decision owner (Req 5.3); the ACK-driven continuation never decides " + "snapshots"); + + printf("test_heartbeat_decision_path_unchanged: checked\n"); +} + +#endif /* XCE_TESTING */ + +/* ── Suite entry point ──────────────────────────────────────────────────── */ + +void test_convergence_verification(void) +{ +#ifdef XCE_TESTING + printf("\n>> ACK-driven catch-up continuation tests (RED, task 4.1)\n\n"); + + RUN_TEST(test_walack_behind_target_schedules_one_continuation); + RUN_TEST(test_walack_single_flight_no_duplicate_continuation); + RUN_TEST(test_walack_reschedules_after_continuation_completes); + RUN_TEST(test_walack_at_parity_completes_catchup_and_clears_in_catch_up); + RUN_TEST(test_heartbeat_decision_path_unchanged); +#else + printf("test_convergence_verification: SKIP (non-test build)\n"); +#endif +}