el/engram/test/test_compaction.c

/* test_compaction.c — M5 gate: ONLINE COMPACTION + background checkpointer.
 *
 * Pure C. Build: gcc -O2 test_compaction.c ../../lang/runtime/engram_store.c -o t
 * Writes ONLY under a throwaway /tmp dir. Never touches ~/.neuron or live ports.
 *
 * Proves:
 *   1) RECLAIM — tombstone/forget a large fraction of nodes + re-put many edges
 *      (dead versions) + orphan large-record overflow chains, then compact:
 *      page count AND file size drop, yet EVERY live record survives bit-exact and
 *      the id + adjacency indexes resolve correctly at the relocated positions.
 *   2) CRASH-DURING-COMPACTION — kill at phases 0/1/2; recovery is always a
 *      consistent store (crc clean, every live record intact), never corrupt.
 *   3) BACKGROUND CHECKPOINTER — a low ops / WAL-bytes threshold fires a checkpoint
 *      automatically on the write path; the WAL prefix is reclaimed; recovery works.
 *   4) POOL COOPERATION — compaction under a tiny ENGRAM_POOL_FRAMES stays correct
 *      with no stale frame surviving for a relocated page.
 */
#include "../../lang/runtime/engram_store.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>

static int g_pass = 0, g_fail = 0;
static void ok(const char* name, int cond){
    printf("  [%s] %s\n", cond ? "PASS" : "FAIL", name);
    if (cond) g_pass++; else g_fail++;
}

static char g_dir[512];
static int g_dseq = 0;
static void mk_dir(void){
    snprintf(g_dir, sizeof g_dir, "/tmp/engram-compact-test-%d-%d", (int)getpid(), g_dseq++);
    mkdir(g_dir, 0700);
}
static void egm_path(char* out, size_t cap){ snprintf(out, cap, "%s/neuron.egm", g_dir); }
static void wal_path(char* out, size_t cap){ snprintf(out, cap, "%s/neuron.wal", g_dir); }
static long file_size(const char* p){ struct stat st; return stat(p,&st)==0 ? (long)st.st_size : -1; }

/* ── deterministic generators (bit-exact regeneration for oracles) ─────────── */
static uint64_t xs(uint64_t* s){ uint64_t x=*s; x^=x<<13; x^=x>>7; x^=x<<17; *s=x; return x; }
static uint64_t node_seed(int i){ return 0x9E3779B97F4A7C15ULL ^ ((uint64_t)(i+1)*0xD1B54A32D192ED03ULL); }
static uint64_t edge_seed(int i){ return 0xC2B2AE3D27D4EB4FULL ^ ((uint64_t)(i+1)*0x165667B19E3779F9ULL); }
static char* rnd_str(uint64_t* st, size_t len){
    char* s = (char*)malloc(len + 1);
    for (size_t i=0;i<len;i++) s[i] = (char)(33 + (xs(st) % 94));
    s[len] = 0; return s;
}

#define N_NODES     1500
#define N_DEAD      1200          /* forget node-0 .. node-1199 (1200 dead / 300 live) */
#define N_EDGES     3000
#define EDGE_REPUT  2000          /* re-put edge-0 .. edge-1999 to version 3 */
#define EMB_DIM     96

static int node_is_live(int i){ return i >= N_DEAD; }
static int edge_live_version(int i){ return (i < EDGE_REPUT) ? 3 : 0; }

