myxce/src/controller/testsuite/walcompare.c

#define _GNU_SOURCE
#include "walcompare.h"

#include <ctype.h>
#include <dirent.h>
#include <errno.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "persist.h"  /* WAL_LEGACY_TERM */
#include "waldump.h"

#define WALCMP_MAX_MEMBERS 16

/* One member's collected log, indexed by LSN. `slot[lsn]` is valid when
 * `present[lsn]` is set. LSNs are 1-based; index 0 is unused. */
typedef struct {
    uint64_t  recovered_tip; /* highest CRC-intact LSN held (waldump tip) */
    uint64_t  committed_lsn;  /* effective committed boundary for this member */
    /* raft-ha-core-safety task 5.10a / ha-entity-snapshots task 7.1: the
     * highest checkpoint LSN this member has installed (0 = never installed
     * a snapshot -- expect raw WAL from LSN 1, the pre-existing behavior,
     * unchanged). A member that caught up via a full-checkpoint snapshot
     * legitimately holds NO raw WAL entries at or below its own floor_lsn --
     * that range was restored via entity content, not WAL replay -- so it
     * must never be flagged MISSING there. Read once per member from its own
     * snap/ directory (scan_snapshot_floor); never uniform across members
     * (unlike committed_lsn, which the caller may force to one shared value). */
    uint64_t  floor_lsn;
    uint64_t  cap;            /* allocated index size (LSN range [0..cap-1]) */
    uint64_t* term;           /* raw on-disk term per LSN (0 = legacy / absent) */
    uint32_t* pcrc;           /* payload CRC per LSN */
    uint8_t*  present;        /* 1 if this LSN is held */
} member_log_t;

/* Visitor context: collect entries into a member_log_t. */
typedef struct {
    member_log_t* log;
    int           oom; /* set on an allocation failure mid-scan */
} collect_ctx_t;

/* Map a raw on-disk term to its matching term: legacy 0 -> WAL_LEGACY_TERM
 * (R1.8) so two pre-feature entries are not spuriously divergent. */
static uint64_t match_term(uint64_t raw)
{
    return raw == 0 ? WAL_LEGACY_TERM : raw;
}

/* Grow the per-LSN index so `lsn` is addressable. Returns 0 or -ENOMEM. */
static int member_log_ensure(member_log_t* m, uint64_t lsn)
{
    if (lsn < m->cap)
        return 0;
    uint64_t newcap = m->cap ? m->cap : 64;
    while (newcap <= lsn)
        newcap *= 2;

    uint64_t* nt = realloc(m->term, newcap * sizeof(*nt));
    uint32_t* np = realloc(m->pcrc, newcap * sizeof(*np));
    uint8_t*  ne = realloc(m->present, newcap * sizeof(*ne));
    if (!nt || !np || !ne) {
        /* realloc may have partially succeeded; keep whatever grew so the
         * eventual free() is safe, but report OOM. */
        if (nt)
            m->term = nt;
        if (np)
            m->pcrc = np;
        if (ne)
            m->present = ne;
        return -ENOMEM;
    }
    /* Zero the newly added tail. */
    memset(nt + m->cap, 0, (newcap - m->cap) * sizeof(*nt));
    memset(np + m->cap, 0, (newcap - m->cap) * sizeof(*np));
    memset(ne + m->cap, 0, (newcap - m->cap) * sizeof(*ne));
    m->term    = nt;
    m->pcrc    = np;
    m->present = ne;
    m->cap     = newcap;
    return 0;
}

static int collect_visit(void* ctx, const waldump_entry_t* e)
{
    collect_ctx_t* cc = ctx;
    member_log_t*  m  = cc->log;
    if (e->lsn == 0)
        return 0; /* LSN 0 is never a real committed entry */
    if (member_log_ensure(m, e->lsn) != 0) {
        cc->oom = 1;
        return 1; /* stop the scan */
    }
    m->term[e->lsn]    = e->term;
    m->pcrc[e->lsn]    = e->pcrc;
    m->present[e->lsn] = 1;
    return 0;
}

static void member_log_free(member_log_t* m)
{
    free(m->term);
    free(m->pcrc);
    free(m->present);
    m->term    = NULL;
    m->pcrc    = NULL;
    m->present = NULL;
    m->cap     = 0;
}

