myxce/src/controller/testsuite/test_repl_apply.c

/*
 * test_repl_apply.c — P2-2: Tests for apply_wal_entry() standby WAL replay.
 *
 * TDD RED phase: written BEFORE the implementation.
 * These tests verify:
 *   - LSN advances past replayed entry
 *   - WAL_CHECKPOINT entries are skipped (rc=0, lsn unchanged)
 *   - Oversized payloads are rejected (rc != 0)
 *   - Unknown types (> WAL_ENTRY_TYPE_MAX) are skipped with rc=0
 */
#include <dirent.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>

#include <unity/unity.h>

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

/* ------------------------------------------------------------------ */
/* Weak stubs for P0 entity-restore functions (job.c / worker.c).      */
/* The unit test binary does not link those translation units — they   */
/* pull in the full cluster.h / netio.h chain.  apply_wal_entry() only */
/* calls these when p->cluster != NULL; tests use NULL cluster, so the */
/* weak stubs are never reached.  The linker needs the symbols to link. */
/* ------------------------------------------------------------------ */
typedef struct cluster_s cluster_t;

__attribute__((weak))
int restore_job(cluster_t* c, uint64_t jobid,
                const void* capnp_payload, size_t len, uint64_t entry_lsn)
{
    (void)c; (void)jobid; (void)capnp_payload; (void)len; (void)entry_lsn;
    return 0;
}

__attribute__((weak))
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
{
    (void)c; (void)capnp_payload; (void)len;
    return 0;
}

__attribute__((weak))
int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state)
{
    (void)c; (void)jobid; (void)state;
    return 0;
}

__attribute__((weak))
int restore_queue_state(cluster_t* c, const char* name, uint32_t status)
{
    (void)c; (void)name; (void)status;
    return 0;
}

/* ------------------------------------------------------------------ */
/* Test helpers                                                         */
/* ------------------------------------------------------------------ */

static persist_t* make_test_persist_for_apply(void)
{
    persist_t* p = calloc(1, sizeof(persist_t));
    if (!p) {
        fprintf(stderr, "calloc failed\n");
        exit(1);
    }
    persist_config_t cfg = persist_config_defaults();
    cfg.wal_sync_ms = 5000; /* slow fsync — we don't need it in unit tests */

    char tmpdir[256];
    snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_apply_XXXXXX");
    if (!mkdtemp(tmpdir)) {
        fprintf(stderr, "mkdtemp failed\n");
        free(p);
        exit(1);
    }

    persist_set_workdir(p, tmpdir, "ctrl-apply");
    persist_init(p, &cfg, NULL); /* NULL cluster — entity stubs gate on p->cluster */
    persist_start(p);
    return p;
}

/* Like make_test_persist_for_apply but reports the chosen tmpdir so a test
 * can re-open the same WAL via a FRESH persist_t (restart/recover proof). */
static persist_t* make_test_persist_in(char* tmpdir_out, size_t cap)
{
    persist_t* p = calloc(1, sizeof(persist_t));
    if (!p) { fprintf(stderr, "calloc failed\n"); exit(1); }

    persist_config_t cfg = persist_config_defaults();
    cfg.wal_sync_ms = 5000;

    snprintf(tmpdir_out, cap, "/tmp/xce_replwr_XXXXXX");
    if (!mkdtemp(tmpdir_out)) {
        fprintf(stderr, "mkdtemp failed\n");
        free(p);
        exit(1);
    }

    persist_set_workdir(p, tmpdir_out, "ctrl-replwr");
    persist_init(p, &cfg, NULL);
    persist_start(p);
    return p;
}

/* Re-open an existing workdir as a fresh persist_t (simulates a restart). */
static persist_t* reopen_test_persist(const char* tmpdir)
{
    persist_t* p = calloc(1, sizeof(persist_t));
    if (!p) { fprintf(stderr, "calloc failed\n"); exit(1); }

    persist_config_t cfg = persist_config_defaults();
    cfg.wal_sync_ms = 5000;

    persist_set_workdir(p, tmpdir, "ctrl-replwr");
    persist_init(p, &cfg, NULL);
    persist_start(p);
    return p;
}

