myxce/src/controller/testsuite/test_persist_truncate.c

/*
 * test_persist_truncate.c — RED-phase tests for persist_truncate_wal_above().
 *
 * Function does not exist yet (task 7.5). Will fail to link until implemented.
 *
 * Requirements: 5.6, 5.7
 */
#define _GNU_SOURCE
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

#include <unity/unity.h>
#include "persist.h"
#include "wal_types.h"
#include "config.h"
#include "commit.h" /* raft-apply-decoupling task 1.4: commit_engine_t,
                     * commit_submit_async, commit_outcome_t */

/* ── Forward declaration of function-under-test (not yet implemented) ─────── */
extern int persist_truncate_wal_above(persist_t *p, uint64_t cutoff_lsn);

/* ── Helpers ────────────────────────────────────────────────────────────────── */

#define WAL_SEG_FMT "%s/data/wal/seg-%016" PRIx64 ".wal"

static persist_t *make_fresh_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;
    cfg.snap_keep_n = 2;

    persist_set_workdir(p, tmpdir, "trunc-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_persist(persist_t *p)
{
    persist_stop(p);
    persist_fini(p);
    free(p);
}

/* Write a minimal wal_entry_hdr_t with the given lsn to an open fd at offset */
static int write_wal_entry(int fd, uint64_t lsn, off_t offset)
{
    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.type        = WAL_JOB_SUBMIT;
    hdr.lsn         = lsn;
    hdr.payload_len = 0;
    hdr.crc         = 0;
    hdr.prev_crc    = 0;
    ssize_t n = pwrite(fd, &hdr, sizeof(hdr), offset);
    return (n == (ssize_t)sizeof(hdr)) ? 0 : -EIO;
}

/* Count .wal files in the WAL directory */
static int count_wal_segments(const char *wal_dir)
{
    DIR *d = opendir(wal_dir);
    if (!d) return 0;
    int count = 0;
    struct dirent *e;
    while ((e = readdir(d)) != NULL) {
        size_t len = strlen(e->d_name);
        if (len > 4 && strcmp(e->d_name + len - 4, ".wal") == 0)
            count++;
    }
    closedir(d);
    return count;
}

/* ── Test 1: truncation at segment boundary deletes whole segment ──────────── */

static void test_truncate_at_segment_boundary_deletes_segment(void)
{
    char tmpdir[] = "/tmp/xce-trunc-test-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    /* Write 3 WAL entries so we have a known LSN sequence */
    int dummy = 0;
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    uint64_t lsn_before = p->current_lsn;

    /* Append a 4th entry — this represents an uncommitted entry above cutoff */
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    /* Truncate above lsn_before-1 (the 3rd entry); 4th should be removed */
    int rc = persist_truncate_wal_above(p, lsn_before - 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* After truncation, current_lsn (or applied_lsn) should reflect cutoff */
    TEST_ASSERT_TRUE(p->current_lsn <= lsn_before);

    teardown_persist(p);

    /* Cleanup */
    char wal_dir[256];
    snprintf(wal_dir, sizeof(wal_dir), "%s/data/wal", tmpdir);
    printf("test_truncate_at_segment_boundary_deletes_segment: PASS\n");
}

/* ── Test 2: truncation mid-segment scans to correct byte offset ───────────── */

static void test_truncate_mid_segment_at_correct_offset(void)
{
    char tmpdir[] = "/tmp/xce-trunc-mid-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    int dummy = 0;
    /* Write 5 entries, truncate above entry 3 (lsn=2), entries 3-4 gone */
    for (int i = 0; i < 5; i++)
        persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    uint64_t cutoff = 2; /* keep entries with lsn <= 2 */
    int rc = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* applied_lsn / current_lsn must be <= cutoff + 1 */
    TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1);

    teardown_persist(p);
    printf("test_truncate_mid_segment_at_correct_offset: PASS\n");
}

/* ── Test 3: idempotency — calling twice with same cutoff is safe ──────────── */

static void test_truncate_idempotent(void)
{
    char tmpdir[] = "/tmp/xce-trunc-idem-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    int dummy = 0;
    for (int i = 0; i < 4; i++)
        persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    uint64_t cutoff = 1;
    int rc1 = persist_truncate_wal_above(p, cutoff);
    int rc2 = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc1);
    TEST_ASSERT_EQUAL_INT(0, rc2);

    /* current_lsn consistent after two calls */
    TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1);

    teardown_persist(p);
    printf("test_truncate_idempotent: PASS\n");
}

