myxce/src/controller/testsuite/test_demotion_convergence.c

#define _GNU_SOURCE
/*
 * test_demotion_convergence.c — RED-phase tests for the demotion-convergence
 * sequence and the self-restart primitive (ha-convergence-recovery task 2.1).
 *
 * Requirements (this spec's IDs — see requirements.md):
 *   1.1 — on demotion from PRIMARY the controller STOPS persisting new state
 *         BEFORE continuing demotion (persistence hooks disabled synchronously,
 *         observed via hooks_registered reset + a hook-counter probe).
 *   1.2 — on demotion the controller removes all log entries ABOVE committed_lsn
 *         (truncate above the committed boundary).
 *   1.3 — when truncation completes the controller self-restarts so in-memory
 *         state is rebuilt by boot recovery from the committed prefix only.
 *   1.4 — if the self-restart cannot be performed, terminate FATALLY rather
 *         than continue (FATAL contract; XCE_TESTING routes to ha_fatal_hook).
 *   1.5 — the demotion is logged with the truncation boundary (committed LSN and
 *         prior log tip) before restarting (observed here via the truncation
 *         cutoff: current_lsn collapses to committed_lsn + 1).
 *   1.6 — under test mode the controller invokes the ha_restart_hook seam in
 *         place of process re-execution.
 *   1.7 — in-memory state is reconstructible from the persisted committed state
 *         only (provisional entries above committed_lsn absent after replay).
 *   2.4 — single-outcome convergence (in-process half): a committed prefix +
 *         uncommitted suffix, run through the demotion seam, replays to a
 *         committed-only state.
 *
 * RED status against the current tree: raft_cb_become_follower (ha.c) neither
 * disables the persistence hooks nor schedules any truncate/restart work on the
 * PRIMARY→follower edge, and ha_self_restart is a RED scaffold (ha_convergence.c)
 * that does not yet honor the restart/fatal seams.  Every case below therefore
 * FAILS until task 2.2 lands the demotion branch + the real primitive.
 *
 * Harness: a partial-but-real cluster_t (real persist over a private tmpdir,
 * wired commit engine, stub standby registry, un-started delaymgr) mirroring
 * test_ha_role_lifecycle.c's lifecycle_cluster_setup, plus the live-cluster
 * become-follower seam raft_cb_become_follower_with_cluster_for_test.
 *
 * AAA structure throughout.
 */
#include <assert.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

#include <unity/unity.h>

#include "commit.h"            /* commit_engine_t, commit_engine_init */
#include "delay.h"             /* delaymgr_t, delay_task_t */
#include "ha.h"                /* raft_cb_become_follower_with_cluster_for_test,
                                 * ha_self_restart, controller_saved_argv,
                                 * ha_restart_hook / ha_fatal_hook seams */
#include "ordq.h"             /* ordq_item_t (drive the scheduled task) */
#include "persist.h"          /* persist_t, persist_truncate_wal_above */
#include "repl.h"             /* repl_add_standby / repl_set_commit_engine */
#include "xce/cluster_def.h"  /* full cluster_t layout */

#ifdef XCE_TESTING

/* ── Broadcast drop-hook (precedent: test_ha_role_lifecycle.c lc_drop_send_hook):
 * the stub int endpoints registered via repl_add_standby are not real
 * endpoint_t, so repl_broadcast must never dereference ep->peer.  Installing
 * this hook drops the wire send harmlessly during priming appends. */
extern int (*repl_broadcast_send_hook)(void *ep,
                                       wal_catchup_entry_t **entries,
                                       int n_entries);
static int dc_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
                             int n_entries)
{
    (void)ep; (void)entries; (void)n_entries;
    return 0;
}

/* ── Restart / fatal seam counters (Req 1.4, 1.6) ───────────────────────── */
static atomic_int g_restart_calls;
static atomic_int g_fatal_calls;
static int        g_restart_hook_rc; /* return code dc_restart_hook reports */

static int dc_restart_hook(cluster_t *c)
{
    (void)c;
    atomic_fetch_add(&g_restart_calls, 1);
    return g_restart_hook_rc;
}

static void dc_fatal_hook(const char *msg)
{
    (void)msg;
    atomic_fetch_add(&g_fatal_calls, 1);
}

/* ── Cluster fixture (mirror of lifecycle_cluster_setup) ────────────────── */
static cluster_t g_dc;            /* large — must not live on the test stack */
static int       g_dc_eps[MAX_STANDBYS];
static char      g_dc_workdir[256]; /* the fixture tmpdir (workdir for replay) */
#define DC_PERSIST_NAME "ctrl-test"

