diff --git a/engram/test/run_compaction_tests.sh b/engram/test/run_compaction_tests.sh new file mode 100755 index 0000000..725f241 --- /dev/null +++ b/engram/test/run_compaction_tests.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +# M5 online-compaction + background-checkpointer gate. Pure C (NOT elb/elc). +# Writes only under /tmp. Runs an -O2 correctness build then an ASan+UBSan build. +set -e +HERE="$(cd "$(dirname "$0")" && pwd)" +SRC="$HERE/../../lang/runtime/engram_store.c" +TST="$HERE/test_compaction.c" + +echo "== compiling (gcc -O2): test_compaction.c engram_store.c ==" +BIN="/tmp/test_compaction.$$" +gcc -O2 -Wall -Wextra -std=c11 "$TST" "$SRC" -o "$BIN" +"$BIN"; rc=$? +rm -f "$BIN"; rm -rf /tmp/engram-compact-test-* +[ $rc -ne 0 ] && exit $rc + +echo +echo "== ASan+UBSan build (memory-error + UB checks; LSan unavailable on macOS) ==" +ABIN="/tmp/test_compaction_asan.$$" +gcc -O1 -g -fsanitize=address,undefined -fno-omit-frame-pointer -std=c11 "$TST" "$SRC" -o "$ABIN" +ASAN_OPTIONS=detect_leaks=0 UBSAN_OPTIONS=halt_on_error=1 "$ABIN"; rc=$? +rm -f "$ABIN"; rm -rf /tmp/engram-compact-test-* +exit $rc diff --git a/engram/test/test_compaction.c b/engram/test/test_compaction.c new file mode 100644 index 0000000..dd13206 --- /dev/null +++ b/engram/test/test_compaction.c @@ -0,0 +1,421 @@ +/* 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 +#include +#include +#include +#include +#include +#include + +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= 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;kaccess_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;kemb[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;kaccess_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 %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; +} diff --git a/lang/runtime/engram_store.c b/lang/runtime/engram_store.c index 9cce290..d477296 100644 --- a/lang/runtime/engram_store.c +++ b/lang/runtime/engram_store.c @@ -141,6 +141,11 @@ struct EngramPagedStore { int recovering; /* set during WAL replay */ uint64_t ops_since_ckpt; /* checkpoint threshold counter */ uint64_t ckpt_threshold; /* auto-checkpoint after this many ops (0 = never) */ + /* ── M5 background-checkpointer triggers (0 = that trigger disabled) ────── */ + size_t ckpt_dirty_threshold; /* auto-checkpoint at this many dirty frames */ + uint64_t ckpt_wal_threshold; /* auto-checkpoint at this many WAL bytes */ + long long ckpt_interval_ms; /* auto-checkpoint after this many ms elapse */ + long long last_ckpt_ms; /* wall-clock time of the last checkpoint */ }; /* M2/M4 buffer-pool hooks (defined in the pool section at the bottom of this file). @@ -176,6 +181,7 @@ struct PgCache { PgEnt* mru; PgEnt* lru; /* MRU (front) → LRU (back) recency list */ unsigned prefetch; /* read-ahead window (pages); 0 = off */ LayerPin* lp; size_t lp_n, lp_cap; /* hot-layer pin bookkeeping */ + size_t dirty_count; /* # dirty frames, maintained incrementally (M5) */ /* stats (introspection only — never affect semantics) */ uint64_t hits, misses, evictions, prefetch_reads; }; @@ -1424,7 +1430,7 @@ static int pc_put(EngramPagedStore* s, uint64_t id, const uint8_t* buf, int dirt } memcpy(e->buf, buf, STORE_PAGE_SIZE); e->lsn = get_u64(buf + 16); - if (dirty) e->dirty = 1; + if (dirty){ if (!e->dirty) c->dirty_count++; e->dirty = 1; } /* clean→dirty transition */ pc_evict_to_budget(c); return 0; } @@ -1434,6 +1440,7 @@ static int pc_flush(EngramPagedStore* s){ for (size_t i=0;inbuckets;i++) for (PgEnt* e=c->buckets[i]; e; e=e->next) if (e->dirty){ if (page_write_raw(s, e->id, e->buf)!=0) return -1; e->dirty=0; } + c->dirty_count = 0; /* all frames clean after flush */ /* Post-checkpoint the just-cleaned frames are now evictable; trim the pool * back to budget so a dirty-heavy burst that transiently overshot cap does * not leave the pool oversized. No-op at the default (unlimited-ish) cap. */ @@ -1601,6 +1608,7 @@ struct EngramWal { uint64_t last_fsync_lsn; long long last_fsync_ms; uint64_t appended_since_fsync; + uint64_t bytes_since_reclaim; /* WAL bytes appended since last reclaim (M5) */ }; static long long now_ms(void){ @@ -1658,12 +1666,14 @@ static int wal_append(EngramPagedStore* s, uint8_t op, const uint8_t* payload, free(fr); if (wr != (ssize_t)fl) return -1; w->appended_since_fsync++; + w->bytes_since_reclaim += fl; wal_maybe_fsync(s, lsn); return 0; } static int wal_reclaim(EngramPagedStore* s, uint64_t ckpt_lsn){ EngramWal* w = s->wal; if (!w) return 0; if (ftruncate(w->fd, 0) != 0) return -1; /* prefix <= ckpt reclaimed */ + w->bytes_since_reclaim = 0; /* WAL just shrank to the marker */ uint8_t p[8]; put_u64(p, ckpt_lsn); if (wal_append(s, OP_CHECKPOINT, p, 8, ckpt_lsn) != 0) return -1; fsync(w->fd); w->last_fsync_ms = now_ms(); @@ -2008,12 +2018,24 @@ static int wal_recover(EngramPagedStore* s){ return rc; } -/* ── checkpoint threshold trigger ────────────────────────────────────────────── */ +/* ── background checkpointer: fire on ops / dirty-frames / WAL-bytes / timer ───── + * Single-threaded model: the triggers are evaluated on the write path (no + * background thread), so a checkpoint fires on the first mutation after any armed + * threshold trips. This reclaims the WAL prefix automatically instead of only at + * an explicit engram_checkpoint. Same checkpoint semantics as M2 (it calls the + * very same engram_checkpoint). */ static void ckpt_maybe(EngramPagedStore* s){ if (s->recovering) return; s->ops_since_ckpt++; - if (s->ckpt_threshold && s->ops_since_ckpt >= s->ckpt_threshold) - engram_checkpoint(s); + int fire = 0; + if (s->ckpt_threshold && s->ops_since_ckpt >= s->ckpt_threshold) fire = 1; + if (!fire && s->ckpt_dirty_threshold && s->cache && + s->cache->dirty_count >= s->ckpt_dirty_threshold) fire = 1; + if (!fire && s->ckpt_wal_threshold && s->wal && + s->wal->bytes_since_reclaim >= s->ckpt_wal_threshold) fire = 1; + if (!fire && s->ckpt_interval_ms && + (now_ms() - s->last_ckpt_ms) >= s->ckpt_interval_ms) fire = 1; + if (fire) engram_checkpoint(s); } /* ── public mutation entry points (log-then-apply when a WAL is attached) ─────── */ @@ -2148,6 +2170,7 @@ int store__checkpoint_crashat(EngramPagedStore* s, int phase){ if (phase == 3){ store__crash(s); return 0; } if (wal_reclaim(s, C) != 0) return -1; /* 4: reclaim WAL prefix */ s->ops_since_ckpt = 0; + s->last_ckpt_ms = now_ms(); /* arm the interval trigger */ if (phase == 4){ store__crash(s); return 0; } return 0; } @@ -2375,6 +2398,26 @@ static int import_snapshot(EngramPagedStore* s, const char* path){ return 0; } +/* Arm the background checkpointer with sensible defaults, overridable by env: + * ENGRAM_CKPT_OPS mutations since last checkpoint (default 100000) + * ENGRAM_CKPT_DIRTY dirty pool frames (default 0 = off) + * ENGRAM_CKPT_WAL_BYTES WAL bytes since reclaim (default 64 MiB) + * ENGRAM_CKPT_INTERVAL_MS wall-clock ms (default 0 = off) + * Any of these tripping on the write path triggers a checkpoint (→ WAL reclaimed). + * The M4 tests set tiny pools but never hit these bounds, so behaviour is unchanged. */ +static void engram__default_ckpt_policy(EngramPagedStore* s){ + s->ckpt_threshold = 100000; + s->ckpt_dirty_threshold = 0; + s->ckpt_wal_threshold = 64u*1024u*1024u; + s->ckpt_interval_ms = 0; + s->last_ckpt_ms = now_ms(); + const char* e; + if ((e=getenv("ENGRAM_CKPT_OPS")) && *e) s->ckpt_threshold = strtoull(e,NULL,10); + if ((e=getenv("ENGRAM_CKPT_DIRTY")) && *e) s->ckpt_dirty_threshold = (size_t)strtoull(e,NULL,10); + if ((e=getenv("ENGRAM_CKPT_WAL_BYTES")) && *e) s->ckpt_wal_threshold = strtoull(e,NULL,10); + if ((e=getenv("ENGRAM_CKPT_INTERVAL_MS")) && *e) s->ckpt_interval_ms = strtoll(e,NULL,10); +} + /* ── durable-engram boot / close ─────────────────────────────────────────────── */ EngramPagedStore* engram_open(const char* data_dir){ if (!data_dir) return NULL; @@ -2398,7 +2441,7 @@ EngramPagedStore* engram_open(const char* data_dir){ if (!s) return NULL; s->wal = wal_open(wal_path, sync); if (!s->wal){ store_close(s); return NULL; } - s->ckpt_threshold = 100000; + engram__default_ckpt_policy(s); wal_recover(s); /* replay post-checkpoint tail */ return s; } @@ -2407,7 +2450,7 @@ EngramPagedStore* engram_open(const char* data_dir){ if (!s) return NULL; s->wal = wal_open(wal_path, sync); if (!s->wal){ store_close(s); return NULL; } - s->ckpt_threshold = 100000; + engram__default_ckpt_policy(s); if (stat(snap_path, &st) == 0) import_snapshot(s, snap_path); engram_checkpoint(s); /* store is now authoritative */ return s; @@ -2417,3 +2460,175 @@ int engram_close(EngramPagedStore* s){ engram_checkpoint(s); return store_close(s); } + +/* ══════════════════════════════════════════════════════════════════════════════ + * M5 — ONLINE COMPACTION + background-checkpointer policy setter + * + * Dead space accrues in three shapes, all reclaimed here: + * 1. DEAD slots on NODE/EDGE pages — tombstones (prune/forget), superseded ids, + * and the stale prior versions a re-put / hebb-batch leaves (apply_*_put + * appends a new record + index entry; the old slot is marked DEAD). + * 2. Duplicate primary/adjacency index entries pointing at those DEAD records. + * 3. OVERFLOW chains orphaned when a large record died (tombstone only flips the + * slot; it never frees the record's overflow pages). + * + * Strategy — copy-live + atomic swap (the safest crash-safe relocation): + * A. Quiesce: checkpoint (or sync) so the on-disk .egm fully reflects state and + * the WAL is reduced to its CHECKPOINT{C} marker (C = current LSN watermark). + * B. Build a brand-new store file `.compact` holding ONLY the live records + * — walked canonically (latest-live per id) and re-placed bit-exact into + * fresh, densely packed pages with fresh id + adjacency B+-trees. Every page + * is stamped with LSN = C and the new superblock records last_checkpoint_lsn + * = C, so it is LSN-consistent with the (unchanged) WAL. fsync it. + * C. Commit by rename(.compact → ) — POSIX-atomic: recovery sees + * either the whole old file or the whole new file, never a torn mix. + * D. Reopen in place: swap the fd, INVALIDATE every pool frame (old page ids now + * hold different data — this is the M4 "remap relocated pages" step), reload + * the superblock, re-autopin. + * + * Crash safety (proven by the test at phases 0/1/2): + * • crash in A/B (before rename): old .egm is byte-for-byte intact and the WAL + * still matches it → recovery = PRE-compaction (all live records present). + * The half-built `.compact` temp is ignored by engram_open and unlinked at the + * next compaction. + * • crash after rename (C/D): the new .egm is fully fsync'd with last_checkpoint + * = C and the WAL (CHECKPOINT{C}, nothing newer) matches it → recovery = + * POST-compaction. No undo ever needed because relocation is copy-then-swap, + * never in-place mutation of a still-referenced page. + * + * Online vs quiesce: the store is single-threaded, so "online" means it is safe + * to interleave between mutations (each mutation is a synchronous call) — NOT that + * it runs concurrently with one. It takes a checkpoint quiesce point at entry. + * + * M4 cooperation: the build writes into a SEPARATE store `d` whose own pool obeys + * ENGRAM_POOL_FRAMES (so a small pool evicts/re-faults throughout the build, + * no-steal + pins honoured there); the live store's pool is fully invalidated on + * reopen, guaranteeing no stale frame maps a relocated page. + * ════════════════════════════════════════════════════════════════════════════ */ + +/* Drop every resident frame (relocated pages are no longer valid) but keep the + * pool object with its cap / prefetch / stats. Layer-pin bookkeeping is cleared + * (those page ids belong to the old image). */ +static void pc_invalidate_all(PgCache* c){ + if (!c) return; + for (size_t i=0;inbuckets;i++){ + PgEnt* e = c->buckets[i]; + while (e){ PgEnt* n=e->next; free(e->buf); free(e); e=n; } + c->buckets[i] = NULL; + } + for (size_t i=0;ilp_n;i++) free(c->lp[i].pages); + c->lp_n = 0; + c->count = 0; c->dirty_count = 0; c->mru = c->lru = NULL; +} + +/* Re-open the store file in place after an atomic swap: swap fd, invalidate the + * pool, reload the superblock, resume the LSN watermark, re-autopin. Keeps the + * attached WAL (it references the same checkpoint LSN the new file carries). */ +static int store__reopen_swapped(EngramPagedStore* s){ + if (s->fd >= 0) close(s->fd); + s->fd = open(s->path, O_RDWR); + if (s->fd < 0) return -1; + pc_invalidate_all(s->cache); /* M4: no stale frame for a relocated page */ + uint8_t b0[STORE_PAGE_SIZE], b1[STORE_PAGE_SIZE]; + uint64_t s0=0, s1=0; + int ok0 = sb_load_one(s, 0, b0, &s0) == 0; + int ok1 = sb_load_one(s, 1, b1, &s1) == 0; + if (!ok0 && !ok1) return -1; + const uint8_t* pick = (ok0&&ok1) ? ((s0>=s1)?b0:b1) : (ok0?b0:b1); + sb_apply(s, pick); + s->next_lsn = (s->last_checkpoint_lsn > s->sb_seq) ? s->last_checkpoint_lsn : s->sb_seq; + s->cur_node_page = 0; s->cur_edge_page = 0; + s->ops_since_ckpt = 0; s->last_ckpt_ms = now_ms(); + store__autopin(s); + return 0; +} + +/* Callback context for copying live records into the compacted store `d`. */ +typedef struct { EngramPagedStore* d; int err; } CompactCtx; +static void compact_node_cb(const StoreNode* n, void* ctx){ + CompactCtx* c = (CompactCtx*)ctx; + if (c->err) return; + if (node_place(c->d, n) != 0) c->err = 1; /* bit-exact re-place; stamped d->stamp_lsn */ +} +static void compact_edge_cb(const StoreEdge* e, void* ctx){ + CompactCtx* c = (CompactCtx*)ctx; + if (c->err) return; + if (edge_place(c->d, e) != 0) c->err = 1; +} + +/* Build the compacted image (only live records, fresh indexes) into a new file. */ +static int compact_build(EngramPagedStore* s, const char* tmp_path){ + unlink(tmp_path); /* drop any temp from a crashed run */ + EngramPagedStore* d = store_create(tmp_path); + if (!d) return -1; + uint64_t W = s->next_lsn; /* LSN watermark == checkpoint LSN */ + memcpy(d->uuid, s->uuid, 16); /* preserve store identity */ + d->last_checkpoint_lsn = W; + d->next_lsn = W; + d->stamp_lsn = W; /* every compacted page → LSN W */ + + int rc = 0; + /* live layers */ + StoreLayer* layers = NULL; size_t nlay = 0; + if (store_list_layers(s, &layers, &nlay) == 0){ + for (size_t i=0;istamp_lsn = 0; + if (rc == 0 && store_sync(d) != 0) rc = -1; /* flush + fsync + both superblocks */ + store_close(d); + return rc; +} + +int store__compact_crashat(EngramPagedStore* s, int phase){ + if (!s) return -1; + /* A. quiesce → on-disk store consistent, WAL reduced to its checkpoint marker */ + if (s->wal){ if (engram_checkpoint(s) != 0) return -1; } + else { if (store_sync(s) != 0) return -1; } + if (phase == 0){ store__crash(s); return 0; } /* → recovers pre-compaction */ + + char tmp[1200]; + snprintf(tmp, sizeof tmp, "%s.compact", s->path); + if (compact_build(s, tmp) != 0){ unlink(tmp); return -1; } + if (phase == 1){ store__crash(s); return 0; } /* built, not renamed → pre-compaction */ + + /* C. atomic commit */ + if (rename(tmp, s->path) != 0){ unlink(tmp); return -1; } + if (phase == 2){ store__crash(s); return 0; } /* renamed, not reopened → post-compaction */ + + /* D. reopen RAM state against the compacted file */ + return store__reopen_swapped(s); +} + +int store_compact(EngramPagedStore* s){ return store__compact_crashat(s, -1); } + +void store_set_checkpoint_policy(EngramPagedStore* s, uint64_t ops, + size_t dirty_pages, uint64_t wal_bytes, + long long interval_ms){ + if (!s) return; + s->ckpt_threshold = ops; + s->ckpt_dirty_threshold = dirty_pages; + s->ckpt_wal_threshold = wal_bytes; + s->ckpt_interval_ms = interval_ms; + s->last_ckpt_ms = now_ms(); +} + +uint64_t store_free_page_count(const EngramPagedStore* s){ + if (!s) return 0; + uint64_t n = 0, id = s->free_list_head; + uint8_t buf[STORE_PAGE_SIZE]; + while (id){ + if (page_read((EngramPagedStore*)s, id, buf) != 0) break; + if (buf[8] != STORE_PT_FREE) break; + n++; + id = get_u64(buf + OVF_NEXT_OFF); + } + return n; +} diff --git a/lang/runtime/engram_store.h b/lang/runtime/engram_store.h index c75592d..fa27b3d 100644 --- a/lang/runtime/engram_store.h +++ b/lang/runtime/engram_store.h @@ -253,4 +253,43 @@ void store__crash(EngramPagedStore* s); int store__flush_pages(EngramPagedStore* s); int store__checkpoint_crashat(EngramPagedStore* s, int phase); +/* ── M5: online compaction + background checkpointer (additive; format UNCHANGED) ── + * + * COMPACTION reclaims the space held by DEAD records — tombstoned nodes/edges + * (telemetry prune, forget), superseded ids, and the stale prior versions a + * re-put/hebb-batch leaves behind — plus the overflow pages they orphaned. It + * rewrites only the LIVE records (bit-exact) into a fresh, densely packed image + * with fresh id + adjacency indexes, then commits the swap atomically, so the + * .egm file physically SHRINKS and the freed pages are reclaimed. Crash-safe: + * a crash at any instant recovers to either the pre- or the post-compaction + * store, never a corrupt mix (atomic rename is the commit point). It cooperates + * with the M4 pool (no-steal, pins) by building into a separate store whose own + * pool honours ENGRAM_POOL_FRAMES, then INVALIDATING every frame of the live + * pool so no stale frame survives for a relocated page. + * + * Requires a quiesce point: store_compact performs a checkpoint (or sync) at + * entry, so it is called between mutations, not concurrently with one. */ +int store_compact(EngramPagedStore* s); + +/* Test hook: run compaction but stop (then power-loss) after `phase`: + * 0 = after the entry checkpoint, before building (→ recovers pre-compaction) + * 1 = after building+fsync the new image, before rename (→ pre-compaction) + * 2 = after the atomic rename, before reopening RAM state (→ post-compaction) + * phase<0 = full compaction. Frees `s` on a crash phase (like the checkpoint hook). */ +int store__compact_crashat(EngramPagedStore* s, int phase); + +/* BACKGROUND CHECKPOINTER policy. A checkpoint fires automatically on the write + * path when ANY armed trigger trips, reclaiming the WAL prefix without an explicit + * engram_checkpoint. 0 disables that trigger. Same checkpoint semantics as M2. + * ops — mutations since last checkpoint (default 100000) + * dirty_pages — dirty (un-checkpointed) pool frames + * wal_bytes — bytes appended to the WAL since it was last reclaimed + * interval_ms — wall-clock ms since the last checkpoint (checked on writes) */ +void store_set_checkpoint_policy(EngramPagedStore* s, uint64_t ops, + size_t dirty_pages, uint64_t wal_bytes, + long long interval_ms); + +/* Introspection: number of pages currently on the free-list. */ +uint64_t store_free_page_count(const EngramPagedStore* s); + #endif /* ENGRAM_STORE_H */