/* Count *.wal files under <tmpdir>/ctrl-replwr/data/wal */
static int count_replwr_wal_files(const char* tmpdir)
{
    char wal_path[512];
    snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir);
    DIR* dir = opendir(wal_path);
    if (!dir) return 0;
    int count = 0;
    struct dirent* ent;
    while ((ent = readdir(dir)) != NULL) {
        size_t len = strlen(ent->d_name);
        if (len > 4 && strcmp(ent->d_name + len - 4, ".wal") == 0)
            count++;
    }
    closedir(dir);
    return count;
}

/* Total bytes across all *.wal segment files (proxy for "entry on disk"). */
static off_t total_replwr_wal_bytes(const char* tmpdir)
{
    char wal_path[512];
    snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir);
    DIR* dir = opendir(wal_path);
    if (!dir) return 0;
    off_t total = 0;
    struct dirent* ent;
    while ((ent = readdir(dir)) != NULL) {
        size_t len = strlen(ent->d_name);
        if (len <= 4 || strcmp(ent->d_name + len - 4, ".wal") != 0)
            continue;
        char full[1024];
        snprintf(full, sizeof(full), "%s/%s", wal_path, ent->d_name);
        struct stat st;
        if (stat(full, &st) == 0)
            total += st.st_size;
    }
    closedir(dir);
    return total;
}

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

/* ------------------------------------------------------------------ */
/* Tests                                                                */
/* ------------------------------------------------------------------ */

static void test_apply_wal_entry_advances_lsn(void)
{
    persist_t* p = make_test_persist_for_apply();

    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.type        = WAL_JOB_UPDATE;
    hdr.lsn         = 42;
    hdr.term        = 1;
    hdr.payload_len = 4;
    hdr.crc         = 0xDEADBEEF;
    hdr.prev_crc    = WAL_CRC_NONE;

    uint8_t payload[4] = {1, 2, 3, 4};

    int rc = apply_wal_entry(p, hdr.type, hdr.lsn, payload, hdr.payload_len);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* current_lsn high-water must have advanced to at least lsn+1 */
    TEST_ASSERT_TRUE(p->current_lsn > 42);

    /* bug-778: apply_wal_entry is LOGICAL restore only — it must NOT touch the
     * on-disk CRC chain link. last_crc stays at its initial value (the physical
     * writer / replay owns it); it must NOT adopt the header's crc. */
    TEST_ASSERT_EQUAL_UINT32(WAL_CRC_NONE, p->last_crc);

    destroy_test_persist(p);
    printf("test_apply_wal_entry_advances_lsn: PASS\n");
}

static void test_apply_wal_entry_skips_checkpoint(void)
{
    persist_t* p = make_test_persist_for_apply();
    uint64_t lsn_before = p->current_lsn;

    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.type        = WAL_CHECKPOINT;
    hdr.lsn         = 100;
    hdr.term        = 1;
    hdr.payload_len = 0;

    int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
    TEST_ASSERT_EQUAL_INT(0, rc); /* skip = success, not error */

    /* Checkpoint entries are local-only markers — standbys never receive them.
     * current_lsn must NOT advance: skip means no-op. */
    TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn);

    destroy_test_persist(p);
    printf("test_apply_wal_entry_skips_checkpoint: PASS\n");
}

static void test_apply_wal_entry_rejects_oversized(void)
{
    persist_t* p = make_test_persist_for_apply();

    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.type        = WAL_JOB_SUBMIT;
    hdr.lsn         = 1;
    hdr.term        = 1;
    hdr.payload_len = MAX_WAL_PAYLOAD_BYTES + 1; /* too large */

    int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
    TEST_ASSERT_NOT_EQUAL(0, rc); /* must be rejected */

    destroy_test_persist(p);
    printf("test_apply_wal_entry_rejects_oversized: PASS\n");
}

static void test_apply_wal_entry_skips_unknown_type(void)
{
    persist_t* p = make_test_persist_for_apply();
    uint64_t lsn_before = p->current_lsn;

    /*
     * Type 200 is above WAL_ENTRY_TYPE_MAX (100).
     * Spec: log WARN and skip with rc=0 (forward compatibility).
     */
    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.type        = 200;
    hdr.lsn         = 5;
    hdr.term        = 1;
    hdr.payload_len = 0;

    int rc = apply_wal_entry(p, hdr.type, hdr.lsn, NULL, hdr.payload_len);
    TEST_ASSERT_EQUAL_INT(0, rc); /* unknown type: skip not error */

    /* LSN must NOT advance for a skipped-unknown type */
    TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn);

    destroy_test_persist(p);
    printf("test_apply_wal_entry_skips_unknown_type: PASS\n");
}

