#define _GNU_SOURCE
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <unity/unity.h>

#include "commit.h"
#include "config.h"
#include "persist.h"

/* ------------------------------------------------------------------ *
 * ha-commit-durability — task 1.2: durability-verification test seam  *
 *                                                                     *
 * These tests exercise persist_test_durable_lsn_on_disk(), the        *
 * XCE_TESTING-only seam that re-reads + CRC-validates the WAL from     *
 * stable storage and reports the highest fully-persisted LSN —         *
 * independent of the in-memory current_lsn watermark.  This lets a     *
 * test assert an acknowledged operation is recoverable from disk after *
 * an immediate ungraceful termination, without relying on the          *
 * background flush timer (Req 3.2).  The seam is purely inspective:    *
 * it changes nothing observable about the acknowledgment contract      *
 * (Req 3.3).                                                           *
 *                                                                     *
 * The seam is compiled ONLY into the controller_unit_test target       *
 * (-DXCE_TESTING).  The production xce_controller target links without *
 * it — see #ifdef XCE_TESTING in persist.c / declaration in persist.h. *
 * ------------------------------------------------------------------ */

#ifdef XCE_TESTING

/* Build a persist instance with a deliberately slow background flush
 * cadence, so the only durability point is an explicit persist_flush()
 * (mirrors make_fresh_persist in test_persist_recovery.c). */
static persist_t* make_durability_persist(const char* tmpdir)
{
    persist_t* p = calloc(1, sizeof(persist_t));
    if (!p)
        return NULL;

    persist_config_t cfg = persist_config_defaults();
    cfg.wal_sync_ms = 5000; /* slow background fsync — durability comes from
                             * the explicit flush, not the timer */
    cfg.snap_keep_n = 2;

    persist_set_workdir(p, tmpdir, "ctrl-test");

    if (persist_init(p, &cfg, NULL) != 0) {
        free(p);
        return NULL;
    }
    if (persist_start(p) != 0) {
        persist_fini(p);
        free(p);
        return NULL;
    }
    return p;
}

static void teardown_durability_persist(persist_t* p)
{
    persist_stop(p);
    persist_fini(p);
    free(p);
}

/* ------------------------------------------------------------------ *
 * Test 1: an empty WAL has no durable entries (floor == 0).           *
 * An empty first-boot WAL is not corruption — it simply has nothing   *
 * persisted, so the seam reports 0.                                   *
 * ------------------------------------------------------------------ */