static void gen_node(int i, StoreNode* n){
    memset(n, 0, sizeof *n);
    uint64_t st = node_seed(i);
    char id[32]; snprintf(id, sizeof id, "node-%d", i);
    n->id = strdup(id);
    /* every 7th record is large → its own overflow chain (orphaned when it dies) */
    size_t clen = (i % 7 == 0) ? (size_t)(18000 + (xs(&st) % 4000)) : (size_t)(xs(&st) % 200);
    n->content   = rnd_str(&st, clen);
    n->node_type = rnd_str(&st, 4 + (xs(&st) % 8));
    n->label     = (i % 2)   ? rnd_str(&st, 3 + (xs(&st) % 10)) : NULL;
    n->tier      = rnd_str(&st, 4 + (xs(&st) % 6));
    n->tags      = rnd_str(&st, xs(&st) % 40);
    n->metadata  = (i % 3)   ? rnd_str(&st, xs(&st) % 60) : NULL;
    n->salience             = (double)(xs(&st) % 1000000) / 997.0;
    n->importance           = (double)(xs(&st) % 1000000) / 131.0;
    n->confidence           = (double)(xs(&st) % 1000000) / 733.0;
    n->temporal_decay_rate  = (double)(xs(&st) % 1000000) / 101.0;
    n->activation_count     = (int64_t)(xs(&st) % 100000);
    n->last_activated       = (int64_t)xs(&st);
    n->created_at           = (int64_t)(1600000000000LL + i);
    n->updated_at           = (int64_t)xs(&st);
    n->background_activation = (double)(xs(&st) % 1000000) / 17.0;
    n->working_memory_weight = (double)(xs(&st) % 1000000) / 29.0;
    n->suppression_count    = (int32_t)(xs(&st) % 50);
    n->layer_id             = (uint32_t)(xs(&st) % 5);
    for (int k=0;k<STORE_BLL_K;k++) n->access_ts[k] = (int64_t)xs(&st);
    n->access_head   = (int32_t)(xs(&st) % STORE_BLL_K);
    n->access_filled = (int32_t)(xs(&st) % (STORE_BLL_K + 1));
    n->wm_anchor     = (double)(xs(&st) % 1000000) / 3.0;
    n->emb = (float*)malloc(EMB_DIM * sizeof(float));
    for (int k=0;k<EMB_DIM;k++){ uint32_t u=(uint32_t)xs(&st); memcpy(&n->emb[k], &u, 4); }
    n->emb_dim = EMB_DIM;
}
/* version alters weight/hebb/last_fired so a re-put is a distinct payload. */
static void gen_edge(int i, int version, StoreEdge* e){
    memset(e, 0, sizeof *e);
    uint64_t st = edge_seed(i);
    char id[32], from[32], to[32];
    snprintf(id, sizeof id, "edge-%d", i);
    /* connect live nodes so adjacency queries on live nodes are meaningful */
    snprintf(from, sizeof from, "node-%d", N_DEAD + (int)(xs(&st) % (N_NODES - N_DEAD)));
    snprintf(to,   sizeof to,   "node-%d", N_DEAD + (int)(xs(&st) % (N_NODES - N_DEAD)));
    e->id = strdup(id); e->from_id = strdup(from); e->to_id = strdup(to);
    e->relation = rnd_str(&st, 3 + (xs(&st) % 12));
    e->metadata = (i % 4) ? rnd_str(&st, xs(&st) % 40) : NULL;
    e->weight     = (double)(xs(&st) % 1000000) / 7.0 + version * 100.0;
    e->hebb       = (double)(xs(&st) % 1000000) / 13.0 + version * 3.0;
    e->confidence = (double)(xs(&st) % 1000000) / 5.0;
    e->created_at = (int64_t)(1600000000000LL + i);
    e->updated_at = (int64_t)xs(&st) + version;
    e->last_fired = (int64_t)xs(&st) + version * 1000;
    e->inhibitory = (int32_t)(xs(&st) % 2);
    e->layer_id   = (uint32_t)(xs(&st) % 5);
}