static void test_apply_wal_entry_null_guard(void)
{
    /* NULL persist must return non-zero (EINVAL) */
    int rc = apply_wal_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0);
    TEST_ASSERT_NOT_EQUAL(0, rc);
    printf("test_apply_wal_entry_null_guard: PASS\n");
}

/* ================================================================== */
/* persist_wal_write_replicated — standby persists replicated entries  */
/* (durability: persist-before-ACK). RED before the fix.               */
/* ================================================================== */

/* A streamed entry written to the standby's WAL must survive a restart:
 * a FRESH persist_recover over the same WAL replays it (crc-valid) and
 * advances current_lsn. Before the fix the segment is empty / nothing
 * is recovered. */
static void test_repl_write_persists_and_recovers(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    /* Fresh standby: first replicated entry is lsn 1 (1-based; reserve-lsn-0). */
    const char payload[] = "replicated_job_update_payload";
    int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE,
                                          payload, (uint32_t)strlen(payload));
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* current_lsn advanced to lsn+1 */
    TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn);

    /* bug-828 secondary: the entry was fdatasync'd inline before returning, so
     * durable_lsn must reflect the synced tip (lsn 1) IMMEDIATELY — not lag
     * behind waiting for the cadence flusher.  A standby advertises this as its
     * durable floor (RequestVoteReply.peerLastLogLsn) during a vote; if it lags,
     * a clean failover's safe-commit-from-votes under-counts. RED before fix. */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1ULL, persist_durable_lsn(p),
        "durable_lsn must advance to the synced tip after persist_wal_write_replicated");

    /* A .wal segment must now exist on disk. */
    TEST_ASSERT_TRUE(count_replwr_wal_files(tmpdir) >= 1);
    TEST_ASSERT_TRUE(total_replwr_wal_bytes(tmpdir) > 0);

    /* Simulate a restart: drop this persist_t, re-open the same workdir,
     * and run recovery. The replicated entry must replay (crc-chain valid)
     * and current_lsn must advance past it. */
    destroy_test_persist(p);

    persist_t* p2 = reopen_test_persist(tmpdir);
    int rec = persist_recover(p2);
    TEST_ASSERT_TRUE(rec >= 0);

    /* Recovery must have advanced current_lsn to 2 (replayed lsn 1). */
    TEST_ASSERT_EQUAL_UINT64(2, p2->current_lsn);

    destroy_test_persist(p2);
    printf("test_repl_write_persists_and_recovers: PASS\n");
}

/* Multiple replicated entries in sequence form a self-consistent crc chain
 * that the standby's OWN recovery validates with no chain break (all replay). */
static void test_repl_write_crc_chain_recovers(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    const char* payloads[] = { "entry-one", "entry-two", "entry-three" };
    for (int i = 0; i < 3; i++) {
        uint64_t lsn = (uint64_t)(i + 1); /* 1,2,3 */
        int rc = persist_wal_write_replicated(p, lsn, WAL_JOB_UPDATE,
                                              payloads[i],
                                              (uint32_t)strlen(payloads[i]));
        TEST_ASSERT_EQUAL_INT(0, rc);
    }
    TEST_ASSERT_EQUAL_UINT64(4, p->current_lsn);

    destroy_test_persist(p);

    /* Fresh recovery must replay ALL three (no crc-chain break) → current_lsn=4. */
    persist_t* p2 = reopen_test_persist(tmpdir);
    int rec = persist_recover(p2);
    TEST_ASSERT_TRUE(rec >= 0);
    TEST_ASSERT_EQUAL_UINT64(4, p2->current_lsn);

    destroy_test_persist(p2);
    printf("test_repl_write_crc_chain_recovers: PASS\n");
}

/* Re-streaming the same lsn must be idempotent: the dedup guard skips the
 * second write (no double-write, no chain corruption). After a dup write,
 * recovery still yields exactly the single entry chain. */