static void test_durable_floor_empty(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durfloor0_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    uint64_t floor = persist_test_durable_lsn_on_disk(p);
    TEST_ASSERT_EQUAL_UINT64(0, floor);

    teardown_durability_persist(p);
    printf("test_durable_floor_empty: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 2 (core Req 3.2): after append + explicit fdatasync, the seam  *
 * sees the entries on disk — the durable floor equals the highest     *
 * appended LSN, recoverable independent of in-memory state.           *
 * ------------------------------------------------------------------ */
static void test_durable_floor_after_flush(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durfloor1_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Append three entries. LSN 0 is reserved, so these take LSN 1,2,3.
     * current_lsn advances to 4 (one past the last appended). */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1));
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "c", 1));

    /* Durability point: fdatasync the segment so the bytes are on stable
     * storage (this is what an acknowledged-before-durable contract must
     * have done before replying ok). */
    TEST_ASSERT_EQUAL_INT(0, persist_flush(p));

    /* The seam re-reads from disk and reports the highest intact LSN. */
    uint64_t floor = persist_test_durable_lsn_on_disk(p);
    TEST_ASSERT_EQUAL_UINT64(p->current_lsn - 1, floor);
    TEST_ASSERT_EQUAL_UINT64(3, floor);

    teardown_durability_persist(p);
    printf("test_durable_floor_after_flush: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 3 (recoverable-after-ungraceful-kill, Req 3.2): a SECOND        *
 * persist instance opened over the same WAL dir — with NO in-memory    *
 * state from the writer — recovers exactly the durable floor the seam  *
 * reported.  This is the "survives an immediate ungraceful            *
 * termination" assertion: durability lives on disk, not in the         *
 * volatile buffers of the process that wrote it.                       *
 * ------------------------------------------------------------------ */
static void test_durable_floor_survives_process_loss(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durfloor2_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    uint64_t durable_floor = 0;

    /* Writer process: append, fdatasync, then "die" (we drop the instance
     * without any further flush — the bytes are already durable). */
    {
        persist_t* p = make_durability_persist(tmp);
        TEST_ASSERT_NOT_NULL(p);

        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1));
        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "y", 1));
        TEST_ASSERT_EQUAL_INT(0, persist_flush(p));

        durable_floor = persist_test_durable_lsn_on_disk(p);
        TEST_ASSERT_EQUAL_UINT64(2, durable_floor);

        teardown_durability_persist(p);
    }

    /* "Restarted" process: a fresh persist_t with zero in-memory state,
     * opened over the same on-disk WAL.  Its durable floor — read straight
     * from disk before any recover/replay — matches what the writer
     * persisted.  Acknowledged-before-durable ops therefore survive the
     * loss of the acking process. */
    {
        persist_t* p2 = make_durability_persist(tmp);
        TEST_ASSERT_NOT_NULL(p2);

        /* current_lsn is still the fresh-boot value (1, LSN 0 reserved):
         * the durable floor does NOT come from in-memory state. */
        uint64_t floor2 = persist_test_durable_lsn_on_disk(p2);
        TEST_ASSERT_EQUAL_UINT64(durable_floor, floor2);
        TEST_ASSERT_EQUAL_UINT64(2, floor2);

        teardown_durability_persist(p2);
    }

    printf("test_durable_floor_survives_process_loss: PASS\n");
}

/* ================================================================== *
 * task 2.1 — durable-LSN tracking via a group-commit flusher          *
 *                                                                     *
 * These tests pin the invariants of persist_t.durable_lsn, the        *
 * prefix watermark advanced by the demand-driven flusher (the         *
 * repurposed fsync_worker):                                           *
 *   - durable_lsn never exceeds the synced tip (<= current_lsn - 1);  *
 *   - durable_lsn is monotonic non-decreasing;                        *
 *   - one fdatasync advances durable_lsn past MULTIPLE appended LSNs   *
 *     (batching / group-commit);                                      *
 *   - durable_lsn advances ONLY after a successful fdatasync;          *
 *   - under concurrent append load the invariants always hold.        *
 * ------------------------------------------------------------------ */

/* Spin until the flusher catches durable_lsn up to the appended tip,
 * or a generous deadline elapses.  Returns the observed durable_lsn. */
static uint64_t wait_durable_reaches(persist_t* p, uint64_t target_lsn,
                                     int max_ms)
{
    for (int i = 0; i < max_ms; i++) {
        uint64_t d = persist_durable_lsn(p);
        if (d >= target_lsn)
            return d;
        usleep(1000); /* 1 ms */
    }
    return persist_durable_lsn(p);
}

/* ------------------------------------------------------------------ *
 * Test 4: a fresh store has durable_lsn == 0 (nothing synced yet).    *
 * ------------------------------------------------------------------ */