static ha_config_t dc_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_dc: 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 forced to CTRL_PRIMARY with the persist hooks registered (so a demotion
 * has hooks to disable).  Leaves committed_lsn at 0; callers set it. */
static void dc_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_dc, 0, sizeof(g_dc));
    atomic_store(&g_restart_calls, 0);
    atomic_store(&g_fatal_calls, 0);
    g_restart_hook_rc = 0;

    ha_config_t cfg = dc_ha_cfg();
    TEST_ASSERT_EQUAL_INT(0, ha_init(&g_dc.ha, &cfg, 0));
    g_dc.ha.phase2_done = 1;       /* no phase2 wiring in this fixture */
    g_dc.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_dc_%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
    snprintf(g_dc_workdir, sizeof(g_dc_workdir), "%s", tmpdir);
    persist_set_workdir(&g_dc.persist, tmpdir, DC_PERSIST_NAME);
    TEST_ASSERT_EQUAL_INT(0, persist_init(&g_dc.persist, &pcfg, &g_dc));
    TEST_ASSERT_EQUAL_INT(0, persist_start(&g_dc.persist));

    /* Prime the appended watermark: current_lsn = appended_lsn + 1. */
    pthread_mutex_lock(&g_dc.persist.lock);
    g_dc.persist.current_lsn = appended_lsn + 1;
    pthread_mutex_unlock(&g_dc.persist.lock);
    atomic_store(&g_dc.persist.committed_lsn, 0);
    /* ha-commit-durability 2.2: self caps at durable_lsn. This fixture primes
     * the WAL tip in-memory (no bytes written → the flusher can't advance
     * durable_lsn itself), so mark the primed log durable up front to preserve
     * the pre-durability "self covers every appended entry" commit model. */
    atomic_store(&g_dc.persist.durable_lsn, appended_lsn);

    for (int i = 0; i < n_standbys; i++) {
        snprintf(name, sizeof(name), "dcsb%d", i);
        TEST_ASSERT_TRUE_MESSAGE(
            repl_add_standby(&g_dc.persist.repl, &g_dc_eps[i], name,
                             7500 + (uint32_t)i) >= 0,
            "repl_add_standby must return a slot index >= 0");
    }
    TEST_ASSERT_EQUAL_INT(0,
        commit_engine_init(&g_dc.commit, &g_dc.persist, n_peers, n_vcpus));
    repl_set_commit_engine(&g_dc.persist.repl, &g_dc.commit);

    repl_broadcast_send_hook = dc_drop_send_hook;

    /* delaymgr initialised but NOT started — the demotion-convergence task,
     * once task 2.2 schedules it, lands in tasks.head and is driven manually
     * by dc_run_pending_task (D1: truncation + restart run off the callback). */
    TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_dc.delaymgr, 100, NULL));

    /* Make this a PRIMARY with hooks registered, so a demotion has the three
     * persistence hooks to disable (Req 1.1).  ha_promote_to_primary runs the
     * shared promotion sequence (ha_do_promote → register hooks → heartbeat →
     * own-term gate); the un-started delaymgr keeps the heartbeat arm skipped. */
    TEST_ASSERT_EQUAL_INT(0, ha_promote_to_primary(&g_dc));
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        (int)CTRL_PRIMARY,
        (int)atomic_load_explicit(&g_dc.ha.role, memory_order_acquire),
        "fixture precondition: node must be CTRL_PRIMARY before demotion");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_dc.ha.hooks_registered,
        "fixture precondition: persist hooks must be registered before demotion");
}