static int streq(const char* a, const char* b){
    if (!a && !b) return 1; if (!a || !b) return 0; return strcmp(a,b)==0;
}
static int cmp_node(const StoreNode* a, const StoreNode* b){
    if (!streq(a->id,b->id) || !streq(a->content,b->content) ||
        !streq(a->node_type,b->node_type) || !streq(a->label,b->label) ||
        !streq(a->tier,b->tier) || !streq(a->tags,b->tags) ||
        !streq(a->metadata,b->metadata)) return 0;
    if (a->salience!=b->salience || a->importance!=b->importance ||
        a->confidence!=b->confidence || a->temporal_decay_rate!=b->temporal_decay_rate ||
        a->activation_count!=b->activation_count || a->last_activated!=b->last_activated ||
        a->created_at!=b->created_at || a->updated_at!=b->updated_at ||
        a->background_activation!=b->background_activation ||
        a->working_memory_weight!=b->working_memory_weight ||
        a->suppression_count!=b->suppression_count || a->layer_id!=b->layer_id ||
        a->access_head!=b->access_head || a->access_filled!=b->access_filled ||
        a->wm_anchor!=b->wm_anchor || a->emb_dim!=b->emb_dim) return 0;
    for (int k=0;k<STORE_BLL_K;k++) if (a->access_ts[k]!=b->access_ts[k]) return 0;
    if ((a->emb==NULL) != (b->emb==NULL)) return 0;
    if (a->emb && memcmp(a->emb, b->emb, (size_t)a->emb_dim*4)!=0) return 0;
    return 1;
}
static int cmp_edge(const StoreEdge* a, const StoreEdge* b){
    if (!streq(a->id,b->id) || !streq(a->from_id,b->from_id) || !streq(a->to_id,b->to_id) ||
        !streq(a->relation,b->relation) || !streq(a->metadata,b->metadata)) return 0;
    if (a->weight!=b->weight || a->hebb!=b->hebb || a->confidence!=b->confidence ||
        a->created_at!=b->created_at || a->updated_at!=b->updated_at ||
        a->last_fired!=b->last_fired || a->inhibitory!=b->inhibitory ||
        a->layer_id!=b->layer_id) return 0;
    return 1;
}

/* Populate a durable store with dead space: all nodes/edges, then forget the first
 * N_DEAD nodes and re-put the first EDGE_REPUT edges three times. */
static void populate_with_dead_space(EngramPagedStore* s){
    for (int i=0;i<N_NODES;i++){ StoreNode n; gen_node(i,&n); store_put_node(s,&n); store_node_free(&n); }
    for (int i=0;i<N_EDGES;i++){ StoreEdge e; gen_edge(i,0,&e); store_put_edge(s,&e); store_edge_free(&e); }
    /* re-put (in-place field mutation) → prior versions become dead records */
    for (int v=1; v<=3; v++)
        for (int i=0;i<EDGE_REPUT;i++){ StoreEdge e; gen_edge(i,v,&e); store_put_edge(s,&e); store_edge_free(&e); }
    /* forget the cold nodes (tombstone; their large overflow chains orphan) */
    for (int i=0;i<N_DEAD;i++){ char id[32]; snprintf(id,sizeof id,"node-%d",i); store_forget(s,id); }
}

/* Assert every live node/edge is present + bit-exact via point reads. */
static int verify_live_set(EngramPagedStore* s){
    int bad = 0;
    for (int i=0;i<N_NODES;i++){
        char id[32]; snprintf(id,sizeof id,"node-%d",i);
        StoreNode got; int hit = store_get_node(s, id, &got);
        if (node_is_live(i)){
            StoreNode want; gen_node(i,&want);
            if (hit!=1 || !cmp_node(&want,&got)) bad++;
            if (hit==1) store_node_free(&got);
            store_node_free(&want);
        } else {
            if (hit!=0) bad++;                 /* forgotten → must be absent */
            if (hit==1) store_node_free(&got);
        }
    }
    for (int i=0;i<N_EDGES;i++){
        char id[32]; snprintf(id,sizeof id,"edge-%d",i);
        StoreEdge got; int hit = store_get_edge(s, id, &got);
        StoreEdge want; gen_edge(i, edge_live_version(i), &want);
        if (hit!=1 || !cmp_edge(&want,&got)) bad++;
        if (hit==1) store_edge_free(&got);
        store_edge_free(&want);
    }
    return bad;
}

/* ════════════════════════════════════════════════════════════════════════════
 * TEST 1 — RECLAIM: dead space is reclaimed; live records + indexes survive.
 * ════════════════════════════════════════════════════════════════════════════ */