static void test_durable_lsn_initial_zero(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durlsn0_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    TEST_ASSERT_EQUAL_UINT64(0, persist_durable_lsn(p));

    teardown_durability_persist(p);
    printf("test_durable_lsn_initial_zero: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 5 (core / batching, Req 1.1): several appends, then the        *
 * demand-driven flusher advances durable_lsn past ALL of them with a  *
 * single fdatasync — and never past the appended tip.  Uses the fast  *
 * cadence so the append-signal (not the timer) drives the flush, but  *
 * the slow-cadence helper would converge too via the bounded fallback.*
 * ------------------------------------------------------------------ */
static void test_durable_lsn_batches_multiple_appends(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durlsn1_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* LSN 0 reserved → these take 1,2,3,4,5; current_lsn ends at 6. */
    for (int i = 0; i < 5; i++)
        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1));

    uint64_t tip = p->current_lsn - 1; /* 5 */
    TEST_ASSERT_EQUAL_UINT64(5, tip);

    uint64_t d = wait_durable_reaches(p, tip, 5000);

    /* The flusher advanced durable_lsn past several appended LSNs with
     * one fdatasync (batching), reaching the synced tip. */
    TEST_ASSERT_EQUAL_UINT64(tip, d);
    /* Invariant: never exceeds the synced tip. */
    TEST_ASSERT_TRUE(d <= p->current_lsn - 1);
    /* The on-disk seam agrees: those LSNs are genuinely fdatasync'd. */
    TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d);

    teardown_durability_persist(p);
    printf("test_durable_lsn_batches_multiple_appends: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 6 (advances-only-after-fdatasync + monotonic): the durable     *
 * floor reported by the on-disk seam must always be >= durable_lsn —  *
 * i.e. durable_lsn never claims an LSN whose bytes are not actually    *
 * on stable storage.  Also asserts monotonic non-decreasing across    *
 * successive append+flush rounds.                                      *
 * ------------------------------------------------------------------ */
static void test_durable_lsn_never_exceeds_synced_tip(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durlsn2_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    uint64_t prev = persist_durable_lsn(p);

    for (int round = 0; round < 4; round++) {
        for (int i = 0; i < 3; i++)
            TEST_ASSERT_EQUAL_INT(0,
                                  persist_wal_async(p, WAL_JOB_UPDATE, "y", 1));

        uint64_t tip = p->current_lsn - 1;
        uint64_t d   = wait_durable_reaches(p, tip, 5000);

        /* Monotonic non-decreasing. */
        TEST_ASSERT_TRUE(d >= prev);
        /* Never exceeds the synced tip. */
        TEST_ASSERT_TRUE(d <= p->current_lsn - 1);
        /* Advances only after a real fdatasync: the independent on-disk
         * scan must confirm every claimed-durable LSN is on disk. */
        TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d);

        prev = d;
    }

    teardown_durability_persist(p);
    printf("test_durable_lsn_never_exceeds_synced_tip: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 7 (UNDER LOAD, Req 1.6 / lost-wakeup): many appends from       *
 * multiple threads racing the flusher.  Asserts the invariants hold   *
 * throughout (durable_lsn <= synced tip, never past disk) AND that    *
 * the flusher eventually catches durable_lsn up to the final tip —    *
 * i.e. no append is left permanently un-synced (no lost wakeup).      *
 * ------------------------------------------------------------------ */
#define LOAD_THREADS    4
#define LOAD_PER_THREAD 200

typedef struct {
    persist_t* p;
    int        n;
} load_arg_t;

static void* load_appender(void* arg)
{
    load_arg_t* a = arg;
    for (int i = 0; i < a->n; i++) {
        /* Ignore rc: a full-segment rotate or transient EIO is not the
         * subject under test; the invariant checks below tolerate it. */
        (void)persist_wal_async(a->p, WAL_JOB_UPDATE, "z", 1);
        /* Sample the invariant mid-flight: durable_lsn must NEVER exceed
         * the appended tip, even while a concurrent flush is in progress. */
        uint64_t d   = persist_durable_lsn(a->p);
        uint64_t cur = atomic_load_explicit(
            (const _Atomic uint64_t*)&a->p->current_lsn, memory_order_acquire);
        /* current_lsn is read without the lock here purely as a coarse
         * upper bound; a torn read can only make `cur` smaller, so the
         * assertion stays conservative. durable_lsn is a prefix watermark
         * that can never outrun the last assigned LSN. */
        TEST_ASSERT_TRUE(d == 0 || d <= cur);
    }
    return NULL;
}

static void test_durable_lsn_invariants_under_load(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durlsn3_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    pthread_t  th[LOAD_THREADS];
    load_arg_t args[LOAD_THREADS];
    for (int i = 0; i < LOAD_THREADS; i++) {
        args[i].p = p;
        args[i].n = LOAD_PER_THREAD;
        TEST_ASSERT_EQUAL_INT(0, pthread_create(&th[i], NULL,
                                                load_appender, &args[i]));
    }
    for (int i = 0; i < LOAD_THREADS; i++)
        pthread_join(th[i], NULL);

    /* Final tip after all appends settled. */
    uint64_t tip = p->current_lsn - 1;

    /* No lost wakeup: the flusher must eventually advance durable_lsn up
     * to the final appended tip (every entry gets synced). */
    uint64_t d = wait_durable_reaches(p, tip, 10000);
    TEST_ASSERT_EQUAL_UINT64(tip, d);
    TEST_ASSERT_TRUE(d <= p->current_lsn - 1);
    TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d);

    teardown_durability_persist(p);
    printf("test_durable_lsn_invariants_under_load: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test 8 (ERROR PATH, Req 1.1/1.2 — fail-loud): when fdatasync FAILS,  *
 * durable_lsn must NOT advance.  Bytes that did not reach stable        *
 * storage must never be counted durable — otherwise the member would   *
 * contribute a non-durable LSN toward the commit majority.             *
 *                                                                     *
 * Mechanism: the XCE_TESTING fault-injection toggle forces every        *
 * fdatasync site to treat the sync as failed.  We append entries while  *
 * the toggle is ON and assert durable_lsn stays put across the full     *
 * cadence window (the flusher runs but its advance is gated on rc==0).  *
 * Then we clear the toggle and confirm the flusher recovers — durable_  *
 * lsn advances to the tip once syncs succeed again.                     *
 * ------------------------------------------------------------------ */
static void test_durable_lsn_holds_on_fsync_failure(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durlsn4_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* First, append one entry and let it become durable normally, so the
     * watermark has a known non-zero baseline that must HOLD during the
     * failure window. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
    uint64_t baseline = wait_durable_reaches(p, 1, 5000);
    TEST_ASSERT_EQUAL_UINT64(1, baseline);

    /* Now inject fdatasync failure and append more entries. */
    persist_test_set_force_fsync_fail(p, 1);

    for (int i = 0; i < 5; i++)
        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1));

    uint64_t tip = p->current_lsn - 1; /* 6 */
    TEST_ASSERT_EQUAL_UINT64(6, tip);

    /* Give the flusher ample time to run several passes (cadence is 5s but
     * the append signal fires it immediately) AND an explicit flush attempt.
     * Under injected failure, durable_lsn MUST remain at the baseline — the
     * new bytes are not on stable storage, so they are not durable. */
    (void)persist_flush(p); /* also gated on rc==0 — must not advance either */
    for (int i = 0; i < 200; i++) {
        TEST_ASSERT_EQUAL_UINT64(baseline, persist_durable_lsn(p));
        usleep(1000); /* 1 ms; 200 ms total — many flusher signal cycles */
    }
    /* Still parked at the baseline after the window. */
    TEST_ASSERT_EQUAL_UINT64(baseline, persist_durable_lsn(p));
    TEST_ASSERT_TRUE(persist_durable_lsn(p) <= p->current_lsn - 1);

    /* Recovery: clear the fault and confirm the flusher catches up — proving
     * the failure path did not corrupt the watermark, only held it. */
    persist_test_set_force_fsync_fail(p, 0);
    /* Nudge the flusher with one more append so the level predicate signals. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "c", 1));
    uint64_t new_tip = p->current_lsn - 1; /* 7 */
    uint64_t d = wait_durable_reaches(p, new_tip, 5000);
    TEST_ASSERT_EQUAL_UINT64(new_tip, d);
    TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d);

    teardown_durability_persist(p);
    printf("test_durable_lsn_holds_on_fsync_failure: PASS\n");
}