/* Scan one directory for `snap-<16-hex-digit-lsn>.{capnp,bin}` files (the
 * on-disk naming write_snapshot()/legacy checkpoints use — persist.c) and
 * return the highest embedded LSN found, or 0 if the directory doesn't exist
 * or holds no matching file. A bare 16-hex-digit match is required (skips
 * `.tmp` in-progress files and anything else) so this never mis-parses an
 * unrelated entry as a checkpoint LSN. */
static uint64_t scan_snap_dir_floor(const char* dir)
{
    DIR* d = opendir(dir);
    if (!d)
        return 0; /* no such dir (no snapshot ever installed here) — floor 0 */

    uint64_t max_lsn = 0;
    struct dirent* de;
    while ((de = readdir(d)) != NULL) {
        unsigned long long lsn = 0;
        char ext[8];
        /* "snap-XXXXXXXXXXXXXXXX.capnp" / ".bin" — %16llx would also accept
         * fewer than 16 digits, so additionally check the literal width via
         * strlen on the numeric run; simplest robust check: require the
         * filename to be exactly "snap-" + 16 hex chars + "." + ext. */
        size_t len = strlen(de->d_name);
        if (len < 5 + 16 + 1 + 3 || strncmp(de->d_name, "snap-", 5) != 0)
            continue;
        int matched = sscanf(de->d_name + 5, "%16llx.%7s", &lsn, ext);
        if (matched != 2)
            continue;
        if (strcmp(ext, "capnp") != 0 && strcmp(ext, "bin") != 0)
            continue;
        /* Confirm the hex run is EXACTLY 16 chars (reject a short-digit name
         * like "snap-1.capnp" that sscanf's %16llx would still happily take). */
        int hexlen = 0;
        while (hexlen < 16 && isxdigit((unsigned char)de->d_name[5 + hexlen]))
            hexlen++;
        if (hexlen != 16 || de->d_name[5 + 16] != '.')
            continue;
        if ((uint64_t)lsn > max_lsn)
            max_lsn = (uint64_t)lsn;
    }
    closedir(d);
    return max_lsn;
}

/* Find this member's own checkpoint floor (raft-ha-core-safety task 5.10a /
 * ha-entity-snapshots task 7.1). `src->path` may be either a "data" dir
 * containing sibling `wal/`+`snap/` subdirs (the soak harness's convention:
 * `$TOP/controller/ctrl-N/data`), or the WAL dir itself passed directly (seen
 * in unit-test call sites, e.g. `persist_t.wal_path`) — in the latter case the
 * sibling snap dir sits at the same parent level with `wal` replaced by
 * `snap`. Try both; a nonexistent candidate just yields floor 0 (harmless). */
static uint64_t member_snapshot_floor(const char* path)
{
    if (!path || !*path)
        return 0;

    char candidate[4160];
    uint64_t floor1 = 0, floor2 = 0;

    /* Candidate 1: <path>/snap (path is a "data" parent dir). */
    snprintf(candidate, sizeof(candidate), "%s/snap", path);
    floor1 = scan_snap_dir_floor(candidate);

    /* Candidate 2: sibling of <path> with a trailing "/wal" (or "wal")
     * component replaced by "/snap" (path is the WAL dir itself). */
    size_t plen = strlen(path);
    const char* base = strrchr(path, '/');
    const char* leaf = base ? base + 1 : path;
    if (strcmp(leaf, "wal") == 0) {
        size_t prefix_len = (size_t)(leaf - path);
        if (prefix_len + 4 < sizeof(candidate)) {
            memcpy(candidate, path, prefix_len);
            snprintf(candidate + prefix_len, sizeof(candidate) - prefix_len, "snap");
            floor2 = scan_snap_dir_floor(candidate);
        }
    }
    (void)plen;

    return floor1 > floor2 ? floor1 : floor2;
}