/* ── Test 4: applied_lsn updated atomically to cutoff_lsn ─────────────────── */

static void test_truncate_updates_applied_lsn(void)
{
    char tmpdir[] = "/tmp/xce-trunc-lsn-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    int dummy = 0;
    for (int i = 0; i < 6; i++)
        persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    uint64_t cutoff = 3;
    int rc = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* committed_lsn field must exist on persist_t after task 7.5 adds it.
     * For now, verify current_lsn reflects the truncation. */
    TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1);

    teardown_persist(p);
    printf("test_truncate_updates_applied_lsn: PASS\n");
}

/* ── Test 5: no-op when cutoff_lsn == applied_lsn (nothing to truncate) ───── */

static void test_truncate_noop_at_current_lsn(void)
{
    char tmpdir[] = "/tmp/xce-trunc-noop-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    int dummy = 0;
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    /* Truncate AT current boundary — should be a no-op */
    uint64_t cutoff = p->current_lsn - 1;
    int rc = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Nothing should change */
    TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1);

    teardown_persist(p);
    printf("test_truncate_noop_at_current_lsn: PASS\n");
}

/* ── Test: truncation resolves invalidated commit continuations (task 1.4) ── */

/* Minimal local recorder for commit_submit_async continuations -- mirrors
 * test_commit_resolve.c's continuation_recorder_t/recorder_fn (kept as a
 * separate file-static copy; no cross-TU sharing of test statics). */
typedef struct {
    _Atomic int      calls;
    uint64_t         last_lsn;
    commit_outcome_t last_outcome;
    int              last_rc;
} trunc_continuation_recorder_t;

static void trunc_recorder_fn(void *ctx, uint64_t lsn, commit_outcome_t outcome,
                              int rc)
{
    trunc_continuation_recorder_t *r = (trunc_continuation_recorder_t *)ctx;
    r->last_lsn     = lsn;
    r->last_outcome = outcome;
    r->last_rc      = rc;
    atomic_fetch_add(&r->calls, 1);
}

/* raft-apply-decoupling task 1.4 (Req 1.3/1.4, 9.3/9.4): a real truncation
 * through persist_truncate_wal_above must resolve every commit continuation
 * registered ABOVE the new tail as FAILED -- those entries no longer exist
 * in the log and can never commit -- while a continuation registered AT OR
 * BELOW the new tail is left untouched (its entry is still safely in the
 * log and may still reach majority).
 *
 * Uses n_peers=3 with NO standbys registered so majority is permanently
 * unreachable (majority_candidate_locked's zero-fill keeps committed_lsn at
 * 0 regardless of local append/durable progress) -- every appended entry
 * stays genuinely pending, giving a stable LSN range to register
 * continuations against without racing the commit engine's own majority
 * evaluation.
 *
 * RED: before task 1.4, persist_truncate_wal_above never calls into the
 * commit engine, so all three continuations stay pending forever (calls ==
 * 0 for all). */