/* ================================================================== *
 * task 2.2 — durable-advance wake drives commit (integration)         *
 *                                                                     *
 * End-to-end over a REAL persist_t + commit_engine_t (single-member,  *
 * NULL cluster): the group-commit flusher advances durable_lsn and    *
 * fires the registered durable-advance callback, which the commit     *
 * engine uses to advance committed_lsn and wake WAIT callers. Proves   *
 * the persist↔commit wiring works against the real flusher thread —    *
 * not just the directly-driven callback (covered in                    *
 * test_commit_engine.c). n_peers=1 means self alone is the majority,   *
 * so committed_lsn tracks durable_lsn with no standby ACKs needed.     *
 * ------------------------------------------------------------------ */
static void test_durable_advance_drives_commit_single_member(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_durcommit_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Single-member commit engine over the real persist. n_vcpus large enough
     * to admit a wait. commit_engine_init registers the durable-advance cb. */
    commit_engine_t ce;
    TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, 1, 8));

    /* committed starts at 0; nothing appended/durable. */
    TEST_ASSERT_EQUAL_UINT64(0, atomic_load(&p->committed_lsn));

    /* Append three entries (LSN 1,2,3). They are in page cache only — the slow
     * 5s cadence means the append-SIGNAL (not the timer) must drive the flush;
     * the real flusher fdatasyncs, advances durable_lsn, and fires the cb,
     * which advances committed_lsn for this single-member quorum. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1));
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "c", 1));

    uint64_t tip = p->current_lsn - 1; /* 3 */
    TEST_ASSERT_EQUAL_UINT64(3, tip);

    /* Wait for the flusher to make the tip durable AND for the wake-driven
     * commit to catch up. committed_lsn must reach the tip — driven by the
     * durable-advance callback, not by any ACK (there are no standbys). */
    uint64_t committed = 0;
    for (int i = 0; i < 5000; i++) {
        committed = atomic_load(&p->committed_lsn);
        if (committed >= tip)
            break;
        usleep(1000);
    }
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        tip, committed,
        "durable-advance callback must drive committed_lsn to the durable tip");
    /* committed never exceeds durable (the invariant the wake preserves). */
    TEST_ASSERT_TRUE(atomic_load(&p->committed_lsn) <= persist_durable_lsn(p));

    /* Tear the engine down BEFORE the persist (unhooks the callback so the
     * flusher cannot fire against a destroyed engine). */
    commit_engine_fini(&ce);
    teardown_durability_persist(p);
    printf("test_durable_advance_drives_commit_single_member: PASS\n");
}

