test(ha-convergence-recovery): RED clamped rejoin reconciliation tests + intake extraction (task 3.1)

.kiro/specs/ha-convergence-recovery/tasks.md

@@ -33,7 +33,7 @@
   - _Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7_
 
 - [ ] 3. Rejoin reconciliation at the CtrlHelloReply intake
-- [ ] 3.1 Write RED tests for clamped rejoin truncation with once-guard
+- [x] 3.1 Write RED tests for clamped rejoin truncation with once-guard
   - Tests in `test_demotion_convergence.c`: `ha_rejoin_reconcile` with local tip > primary boundary → truncation at `max(primary_committed, local committed_lsn)` (clamp verified with a local committed_lsn ABOVE the reported boundary) and `ha_restart_hook` invoked once; tip <= boundary → no truncation, no restart, returns 0; second reconcile call in the same process lifetime → no second restart (defensive once-guard); intake integration: delivering a CtrlHelloReply through the cluster.c handler path stores `last_known_committed_lsn` (existing) AND triggers reconciliation
   - Observable completion: new cases FAIL (red) — no reconciliation exists at the intake (cluster.c:283-298 anchor)
   - _Requirements: 2.1, 2.2, 2.3_

src/controller/cluster.c

@@ -287,14 +287,9 @@ static int myreq_handler(vtask_t* task)
             /* Standby receives CtrlHelloReply from primary — update last known
              * committed LSN and config tag for use in RDHUP failover path. */
             ctrl_hello_reply_t *chrep = request.body.replymsg->data.ctrlhelloreplymsg;
-            if (chrep) {
-                atomic_store_explicit(&c->ha.last_known_committed_lsn,
-                                       chrep->committed_lsn, memory_order_release);
-                c->ha.last_known_committed_config_tag = chrep->config_tag;
-                INFO("HA: CtrlHelloReply from primary: committed_lsn=%lu config_tag=%lu",
-                     (unsigned long)chrep->committed_lsn,
-                     (unsigned long)chrep->config_tag);
-            }
+            if (chrep)
+                cluster_apply_ctrlhello_reply(c, chrep->committed_lsn,
+                                              chrep->config_tag);
         } else {
             /* Legacy plain OP_hello (nop path) */
             reply.status     = 0;

src/controller/ha.h

@@ -193,6 +193,13 @@ typedef struct {
                                         * repl_active_standby_count (ep != NULL && !in_catch_up);
                                         * NULL → legacy standby_endpoints->count fallback.
                                         * (Req 3.1-3.3, 6.1, 6.2; wired at init by task 5.3.) */
+    /* ── New field added by ha-convergence-recovery spec (Task 3.x) ──────── */
+    _Atomic int      rejoin_reconciled; /* Defensive once-guard for rejoin
+                                         * reconciliation (Req 2.1): set after the
+                                         * first truncate+self-restart so a second
+                                         * CtrlHelloReply in the same process
+                                         * lifetime cannot re-fire the restart.
+                                         * 0 at init (cleared by ha_init memset). */
 } ha_state_t;
 
 /* ── Public API ────────────────────────────────────────────── */
@@ -293,6 +300,44 @@ int ha_schedule_demotion_convergence(cluster_t *c);
 int ha_demotion_convergence_fn(void *arg);
 #endif
 
+/*
+ * 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,
+ * design "Rejoin Reconciliation", research D5).
+ *
+ * Called from the standby's CtrlHelloReply intake (cluster.c) after the existing
+ * last_known_committed_lsn store, with primary_committed_lsn = the boundary the
+ * primary reported.
+ *
+ * Contract (task 3.2 GREEN implements; task 3.1 ships a RED no-op stub):
+ *   - local_tip = persist.current_lsn - 1.
+ *   - If local_tip > primary_committed_lsn AND the rejoin_reconciled once-guard
+ *     is clear: cutoff = max(primary_committed_lsn, local committed_lsn) — the
+ *     CLAMP that never discards the locally known committed prefix (Req 2.2) —
+ *     persist_truncate_wal_above(cutoff), WARN the discarded range, set the
+ *     once-guard, then ha_self_restart (deferred to delaymgr if the intake runs
+ *     on iomgr, same rule as demotion).  Returns the restart result.
+ *   - Else (tip <= boundary): return 0 with NO side effects (Req 2.3).
+ *   - A second call in the same process lifetime takes no restart (Req 2.1 —
+ *     defensive once-guard; "once" is also structural since post-restart the tip
+ *     equals the boundary).
+ *
+ * Generalizes the RDHUP truncation pattern (ha_on_primary_ctrl_fd_rdhup), which
+ * remains in place unchanged for the failover path.
+ */
+int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn);
+
+/*
+ * cluster_apply_ctrlhello_reply: standby-side CtrlHelloReply intake body
+ * (ha-convergence-recovery task 3.1).  Stores the primary-reported committed
+ * boundary + config tag; called from cluster.c's myreq_handler on every
+ * CtrlHelloReply.  Defined in ha_convergence.c (a unit-linkable TU) so the
+ * rejoin intake-integration test can drive the real path.  Task 3.2 adds the
+ * ha_rejoin_reconcile trigger after the store.
+ */
+void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn,
+                                   uint64_t config_tag);
+
 
 /* ── Redirect helpers used by join.c ──────────────────────── */
 