static void dc_teardown(void)
{
    repl_broadcast_send_hook = NULL;
    ha_restart_hook = NULL;
    ha_fatal_hook   = NULL;
    commit_engine_fini(&g_dc.commit);
    delaymgr_fini(&g_dc.delaymgr);
    persist_stop(&g_dc.persist);
    persist_fini(&g_dc.persist);
    repl_fini(&g_dc.persist.repl);
    ha_fini(&g_dc.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).  Same seam the lifecycle
 * suite uses to assert promotion arms no watchdog. */
static int dc_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 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(delaymgr_t *mgr)
{
    ordq_item_t *it = ordq_get(&mgr->tasks, NULL);
    if (it == NULL || it->data == NULL) {
        if (it != NULL)
            free(it);
        return DC_NO_TASK;
    }
    delay_task_t *t = (delay_task_t *)it->data;
    if (t->entry == NULL) {
        free(t);
        free(it);
        return DC_NO_TASK;
    }
    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 ──────────────────────────────────────────────────────────────── */

/* 1.1: demotion from PRIMARY disables the persistence hooks synchronously in
 * the callback (before any further event could be persisted), observed via
 * hooks_registered transitioning 1 → 0.
 * RED: raft_cb_become_follower never disables hooks on the PRIMARY edge. */
static void test_demotion_from_primary_disables_hooks(void)
{
    dc_setup("hooks", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/5, /*n_vcpus=*/8);
    atomic_store(&g_dc.persist.committed_lsn, 3);

    /* Act: drive become-follower from PRIMARY with a LIVE cluster. */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7);

    /* Assert: hooks disabled synchronously (Req 1.1). */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_dc.ha.hooks_registered,
        "demotion from PRIMARY must disable the persistence hooks "
        "synchronously (hooks_registered 1→0, Req 1.1)");

    dc_teardown();
    printf("test_demotion_from_primary_disables_hooks: checked\n");
}

/* 1.2/1.3/1.5/1.6: demotion from PRIMARY schedules exactly one convergence task
 * (off the callback thread, D1); running it truncates above committed_lsn and
 * invokes the restart seam exactly once.
 * RED: no task is scheduled, so dc_run_pending_task returns DC_NO_TASK and the
 * truncation/restart assertions fail. */
static void test_demotion_schedules_convergence_truncate_and_restart(void)
{
    dc_setup("conv", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/9, /*n_vcpus=*/8);
    /* committed prefix ends at LSN 4; entries 5..9 are the uncommitted suffix. */
    atomic_store(&g_dc.persist.committed_lsn, 4);

    /* Install the test restart seam so the eventual self-restart is observable
     * deterministically rather than exec'ing the test binary (Req 1.6). */
    ha_restart_hook  = dc_restart_hook;
    ha_fatal_hook    = dc_fatal_hook;
    g_restart_hook_rc = 0; /* model a successful restart */

    int tasks_before = dc_task_count(&g_dc.delaymgr);

    /* Act: demotion edge. */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7);

    /* The convergence work runs OFF the callback (D1) — exactly one one-shot
     * delaymgr task must have been scheduled (Req 1.2/1.3 deferred). */
    int tasks_after = dc_task_count(&g_dc.delaymgr);
    TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before + 1, tasks_after,
        "demotion from PRIMARY must schedule exactly one convergence task "
        "(truncate+restart off the callback thread, D1)");

    /* Run the scheduled convergence task synchronously. */
    int rc = dc_run_pending_task(&g_dc.delaymgr);
    TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc,
        "no convergence task was scheduled on the demotion edge (Req 1.2/1.3)");

    /* Assert (Req 1.2/1.5): truncation removed everything above committed_lsn —
     * current_lsn collapses to committed_lsn + 1 (the truncation boundary). */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        atomic_load(&g_dc.persist.committed_lsn) + 1, g_dc.persist.current_lsn,
        "demotion convergence must truncate above committed_lsn (Req 1.2): "
        "current_lsn must collapse to committed_lsn + 1");

    /* Assert (Req 1.3/1.6): the restart seam was invoked exactly once. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
        "demotion convergence must invoke the restart seam exactly once "
        "after truncation (Req 1.3, 1.6)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls),
        "a successful restart must NOT invoke the fatal seam");

    dc_teardown();
    printf("test_demotion_schedules_convergence_truncate_and_restart: checked\n");
}

/* 1.4: ha_self_restart with a FAILING restart hook must invoke the fatal seam
 * (the FATAL contract — a divergent ex-primary must never continue serving).
 * RED: the ha_self_restart scaffold (ha_convergence.c) does not route to the
 * seams yet, so neither hook fires. */
static void test_self_restart_failure_invokes_fatal_hook(void)
{
    dc_setup("fatal", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/5, /*n_vcpus=*/8);

    ha_restart_hook   = dc_restart_hook;
    ha_fatal_hook     = dc_fatal_hook;
    g_restart_hook_rc = -EIO; /* model an exec failure */

    /* Act: invoke the primitive directly. */
    (void)ha_self_restart(&g_dc);

    /* Assert (Req 1.4/1.6): the restart seam was attempted, and because it
     * failed, the fatal seam fired in place of abort(). */
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
        "ha_self_restart must invoke the restart seam under XCE_TESTING (Req 1.6)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_fatal_calls),
        "a failing restart must invoke the fatal seam — the FATAL contract "
        "(Req 1.4); the ex-primary must never continue with divergent state");

    dc_teardown();
    printf("test_self_restart_failure_invokes_fatal_hook: checked\n");
}