/* ================================================================== *
 * task 2.3 — standby persist-before-ACK regression (Req 1.3)          *
 *                                                                     *
 * Req 1.3: a standby durably persists a replicated entry to local      *
 * stable storage BEFORE sending its acknowledgment. The ACK is sent by *
 * the CALLER (ha_apply_single_entry) only after                        *
 * persist_wal_write_replicated returns success — so to pin "persist     *
 * before ACK" we prove a durable fdatasync ran INSIDE the persist call  *
 * before it returned.                                                  *
 *                                                                     *
 * Mechanism: a disk re-read (persist_test_durable_lsn_on_disk) cannot   *
 * distinguish an fsync'd page from a page-cache-only page, so it CANNOT *
 * prove the sync ran. Instead we read the XCE_TESTING fdatasync counter *
 * incremented at the inline sync site: snapshot it before the call,     *
 * assert it incremented after — the sync provably ran within the call,  *
 * before the caller could ACK. The test FAILS if someone removed or     *
 * moved the fdatasync out of persist_wal_write_replicated.             *
 * ------------------------------------------------------------------ */
static void test_replicated_persist_syncs_before_return(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_replsync_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Snapshot the sync counter, then write a replicated entry. The standby
     * path must fdatasync inline BEFORE returning — so the counter must
     * advance during the single call. */
    unsigned before = persist_test_replicated_fdatasync_count(p);

    int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE,
                                          "x", 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    unsigned after = persist_test_replicated_fdatasync_count(p);
    TEST_ASSERT_TRUE_MESSAGE(
        after > before,
        "persist_wal_write_replicated must fdatasync inline (persist-before-"
        "ACK): the sync counter did not advance during the call");

    /* And the entry is genuinely on stable storage by the time the call
     * returned (would survive an immediate process loss → recoverable ACK). */
    TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= 1);

    teardown_durability_persist(p);
    printf("test_replicated_persist_syncs_before_return: PASS\n");
}