src/controller/ha_convergence.c

@@ -210,3 +210,49 @@ int ha_schedule_demotion_convergence(cluster_t *c)
     }
     return 0;
 }
+
+/*
+ * ha_rejoin_reconcile — RED scaffold (ha-convergence-recovery task 3.1).
+ *
+ * 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).
+ *
+ * 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.
+ */
+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 */
+}
+
+/*
+ * cluster_apply_ctrlhello_reply — standby-side CtrlHelloReply intake body.
+ *
+ * Lives in this (unit-linkable) TU rather than cluster.c so the rejoin
+ * intake-integration test can drive the real path: cluster.c (the daemon-only
+ * vtask handler graph) is NOT part of controller_unit_test_src, but
+ * ha_convergence.c is.  cluster.c's myreq_handler calls this on every
+ * 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.
+ */
+void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn,
+                                   uint64_t config_tag)
+{
+    atomic_store_explicit(&c->ha.last_known_committed_lsn, committed_lsn,
+                          memory_order_release);
+    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 */
+}

src/controller/testsuite/test_demotion_convergence.c

@@ -514,6 +514,204 @@ static void test_demotion_replay_yields_committed_only(void)
     printf("test_demotion_replay_yields_committed_only: checked\n");
 }
 
+/* ════════════════════════════════════════════════════════════════════════════
+ * Rejoin reconciliation (ha-convergence-recovery task 3.1 — RED)
+ *
+ * Requirements (this spec's IDs — see requirements.md):
+ *   2.1 — when a standby learns the primary's committed boundary at (re)join and
+ *         its local log tip EXCEEDS that boundary, it truncates above the boundary
+ *         and self-restarts ONCE.  A second reconcile call in the same process
+ *         lifetime must NOT restart again (defensive once-guard).
+ *   2.2 — the standby must NOT remove entries at/below its own locally known
+ *         committed_lsn during reconciliation (clamp): the truncation cutoff is
+ *         max(primary_committed, local committed_lsn).  Verified by setting the
+ *         local committed_lsn ABOVE the reported boundary and asserting the cutoff
+ *         is the higher local value, NOT the reported boundary.
+ *   2.3 — while the local tip does NOT exceed the boundary, the standby joins with
+ *         NO truncation and NO restart, returning 0.
+ *   (intake integration) — delivering a CtrlHelloReply through the real cluster.c
+ *         intake stores last_known_committed_lsn (existing) AND triggers
+ *         reconciliation.
+ *
+ * RED status against the current tree: ha_rejoin_reconcile is a no-op scaffold
+ * (ha_convergence.c) and the cluster.c CtrlHelloReply intake does NOT call it yet
+ * (task 3.2 GREEN wires both).  Every case below therefore FAILS until 3.2 lands.
+ *
+ * Harness: reuses the dc_* fixture (real persist over a tmpdir, restart/fatal
+ * seam counters).  The fixture comes up CTRL_PRIMARY; rejoin reconciliation is a
+ * STANDBY-side concern, so each rejoin test first demotes to STANDBY (which also
+ * clears the persist hooks — irrelevant to reconciliation) before exercising it.
+ * AAA structure throughout.
+ * ════════════════════════════════════════════════════════════════════════════ */
+
+/* Set the local on-disk WAL tip to `tip_lsn` (current_lsn = tip_lsn + 1) and the
+ * local committed_lsn to `committed`.  Mirrors dc_setup's priming convention. */
+static void rj_set_local_state(uint64_t tip_lsn, uint64_t committed)
+{
+    pthread_mutex_lock(&g_dc.persist.lock);
+    g_dc.persist.current_lsn = tip_lsn + 1;
+    pthread_mutex_unlock(&g_dc.persist.lock);
+    atomic_store(&g_dc.persist.committed_lsn, committed);
+}
+
+/* Demote the fixture to STANDBY (the role rejoin reconciliation runs in) and
+ * install the restart/fatal seams so the self-restart is observable. */
+static void rj_make_standby_with_seams(void)
+{
+    TEST_ASSERT_EQUAL_INT(0, ha_demote_to_standby(&g_dc));
+    TEST_ASSERT_EQUAL_INT((int)CTRL_STANDBY,
+        (int)atomic_load_explicit(&g_dc.ha.role, memory_order_acquire));
+    ha_restart_hook   = dc_restart_hook;
+    ha_fatal_hook     = dc_fatal_hook;
+    g_restart_hook_rc = 0; /* model a successful restart */
+}
+
+/* 2.1/2.2: local tip ABOVE the primary boundary AND local committed_lsn ABOVE the
+ * boundary → reconcile truncates at the CLAMP max(boundary, local committed) (the
+ * higher local value, never the lower reported boundary) and self-restarts once.
+ * RED: ha_rejoin_reconcile is a no-op stub → no truncation, no restart. */
+static void test_rejoin_clamp_truncates_at_local_committed_and_restarts_once(void)
+{
+    dc_setup("rjclamp", /*n_peers=*/3, /*n_standbys=*/0,
+             /*appended_lsn=*/9, /*n_vcpus=*/8);
+    rj_make_standby_with_seams();
+
+    /* Local tip = 9 (entries 1..9 on disk).  Local committed_lsn = 6 — ABOVE the
+     * primary-reported boundary of 4.  The clamp must protect [1..6] and truncate
+     * only above 6 → cutoff = max(4, 6) = 6 → current_lsn collapses to 7. */
+    rj_set_local_state(/*tip_lsn=*/9, /*committed=*/6);
+    const uint64_t primary_boundary = 4;
+    const uint64_t local_committed   = 6;
+
+    /* Act. */
+    int rc = ha_rejoin_reconcile(&g_dc, primary_boundary);
+    /* Drive any deferred convergence task (the intake may defer to delaymgr,
+     * same rule as demotion); harmless if 3.2 truncates inline. */
+    (void)dc_run_pending_task(&g_dc.delaymgr);
+
+    /* Assert (Req 2.2 clamp): truncation cutoff is the LOCAL committed (6), not the
+     * lower reported boundary (4) — current_lsn collapses to local_committed + 1. */
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(local_committed + 1, g_dc.persist.current_lsn,
+        "rejoin reconcile must clamp the cutoff to max(boundary, local committed) "
+        "(Req 2.2): with boundary=4 and local committed=6 it must truncate above 6, "
+        "NOT above the lower reported boundary 4");
+
+    /* Assert (Req 2.1): self-restart fired exactly once, fatal seam never. */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
+        "rejoin reconcile with tip > boundary must self-restart exactly once "
+        "(Req 2.1)");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls),
+        "a successful rejoin restart must NOT invoke the fatal seam");
+    (void)rc;
+
+    dc_teardown();
+    printf("test_rejoin_clamp_truncates_at_local_committed_and_restarts_once: checked\n");
+}
+
+/* 2.3: local tip AT/BELOW the boundary → no truncation, no restart, returns 0.
+ * RED: the no-op stub returns 0 with no side effects, so the no-action assertions
+ * PASS but the suite as a whole is RED via the other cases; this case pins the
+ * negative contract so 3.2 cannot over-truncate.  (It is the one rejoin case that
+ * a correct no-op stub already satisfies — kept to lock the boundary for GREEN.) */
+static void test_rejoin_tip_at_or_below_boundary_no_action(void)
+{
+    dc_setup("rjnoop", /*n_peers=*/3, /*n_standbys=*/0,
+             /*appended_lsn=*/4, /*n_vcpus=*/8);
+    rj_make_standby_with_seams();
+
+    /* Local tip = 4, exactly the primary boundary; nothing to discard. */
+    rj_set_local_state(/*tip_lsn=*/4, /*committed=*/4);
+    uint64_t tip_before = g_dc.persist.current_lsn;
+
+    /* Act. */
+    int rc = ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
+
+    /* Assert (Req 2.3): no side effects, returns 0. */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
+        "rejoin reconcile with tip <= boundary must return 0 (Req 2.3)");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
+        "rejoin reconcile with tip <= boundary must NOT truncate (Req 2.3)");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
+        "rejoin reconcile with tip <= boundary must NOT self-restart (Req 2.3)");
+
+    dc_teardown();
+    printf("test_rejoin_tip_at_or_below_boundary_no_action: checked\n");
+}
+
+/* 2.1 once-guard: a SECOND reconcile call in the same process lifetime must NOT
+ * self-restart again, even if the tip still exceeds the boundary (defensive guard;
+ * "once" is also structural post-restart).
+ * RED: the no-op stub never restarts even once, so the "exactly one restart total"
+ * assertion fails. */
+static void test_rejoin_second_call_no_second_restart(void)
+{
+    dc_setup("rjonce", /*n_peers=*/3, /*n_standbys=*/0,
+             /*appended_lsn=*/9, /*n_vcpus=*/8);
+    rj_make_standby_with_seams();
+    rj_set_local_state(/*tip_lsn=*/9, /*committed=*/4);
+
+    /* Act: first reconcile (tip 9 > boundary 4 → truncate + restart once). */
+    (void)ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
+    (void)dc_run_pending_task(&g_dc.delaymgr);
+    int restarts_after_first = atomic_load(&g_restart_calls);
+
+    /* Act: second reconcile in the SAME lifetime.  The once-guard must suppress a
+     * second restart regardless of the reported boundary. */
+    (void)ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
+    (void)dc_run_pending_task(&g_dc.delaymgr);
+
+    /* Assert (Req 2.1): exactly ONE restart across both calls. */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, restarts_after_first,
+        "the FIRST rejoin reconcile with tip > boundary must self-restart once "
+        "(Req 2.1)");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
+        "a SECOND rejoin reconcile in the same process lifetime must NOT restart "
+        "again — defensive once-guard (Req 2.1)");
+
+    dc_teardown();
+    printf("test_rejoin_second_call_no_second_restart: checked\n");
+}
+
+/* Intake integration: delivering a CtrlHelloReply through the REAL cluster.c
+ * intake (cluster_apply_ctrlhello_reply) stores last_known_committed_lsn AND
+ * triggers reconciliation.
+ * RED: the intake stores the boundary (passes) but does NOT call
+ * ha_rejoin_reconcile yet (task 3.2 wires it), so the restart assertion fails. */
+static void test_ctrlhello_reply_intake_stores_and_reconciles(void)
+{
+    dc_setup("rjintake", /*n_peers=*/3, /*n_standbys=*/0,
+             /*appended_lsn=*/9, /*n_vcpus=*/8);
+    rj_make_standby_with_seams();
+    /* Local tip 9 > the boundary the primary will report (4) → reconcile must fire
+     * from inside the intake. */
+    rj_set_local_state(/*tip_lsn=*/9, /*committed=*/4);
+
+    const uint64_t reported_committed = 4;
+    const uint64_t reported_tag       = 77;
+
+    /* Act: drive the real standby-side CtrlHelloReply intake. */
+    cluster_apply_ctrlhello_reply(&g_dc, reported_committed, reported_tag);
+    (void)dc_run_pending_task(&g_dc.delaymgr);
+
+    /* Assert (existing behavior): the boundary + config tag are stored. */
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(reported_committed,
+        atomic_load(&g_dc.ha.last_known_committed_lsn),
+        "CtrlHelloReply intake must store last_known_committed_lsn (existing)");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(reported_tag,
+        g_dc.ha.last_known_committed_config_tag,
+        "CtrlHelloReply intake must store the config tag (existing)");
+
+    /* Assert (intake integration): the intake also TRIGGERED reconciliation —
+     * tip 9 > boundary 4 → exactly one self-restart. */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
+        "CtrlHelloReply intake must trigger rejoin reconciliation when the local "
+        "tip exceeds the reported boundary — exactly one self-restart (task 3.2 "
+        "wires ha_rejoin_reconcile into the intake)");
+
+    dc_teardown();
+    printf("test_ctrlhello_reply_intake_stores_and_reconciles: checked\n");
+}
+
 #endif /* XCE_TESTING */
 
 /* ── Suite entry point ──────────────────────────────────────────────────── */
@@ -530,6 +728,13 @@ void test_demotion_convergence(void)
     RUN_TEST(test_demotion_from_electing_no_truncation_no_restart);
     RUN_TEST(test_demotion_from_init_no_truncation_no_restart);
     RUN_TEST(test_demotion_replay_yields_committed_only);
+
+    printf("\n>> rejoin reconciliation tests (RED, task 3.1)\n\n");
+
+    RUN_TEST(test_rejoin_clamp_truncates_at_local_committed_and_restarts_once);
+    RUN_TEST(test_rejoin_tip_at_or_below_boundary_no_action);
+    RUN_TEST(test_rejoin_second_call_no_second_restart);
+    RUN_TEST(test_ctrlhello_reply_intake_stores_and_reconciles);
 #else
     printf("test_demotion_convergence: SKIP (non-test build)\n");
 #endif