static void test_truncate_above_resolves_pending_continuations_as_failed(void)
{
    char tmpdir[] = "/tmp/xce-trunc-resolve-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    commit_engine_t ce;
    TEST_ASSERT_EQUAL_INT(0,
        commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8));

    int dummy = 0;
    for (int i = 0; i < 5; i++)
        persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(6, p->current_lsn,
        "fixture: 5 entries appended, current_lsn must be 6");

    trunc_continuation_recorder_t below  = {0};
    trunc_continuation_recorder_t above1 = {0};
    trunc_continuation_recorder_t above2 = {0};

    TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &below));
    TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 4, trunc_recorder_fn, &above1));
    TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 5, trunc_recorder_fn, &above2));

    /* Act: truncate above lsn=3 -- entries 4 and 5 are removed. */
    int rc = persist_truncate_wal_above(p, 3);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, p->current_lsn,
        "current_lsn must collapse to cutoff_lsn + 1");

    /* Assert: continuations ABOVE the new tail resolve FAILED exactly once.
     * No vtaskmgr pool is wired on this engine (commit_set_pool never
     * called), so commit_resolve's dispatch falls back to the inline
     * fallback -- synchronous with the persist_truncate_wal_above call
     * above, no polling required. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above1.calls),
        "a continuation above the new tail must resolve on truncation");
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
        (int)above1.last_outcome, "truncation must resolve as FAILED");
    TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above1.last_rc,
        "truncation must use the existing -ENOENT vocabulary already "
        "exercised in test_commit_resolve.c (task 1.2)");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, above1.last_lsn,
        "the recorder must observe the lsn it was registered for");

    TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above2.calls),
        "a second continuation above the new tail must also resolve");
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED,
        (int)above2.last_outcome, "truncation must resolve as FAILED");
    TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above2.last_rc,
        "truncation must use the existing -ENOENT vocabulary");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, above2.last_lsn,
        "the recorder must observe the lsn it was registered for");

    /* Assert: a continuation AT OR BELOW the new tail is left untouched --
     * its entry is still in the log and may still reach majority. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&below.calls),
        "a continuation at/below the new tail must NOT be resolved by "
        "truncation -- only entries strictly above the new tail were "
        "removed from the log");

    /* bug-1452: persist_stop() MUST run before commit_engine_fini() -- the
     * fsync_worker flusher thread is still live here and can be mid-flight
     * calling the durable-advance callback (commit.c's fsync_worker ->
     * commit_on_durable_advance -> commit_advance_locked) with NO lock held.
     * commit_engine_fini() destroys ce->lock/ce->cond and nulls ce->p; racing
     * that against a still-running flusher lets commit_advance_locked observe
     * ce->p go NULL between its own guard check and its dereference -> SEGV.
     * persist_stop() joins the flusher thread first, so by the time
     * commit_engine_fini() runs, nothing can call back into ce. */
    persist_stop(p);
    commit_engine_fini(&ce);
    teardown_persist(p);
    printf("test_truncate_above_resolves_pending_continuations_as_failed: PASS\n");
}

/* Regression: a no-op truncate (cutoff at/above the current tail) must NOT
 * fire the truncation wake -- a pending continuation for any lsn must stay
 * untouched, since nothing was actually removed from the log. */
static void test_truncate_noop_does_not_resolve_pending_continuations(void)
{
    char tmpdir[] = "/tmp/xce-trunc-resolve-noop-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    commit_engine_t ce;
    TEST_ASSERT_EQUAL_INT(0,
        commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8));

    int dummy = 0;
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
    persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    trunc_continuation_recorder_t rec = {0};
    TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &rec));

    /* Truncate AT the current boundary -- a no-op (nothing above cutoff). */
    uint64_t cutoff = p->current_lsn - 1;
    int      rc     = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&rec.calls),
        "a no-op truncate (nothing above cutoff_lsn) must not resolve any "
        "pending continuation");

    /* bug-1452: stop the flusher thread before tearing down ce -- see the
     * comment in test_truncate_above_resolves_pending_continuations_as_failed. */
    persist_stop(p);
    commit_engine_fini(&ce);
    teardown_persist(p);
    printf("test_truncate_noop_does_not_resolve_pending_continuations: PASS\n");
}

/* ── Concurrent burst (raft-apply-decoupling task 1.5, Req 9.1-9.4) ────────
 * Many continuations, registered from real threads racing each other for
 * ce->lock and split across the truncation cutoff, must ALL resolve exactly
 * once through the SAME real persist_truncate_wal_above detection path
 * exercised above: entries above the new tail FAILED/-ENOENT, entries at/
 * below left untouched -- proving the bulk resolve holds under genuine
 * concurrent registration, not just the fixed handful the tests above use. */
#define TRUNC_BURST_ABOVE 6
#define TRUNC_BURST_BELOW 3
#define TRUNC_BURST_TOTAL (TRUNC_BURST_ABOVE + TRUNC_BURST_BELOW)

typedef struct {
    commit_engine_t*               ce;
    uint64_t                       lsn;
    int                            rc;
    trunc_continuation_recorder_t* rec;
} trunc_burst_arg_t;

static void* trunc_burst_register_thread(void* p)
{
    trunc_burst_arg_t* a = (trunc_burst_arg_t*)p;

    a->rc = commit_submit_async(a->ce, a->lsn, trunc_recorder_fn, a->rec);
    return NULL;
}