/* ------------------------------------------------------------------ *
 * task 2.3 — forced standby-sync failure ⇒ error, NO success/ACK      *
 *                                                                     *
 * Closes the Req-1.3 hole: if the inline fdatasync FAILS, the standby   *
 * must NOT report success — otherwise the caller would ACK an entry     *
 * that is not durable (durable AFTER, not BEFORE, the ACK). With the     *
 * force-fail seam ON, persist_wal_write_replicated must return a         *
 * negative error; the caller (ha_apply_single_entry) already treats any  *
 * negative return as "do not apply, do not ACK".                        *
 *                                                                     *
 * RED evidence: against the pre-2.3 code (which ignored the fdatasync    *
 * return at the standby path) this returned 0 (success) under forced     *
 * failure — a non-durable ACK. The rc-gate makes it return -EIO.        *
 * ------------------------------------------------------------------ */
static void test_replicated_sync_failure_no_ack(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_replsyncfail_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Force the fdatasync to be treated as failed, then attempt a replicated
     * write. The durable sync cannot succeed, so the call MUST fail (no ACK).
     * A negative return is what ha_apply_single_entry routes to its
     * "not applying/acking" path. */
    persist_test_set_force_fsync_fail(p, 1);

    int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE,
                                          "x", 1);
    TEST_ASSERT_TRUE_MESSAGE(
        rc < 0,
        "forced standby fdatasync failure must make persist_wal_write_"
        "replicated FAIL (negative rc) so the standby does NOT ACK a "
        "non-durable entry");

    /* The sync was still attempted (counter advanced) — we fail loud on the
     * RESULT, not by skipping the sync. */
    TEST_ASSERT_TRUE(persist_test_replicated_fdatasync_count(p) >= 1);

    /* Recovery: clear the fault; a fresh replicated write (new lsn, not the
     * deduped lsn=1) now succeeds and is durable — proving the failure path
     * did not corrupt persist state and the sync runs again on success.
     * (lsn=1 already advanced current_lsn to 2, so re-using 1 would dedup;
     * lsn=2 is the genuine next entry.) */
    persist_test_set_force_fsync_fail(p, 0);
    int rc2 = persist_wal_write_replicated(p, /*lsn=*/2, WAL_JOB_UPDATE,
                                           "x", 1);
    TEST_ASSERT_EQUAL_INT(0, rc2);
    TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= 2);

    teardown_durability_persist(p);
    printf("test_replicated_sync_failure_no_ack: PASS\n");
}

/* ================================================================== *
 * task 2.4 — reconcile the dormant synchronous-append commit path      *
 *                                                                     *
 * persist_wal_sync_append's LEGACY branch (ce == NULL, or the LSN-0    *
 * first entry on the engine path) historically advanced committed_lsn  *
 * DIRECTLY on a majority ACK (atomic_store committed_lsn = written_lsn) *
 * with NO durable precondition. Group-commit removed the premise that  *
 * the append already fsync'd inline, so a majority ACK no longer        *
 * implies the local entry is durable. This branch is the structural    *
 * ACK-without-durability hazard task 2.4 must close: NO path may        *
 * advance committed_lsn on a majority ACK unless the entry is durable   *
 * locally (committed_lsn <= durable_lsn).                              *
 *                                                                     *
 * NOTE: persist_wal_sync_append is NOT dormant — it is the LIVE         *
 * job-submit path (persist_job_hook_fun, WAL_JOB_SUBMIT). So the fix is *
 * to ROUTE the legacy direct-store through the durable precondition,    *
 * not to delete the function.                                          *
 * ------------------------------------------------------------------ */

/* Build an ENGINE-LESS persist with standbys registered, so
 * persist_wal_sync_append takes the legacy branch (ce == NULL) and its
 * has_standbys path: repl_wait_majority_ack returns 0 immediately for
 * the LSN-0 first entry, but for a NON-zero LSN it surfaces -EINVAL via
 * commit_submit (engine-less). The hazard we guard is the direct
 * committed_lsn store that the legacy branch performs on a 0 return. */