/* Load one member's WAL into `*m`. Returns 0 or a negative errno. */
static int load_member(const walcompare_member_t* src, member_log_t* m, FILE* out)
{
    memset(m, 0, sizeof(*m));

    waldump_stats_t st;
    memset(&st, 0, sizeof(st));
    collect_ctx_t cc = { .log = m, .oom = 0 };

    /* Send the human dump to the caller's `out` (or a discard sink). The
     * visitor harvests the (lsn -> term, pcrc) index from the same decode. */
    FILE* dump_sink = out ? out : fopen("/dev/null", "w");
    int   own_sink  = (out == NULL);
    if (!dump_sink)
        return -errno;

    int rc = waldump_path_visit(src->path, dump_sink, &st, collect_visit, &cc);
    if (own_sink)
        fclose(dump_sink);

    if (rc != 0) {
        member_log_free(m);
        return rc; /* hard load error (missing path / no segments) */
    }
    if (cc.oom) {
        member_log_free(m);
        return -ENOMEM;
    }

    m->recovered_tip = st.recovered_tip;
    /* An explicit committed_lsn wins; 0 means "derive from the shared
     * boundary" (filled by the caller after all members are loaded). */
    m->committed_lsn = src->committed_lsn;
    /* Always PER-MEMBER, never uniform/shared (unlike committed_lsn) — a
     * member's own checkpoint floor has nothing to do with any other
     * member's state. */
    m->floor_lsn = member_snapshot_floor(src->path);
    return 0;
}

/* Loaded cluster state shared by walcompare() and walcompare_report(). */
typedef struct {
    member_log_t logs[WALCMP_MAX_MEMBERS];
    int          n;
    int          loaded;   /* members successfully loaded (for cleanup) */
    uint64_t     boundary; /* highest committed LSN compared (committed edge) */
    uint64_t     max_tip;  /* highest recovered tip across members (tail edge) */
} wc_cluster_t;

/* Stable status token for one member at one LSN. */
static const char* wc_status(const member_log_t* m, uint64_t lsn)
{
    if ((lsn < m->cap) && m->present[lsn])
        return "present";
    if (lsn <= m->floor_lsn)
        return "compacted"; /* below this member's OWN checkpoint floor —
                              * legitimately absent (entity-restored, not
                              * WAL-held), never a divergence signal */
    if (lsn <= m->committed_lsn)
        return "MISSING"; /* a committed entry this member SHOULD hold (R4.2) */
    return "beyond";      /* legitimately not held (above its committed prefix) */
}

/* Load all members and compute the committed boundary + tail edge. On a load
 * error, frees what loaded, reports to `out`, returns a negative errno. */
static int wc_load(const walcompare_member_t* members, int n, FILE* out,
                   wc_cluster_t* cl)
{
    memset(cl, 0, sizeof(*cl));
    cl->n = n;

    for (int i = 0; i < n; i++) {
        int rc = load_member(&members[i], &cl->logs[i], out);
        if (rc != 0) {
            if (out)
                fprintf(out, "walcompare: cannot load member %d ('%s'): %s\n", i,
                        members[i].path ? members[i].path : "(null)",
                        strerror(-rc));
            for (int j = 0; j < cl->loaded; j++)
                member_log_free(&cl->logs[j]);
            return rc;
        }
        cl->loaded = i + 1;
    }

    /* Shared committed boundary = the minimum recovered tip across members:
     * the largest committed prefix every member can be expected to hold. A
     * member with an explicit committed_lsn keeps its own. */
    uint64_t shared = cl->logs[0].recovered_tip;
    for (int i = 1; i < n; i++)
        if (cl->logs[i].recovered_tip < shared)
            shared = cl->logs[i].recovered_tip;

    for (int i = 0; i < n; i++) {
        if (cl->logs[i].committed_lsn == 0)
            cl->logs[i].committed_lsn = shared;
        if (cl->logs[i].committed_lsn > cl->boundary)
            cl->boundary = cl->logs[i].committed_lsn;
        if (cl->logs[i].recovered_tip > cl->max_tip)
            cl->max_tip = cl->logs[i].recovered_tip;
    }
    return 0;
}

static void wc_unload(wc_cluster_t* cl)
{
    for (int i = 0; i < cl->loaded; i++)
        member_log_free(&cl->logs[i]);
}