/* Regression: demotion from STANDBY must NOT truncate, NOT restart, NOT disable
 * hooks beyond today's behavior — byte-equivalent to the pre-spec path.  A
 * STANDBY has no registered hooks and no convergence work. */
static void test_demotion_from_standby_no_truncation_no_restart(void)
{
    dc_setup("standby", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/9, /*n_vcpus=*/8);
    atomic_store(&g_dc.persist.committed_lsn, 4);

    /* Force role to STANDBY and disable hooks (the non-PRIMARY precondition). */
    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;

    uint64_t tip_before   = g_dc.persist.current_lsn;
    int      tasks_before = dc_task_count(&g_dc.delaymgr);

    /* Act: a second become-follower from a non-PRIMARY role. */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/8);

    /* Assert: no convergence task, no truncation, no restart. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before,
        dc_task_count(&g_dc.delaymgr),
        "demotion from STANDBY must schedule NO convergence task (regression)");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
        "demotion from STANDBY must NOT truncate the WAL (regression)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
        "demotion from STANDBY must NOT self-restart (regression)");

    dc_teardown();
    printf("test_demotion_from_standby_no_truncation_no_restart: checked\n");
}

/* Regression: demotion from ELECTING must NOT truncate or restart. */
static void test_demotion_from_electing_no_truncation_no_restart(void)
{
    dc_setup("electing", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/9, /*n_vcpus=*/8);
    atomic_store(&g_dc.persist.committed_lsn, 4);

    /* No registered hooks on a candidate; force role to ELECTING. */
    TEST_ASSERT_EQUAL_INT(0, ha_demote_to_standby(&g_dc)); /* clears hooks */
    atomic_store_explicit(&g_dc.ha.role, (int32_t)CTRL_ELECTING,
                          memory_order_release);

    ha_restart_hook = dc_restart_hook;
    ha_fatal_hook   = dc_fatal_hook;

    uint64_t tip_before   = g_dc.persist.current_lsn;
    int      tasks_before = dc_task_count(&g_dc.delaymgr);

    /* Act. */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/8);

    /* Assert. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before,
        dc_task_count(&g_dc.delaymgr),
        "demotion from ELECTING must schedule NO convergence task (regression)");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
        "demotion from ELECTING must NOT truncate the WAL (regression)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
        "demotion from ELECTING must NOT self-restart (regression)");

    dc_teardown();
    printf("test_demotion_from_electing_no_truncation_no_restart: checked\n");
}

/* Regression: demotion from INIT must NOT truncate or restart (mirror of the
 * ELECTING case — a CTRL_INIT node has no registered persistence hooks and no
 * provisional suffix, so the was-PRIMARY convergence branch must not fire). */
static void test_demotion_from_init_no_truncation_no_restart(void)
{
    dc_setup("init", /*n_peers=*/3, /*n_standbys=*/2,
             /*appended_lsn=*/9, /*n_vcpus=*/8);
    atomic_store(&g_dc.persist.committed_lsn, 4);

    /* No registered hooks on a starting node; force role to INIT. */
    TEST_ASSERT_EQUAL_INT(0, ha_demote_to_standby(&g_dc)); /* clears hooks */
    atomic_store_explicit(&g_dc.ha.role, (int32_t)CTRL_INIT,
                          memory_order_release);

    ha_restart_hook = dc_restart_hook;
    ha_fatal_hook   = dc_fatal_hook;

    uint64_t tip_before   = g_dc.persist.current_lsn;
    int      tasks_before = dc_task_count(&g_dc.delaymgr);

    /* Act. */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/8);

    /* Assert. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before,
        dc_task_count(&g_dc.delaymgr),
        "demotion from INIT must schedule NO convergence task (regression)");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
        "demotion from INIT must NOT truncate the WAL (regression)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
        "demotion from INIT must NOT self-restart (regression)");

    dc_teardown();
    printf("test_demotion_from_init_no_truncation_no_restart: checked\n");
}

/* 1.7/2.4 (integration, in-process half): append a committed prefix + an
 * uncommitted suffix, run the demotion seam, then replay the SURVIVING WAL into
 * a FRESH persist context.  The replayed state must contain the committed prefix
 * only — provisional entries above committed_lsn must be absent.
 * RED: the demotion seam does not truncate, so the suffix survives into the
 * fresh replay (current_lsn after recovery exceeds committed_lsn + 1). */