/* Drive the legacy branch directly: a single-member commit engine makes
 * the ENGINE path durable-safe, so to exercise the legacy direct-store
 * we use an engine-less persist and a registered standby that ACKs.
 * Rather than stand up the replication wire (out of boundary), we test
 * the invariant the fix establishes: committed_lsn must never exceed
 * durable_lsn after a commit advance, on EVERY path. */

/* ------------------------------------------------------------------ *
 * Test (RED before fix): the legacy direct-store path must NOT advance  *
 * committed_lsn past durable_lsn. We drive the legacy branch via the    *
 * LSN-0-style direct store by appending one entry that is NOT yet       *
 * durable (slow cadence, no flush) and confirming committed_lsn stays   *
 * at/below durable_lsn.                                                 *
 *                                                                     *
 * Mechanism: persist_test_sync_append_legacy_store() is an XCE_TESTING  *
 * seam that invokes the SAME guarded store the legacy branch performs   *
 * (advance committed_lsn to written_lsn only if written_lsn <=          *
 * durable_lsn). Pre-fix the store was unconditional, so this over-       *
 * advanced committed past durable (RED). Post-fix it holds.             *
 * ------------------------------------------------------------------ */
static void test_legacy_sync_append_holds_committed_at_durable(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_legacycommit_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Append one entry (LSN 1) but do NOT flush — durable_lsn lags at 0,
     * while the appended tip is 1. The legacy direct-store would advance
     * committed_lsn to the ACK'd written_lsn (1) — over-advancing past
     * the durable watermark (0). The durable precondition must hold it. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
    uint64_t written_lsn = p->current_lsn - 1; /* 1 */
    TEST_ASSERT_EQUAL_UINT64(1, written_lsn);
    TEST_ASSERT_EQUAL_UINT64(0, persist_durable_lsn(p));
    TEST_ASSERT_EQUAL_UINT64(0, atomic_load(&p->committed_lsn));

    /* Invoke the guarded legacy store directly (simulates a majority ACK
     * for an entry that is not yet durable locally). It must NOT advance
     * committed_lsn past durable_lsn. */
    persist_test_sync_append_legacy_commit_store(p, written_lsn);

    TEST_ASSERT_TRUE_MESSAGE(
        atomic_load(&p->committed_lsn) <= persist_durable_lsn(p),
        "legacy sync-append commit store must NOT advance committed_lsn past "
        "durable_lsn (ACK-without-durability hazard)");
    /* Concretely: committed stays at 0 because durable_lsn is still 0. */
    TEST_ASSERT_EQUAL_UINT64(0, atomic_load(&p->committed_lsn));

    /* Now make the entry durable and re-invoke: the store is now allowed
     * (written_lsn <= durable_lsn) and committed advances to it. */
    TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
    TEST_ASSERT_TRUE(persist_durable_lsn(p) >= written_lsn);
    persist_test_sync_append_legacy_commit_store(p, written_lsn);
    TEST_ASSERT_EQUAL_UINT64(written_lsn, atomic_load(&p->committed_lsn));
    TEST_ASSERT_TRUE(atomic_load(&p->committed_lsn) <= persist_durable_lsn(p));

    teardown_durability_persist(p);
    printf("test_legacy_sync_append_holds_committed_at_durable: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Test: the LSN-0 first-entry edge stays valid. durable_lsn and        *
 * committed_lsn both start at 0; the legacy store of written_lsn == 0   *
 * (0 <= 0) is permitted and is a harmless no-op store of 0. This is the *
 * once-per-cluster-lifetime path the engine branch routes here.         *
 * ------------------------------------------------------------------ */
static void test_legacy_sync_append_lsn0_edge_valid(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_legacylsn0_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    /* Fresh store: durable_lsn == 0, committed_lsn == 0. */
    TEST_ASSERT_EQUAL_UINT64(0, persist_durable_lsn(p));
    TEST_ASSERT_EQUAL_UINT64(0, atomic_load(&p->committed_lsn));

    /* Storing written_lsn == 0 (0 <= durable 0) is allowed — a no-op. */
    persist_test_sync_append_legacy_commit_store(p, 0);
    TEST_ASSERT_EQUAL_UINT64(0, atomic_load(&p->committed_lsn));
    TEST_ASSERT_TRUE(atomic_load(&p->committed_lsn) <= persist_durable_lsn(p));

    teardown_durability_persist(p);
    printf("test_legacy_sync_append_lsn0_edge_valid: PASS\n");
}

/* ------------------------------------------------------------------ *
 * Guard test (the Observable's "test guards against a future caller"):  *
 * the committed_lsn <= durable_lsn invariant holds across the commit-   *
 * engine advance path too. Drives the real engine + flusher and asserts *
 * that at every observed point committed never exceeds durable — the    *
 * future-proof guard against reintroducing an ACK-without-durability    *
 * commit on ANY path. In a debug build the in-code assert would abort   *
 * if a path ever advanced committed past durable.                       *
 * ------------------------------------------------------------------ */
static void test_commit_never_exceeds_durable_invariant(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_cinvariant_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    persist_t* p = make_durability_persist(tmp);
    TEST_ASSERT_NOT_NULL(p);

    commit_engine_t ce;
    TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, 1, 8));

    /* Append a burst while sampling the invariant: committed must never
     * outrun durable, even mid-flight while the flusher advances both. */
    for (int round = 0; round < 6; round++) {
        for (int i = 0; i < 4; i++)
            TEST_ASSERT_EQUAL_INT(
                0, persist_wal_async(p, WAL_JOB_UPDATE, "q", 1));
        for (int s = 0; s < 50; s++) {
            uint64_t committed = atomic_load(&p->committed_lsn);
            uint64_t durable   = persist_durable_lsn(p);
            TEST_ASSERT_TRUE_MESSAGE(
                committed <= durable,
                "committed_lsn must never exceed durable_lsn on any commit "
                "advance path");
            usleep(1000);
        }
    }

    /* And eventually committed catches up to the durable tip. */
    uint64_t tip = p->current_lsn - 1;
    uint64_t committed = 0;
    for (int i = 0; i < 5000; i++) {
        committed = atomic_load(&p->committed_lsn);
        if (committed >= tip)
            break;
        usleep(1000);
    }
    TEST_ASSERT_EQUAL_UINT64(tip, committed);
    TEST_ASSERT_TRUE(atomic_load(&p->committed_lsn) <= persist_durable_lsn(p));

    commit_engine_fini(&ce);
    teardown_durability_persist(p);
    printf("test_commit_never_exceeds_durable_invariant: PASS\n");
}