static void test_reclaim(void){
    printf("\n== 1) reclaim: forget %d nodes + re-put %d edges x3, then compact ==\n",
           N_DEAD, EDGE_REPUT);
    mk_dir();
    char egm[600]; egm_path(egm, sizeof egm);
    EngramPagedStore* s = engram_open(g_dir);
    ok("engram_open", s != NULL);
    if (!s) return;

    populate_with_dead_space(s);
    engram_checkpoint(s);                       /* flush so file size reflects state */

    uint64_t pc_before   = store_page_count(s);
    uint64_t free_before = store_free_page_count(s);
    long     sz_before   = file_size(egm);
    printf("   BEFORE: page_count=%llu  free_pages=%llu  file=%ld bytes  (live records intact?)\n",
           (unsigned long long)pc_before, (unsigned long long)free_before, sz_before);
    ok("pre-compaction live set intact", verify_live_set(s)==0);

    /* capture adjacency for a sample of live from-ids to compare post-compaction */
    #define NSAMP 12
    char samp[NSAMP][32]; size_t pre_cnt[NSAMP];
    for (int k=0;k<NSAMP;k++){
        snprintf(samp[k], sizeof samp[k], "node-%d", N_DEAD + k*20);
        StoreEdge* arr=NULL; size_t cnt=0;
        store_get_edges_from(s, samp[k], &arr, &cnt);
        pre_cnt[k]=cnt; store_edges_free(arr,cnt);
    }

    int rc = store_compact(s);
    ok("store_compact returns 0", rc==0);

    uint64_t pc_after   = store_page_count(s);
    uint64_t free_after = store_free_page_count(s);
    long     sz_after   = file_size(egm);
    printf("   AFTER : page_count=%llu  free_pages=%llu  file=%ld bytes\n",
           (unsigned long long)pc_after, (unsigned long long)free_after, sz_after);
    printf("   RECLAIMED: %llu pages, %ld bytes (%.1f%% of file)\n",
           (unsigned long long)(pc_before - pc_after), sz_before - sz_after,
           sz_before ? 100.0*(sz_before-sz_after)/sz_before : 0.0);

    ok("page count dropped (dead pages reclaimed)", pc_after < pc_before);
    ok("file size dropped (store physically shrank)", sz_after < sz_before);
    ok("store_check crc clean after compaction", store_check(s, STORE_CHECK_CRC)==0);
    ok("every LIVE record present + bit-exact at new locations", verify_live_set(s)==0);

    /* adjacency index correct at relocated positions */
    int adj_bad = 0;
    for (int k=0;k<NSAMP;k++){
        StoreEdge* arr=NULL; size_t cnt=0;
        store_get_edges_from(s, samp[k], &arr, &cnt);
        if (cnt != pre_cnt[k]) adj_bad++;
        for (size_t j=0;j<cnt;j++){
            if (!streq(arr[j].from_id, samp[k])) { adj_bad++; break; }
            /* the returned edge must be the canonical latest live edge, bit-exact */
            int idx = atoi(arr[j].id + 5);
            StoreEdge want; gen_edge(idx, edge_live_version(idx), &want);
            if (!cmp_edge(&want,&arr[j])) adj_bad++;
            store_edge_free(&want);
        }
        store_edges_free(arr,cnt);
    }
    ok("adjacency (get_edges_from) correct + bit-exact post-compaction", adj_bad==0);

    /* second compaction is a near no-op (no new dead space) and stays correct */
    uint64_t pc2_before = store_page_count(s);
    ok("compact again returns 0", store_compact(s)==0);
    ok("idempotent-ish: no growth on re-compact", store_page_count(s) <= pc2_before);
    ok("live set still intact after 2nd compaction", verify_live_set(s)==0);

    engram_close(s);
}

/* ════════════════════════════════════════════════════════════════════════════
 * TEST 2 — CRASH DURING COMPACTION: kill at phases 0/1/2 → consistent recovery.
 * Live set is identical whether we recover pre- or post-compaction, so the same
 * oracle must hold, and crc must always be clean (never corrupt).
 * ════════════════════════════════════════════════════════════════════════════ */