/* One machine line for member `i` at `lsn`. Fixed field order, stable tokens. */
static void wc_emit_member(const wc_cluster_t* cl, int i, uint64_t lsn,
                           int committed, FILE* out)
{
    const member_log_t* m     = &cl->logs[i];
    int                 holds = (lsn < m->cap) && m->present[lsn];
    fprintf(out,
            "WALCMP member=%d lsn=%" PRIu64 " term=%" PRIu64
            " pcrc=%08x status=%s committed=%d\n",
            i, lsn, holds ? match_term(m->term[lsn]) : 0,
            holds ? m->pcrc[lsn] : 0u, wc_status(m, lsn), committed);
}

/* Full per-member machine table over [1..max_tip]; committed=1 within the
 * committed boundary, 0 in the uncommitted tail. */
static void wc_emit_table(const wc_cluster_t* cl, FILE* out)
{
    for (uint64_t lsn = 1; lsn <= cl->max_tip; lsn++) {
        int committed = (lsn <= cl->boundary) ? 1 : 0;
        for (int i = 0; i < cl->n; i++)
            wc_emit_member(cl, i, lsn, committed, out);
    }
}

/* Emit the classification + per-member identity for ONE diverging LSN. */
static void wc_emit_divergence(const wc_cluster_t* cl, uint64_t lsn,
                               int committed, int compared, int present,
                               FILE* out)
{
    fprintf(out,
            "WALCMP_DIVERGENCE lsn=%" PRIu64 " class=%s compared=%d present=%d\n",
            lsn, committed ? "committed" : "uncommitted", compared, present);
    for (int i = 0; i < cl->n; i++)
        wc_emit_member(cl, i, lsn, committed, out);
}

/* Compare one committed LSN. Fills `div` (committed=1) and prints the human +
 * machine report when divergent. Returns 1 if divergent, else 0. */
static int wc_compare_committed(const wc_cluster_t* cl,
                                const walcompare_member_t* members, uint64_t lsn,
                                FILE* out, walcompare_divergence_t* div)
{
    int      have = 0, present = 0, have_ref = 0, divergent = 0;
    uint64_t ref_term = 0;
    uint32_t ref_pcrc = 0;

    for (int i = 0; i < cl->n; i++) {
        if (lsn > cl->logs[i].committed_lsn)
            continue; /* beyond member i's committed prefix */
        if (lsn <= cl->logs[i].floor_lsn)
            continue; /* below member i's OWN checkpoint floor — legitimately
                       * compacted (entity-restored), never a divergence */
        have++;
        const member_log_t* m     = &cl->logs[i];
        int                 holds = (lsn < m->cap) && m->present[lsn];
        if (!holds) {
            divergent = 1; /* member should hold this committed LSN but doesn't */
            continue;
        }
        present++;
        uint64_t mt = match_term(m->term[lsn]);
        uint32_t mp = m->pcrc[lsn];
        if (!have_ref) {
            ref_term = mt;
            ref_pcrc = mp;
            have_ref = 1;
        } else if (mt != ref_term || mp != ref_pcrc) {
            divergent = 1;
        }
    }

    if (have < 2 || !divergent)
        return 0;

    if (div) {
        memset(div, 0, sizeof(*div));
        div->lsn       = lsn;
        div->n_members = cl->n;
        div->committed = 1;
        for (int i = 0; i < cl->n && i < 16; i++) {
            const member_log_t* m = &cl->logs[i];
            int holds = (lsn <= m->committed_lsn) && (lsn > m->floor_lsn) &&
                        (lsn < m->cap) && m->present[lsn];
            div->member_present[i] = holds ? 1 : 0;
            div->member_term[i]    = holds ? match_term(m->term[lsn]) : 0;
            div->member_pcrc[i]    = holds ? m->pcrc[lsn] : 0;
        }
    }
    if (out) {
        fprintf(out,
                "*** DIVERGENCE at committed LSN %" PRIu64
                " (compared %d, present %d) ***\n",
                lsn, have, present);
        for (int i = 0; i < cl->n; i++) {
            if (lsn > cl->logs[i].committed_lsn) {
                fprintf(out,
                        "  member %d ('%s'): LSN beyond its committed prefix"
                        " (committed=%" PRIu64 ")\n",
                        i, members[i].path, cl->logs[i].committed_lsn);
                continue;
            }
            if (lsn <= cl->logs[i].floor_lsn) {
                fprintf(out,
                        "  member %d ('%s'): LSN below its own checkpoint floor"
                        " (floor=%" PRIu64 ") — legitimately compacted\n",
                        i, members[i].path, cl->logs[i].floor_lsn);
                continue;
            }
            int holds = (lsn < cl->logs[i].cap) && cl->logs[i].present[lsn];
            if (holds)
                fprintf(out, "  member %d ('%s'): term=%" PRIu64 " pcrc=%08x\n",
                        i, members[i].path, match_term(cl->logs[i].term[lsn]),
                        cl->logs[i].pcrc[lsn]);
            else
                fprintf(out, "  member %d ('%s'): MISSING committed entry\n", i,
                        members[i].path);
        }
        wc_emit_divergence(cl, lsn, 1, have, present, out);
    }
    return 1;
}