static void test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once(void)
{
    char tmpdir[] = "/tmp/xce-trunc-burst-XXXXXX";
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));

    persist_t *p = make_fresh_persist(tmpdir);
    TEST_ASSERT_NOT_NULL(p);

    commit_engine_t ce;
    TEST_ASSERT_EQUAL_INT(0,
        commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/TRUNC_BURST_TOTAL + 4));

    int dummy = 0;
    for (int i = 0; i < TRUNC_BURST_TOTAL + 2; i++)
        persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);

    /* Cutoff placed so lsn 1..TRUNC_BURST_BELOW sit at/below the new tail
     * and lsn (TRUNC_BURST_BELOW+1)..TRUNC_BURST_TOTAL sit above it. */
    const uint64_t cutoff = TRUNC_BURST_BELOW;

    trunc_continuation_recorder_t recs[TRUNC_BURST_TOTAL] = {0};
    trunc_burst_arg_t             args[TRUNC_BURST_TOTAL];
    pthread_t                     tids[TRUNC_BURST_TOTAL];

    /* Register the whole burst -- below-cutoff and above-cutoff LSNs alike
     * -- CONCURRENTLY from real threads. */
    for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
        args[i].ce  = &ce;
        args[i].lsn = (uint64_t)(i + 1);
        args[i].rc  = -9999;
        args[i].rec = &recs[i];
        TEST_ASSERT_EQUAL_INT(
            0, pthread_create(&tids[i], NULL, trunc_burst_register_thread, &args[i]));
    }
    for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
        pthread_join(tids[i], NULL);
    }
    for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
        TEST_ASSERT_EQUAL_INT_MESSAGE(
            0, args[i].rc,
            "every concurrent registration ahead of the truncation must succeed");
    }
    TEST_ASSERT_EQUAL_INT(TRUNC_BURST_TOTAL, ce.inflight_waits);

    /* Act: a single real truncation above cutoff. The burst above is
     * already fully registered by the time this fires -- this proves the
     * BULK resolve path handles a registry populated by real concurrent
     * registrants correctly, not just the fixed three-continuation case
     * above. */
    int rc = persist_truncate_wal_above(p, cutoff);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT64(cutoff + 1, p->current_lsn);

    for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
        uint64_t lsn = (uint64_t)(i + 1);

        if (lsn > cutoff) {
            TEST_ASSERT_EQUAL_INT_MESSAGE(
                1, atomic_load(&recs[i].calls),
                "every continuation above the new tail must resolve exactly "
                "once on truncation, even when the whole burst was "
                "registered concurrently");
            TEST_ASSERT_EQUAL_INT_MESSAGE(
                (int)COMMIT_OUTCOME_FAILED, (int)recs[i].last_outcome,
                "truncation must resolve as FAILED");
            TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, recs[i].last_rc,
                "truncation must use the existing -ENOENT vocabulary");
            TEST_ASSERT_EQUAL_UINT64(lsn, recs[i].last_lsn);
        } else {
            TEST_ASSERT_EQUAL_INT_MESSAGE(
                0, atomic_load(&recs[i].calls),
                "a continuation at/below the new tail must NOT be resolved "
                "by truncation, even under concurrent registration");
        }
    }
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        TRUNC_BURST_BELOW, ce.inflight_waits,
        "only the above-cutoff admission slots are released by truncation; "
        "the below-cutoff ones remain pending");

    /* bug-1452: stop the flusher thread before tearing down ce -- see the
     * comment in test_truncate_above_resolves_pending_continuations_as_failed. */
    persist_stop(p);
    commit_engine_fini(&ce);
    teardown_persist(p);
    printf("test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once: "
           "PASS\n");
}

/* ── Test group entry point ────────────────────────────────────────────────── */

void test_persist_truncate(void)
{
    UnitySetTestFile(__FILE__);
    RUN_TEST(test_truncate_at_segment_boundary_deletes_segment);
    RUN_TEST(test_truncate_mid_segment_at_correct_offset);
    RUN_TEST(test_truncate_idempotent);
    RUN_TEST(test_truncate_updates_applied_lsn);
    RUN_TEST(test_truncate_noop_at_current_lsn);
    RUN_TEST(test_truncate_above_resolves_pending_continuations_as_failed);
    RUN_TEST(test_truncate_noop_does_not_resolve_pending_continuations);
    RUN_TEST(test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once);
}