From e399328339b377834ed42f91eaa568495113c73f Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 14 Jun 2026 06:56:57 +0000 Subject: [PATCH] test(ha-convergence-recovery): regression pins for fan-out, snapshot, degraded-gate, re-inclusion (task 6.1) --- .kiro/specs/ha-convergence-recovery/tasks.md | 4 +- .../testsuite/test_convergence_verification.c | 296 ++++++++++++++++++ 2 files changed, 298 insertions(+), 2 deletions(-) diff --git a/.kiro/specs/ha-convergence-recovery/tasks.md b/.kiro/specs/ha-convergence-recovery/tasks.md index 0b87b0b2..f058d7a6 100644 --- a/.kiro/specs/ha-convergence-recovery/tasks.md +++ b/.kiro/specs/ha-convergence-recovery/tasks.md @@ -71,8 +71,8 @@ - Observable completion: all task 5.1 tests PASS; no log line is emitted per rejected operation - _Requirements: 8.1, 8.2, 8.3_ -- [ ] 6. Verification of existing behaviors (regression pins — no production code) -- [ ] 6.1 Write regression tests pinning full fan-out, snapshot fallback, and degraded-mode service +- [x] 6. Verification of existing behaviors (regression pins — no production code) +- [x] 6.1 Write regression tests pinning full fan-out, snapshot fallback, and degraded-mode service - Tests in `test_convergence_verification.c`: `repl_broadcast` enqueues to every registered active standby and skips only empty slots and `in_catch_up` (umbrella 6.1 pin); `ha_evaluate_repl_action` returns HA_REPL_SEND_SNAPSHOT when `standby_lsn < oldest_wal_lsn` and the snapshot path takes a fresh checkpoint before sending (umbrella 6.3 pin, decision boundary only); `ha_dispatch_gate` without write quorum admits `OPCode_get`, heartbeat, the replication opcode allowlist, and the fact-shaped opcodes (`OPCode_notify`/`update`/`setpreason`/`clrpreason` — admitted without quorum per ha-commit-barrier-rollout's fact allowlist) while rejecting request-shaped opcodes only (`OPCode_jobsub`/`jobcontrol`/`queuecontrol`/`hostcontrol`/`jobdispatch`/`join`) with `-XCE_ERR_NOT_READY` (umbrella 7.3 pin); `repl_complete_catchup` clears `in_catch_up` and the standby re-enters broadcast iteration (umbrella 6.4 mechanism pin, complements task 4's trigger test) - Observable completion: all pins PASS against current code without production changes (these encode behavior that already exists; any failure is a real regression or a wrong assumption to escalate) - _Requirements: 4.1, 6.1, 7.1, 9.1, 9.2_ diff --git a/src/controller/testsuite/test_convergence_verification.c b/src/controller/testsuite/test_convergence_verification.c index 3573109e..0c2991c4 100644 --- a/src/controller/testsuite/test_convergence_verification.c +++ b/src/controller/testsuite/test_convergence_verification.c @@ -55,6 +55,7 @@ #include #include "commit.h" /* commit_engine_t, commit_engine_init */ +#include "controller.capnp.h" /* OPCode_* (task 6.1 (c) dispatch-gate pin) */ #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_* */ @@ -63,6 +64,7 @@ #include "repl.h" /* repl_add_standby / repl_remove_standby / * repl_active_standby_count / standby slots */ #include "xce/cluster_def.h" /* full cluster_t layout */ +#include "xce/errnox.h" /* XCE_ERR_NOT_READY (task 6.1 (c) gate pin) */ #ifdef XCE_TESTING @@ -104,6 +106,37 @@ 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" +/* ── Fan-out counting send-hook (task 6.1 (a)/(d), umbrella 6.1/6.4 pins) ───── + * repl_broadcast routes each enqueue through repl_broadcast_send_hook under + * XCE_TESTING; this counting variant records, per standby slot, how many entries + * repl_broadcast tried to send to that slot's ep. An active, non-catch-up + * standby must receive exactly one send; an empty slot or an in_catch_up slot + * must receive ZERO. Indexed by matching ep against the g_cv_eps[] addresses. + * (Declared after g_cv_eps so the slot match resolves.) */ +static atomic_int g_bcast_total_calls; +static atomic_int g_bcast_slot_calls[MAX_STANDBYS]; + +static int cv_count_send_hook(void *ep, wal_catchup_entry_t **entries, + int n_entries) +{ + (void)entries; (void)n_entries; + atomic_fetch_add(&g_bcast_total_calls, 1); + for (int i = 0; i < MAX_STANDBYS; i++) { + if (ep == (void *)&g_cv_eps[i]) { + atomic_fetch_add(&g_bcast_slot_calls[i], 1); + break; + } + } + return 0; +} + +static void cv_bcast_reset_counters(void) +{ + atomic_store(&g_bcast_total_calls, 0); + for (int i = 0; i < MAX_STANDBYS; i++) + atomic_store(&g_bcast_slot_calls[i], 0); +} + static ha_config_t cv_ha_cfg(void) { ha_config_t cfg = { @@ -687,6 +720,261 @@ static void test_quorum_second_cycle_logs_again(void) printf("test_quorum_second_cycle_logs_again: checked\n"); } +/* ════════════════════════════════════════════════════════════════════════════ + * Task 6.1 — REGRESSION PINS (verification of existing behavior, NO production + * change). These encode behavior already present in the current tree; each pin + * PASSES against the code as-is. A failure here is a real regression, not a + * RED-phase placeholder. + * + * (a) repl_broadcast full fan-out — umbrella 6.1 pin (Req 6.1) + * (b) ha_evaluate_repl_action snapshot fallback boundary — umbrella 6.3 pin + * (Req 6.1, decision boundary only) + * (c) ha_dispatch_gate degraded-mode allowlist — umbrella 7.3 pin (Req 7.1) + * (d) repl_complete_catchup re-inclusion mechanism — umbrella 6.4 pin + * (Req 7.1, complements task 4's ACK-driven trigger test) + * ════════════════════════════════════════════════════════════════════════════ */ + +/* Drive one repl_broadcast of a single WAL_NOOP entry (no payload). */ +static int cv_broadcast_one(replicator_t *r, uint64_t lsn) +{ + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.lsn = lsn; + hdr.type = WAL_NOOP; + hdr.payload_len = 0; + return repl_broadcast(r, &hdr, NULL, 0); +} + +/* (a) umbrella 6.1: repl_broadcast enqueues to EVERY registered active standby + * and skips ONLY empty slots and in_catch_up standbys. + * + * Fixture: three standby slots — slot 0 active, slot 1 emptied (removed), slot 2 + * active but in_catch_up. A single broadcast must hit slot 0 exactly once and + * NEITHER slot 1 (empty) NOR slot 2 (catch-up). Pins existing repl.c:259-307. */ +static void test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup(void) +{ + cv_setup("fanout", /*n_peers=*/3, /*n_standbys=*/3, + /*appended_lsn=*/0, /*n_vcpus=*/8); + + /* Swap the drop hook for the counting hook and reset its tallies. */ + cv_bcast_reset_counters(); + repl_broadcast_send_hook = cv_count_send_hook; + + /* Arrange the three slots into the three states under test. */ + repl_remove_standby(&g_cv.persist.repl, &g_cv_eps[1]); /* slot 1 → empty */ + repl_start_catchup(&g_cv.persist.repl, &g_cv_eps[2], + /*join_applied_lsn=*/0); /* slot 2 → catch-up */ + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, cv_standby_in_catch_up(&g_cv_eps[2]), + "fixture precondition: slot 2 must be in_catch_up before broadcast"); + + /* Act: one broadcast. */ + int rc = cv_broadcast_one(&g_cv.persist.repl, /*lsn=*/1); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "repl_broadcast must succeed"); + + /* Assert (Req 6.1): the active non-catch-up standby got exactly one send. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_slot_calls[0]), + "repl_broadcast must enqueue to the active non-catch-up standby (slot 0) " + "exactly once (umbrella 6.1 full fan-out)"); + + /* Assert: the EMPTY slot was skipped (no ep → no send). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[1]), + "repl_broadcast must skip the empty slot (slot 1) (umbrella 6.1)"); + + /* Assert: the in_catch_up standby was skipped. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[2]), + "repl_broadcast must skip the in_catch_up standby (slot 2) " + "(umbrella 6.1)"); + + /* Assert: exactly one send total — no over-fan-out. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_total_calls), + "repl_broadcast must enqueue exactly once across all slots in this " + "fixture (one active, one empty, one catch-up) (umbrella 6.1)"); + + cv_teardown(); + printf("test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup: checked\n"); +} + +/* (b) umbrella 6.3: ha_evaluate_repl_action returns HA_REPL_SEND_SNAPSHOT when + * standby_lsn < oldest_wal_lsn (the standby predates the oldest retained WAL, so + * streaming cannot reach it — a fresh snapshot is required). Decision boundary + * only, per the task: at/above oldest_wal_lsn the verdict is NOT snapshot. + * Pins existing ha.c:1274-1275. (The snapshot send path itself — + * ha_send_snapshot_to_standby, config_repl.c:1857 — takes a fresh checkpoint + * before sending; that send-internals behavior is out of scope for this + * decision-boundary pin.) */ +static void test_pin_evaluate_repl_action_snapshot_below_oldest_wal(void) +{ + /* Below the oldest retained WAL → SEND_SNAPSHOT. tags equal (no config push + * shortcut); committed_lsn high so the only deciding factor is the snapshot + * boundary. */ + ha_repl_action_t below = 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)below, + "standby_lsn (2) below oldest_wal_lsn (5) must decide SEND_SNAPSHOT " + "(umbrella 6.3 fallback boundary)"); + + /* Exactly AT oldest_wal_lsn is inside the retained window → NOT snapshot + * (streamable). The boundary is strict-less-than. */ + ha_repl_action_t at_boundary = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/5, + /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5); + TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)at_boundary, + "standby_lsn == oldest_wal_lsn is within the retained WAL window — the " + "snapshot boundary is strict-less-than, so this must NOT decide " + "SEND_SNAPSHOT (umbrella 6.3 boundary)"); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)at_boundary, + "at the snapshot boundary with lsn (5) below committed (9) the standby is " + "streamable — HA_REPL_STREAM_WAL (umbrella 6.3 boundary)"); + + /* Above the oldest WAL and at committed → CONVERGED, never snapshot. */ + ha_repl_action_t converged = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/9, + /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5); + TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)converged, + "a fully caught-up standby must never decide SEND_SNAPSHOT " + "(umbrella 6.3 boundary)"); + + printf("test_pin_evaluate_repl_action_snapshot_below_oldest_wal: checked\n"); +} + +/* ── Degraded-mode dispatch-gate fixture (task 6.1 (c)) ────────────────────── + * Self-contained ha_state_t with role=CTRL_PRIMARY and write quorum ABSENT + * (n_peers=3, zero connected standby endpoints → ha_has_write_quorum false via + * the standby_endpoints fallback count). Mirrors test_commit_barrier_rollout.c's + * fact_gate_setup. No persist/repl needed — the gate reads role + quorum only. */ +static ha_state_t g_cvg_ha; +static int g_cvg_ep; /* dummy endpoint (zarray stores its address) */ + +static void cv_gate_setup(int n_peers, int n_endpoints) +{ + ha_config_t cfg = cv_ha_cfg(); + TEST_ASSERT_EQUAL_INT(0, ha_init(&g_cvg_ha, &cfg, 0)); + g_cvg_ha.n_peers = n_peers; + atomic_store_explicit(&g_cvg_ha.role, (int32_t)CTRL_PRIMARY, + memory_order_release); + g_cvg_ha.standby_endpoints = calloc(1, sizeof(zarray_t)); + TEST_ASSERT_NOT_NULL(g_cvg_ha.standby_endpoints); + zarray_init(g_cvg_ha.standby_endpoints, 4, NULL); + for (int i = 0; i < n_endpoints; i++) + zarray_store(g_cvg_ha.standby_endpoints, &g_cvg_ep); +} + +static void cv_gate_teardown(void) +{ + if (g_cvg_ha.standby_endpoints) { + zarray_fini(g_cvg_ha.standby_endpoints); + free(g_cvg_ha.standby_endpoints); + g_cvg_ha.standby_endpoints = NULL; + } + ha_fini(&g_cvg_ha); +} + +/* (c) umbrella 7.3: without write quorum, ha_dispatch_gate admits OPCode_get, + * heartbeat, the replication-opcode allowlist, and the fact-shaped opcodes + * (notify/update/setpreason/clrpreason — admitted per ha-commit-barrier-rollout's + * fact allowlist), while rejecting request-shaped opcodes + * (jobsub/jobcontrol/queuecontrol/hostcontrol/jobdispatch/join) with + * -XCE_ERR_NOT_READY. Pins existing ha.c:1516-1551. */ +static void test_pin_dispatch_gate_degraded_allowlist(void) +{ + /* Arrange: CTRL_PRIMARY, 3-peer cluster, zero standbys → quorum absent. */ + cv_gate_setup(/*n_peers=*/3, /*n_endpoints=*/0); + + /* ── ADMITTED (return 0) ───────────────────────────────────────────── */ + static const struct { int opcode; const char *name; } admitted[] = { + /* read-only query + probe */ + { OPCode_get, "OPCode_get" }, + { OPCode_probeConfigTag, "OPCode_probeConfigTag" }, + /* heartbeat */ + { OPCode_heartbeat, "OPCode_heartbeat" }, + /* replication allowlist */ + { OPCode_replConfig, "OPCode_replConfig" }, + { OPCode_replConfigPush, "OPCode_replConfigPush" }, + { OPCode_upgradeConfigAck, "OPCode_upgradeConfigAck" }, + { OPCode_hello, "OPCode_hello" }, + { OPCode_walAck, "OPCode_walAck" }, + { OPCode_replEntry, "OPCode_replEntry" }, + { OPCode_replAck, "OPCode_replAck" }, + { OPCode_replCatchUpDone, "OPCode_replCatchUpDone" }, + { OPCode_walCatchupEntry, "OPCode_walCatchupEntry" }, + { OPCode_snapshotTransfer, "OPCode_snapshotTransfer" }, + /* fact-shaped ops (admitted per the fact allowlist) */ + { OPCode_notify, "OPCode_notify" }, + { OPCode_update, "OPCode_update" }, + { OPCode_setpreason, "OPCode_setpreason" }, + { OPCode_clrpreason, "OPCode_clrpreason" }, + }; + for (size_t i = 0; i < sizeof(admitted) / sizeof(admitted[0]); i++) { + int rc = ha_dispatch_gate(&g_cvg_ha, admitted[i].opcode); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, admitted[i].name); + } + + /* ── REJECTED with -XCE_ERR_NOT_READY (request-shaped ops only) ────── */ + static const struct { int opcode; const char *name; } rejected[] = { + { OPCode_jobsub, "OPCode_jobsub" }, + { OPCode_jobcontrol, "OPCode_jobcontrol" }, + { OPCode_queuecontrol, "OPCode_queuecontrol" }, + { OPCode_hostcontrol, "OPCode_hostcontrol" }, + { OPCode_jobdispatch, "OPCode_jobdispatch" }, + { OPCode_join, "OPCode_join" }, + }; + for (size_t i = 0; i < sizeof(rejected) / sizeof(rejected[0]); i++) { + int rc = ha_dispatch_gate(&g_cvg_ha, rejected[i].opcode); + TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc, rejected[i].name); + } + + cv_gate_teardown(); + printf("test_pin_dispatch_gate_degraded_allowlist: checked\n"); +} + +/* (d) umbrella 6.4 mechanism: repl_complete_catchup clears in_catch_up and the + * standby re-enters repl_broadcast iteration (i.e. it is sent to again). + * Complements task 4's ACK-driven *trigger* test by pinning the *mechanism*. + * Pins existing repl.c:602-628 + the broadcast re-inclusion at repl.c:265. */ +static void test_pin_repl_complete_catchup_reincludes_in_broadcast(void) +{ + cv_setup("reinc", /*n_peers=*/3, /*n_standbys=*/1, + /*appended_lsn=*/0, /*n_vcpus=*/8); + + cv_bcast_reset_counters(); + repl_broadcast_send_hook = cv_count_send_hook; + + repl_ep_t ep = &g_cv_eps[0]; + + /* Arrange: put the only standby into catch-up → excluded from broadcast. */ + 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"); + + /* While in catch-up, a broadcast must NOT reach it. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_broadcast_one(&g_cv.persist.repl, 1), + "repl_broadcast must succeed"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[0]), + "an in_catch_up standby must be skipped by repl_broadcast (umbrella 6.4 " + "precondition)"); + + /* Act: complete catch-up → re-inclusion. */ + repl_complete_catchup(&g_cv.persist.repl, ep); + + /* Assert: in_catch_up cleared. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_standby_in_catch_up(ep), + "repl_complete_catchup must clear in_catch_up (umbrella 6.4)"); + + /* Assert: the standby re-enters broadcast iteration — a fresh broadcast now + * reaches it. */ + cv_bcast_reset_counters(); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_broadcast_one(&g_cv.persist.repl, 2), + "repl_broadcast must succeed"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_slot_calls[0]), + "after repl_complete_catchup the standby must re-enter repl_broadcast " + "iteration and receive the next entry (umbrella 6.4 re-inclusion)"); + + cv_teardown(); + printf("test_pin_repl_complete_catchup_reincludes_in_broadcast: checked\n"); +} + #endif /* XCE_TESTING */ /* ── Suite entry point ──────────────────────────────────────────────────── */ @@ -708,6 +996,14 @@ void test_convergence_verification(void) RUN_TEST(test_quorum_loss_no_repeat_log_while_lost); RUN_TEST(test_quorum_restore_logged_once); RUN_TEST(test_quorum_second_cycle_logs_again); + + printf("\n>> Regression pins: fan-out / snapshot fallback / degraded gate / " + "re-inclusion (task 6.1 — PASS against current code)\n\n"); + + RUN_TEST(test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup); + RUN_TEST(test_pin_evaluate_repl_action_snapshot_below_oldest_wal); + RUN_TEST(test_pin_dispatch_gate_degraded_allowlist); + RUN_TEST(test_pin_repl_complete_catchup_reincludes_in_broadcast); #else printf("test_convergence_verification: SKIP (non-test build)\n"); #endif