/* Scan the uncommitted tail (boundary < lsn <= max_tip): emit a NON-fatal
 * class=uncommitted classification for any LSN where >=2 members hold the entry
 * but disagree on (term, pcrc). Never changes the verdict. */
static void wc_scan_uncommitted(const wc_cluster_t* cl, FILE* out)
{
    for (uint64_t lsn = cl->boundary + 1; lsn <= cl->max_tip; lsn++) {
        int      present = 0, have_ref = 0, divergent = 0;
        uint64_t ref_term = 0;
        uint32_t ref_pcrc = 0;

        for (int i = 0; i < cl->n; i++) {
            const member_log_t* m     = &cl->logs[i];
            int                 holds = (lsn < m->cap) && m->present[lsn];
            if (!holds)
                continue;
            present++;
            uint64_t mt = match_term(m->term[lsn]);
            uint32_t mp = m->pcrc[lsn];
            if (!have_ref) {
                ref_term = mt;
                ref_pcrc = mp;
                have_ref = 1;
            } else if (mt != ref_term || mp != ref_pcrc) {
                divergent = 1;
            }
        }
        if (present < 2 || !divergent)
            continue;
        wc_emit_divergence(cl, lsn, 0, present, present, out);
    }
}

/* Shared driver for walcompare() (full_report=0) and walcompare_report()
 * (full_report=1). */
static int wc_run(const walcompare_member_t* members, int n, FILE* out,
                  walcompare_divergence_t* div, int full_report)
{
    if (!members || n < 2)
        return -EINVAL;
    if (n > WALCMP_MAX_MEMBERS)
        return -E2BIG;

    wc_cluster_t cl;
    int          rc = wc_load(members, n, out, &cl);
    if (rc != 0)
        return rc;

    if (out)
        fprintf(out,
                "== walcompare: %d members, committed boundary up to LSN %" PRIu64
                " ==\n",
                n, cl.boundary);

    if (full_report && out)
        wc_emit_table(&cl, out);

    /* Walk every committed LSN. At each LSN, compare the (term, pcrc) of the
     * members whose own committed boundary covers it and that hold the entry.
     * A member that does not hold a committed LSN it SHOULD is a committed-
     * prefix divergence: a committed entry is missing (R4.2 / R6.4). */
    rc = WALCMP_OK;
    for (uint64_t lsn = 1; lsn <= cl.boundary; lsn++) {
        if (wc_compare_committed(&cl, members, lsn, out, div)) {
            rc = WALCMP_DIVERGENT; /* first committed divergence is the verdict */
            goto done;
        }
    }

    /* The committed prefix is identical. In full-report mode also classify any
     * uncommitted-tail divergence (informational, never fatal). */
    if (full_report && out)
        wc_scan_uncommitted(&cl, out);

    if (out)
        fprintf(out, "walcompare: OK — committed prefixes identical\n");

done:
    wc_unload(&cl);
    return rc;
}

int walcompare(const walcompare_member_t* members, int n, FILE* out,
               walcompare_divergence_t* div)
{
    return wc_run(members, n, out, div, 0);
}

int walcompare_report(const walcompare_member_t* members, int n, FILE* out,
                      walcompare_divergence_t* div)
{
    return wc_run(members, n, out, div, 1);
}