static void test_demotion_replay_yields_committed_only(void)
{
    dc_setup("replay", /*n_peers=*/1, /*n_standbys=*/0,
             /*appended_lsn=*/0, /*n_vcpus=*/8);

    /* Build a real on-disk log: priming + a committed prefix + an uncommitted
     * suffix.  With n_standbys=0 the sync append returns without majority but
     * the LOCAL WAL write lands (production keeps the entry on replication
     * timeout); we only need the bytes on disk. */
    for (int i = 0; i < 4; i++)
        (void)persist_wal_async(&g_dc.persist, WAL_JOB_UPDATE, "p", 1);
    uint64_t committed_boundary = g_dc.persist.current_lsn - 1; /* committed tip */
    atomic_store(&g_dc.persist.committed_lsn, committed_boundary);

    /* Uncommitted suffix (entries above the committed boundary). */
    for (int i = 0; i < 3; i++)
        (void)persist_wal_async(&g_dc.persist, WAL_JOB_UPDATE, "u", 1);
    uint64_t tip_before = g_dc.persist.current_lsn - 1;
    TEST_ASSERT_GREATER_THAN_UINT64_MESSAGE(committed_boundary, tip_before,
        "fixture: the uncommitted suffix must sit above the committed boundary");

    /* Persist the on-disk WAL so a fresh context can replay the same segments. */
    (void)persist_flush(&g_dc.persist);

    /* Capture the workdir BEFORE teardown (the demotion seam restarts/converges
     * the live context; we then boot-recover a fresh persist over the surviving
     * segments using the SAME workdir + name — the boot-recovery contract). */
    char workdir[256];
    snprintf(workdir, sizeof(workdir), "%s", g_dc_workdir);

    ha_restart_hook   = dc_restart_hook;
    ha_fatal_hook     = dc_fatal_hook;
    g_restart_hook_rc = 0;

    /* Act: run the demotion seam (schedules + we drive the convergence task). */
    raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7);
    int rc = dc_run_pending_task(&g_dc.delaymgr);
    TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc,
        "demotion must schedule a convergence task that truncates (Req 1.2)");

    /* Boot-recover a FRESH persist context over the surviving WAL (same workdir +
     * name → same <workdir>/<name>/data/wal segments the demotion seam left).
     * This is the boot-recovery path the real self-restart re-enters (Req 1.7). */
    persist_t *fresh = calloc(1, sizeof(persist_t));
    TEST_ASSERT_NOT_NULL(fresh);
    persist_config_t fcfg = persist_config_defaults();
    fcfg.wal_sync_ms = 5000;
    persist_set_workdir(fresh, workdir, DC_PERSIST_NAME);
    TEST_ASSERT_EQUAL_INT(0, persist_init(fresh, &fcfg, NULL));

    int recov = persist_recover(fresh);
    TEST_ASSERT_TRUE_MESSAGE(recov >= 0, "boot recovery over surviving WAL");

    /* Assert (Req 1.7/2.4): the replayed tip is the committed boundary only —
     * provisional suffix entries are absent. */
    TEST_ASSERT_TRUE_MESSAGE(
        fresh->current_lsn <= committed_boundary + 1,
        "boot recovery over the surviving WAL must yield the committed prefix "
        "ONLY (Req 1.7/2.4): the uncommitted suffix must have been truncated by "
        "the demotion convergence seam before restart");

    persist_stop(fresh);
    persist_fini(fresh);
    free(fresh);

    dc_teardown();
    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);
    /* ha-commit-durability 2.2: keep durable_lsn at the primed tip so the
     * commit-majority self-cap matches this in-memory-primed WAL state. */
    atomic_store(&g_dc.persist.durable_lsn, tip_lsn);
}

/* 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 ──────────────────────────────────────────────────── */

void test_demotion_convergence(void)
{
#ifdef XCE_TESTING
    printf("\n>> demotion convergence + self-restart tests (RED, task 2.1)\n\n");

    RUN_TEST(test_demotion_from_primary_disables_hooks);
    RUN_TEST(test_demotion_schedules_convergence_truncate_and_restart);
    RUN_TEST(test_self_restart_failure_invokes_fatal_hook);
    RUN_TEST(test_demotion_from_standby_no_truncation_no_restart);
    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
}