static void test_repl_write_dedup_idempotent(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    const char payload[] = "dup-entry";
    int rc1 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE,
                                           payload, (uint32_t)strlen(payload));
    TEST_ASSERT_EQUAL_INT(0, rc1);
    uint64_t lsn_after_first = p->current_lsn;        /* expect 2 */
    off_t    bytes_after_first = total_replwr_wal_bytes(tmpdir);

    /* Re-stream lsn 1 (already persisted): must be a no-op success. */
    int rc2 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE,
                                           payload, (uint32_t)strlen(payload));
    TEST_ASSERT_EQUAL_INT(0, rc2);

    /* current_lsn must NOT advance further; no extra bytes written. */
    TEST_ASSERT_EQUAL_UINT64(lsn_after_first, p->current_lsn);
    TEST_ASSERT_EQUAL_INT64((long long)bytes_after_first,
                            (long long)total_replwr_wal_bytes(tmpdir));

    /* Now legitimately stream lsn 2 — should append fine after the dup. */
    int rc3 = persist_wal_write_replicated(p, 2, WAL_JOB_UPDATE,
                                           payload, (uint32_t)strlen(payload));
    TEST_ASSERT_EQUAL_INT(0, rc3);
    TEST_ASSERT_EQUAL_UINT64(3, p->current_lsn);

    destroy_test_persist(p);

    /* Recovery must replay lsn 1 and lsn 2 (no duplicate / no break). */
    persist_t* p2 = reopen_test_persist(tmpdir);
    int rec = persist_recover(p2);
    TEST_ASSERT_TRUE(rec >= 0);
    TEST_ASSERT_EQUAL_UINT64(3, p2->current_lsn);

    destroy_test_persist(p2);
    printf("test_repl_write_dedup_idempotent: PASS\n");
}

/* bug-850 contiguity guard: a replicated entry whose lsn is BEYOND the next-
 * expected current_lsn (a forward gap) must be REJECTED with -EAGAIN without
 * writing or advancing current_lsn — otherwise it would advance the watermark
 * past the hole and the later in-order catch-up of the prefix would be silently
 * dropped by the idempotency guard (the failover prefix-loss mechanism). The
 * entry becomes writable only once the gap is filled contiguously. */
static void test_repl_write_rejects_forward_gap(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    /* Contiguous first entry: lsn 1 == current_lsn → written. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
        p, 1, WAL_JOB_UPDATE, "a", 1));
    TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn);

    /* Forward gap: lsn 5 > current_lsn(2) — like a live broadcast arriving
     * ahead of catch-up. Must be refused (-EAGAIN) and leave current_lsn put. */
    int rc_gap = persist_wal_write_replicated(p, 5, WAL_JOB_UPDATE, "e", 1);
    TEST_ASSERT_EQUAL_INT(-EAGAIN, rc_gap);
    TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn); /* did NOT jump past the gap */

    /* Fill the gap in order: 2,3,4 contiguous → accepted. */
    for (uint64_t lsn = 2; lsn <= 4; lsn++)
        TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
            p, lsn, WAL_JOB_UPDATE, "x", 1));
    TEST_ASSERT_EQUAL_UINT64(5, p->current_lsn);

    /* Now lsn 5 is contiguous (== current_lsn) → accepted. */
    TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
        p, 5, WAL_JOB_UPDATE, "e", 1));
    TEST_ASSERT_EQUAL_UINT64(6, p->current_lsn);

    destroy_test_persist(p);

    /* Recovery yields a clean contiguous chain [1..5] → current_lsn=6. */
    persist_t* p2 = reopen_test_persist(tmpdir);
    TEST_ASSERT_TRUE(persist_recover(p2) >= 0);
    TEST_ASSERT_EQUAL_UINT64(6, p2->current_lsn);
    destroy_test_persist(p2);

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

/* WAL_CHECKPOINT must never be persisted via the replicated-write path
 * (standbys never receive it; it must be rejected, not written). */
static void test_repl_write_rejects_checkpoint(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    int rc = persist_wal_write_replicated(p, 1, WAL_CHECKPOINT, NULL, 0);
    TEST_ASSERT_NOT_EQUAL(0, rc);

    /* current_lsn must not advance for a rejected checkpoint. */
    TEST_ASSERT_EQUAL_UINT64(1, p->current_lsn);

    destroy_test_persist(p);
    printf("test_repl_write_rejects_checkpoint: PASS\n");
}

/* ================================================================== */
/* bug-828 final slice: a catch-up'd standby durable at LSN 5 must     */
/* advertise its REAL durable tip (5) through a failover — the         */
/* pre-election truncate must NOT discard durably-committed entries    */
/* based on a STALE-LOW locally-known committed watermark.             */
/* ================================================================== */