#endif /* XCE_TESTING */

/* ------------------------------------------------------------------ */
/* Test suite entry point                                               */
/* ------------------------------------------------------------------ */
void test_commit_durability(void)
{
#ifdef XCE_TESTING
    RUN_TEST(test_durable_floor_empty);
    RUN_TEST(test_durable_floor_after_flush);
    RUN_TEST(test_durable_floor_survives_process_loss);
    RUN_TEST(test_durable_lsn_initial_zero);
    RUN_TEST(test_durable_lsn_batches_multiple_appends);
    RUN_TEST(test_durable_lsn_never_exceeds_synced_tip);
    RUN_TEST(test_durable_lsn_invariants_under_load);
    RUN_TEST(test_durable_lsn_holds_on_fsync_failure);
    RUN_TEST(test_durable_advance_drives_commit_single_member);
    RUN_TEST(test_replicated_persist_syncs_before_return);
    RUN_TEST(test_replicated_sync_failure_no_ack);
    RUN_TEST(test_legacy_sync_append_holds_committed_at_durable);
    RUN_TEST(test_legacy_sync_append_lsn0_edge_valid);
    RUN_TEST(test_commit_never_exceeds_durable_invariant);
#else
    printf("test_commit_durability: skipped (no XCE_TESTING)\n");
#endif
}