static void test_crash_during_compaction(void){
    printf("\n== 2) crash during compaction at phases 0,1,2 → consistent store ==\n");
    for (int phase=0; phase<=2; phase++){
        mk_dir();
        char egm[600]; egm_path(egm, sizeof egm);
        EngramPagedStore* s = engram_open(g_dir);
        if (!s){ ok("engram_open", 0); continue; }
        populate_with_dead_space(s);
        engram_close(s);                          /* durable baseline on disk */

        uint64_t pc_pre = 0;
        { EngramPagedStore* p = engram_open(g_dir); pc_pre = store_page_count(p); engram_close(p); }

        EngramPagedStore* c = engram_open(g_dir);
        store__compact_crashat(c, phase);         /* crashes mid-compaction (frees c) */

        EngramPagedStore* r = engram_open(g_dir); /* recover */
        char nm[80];
        snprintf(nm, sizeof nm, "phase %d: recovers, crc clean", phase);
        ok(nm, r && store_check(r, STORE_CHECK_CRC)==0);
        snprintf(nm, sizeof nm, "phase %d: every live record intact (not corrupt)", phase);
        ok(nm, r && verify_live_set(r)==0);
        if (r){
            uint64_t pc_now = store_page_count(r);
            if (phase < 2){
                snprintf(nm, sizeof nm, "phase %d: recovered PRE-compaction image", phase);
                ok(nm, pc_now == pc_pre);
            } else {
                snprintf(nm, sizeof nm, "phase %d: recovered POST-compaction (shrunk)", phase);
                ok(nm, pc_now < pc_pre);
            }
            /* store stays writable + durable after recovery */
            StoreNode n; gen_node(N_NODES+phase, &n); free(n.id);
            n.id = strdup("post-recovery-node");
            store_put_node(r, &n); store_node_free(&n);
            StoreNode g; int hit = store_get_node(r, "post-recovery-node", &g);
            snprintf(nm, sizeof nm, "phase %d: store writable after recovery", phase);
            ok(nm, hit==1);
            if (hit==1) store_node_free(&g);
            engram_close(r);
        }
    }
}

/* ════════════════════════════════════════════════════════════════════════════
 * TEST 3 — BACKGROUND CHECKPOINTER: a low threshold fires checkpoints on the
 * write path, reclaiming the WAL prefix automatically; recovery still correct.
 * ════════════════════════════════════════════════════════════════════════════ */
static void test_background_checkpointer(void){
    printf("\n== 3) background checkpointer: auto-checkpoint on threshold ==\n");

    /* (a) ops trigger */
    {
        mk_dir();
        char wal[600]; wal_path(wal, sizeof wal);
        EngramPagedStore* s = engram_open(g_dir);
        if (!s){ ok("engram_open", 0); return; }
        store_set_checkpoint_policy(s, /*ops*/50, /*dirty*/0, /*wal_bytes*/0, /*ms*/0);
        uint64_t ckpt0 = engram_last_checkpoint_lsn(s);
        for (int i=0;i<600;i++){ StoreNode n; gen_node(i,&n); store_put_node(s,&n); store_node_free(&n); }
        uint64_t ckpt1 = engram_last_checkpoint_lsn(s);
        long wsz = file_size(wal);
        printf("   ops-trigger: ckpt_lsn %llu -> %llu, WAL=%ld bytes after 600 puts\n",
               (unsigned long long)ckpt0, (unsigned long long)ckpt1, wsz);
        ok("ops trigger fired an automatic checkpoint", ckpt1 > ckpt0);
        ok("WAL prefix reclaimed (WAL stays small)", wsz >= 0 && wsz < 200000);
        /* crash (abandon RAM) then recover — everything durable via WAL+checkpoint */
        store__crash(s);
        EngramPagedStore* r = engram_open(g_dir);
        int bad=0;
        for (int i=0;i<600;i++){ char id[32]; snprintf(id,sizeof id,"node-%d",i);
            StoreNode w; gen_node(i,&w); StoreNode g; int hit=store_get_node(r,id,&g);
            if (hit!=1 || !cmp_node(&w,&g)) bad++; if(hit==1) store_node_free(&g); store_node_free(&w); }
        ok("recovery correct after auto-checkpoints (ops)", r && bad==0);
        ok("crc clean after recovery (ops)", r && store_check(r,STORE_CHECK_CRC)==0);
        if (r) engram_close(r);
    }
    /* (b) WAL-bytes trigger */
    {
        mk_dir();
        char wal[600]; wal_path(wal, sizeof wal);
        EngramPagedStore* s = engram_open(g_dir);
        if (!s){ ok("engram_open", 0); return; }
        store_set_checkpoint_policy(s, /*ops*/0, /*dirty*/0, /*wal_bytes*/64*1024, /*ms*/0);
        uint64_t ckpt0 = engram_last_checkpoint_lsn(s);
        for (int i=0;i<600;i++){ StoreNode n; gen_node(i,&n); store_put_node(s,&n); store_node_free(&n); }
        uint64_t ckpt1 = engram_last_checkpoint_lsn(s);
        long wsz = file_size(wal);
        printf("   wal-bytes-trigger: ckpt_lsn %llu -> %llu, WAL=%ld bytes\n",
               (unsigned long long)ckpt0, (unsigned long long)ckpt1, wsz);
        ok("wal-bytes trigger fired an automatic checkpoint", ckpt1 > ckpt0);
        ok("WAL kept bounded by byte threshold", wsz >= 0 && wsz < 2*1024*1024);
        engram_close(s);
    }
    /* (c) dirty-frames trigger (under a bounded pool) */
    {
        mk_dir();
        EngramPagedStore* s = engram_open(g_dir);
        if (!s){ ok("engram_open", 0); return; }
        store_set_checkpoint_policy(s, /*ops*/0, /*dirty*/16, /*wal_bytes*/0, /*ms*/0);
        uint64_t ckpt0 = engram_last_checkpoint_lsn(s);
        for (int i=0;i<400;i++){ StoreNode n; gen_node(i,&n); store_put_node(s,&n); store_node_free(&n); }
        uint64_t ckpt1 = engram_last_checkpoint_lsn(s);
        ok("dirty-frames trigger fired an automatic checkpoint", ckpt1 > ckpt0);
        engram_close(s);
    }
}

