neuron-technologies/el
3
0
Fork 0
Code Issues 4 Pull Requests 4 Actions Packages Projects Releases 5 Wiki Activity

engram tiered storage M5: online compaction + background checkpointer (Phase 3, additive)

Browse Source 
Reclaims space held by dead records (tombstoned prune/forget nodes, superseded
ids, stale re-put/hebb versions, and their orphaned overflow chains). On-disk
format UNCHANGED — pure behavior.

COMPACTION (store_compact): copy-live + atomic-swap.
  A. checkpoint/sync to quiesce (WAL reduced to CHECKPOINT{C}); crash here => pre.
  B. build <path>.compact with only the live records, re-placed bit-exact into
     fresh densely-packed pages + fresh id/adjacency B+-trees, every page stamped
     LSN=C, new SB last_checkpoint_lsn=C; fsync. crash here => pre-compaction.
  C. rename(<path>.compact -> <path>) — POSIX-atomic commit; crash after => post.
  D. reopen in place: swap fd, INVALIDATE every pool frame (M4 remap of relocated
     pages), reload SB, re-autopin.
Crash at any instant recovers to pre- OR post-compaction, never a corrupt mix.
Single-threaded => "online" = safe between mutations; takes a checkpoint quiesce
at entry. M4 cooperation: temp build has its own pool honoring ENGRAM_POOL_FRAMES
(evict/re-fault + no-steal + pins); live pool fully invalidated on reopen.

BACKGROUND CHECKPOINTER: ckpt_maybe now fires on ANY armed trigger — ops (default
100000), dirty pool frames, WAL bytes-since-reclaim (default 64 MiB), or a
wall-clock interval (checked on the write path; no extra thread). Same M2
checkpoint semantics (calls engram_checkpoint). Env: ENGRAM_CKPT_OPS/_DIRTY/
_WAL_BYTES/_INTERVAL_MS; runtime setter store_set_checkpoint_policy().

Tests: engram/test/test_compaction.c (+runner). Plain gcc, ASan/UBSan clean.
  36 passed, 0 failed (O2 and ASan+UBSan builds).
  reclaim: page_count 655 -> 153, file 10731520 -> 2506752 bytes (76.6% reclaimed),
    every live record + adjacency bit-exact at new locations.
  crash-during-compaction phases 0/1/2: crc clean, live set intact, writable.
  background checkpointer: ops / WAL-bytes / dirty triggers each auto-fire; WAL
    prefix reclaimed (30 B after 600 puts); recovery correct.
  pool cooperation: compact under 24-frame pool correct, no stale frames.
No regression: M1 33, M2 36, M3 parity, M3.5, M4 bufpool 37 — all green.
On-disk format unchanged (additive).
This commit is contained in:
Will Anderson
2026-08-12 16:03:14 -05:00
parent 02dc12d785
commit ce0d33ba93
4 changed files with 703 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engram/test/run_compaction_tests.sh
Executable
+22
View File
@@ -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
engram/test/test_compaction.c
+421
View File
@@ -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 <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;
}
lang/runtime/engram_store.c
+221 -6
View File
@@ -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;i<c->nbuckets;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 `<path>.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(<path>.compact → <path>) — 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;i<c->nbuckets;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;i<c->lp_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;i<nlay && rc==0;i++)
if (apply_layer_put(d, &layers[i], W) != 0) rc = -1;
store_layers_free(layers, nlay);
}
/* live nodes + edges (canonical latest-live, dedup by id — see store_scan_*) */
CompactCtx ctx = { d, 0 };
if (rc == 0 && store_scan_nodes(s, compact_node_cb, &ctx) < 0) rc = -1;
if (rc == 0 && store_scan_edges(s, compact_edge_cb, &ctx) < 0) rc = -1;
if (ctx.err) rc = -1;
d->stamp_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;
}
lang/runtime/engram_store.h
+39
View File
@@ -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 */