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, payload);
    TEST_ASSERT_EQUAL_INT(0, rc);

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

    /* last_crc must reflect the entry's crc */
    TEST_ASSERT_EQUAL_UINT32(0xDEADBEEF, 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, NULL);
    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, NULL);
    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, NULL);
    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)
{
    /* Both NULL inputs must return non-zero (EINVAL) */
    int rc = apply_wal_entry(NULL, NULL, NULL);
    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);

    /* 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");
}

/* 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");
}

/* ------------------------------------------------------------------ */
/* 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_checkpoint);
}