/* A standby streams catch-up entries up to LSN 5 (all fdatasync'd durable),
 * so persist_durable_floor == 5.  This is the LSN the candidate advertises
 * (ha_candidate_advertise_lsn == persist_durable_floor) and the voter floor
 * (ha_vote_self_freshness over persist_durable_floor) at the failover vote,
 * which feeds safe-commit-from-votes.
 *
 * The clean-failover job-loss bug was that ha_enter_real_election truncated
 * the WAL above the standby's STALE locally-known last_known_committed_lsn
 * (e.g. 1 — the primary committed 2..5 but died before a heartbeat carried
 * committed=5 to this standby) BEFORE the vote.  persist_truncate_wal_above
 * clamps durable_lsn down to the cutoff, so the standby then advertised 1,
 * safe-commit computed 1, and the AUTHORITATIVE post-election truncate kept
 * only LSN 1 — the committed jobs 2..5 (durable on this very node) were lost.
 *
 * This test pins the fix at the persist seam: after catch-up to 5 the durable
 * floor is 5, and a truncate at the genuine safe-commit point (5, derived from
 * the un-clamped voter floors) PRESERVES all five entries.  A truncate at the
 * stale-low watermark (1) — the removed pre-election step — would instead
 * destroy LSN 2..5; this test asserts the durable committed prefix survives. */
static void test_repl_catchup_durable_floor_survives_failover(void)
{
    char tmpdir[256];
    persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));

    /* Stream the committed prefix [1..5] as a standby does on catch-up — each
     * persist_wal_write_replicated fdatasyncs inline (persist-before-ACK). */
    for (uint64_t lsn = 1; lsn <= 5; lsn++) {
        int rc = persist_wal_write_replicated(p, lsn, WAL_JOB_UPDATE,
                                              "committed-job",
                                              (uint32_t)strlen("committed-job"));
        TEST_ASSERT_EQUAL_INT(0, rc);
    }

    /* All five are fdatasync'd durable, so the advertised floor is 5 — the
     * value the failover vote must carry (RED before the partial fix: 1). */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(5ULL, persist_durable_floor(p),
        "catch-up'd standby must advertise its real fsync'd durable tip (5)");

    /* The genuine safe-commit point (from the un-clamped voter floors) is 5;
     * truncating above it is a no-op and preserves the committed prefix. This
     * is the ONLY truncate the fixed failover path runs (post-election). */
    TEST_ASSERT_EQUAL_INT(0, persist_truncate_wal_above(p, 5));
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(5ULL, persist_durable_floor(p),
        "safe-commit truncate at the real tip must preserve all committed entries");
    TEST_ASSERT_EQUAL_UINT64(6, p->current_lsn);

    /* Recovery over the same WAL replays the full committed prefix [1..5]. */
    destroy_test_persist(p);
    persist_t* p2 = reopen_test_persist(tmpdir);
    TEST_ASSERT_TRUE(persist_recover(p2) >= 0);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(6, p2->current_lsn,
        "the durable committed prefix [1..5] must survive a restart");
    TEST_ASSERT_EQUAL_UINT64(5ULL, persist_durable_floor(p2));
    destroy_test_persist(p2);

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

/* ------------------------------------------------------------------ */
/* Suite entry point (registered in controller_test.c)                 */
/* ------------------------------------------------------------------ */

void test_repl_apply(void)
{
    RUN_TEST(test_apply_wal_entry_advances_lsn);
    RUN_TEST(test_apply_wal_entry_skips_checkpoint);
    RUN_TEST(test_apply_wal_entry_rejects_oversized);
    RUN_TEST(test_apply_wal_entry_skips_unknown_type);
    RUN_TEST(test_apply_wal_entry_null_guard);

    /* persist-before-ACK durability (standby persists replicated entries) */
    RUN_TEST(test_repl_write_persists_and_recovers);
    RUN_TEST(test_repl_write_crc_chain_recovers);
    RUN_TEST(test_repl_write_dedup_idempotent);
    RUN_TEST(test_repl_write_rejects_forward_gap);
    RUN_TEST(test_repl_write_rejects_checkpoint);

    /* bug-828 final: durable committed prefix survives the failover vote */
    RUN_TEST(test_repl_catchup_durable_floor_survives_failover);
}