/* ════════════════════════════════════════════════════════════════════════════
 * TEST 4 — POOL COOPERATION: compact under a tiny frame budget (constant eviction
 * + re-fault); correctness holds and no stale frame survives a relocated page.
 * ════════════════════════════════════════════════════════════════════════════ */
static void test_pool_cooperation(void){
    printf("\n== 4) compaction under a small buffer pool (forced eviction) ==\n");
    setenv("ENGRAM_POOL_FRAMES", "24", 1);        /* pool << store, and the temp build too */
    mk_dir();
    char egm[600]; egm_path(egm, sizeof egm);
    EngramPagedStore* s = engram_open(g_dir);
    ok("engram_open (24-frame pool)", s != NULL);
    if (!s){ unsetenv("ENGRAM_POOL_FRAMES"); return; }
    store__set_pool_frames(s, 24);

    populate_with_dead_space(s);
    engram_checkpoint(s);
    uint64_t pc_before = store_page_count(s);

    int rc = store_compact(s);
    ok("store_compact under tiny pool returns 0", rc==0);

    StorePoolStats st; store_pool_stats(s, &st);
    printf("   post-compaction pool: cap=%zu resident=%zu pinned=%zu dirty=%zu\n",
           st.cap, st.resident, st.pinned, st.dirty);
    ok("pool respected budget after compaction (resident<=cap)", st.resident <= st.cap);
    ok("page count dropped under small pool", store_page_count(s) < pc_before);
    ok("crc clean under small pool", store_check(s, STORE_CHECK_CRC)==0);
    /* If any relocated page had a stale frame, a read would return wrong bytes. */
    ok("every live record bit-exact under small pool (no stale frames)", verify_live_set(s)==0);

    engram_close(s);
    unsetenv("ENGRAM_POOL_FRAMES");
}

int main(void){
    printf("=== M5 COMPACTION + BACKGROUND CHECKPOINTER GATE ===\n");
    test_reclaim();
    test_crash_during_compaction();
    test_background_checkpointer();
    test_pool_cooperation();
    printf("\n=== RESULT: %d passed, %d failed ===\n", g_pass, g_fail);
    /* cleanup */
    return g_fail ? 1 : 0;
}