From 8affb1d6e04c0b9cd59243e910ca65293c5ca0f9 Mon Sep 17 00:00:00 2001 From: Will Anderson Date: Tue, 11 Aug 2026 23:00:20 -0500 Subject: [PATCH] engram tiered storage M2: WAL + checkpoint + crash recovery + legacy import MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Write-back no-steal buffer pool makes the fsync'd WAL load-bearing (M1 was write-through). Logical WAL with record-granularity page-LSN redo idempotency. Checkpoint = flush dirty pages, fsync store, advance last_checkpoint_lsn, reclaim WAL prefix. One-time snapshot.json import only when store absent; JSON never read as the ongoing store thereafter. Gates: 33/33 M1 (no regression) + 36/36 M2 — replay parity, torn-tail fuzz (every byte offset), checkpoint-crash at all 5 phases, torn-page+WAL redo, legacy-import parity, hebb-survives-crash. --- engram/test/run_wal_store_tests.sh | 14 + engram/test/test_wal_store.c | 466 +++++++++++++ lang/runtime/engram_store.c | 1000 +++++++++++++++++++++++++++- lang/runtime/engram_store.h | 75 +++ 4 files changed, 1542 insertions(+), 13 deletions(-) create mode 100755 engram/test/run_wal_store_tests.sh create mode 100644 engram/test/test_wal_store.c diff --git a/engram/test/run_wal_store_tests.sh b/engram/test/run_wal_store_tests.sh new file mode 100755 index 0000000..968b211 --- /dev/null +++ b/engram/test/run_wal_store_tests.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# M2 WAL + checkpoint + recovery gate. Pure C (NOT elb/elc). Writes only under /tmp. +# Recovery tests use ENGRAM_WAL_SYNC=always so every WAL record is durable at crash. +set -e +HERE="$(cd "$(dirname "$0")" && pwd)" +SRC="$HERE/../../lang/runtime/engram_store.c" +BIN="/tmp/test_wal_store.$$" +echo "compiling: gcc test_wal_store.c engram_store.c" +gcc -O2 -Wall -Wextra -std=c11 "$HERE/test_wal_store.c" "$SRC" -o "$BIN" +ENGRAM_WAL_SYNC=always "$BIN" +rc=$? +rm -f "$BIN" +rm -rf /tmp/engram-wal-test-* +exit $rc diff --git a/engram/test/test_wal_store.c b/engram/test/test_wal_store.c new file mode 100644 index 0000000..faae6f4 --- /dev/null +++ b/engram/test/test_wal_store.c @@ -0,0 +1,466 @@ +/* test_wal_store.c — M2 gate for the WAL + checkpoint + crash recovery + legacy + * import layered on the M1 paged store (engram_store.{c,h}). + * + * Pure C. Build: gcc -O2 test_wal_store.c ../../lang/runtime/engram_store.c -o t + * Writes ONLY under a throwaway /tmp dir. Never touches ~/.neuron or live ports. + * + * Covers §7/M2 gates: + * 1 replay parity — random op stream: normal-durable path == crash-recover path + * 2 torn-tail fuzz — truncate engram.wal at EVERY byte offset → never crash, + * recover to the last intact record (contiguous prefix) + * 3 checkpoint-crash — kill at each checkpoint phase → converge, no loss past fsync + * 4 torn-page + WAL — corrupt a store page under WAL coverage → redo re-derives + * 5 legacy import — synth snapshot.json (emb+hebb, edges, layers) → import once, + * bit-exact readback; JSON never re-read as the store + * 6 hebb survives crash— hebb via WAL, crash before checkpoint → hebb recovered + */ +#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_base[512]; +static void mk_base(void){ + snprintf(g_base, sizeof g_base, "/tmp/engram-wal-test-%d", (int)getpid()); + mkdir(g_base, 0700); +} +static void mk_dir(const char* name, char* out, size_t cap){ + snprintf(out, cap, "%s/%s", g_base, name); + mkdir(out, 0700); +} + +/* deterministic RNG */ +static uint64_t xs(uint64_t* s){ uint64_t x=*s; x^=x<<13; x^=x>>7; x^=x<<17; *s=x; return x; } + +/* ── small node/edge generators (kept compact so WAL frames stay small) ─────── */ +static void gen_node(int i, int with_emb, StoreNode* n){ + memset(n, 0, sizeof *n); + uint64_t st = 0x1234ULL ^ ((uint64_t)(i+1)*0x9E3779B97F4A7C15ULL); + char id[32]; snprintf(id, sizeof id, "n%d", i); n->id = strdup(id); + char c[64]; snprintf(c, sizeof c, "content-of-node-%d-%llu", i, (unsigned long long)(xs(&st)%9999)); + n->content = strdup(c); + n->node_type = strdup("concept"); + n->tier = strdup("Working"); + n->salience = (double)(xs(&st)%100000)/7.0; + n->importance = (double)(xs(&st)%100000)/11.0; + n->confidence = (double)(xs(&st)%100000)/13.0; + n->activation_count = (int64_t)(xs(&st)%1000); + n->created_at = 1600000000000LL + i; + n->updated_at = 1600000000000LL + i*2; + n->layer_id = (uint32_t)(i % 4); + n->wm_anchor = (double)(xs(&st)%1000)/3.0; + if (with_emb){ + n->emb_dim = 32; + n->emb = (float*)malloc(sizeof(float)*n->emb_dim); + for (int k=0;kemb_dim;k++){ uint32_t u=(uint32_t)xs(&st); memcpy(&n->emb[k],&u,4); } + } +} +static void gen_edge(int i, const char* from, const char* to, StoreEdge* e){ + memset(e, 0, sizeof *e); + uint64_t st = 0xABCDULL ^ ((uint64_t)(i+1)*0xD1B54A32D192ED03ULL); + char id[32]; snprintf(id, sizeof id, "e%d", i); e->id = strdup(id); + e->from_id = strdup(from); e->to_id = strdup(to); + e->relation = strdup("relates_to"); + e->weight = (double)(xs(&st)%100000)/17.0; + e->hebb = (double)(xs(&st)%100000)/100000.0; + e->confidence = (double)(xs(&st)%100000)/19.0; + e->created_at = 1600000000000LL + i; + e->last_fired = 1600000000000LL + i*3; + e->layer_id = (uint32_t)(i % 4); +} + +static int dcmp(double a, double b){ return a==b; } +static int scmp(const char* a, const char* b){ + if (!a && !b) return 1; if (!a || !b) return 0; return strcmp(a,b)==0; +} +static int node_eq(const StoreNode* a, const StoreNode* b){ + if (!scmp(a->id,b->id) || !scmp(a->content,b->content) || !scmp(a->node_type,b->node_type) || + !scmp(a->tier,b->tier)) return 0; + if (!dcmp(a->salience,b->salience) || !dcmp(a->importance,b->importance) || + !dcmp(a->confidence,b->confidence) || a->activation_count!=b->activation_count || + a->created_at!=b->created_at || a->updated_at!=b->updated_at || + a->layer_id!=b->layer_id || !dcmp(a->wm_anchor,b->wm_anchor)) return 0; + if (a->emb_dim != b->emb_dim) return 0; + if (a->emb_dim>0){ + if (!a->emb || !b->emb) return 0; + if (memcmp(a->emb, b->emb, sizeof(float)*a->emb_dim)!=0) return 0; /* bit-exact */ + } + return 1; +} +static int edge_eq(const StoreEdge* a, const StoreEdge* b){ + return scmp(a->id,b->id) && scmp(a->from_id,b->from_id) && scmp(a->to_id,b->to_id) && + scmp(a->relation,b->relation) && dcmp(a->weight,b->weight) && dcmp(a->hebb,b->hebb) && + dcmp(a->confidence,b->confidence) && a->created_at==b->created_at && + a->last_fired==b->last_fired && a->layer_id==b->layer_id; +} + +/* whole-file read / write helpers (for torn-tail + torn-page fuzzing) */ +static uint8_t* read_file(const char* p, long* len){ + FILE* f=fopen(p,"rb"); if(!f) return NULL; + fseek(f,0,SEEK_END); long n=ftell(f); fseek(f,0,SEEK_SET); + uint8_t* b=malloc(n?n:1); if(fread(b,1,n,f)!=(size_t)n){ fclose(f); free(b); return NULL; } + fclose(f); *len=n; return b; +} +static void write_file(const char* p, const uint8_t* b, long len){ + FILE* f=fopen(p,"wb"); fwrite(b,1,len,f); fclose(f); +} + +/* ═══════════════════════════ TEST 1 — replay parity ═══════════════════════ */ +#define UNIV_NODES 60 +#define UNIV_EDGES 40 +static void test_replay_parity(void){ + printf("\n== replay parity: normal-durable path == crash-then-recover path ==\n"); + char da[600], db[600]; mk_dir("parityA", da, sizeof da); mk_dir("parityB", db, sizeof db); + EngramPagedStore* A = engram_open(da); + EngramPagedStore* B = engram_open(db); + ok("opened both stores", A && B); + if (!A || !B) return; + + uint64_t rng = 0xF00DFACEULL; + int OPS = 800; + for (int step=0; step0); + printf(" WAL bytes fuzzed=%ld full-recover offsets=%d\n", wlen, full_recovered); + free(sb); free(wb); +} + +/* ═══════════════════════════ TEST 3 — checkpoint-crash ═══════════════════════ */ +#define CK_NODES 30 +#define CK_EDGES 20 +static int build_and_crash_at_phase(const char* dir, int phase){ + EngramPagedStore* s = engram_open(dir); + if (!s) return -1; + for (int i=0;i=0); + if (victim>=0){ + for (int k=0;k<64;k++) sb[victim*16384 + 200 + k] ^= 0xA5; /* trash record area → bad crc */ + write_file(sp, sb, slen); + } + free(sb); + + EngramPagedStore* r = engram_open(dir); /* heal torn page + replay WAL */ + ok("reopened after page corruption", r!=NULL); + if (r){ + int miss=0; + for (int i=0;iemb_dim;k++) n->emb[k] = (float)((double)(xs(&es)%2000001)/1000000.0 - 1.0); } + fprintf(f, "%s{\"id\":\"%s\",\"content\":\"%s\",\"node_type\":\"%s\",\"tier\":\"%s\"," + "\"salience\":%.17g,\"importance\":%.17g,\"confidence\":%.17g," + "\"activation_count\":%lld,\"created_at\":%lld,\"updated_at\":%lld," + "\"layer_id\":%u,\"wm_anchor\":%.17g,\"emb\":\"", + i?",":"", n->id, n->content, n->node_type, n->tier, + n->salience, n->importance, n->confidence, + (long long)n->activation_count, (long long)n->created_at, (long long)n->updated_at, + n->layer_id, n->wm_anchor); + for (int k=0;kemb_dim;k++) fprintf(f, "%s%.9g", k?",":"", (double)n->emb[k]); /* exact float32 repr */ + fprintf(f, "\"}"); + } + fprintf(f, "],\"edges\":["); + for (int i=0;iid, e->from_id, e->to_id, e->relation, + e->weight, e->hebb, e->confidence, (long long)e->created_at, (long long)e->last_fired, e->layer_id); + } + fprintf(f, "],\"layers\":["); + fprintf(f, "{\"layer_id\":0,\"name\":\"SAFETY\",\"activation_priority\":9,\"suppressible\":0,\"transparent\":0,\"injectable\":0},"); + fprintf(f, "{\"layer_id\":1,\"name\":\"CORE_IDENTITY\",\"activation_priority\":8,\"suppressible\":0,\"transparent\":1,\"injectable\":1}"); + fprintf(f, "]}"); + fclose(f); + + EngramPagedStore* s = engram_open(dir); /* store absent + snapshot present → import */ + ok("engram_open imported the snapshot", s!=NULL); + char sp[700]; snprintf(sp,sizeof sp,"%s/engram.store",dir); struct stat st; + ok("engram.store created by import", stat(sp,&st)==0); + if (!s) return; + + int nmiss=0, embmiss=0; + for (int i=0;i0 && memcmp(got.emb,onodes[i].emb,sizeof(float)*got.emb_dim)!=0)) embmiss++; store_node_free(&got); } + } + int emiss=0, hebbmiss=0; + for (int i=0;iloc B+-tree root */ uint64_t adj_index_page; /* from/to adjacency B+-tree root */ uint64_t layer_registry_page; - uint64_t last_checkpoint_lsn; /* 0 in M1 (no checkpoints yet) */ + uint64_t last_checkpoint_lsn; /* store on disk reflects all ops with lsn<=this */ uint64_t sb_seq; uint8_t uuid[16]; - uint64_t next_lsn; + uint64_t next_lsn; /* monotonic LSN counter (WAL + page stamp share it) */ uint64_t cur_node_page; /* last NODE page with room, 0 = none */ uint64_t cur_edge_page; /* last EDGE page with room, 0 = none */ int leaf_max; /* test hook; 0 = natural */ int int_max; + /* ── M2 additions ─────────────────────────────────────────────────────── */ + PgCache* cache; /* write-back buffer pool (no-steal) */ + EngramWal* wal; /* attached WAL, or NULL (M1 direct mode) */ + uint64_t stamp_lsn; /* LSN to stamp on the next page_write (0 = auto) */ + 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) */ }; +/* M2 buffer-pool hooks (defined in the M2 section at the bottom of this file). */ +typedef struct PgEnt { uint64_t id; uint8_t* buf; uint64_t lsn; int dirty; struct PgEnt* next; } PgEnt; +static PgCache* pc_new(void); +static void pc_free(PgCache* c); +static PgEnt* pc_get(EngramPagedStore* s, uint64_t id); +static int pc_put(EngramPagedStore* s, uint64_t id, const uint8_t* buf, int dirty); +static int pc_flush(EngramPagedStore* s); /* pwrite all dirty → clean */ + /* ── little-endian scalar codecs ──────────────────────────────────────────── */ static void put_u16(uint8_t* p, uint16_t v){ p[0]=(uint8_t)v; p[1]=(uint8_t)(v>>8); } static void put_u32(uint8_t* p, uint32_t v){ for(int i=0;i<4;i++) p[i]=(uint8_t)(v>>(8*i)); } @@ -172,10 +191,18 @@ static uint64_t id_hash(const char* s){ } /* ── raw page I/O ─────────────────────────────────────────────────────────── */ +/* Read a page: served from the write-back cache if resident, else from disk (and + * cached clean). This is the ONLY page-read path, so a dirty (not-yet-flushed) + * page is always observed with its in-RAM contents. */ static int page_read(EngramPagedStore* s, uint64_t id, uint8_t* buf){ + if (s->cache){ + PgEnt* e = pc_get(s, id); + if (e){ memcpy(buf, e->buf, STORE_PAGE_SIZE); return 0; } + } off_t off = (off_t)id * STORE_PAGE_SIZE; ssize_t r = pread(s->fd, buf, STORE_PAGE_SIZE, off); if (r != (ssize_t)STORE_PAGE_SIZE) return -1; + if (s->cache) pc_put(s, id, buf, 0); /* cache clean */ return 0; } static int page_write_raw(EngramPagedStore* s, uint64_t id, const uint8_t* buf){ @@ -184,14 +211,19 @@ static int page_write_raw(EngramPagedStore* s, uint64_t id, const uint8_t* buf){ if (w != (ssize_t)STORE_PAGE_SIZE) return -1; return 0; } -/* Stamp lsn + crc into a generic page header, then write. */ +/* Stamp page_id + lsn + crc into the generic header, then write BACK to the cache + * (dirty). No-steal: the page reaches disk only at the next checkpoint. The stamped + * LSN is the current op's WAL LSN (s->stamp_lsn) when set, else a fresh counter + * value — this is the page LSN used for ARIES-style redo idempotency. */ static int page_write(EngramPagedStore* s, uint64_t id, uint8_t* buf){ put_u64(buf + 0, id); - put_u64(buf + 16, ++s->next_lsn); + uint64_t lsn = s->stamp_lsn ? s->stamp_lsn : ++s->next_lsn; + put_u64(buf + 16, lsn); put_u32(buf + 24, 0); uint32_t crc = crc32_buf(buf, STORE_PAGE_SIZE); put_u32(buf + 24, crc); - return page_write_raw(s, id, buf); + if (s->cache) return pc_put(s, id, buf, 1); + return page_write_raw(s, id, buf); /* pre-cache bootstrap (should not happen) */ } static int page_crc_ok(const uint8_t* buf){ uint8_t tmp[STORE_PAGE_SIZE]; @@ -741,6 +773,10 @@ static void sb_apply(EngramPagedStore* s, const uint8_t* buf){ int store_sync(EngramPagedStore* s){ if (!s) return -1; + /* flush the write-back cache so the store file reflects the SB we are about + * to stamp (this is the checkpoint page-flush + fsync). */ + if (pc_flush(s) != 0) return -1; + if (fsync(s->fd) != 0) return -1; s->sb_seq++; /* write primary, fsync, then mirror, fsync — so a torn write of one leaves * the other valid; recovery prefers the higher valid sb_seq. */ @@ -778,6 +814,8 @@ EngramPagedStore* store_create(const char* path){ if (!s) return NULL; s->fd = open(path, O_RDWR | O_CREAT | O_EXCL, 0600); if (s->fd < 0){ free(s); return NULL; } + s->cache = pc_new(); + if (!s->cache){ close(s->fd); free(s); return NULL; } snprintf(s->path, sizeof s->path, "%s", path); s->format_version = STORE_FORMAT_VERSION; s->page_size = STORE_PAGE_SIZE; @@ -805,6 +843,8 @@ EngramPagedStore* store_open(const char* path){ if (!s) return NULL; s->fd = open(path, O_RDWR); if (s->fd < 0){ free(s); return NULL; } + s->cache = pc_new(); + if (!s->cache){ close(s->fd); free(s); return NULL; } snprintf(s->path, sizeof s->path, "%s", path); uint8_t b0[STORE_PAGE_SIZE], b1[STORE_PAGE_SIZE]; uint64_t s0 = 0, s1 = 0; @@ -816,15 +856,22 @@ EngramPagedStore* store_open(const char* path){ else pick = ok0 ? b0 : b1; sb_apply(s, pick); if (s->page_size != STORE_PAGE_SIZE){ close(s->fd); free(s); return NULL; } - s->next_lsn = s->sb_seq; /* resume LSNs above the last synced value */ + /* resume the LSN counter above both the last checkpoint and the SB seq (M1 + * stores have last_checkpoint_lsn==0 → falls back to sb_seq as before). */ + 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; return s; } +/* wal_close is defined in the M2 section. */ +static void wal_close(EngramWal* w); + int store_close(EngramPagedStore* s){ if (!s) return -1; int rc = store_sync(s); + if (s->wal){ wal_close(s->wal); s->wal = NULL; } + if (s->cache){ pc_free(s->cache); s->cache = NULL; } if (s->fd >= 0) close(s->fd); free(s); return rc; @@ -880,7 +927,9 @@ static int place_record(EngramPagedStore* s, uint8_t ptype, const uint8_t* rec, return 0; } -int store_put_node(EngramPagedStore* s, const StoreNode* n){ +/* Pure record placement + indexing (no LSN/WAL/idempotency). Pages are stamped + * with s->stamp_lsn by page_write. Used by both the M1-direct and M2-WAL paths. */ +static int node_place(EngramPagedStore* s, const StoreNode* n){ if (!s || !n || !n->id) return -1; size_t blen; uint8_t* body = node_serialize(n, &blen); if (!body) return -1; @@ -895,7 +944,7 @@ int store_put_node(EngramPagedStore* s, const StoreNode* n){ return btree_put(s, TREE_PRIMARY, id_hash(n->id), payload); } -int store_put_edge(EngramPagedStore* s, const StoreEdge* e){ +static int edge_place(EngramPagedStore* s, const StoreEdge* e){ if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1; size_t blen; uint8_t* body = edge_serialize(e, &blen); if (!body) return -1; @@ -967,16 +1016,21 @@ int store_get_node(EngramPagedStore* s, const char* id, StoreNode* out){ return found ? 1 : 0; } -int store_tombstone(EngramPagedStore* s, const char* id){ +/* Tombstone the live record(s) for `id`, but only where the record's home page + * LSN < `lsn` (ARIES redo idempotency: a replayed tombstone whose effect is + * already on the page is skipped). Stamps killed pages with `lsn`. */ +static int tombstone_core(EngramPagedStore* s, const char* id, uint64_t lsn){ if (!s || !id) return -1; uint8_t* locs; size_t n; if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n)!=0) return -1; - int hit = 0; + uint64_t prev_stamp = s->stamp_lsn; + s->stamp_lsn = lsn; for (size_t i=0;i= lsn) continue; /* already covered by >= this LSN */ if (slot >= slp_count(buf)) continue; uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl); if (fl != SLOT_LIVE) continue; @@ -988,8 +1042,7 @@ int store_tombstone(EngramPagedStore* s, const char* id){ store_node_free(&cand); if (!match) continue; slp_set_slot(buf, slot, off, len, SLOT_DEAD); - if (page_write(s, page, buf)!=0){ free(locs); return -1; } - hit = 1; + if (page_write(s, page, buf)!=0){ s->stamp_lsn = prev_stamp; free(locs); return -1; } /* if the page is now empty, reclaim it to the free list */ if (page_read(s, page, buf)==0 && slp_live_count(buf)==0){ if (s->cur_node_page == page) s->cur_node_page = 0; @@ -997,8 +1050,9 @@ int store_tombstone(EngramPagedStore* s, const char* id){ page_free(s, page); } } + s->stamp_lsn = prev_stamp; free(locs); - return hit ? 0 : 0; /* absent id is a no-op success */ + return 0; /* absent id is a no-op success */ } static int get_edges_dir(EngramPagedStore* s, const char* id, uint8_t want_dir, @@ -1082,3 +1136,923 @@ void store__set_btree_order(EngramPagedStore* s, int leaf_max, int internal_max) s->leaf_max = leaf_max; s->int_max = internal_max; } uint64_t store_page_count(const EngramPagedStore* s){ return s ? s->page_count : 0; } + +/* ══════════════════════════════════════════════════════════════════════════════ + * M2 — WAL + write-back buffer pool + checkpoint + crash recovery + legacy import + * + * Durability model (design §2.2/§4, ARIES-lite): + * • Buffer pool is WRITE-BACK, no-steal: a mutation dirties a page in RAM; the + * page reaches engram.store ONLY at a checkpoint. So after a crash the store + * file reflects exactly `last_checkpoint_lsn`, and everything since lives in + * the WAL. This is what makes the WAL load-bearing (durability = fsync'd WAL, + * not the page). + * • Each mutation: assign LSN → append WAL record (fsync per policy) → apply to + * the page(s), stamping page-LSN = record LSN. + * • Recovery: open store (at checkpoint), heal torn data pages, replay WAL + * forward; a record is REDONE only where the target record's home-page LSN is + * < the record LSN (idempotent — safe to replay any number of times, and safe + * across torn checkpoints where some dirty pages reached disk). + * + * DELTA vs the design sketch (flagged, becoming permanent): the WAL is LOGICAL + * (op + node/edge/layer payload), not physical page images, so the "redo only if + * rec.lsn > page.lsn" test is applied at RECORD granularity — against the LSN of + * the page currently holding the record for that id — rather than against a + * single physical target page. Records are never relocated once placed, so a + * record placed by op L keeps its page (whose LSN only rises), making + * "page-LSN ≥ L ⇒ op L already durable" a sound, precise idempotency test. + * ════════════════════════════════════════════════════════════════════════════ */ + +#include + +/* ── write-back buffer pool ────────────────────────────────────────────────── */ +struct PgCache { PgEnt** buckets; size_t nbuckets; size_t count; }; + +static PgCache* pc_new(void){ + PgCache* c = (PgCache*)calloc(1, sizeof *c); + if (!c) return NULL; + c->nbuckets = 1024; + c->buckets = (PgEnt**)calloc(c->nbuckets, sizeof(PgEnt*)); + if (!c->buckets){ free(c); return NULL; } + return c; +} +static void pc_free(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; } + } + free(c->buckets); free(c); +} +static PgEnt* pc_get(EngramPagedStore* s, uint64_t id){ + PgCache* c = s->cache; + PgEnt* e = c->buckets[id % c->nbuckets]; + while (e){ if (e->id==id) return e; e=e->next; } + return NULL; +} +static void pc_maybe_grow(PgCache* c){ + if (c->count <= c->nbuckets*4) return; + size_t nn = c->nbuckets*2; + PgEnt** nb = (PgEnt**)calloc(nn, sizeof(PgEnt*)); + if (!nb) return; + for (size_t i=0;inbuckets;i++){ + PgEnt* e = c->buckets[i]; + while (e){ PgEnt* nx=e->next; size_t b=e->id%nn; e->next=nb[b]; nb[b]=e; e=nx; } + } + free(c->buckets); c->buckets=nb; c->nbuckets=nn; +} +static int pc_put(EngramPagedStore* s, uint64_t id, const uint8_t* buf, int dirty){ + PgCache* c = s->cache; + PgEnt* e = pc_get(s, id); + if (!e){ + e = (PgEnt*)calloc(1, sizeof *e); + if (!e) return -1; + e->buf = (uint8_t*)malloc(STORE_PAGE_SIZE); + if (!e->buf){ free(e); return -1; } + e->id = id; + size_t b = id % c->nbuckets; + e->next = c->buckets[b]; c->buckets[b] = e; c->count++; + pc_maybe_grow(c); + } + memcpy(e->buf, buf, STORE_PAGE_SIZE); + e->lsn = get_u64(buf + 16); + if (dirty) e->dirty = 1; + return 0; +} +static int pc_flush(EngramPagedStore* s){ + if (!s->cache) return 0; + PgCache* c = s->cache; + 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; } + return 0; +} +/* ── WAL log ───────────────────────────────────────────────────────────────── */ +enum { OP_NODE_PUT=1, OP_EDGE_PUT, OP_TOMBSTONE, OP_SUPERSEDE, + OP_LAYER_PUT, OP_LAYER_DEL, OP_FORGET, OP_HEBB_BATCH, OP_CHECKPOINT }; + +#define WAL_MAGIC 0x314C5745u /* 'E''W''L''1' little-endian */ +#define WAL_HDR 22u /* magic4 len4 op1 flags1 lsn8 crc4 */ + +struct EngramWal { + int fd; + char path[1024]; + EngramWalSync sync; + uint64_t last_fsync_lsn; + long long last_fsync_ms; + uint64_t appended_since_fsync; +}; + +static long long now_ms(void){ + struct timeval tv; gettimeofday(&tv, NULL); + return (long long)tv.tv_sec*1000 + tv.tv_usec/1000; +} +static EngramWal* wal_open(const char* path, EngramWalSync sync){ + EngramWal* w = (EngramWal*)calloc(1, sizeof *w); + if (!w) return NULL; + w->fd = open(path, O_RDWR | O_CREAT | O_APPEND, 0600); + if (w->fd < 0){ free(w); return NULL; } + snprintf(w->path, sizeof w->path, "%s", path); + w->sync = sync; w->last_fsync_ms = now_ms(); + return w; +} +static void wal_close(EngramWal* w){ + if (!w) return; + if (w->fd >= 0){ fsync(w->fd); close(w->fd); } + free(w); +} +static void wal_maybe_fsync(EngramPagedStore* s, uint64_t lsn){ + EngramWal* w = s->wal; if (!w) return; + if (w->sync == ENGRAM_WAL_OFF) return; + if (w->sync == ENGRAM_WAL_ALWAYS){ + fsync(w->fd); w->last_fsync_lsn=lsn; w->last_fsync_ms=now_ms(); w->appended_since_fsync=0; return; + } + /* GROUP: fsync at most every ~50 ms or every 256 records. */ + long long now = now_ms(); + if (now - w->last_fsync_ms >= 50 || w->appended_since_fsync >= 256){ + fsync(w->fd); w->last_fsync_lsn=lsn; w->last_fsync_ms=now; w->appended_since_fsync=0; + } +} +static int wal_append(EngramPagedStore* s, uint8_t op, const uint8_t* payload, + uint32_t plen, uint64_t lsn){ + EngramWal* w = s->wal; if (!w) return 0; + size_t fl = WAL_HDR + plen; + uint8_t* fr = (uint8_t*)malloc(fl); + if (!fr) return -1; + put_u32(fr + 0, WAL_MAGIC); + put_u32(fr + 4, plen); + fr[8] = op; fr[9] = 0; + put_u64(fr + 10, lsn); + put_u32(fr + 18, 0); + if (plen) memcpy(fr + WAL_HDR, payload, plen); + /* crc over op|flags|lsn|payload — these are NOT contiguous in the frame (the + * crc field sits between), so hash a scratch copy that omits it. */ + uint8_t* cb = (uint8_t*)malloc(10 + plen); + if (!cb){ free(fr); return -1; } + memcpy(cb, fr + 8, 10); + if (plen) memcpy(cb + 10, payload, plen); + uint32_t crc = crc32_buf(cb, 10 + plen); + free(cb); + put_u32(fr + 18, crc); + ssize_t wr = write(w->fd, fr, fl); + free(fr); + if (wr != (ssize_t)fl) return -1; + w->appended_since_fsync++; + 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 */ + 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(); + return 0; +} + +/* ── idempotency: max home-page LSN of the live/dead record(s) for `id` ──────── */ +static uint64_t max_page_lsn_for_id(EngramPagedStore* s, const char* id, int want_edge){ + uint8_t* locs; size_t n; + if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return 0; + uint64_t best = 0; + for (size_t i=0;i best) best = l; + } + free(locs); + return best; +} + +/* Mark live record(s) for `id` DEAD where the home page LSN < lsn. `want_edge` + * disambiguates node vs edge records (page type). Caller sets s->stamp_lsn. */ +static void kill_live_id(EngramPagedStore* s, const char* id, int want_edge, uint64_t lsn){ + uint8_t* locs; size_t n; + if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return; + for (size_t i=0;i= lsn) continue; + if (buf[8] != (want_edge ? STORE_PT_EDGE : STORE_PT_NODE)) continue; + if (slot >= slp_count(buf)) continue; + uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl); + if (fl != SLOT_LIVE) continue; + uint8_t* body; size_t blen; int live; + if (read_body(s, page, slot, &body, &blen, &live) != 0) continue; + int match; + if (want_edge){ StoreEdge e; edge_parse(body,blen,&e); match = e.id && !strcmp(e.id,id); store_edge_free(&e); } + else { StoreNode nn; node_parse(body,blen,&nn); match = nn.id && !strcmp(nn.id,id); store_node_free(&nn); } + free(body); + if (!match) continue; + slp_set_slot(buf, slot, off, len, SLOT_DEAD); + page_write(s, page, buf); + } + free(locs); +} + +/* ── apply routines (used by both the normal write path and WAL replay) ──────── */ +static int apply_node_put(EngramPagedStore* s, const StoreNode* n, uint64_t lsn){ + if (!s || !n || !n->id) return -1; + if (max_page_lsn_for_id(s, n->id, 0) >= lsn) return 0; /* already durable */ + uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; + int r = node_place(s, n); /* re-put appends; reads dedup */ + s->stamp_lsn = prev; + return r; +} +static int apply_edge_put(EngramPagedStore* s, const StoreEdge* e, uint64_t lsn){ + if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1; + if (max_page_lsn_for_id(s, e->id, 1) >= lsn) return 0; /* already durable */ + uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; + kill_live_id(s, e->id, 1, lsn); /* supersede prior versions (no adjacency dup) */ + int r = edge_place(s, e); + s->stamp_lsn = prev; + return r; +} + +/* ── layer registry (single slotted page rooted at layer_registry_page) ──────── */ +enum { LT_ID=1, LT_NAME, LT_PRIO, LT_SUPPRESS, LT_TRANSPARENT, LT_INJECTABLE }; + +static uint8_t* layer_serialize(const StoreLayer* L, size_t* out_len){ + Buf b = {0,0,0}; + tlv_u32(&b, LT_ID, L->layer_id); + tlv_str(&b, LT_NAME, L->name); + tlv_u32(&b, LT_PRIO, L->activation_priority); + tlv_i32(&b, LT_SUPPRESS, L->suppressible); + tlv_i32(&b, LT_TRANSPARENT, L->transparent); + tlv_i32(&b, LT_INJECTABLE, L->injectable); + if (L->unknown && L->unknown_len) buf_raw(&b, L->unknown, L->unknown_len); + *out_len = b.len; return b.p; +} +static void layer_parse(const uint8_t* body, size_t len, StoreLayer* L){ + memset(L, 0, sizeof *L); + Buf unk = {0,0,0}; + size_t i = 0; + while (i + 5 <= len){ + uint8_t tag = body[i]; uint32_t flen = get_u32(body+i+1); + if (i + 5 + flen > len) break; + const uint8_t* v = body + i + 5; + switch (tag){ + case LT_ID: L->layer_id = get_u32(v); break; + case LT_NAME: L->name = dup_str(v, flen); break; + case LT_PRIO: L->activation_priority = get_u32(v); break; + case LT_SUPPRESS: L->suppressible = (int32_t)get_u32(v); break; + case LT_TRANSPARENT: L->transparent = (int32_t)get_u32(v); break; + case LT_INJECTABLE: L->injectable = (int32_t)get_u32(v); break; + default: buf_raw(&unk, body+i, 5+flen); break; + } + i += 5 + flen; + } + L->unknown = unk.p; L->unknown_len = unk.len; +} +void store_layer_free(StoreLayer* L){ + if (!L) return; free(L->name); free(L->unknown); memset(L, 0, sizeof *L); +} +void store_layers_free(StoreLayer* arr, size_t n){ + if (!arr) return; for (size_t i=0;ilayer_registry_page; + uint8_t buf[STORE_PAGE_SIZE]; + if (page_read(s, pg, buf) != 0) return -1; + /* already-applied? a live record with this layer_id and page LSN >= lsn. */ + if (get_u64(buf + 16) >= lsn){ + int nslot = slp_count(buf); + for (int i=0;ilayer_id); store_layer_free(&c); + if (hit) return 0; + } + } + uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; + /* supersede prior live record(s) with this layer_id */ + int nslot = slp_count(buf); + for (int i=0;ilayer_id); store_layer_free(&c); + if (hit) slp_set_slot(buf, i, off, len, SLOT_DEAD); + } + /* append new layer record */ + size_t blen; uint8_t* body = layer_serialize(L, &blen); + if (!body){ s->stamp_lsn=prev; return -1; } + uint16_t reclen = (uint16_t)(REC_HDR + blen); + uint8_t* rec = (uint8_t*)malloc(reclen); + if (!rec){ free(body); s->stamp_lsn=prev; return -1; } + put_u16(rec, reclen); rec[2]=REC_VER; rec[3]=0; + memcpy(rec + REC_HDR, body, blen); free(body); + int slot = slp_put(buf, rec, reclen); free(rec); + if (slot < 0){ s->stamp_lsn=prev; return -1; } /* registry page full (M2: single page) */ + int rc = page_write(s, pg, buf); + s->stamp_lsn = prev; + return rc; +} +static int apply_layer_del(EngramPagedStore* s, uint32_t layer_id, uint64_t lsn){ + uint64_t pg = s->layer_registry_page; + uint8_t buf[STORE_PAGE_SIZE]; + if (page_read(s, pg, buf) != 0) return -1; + if (get_u64(buf + 16) >= lsn) return 0; + uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; + int nslot = slp_count(buf), changed=0; + for (int i=0;istamp_lsn = prev; + return rc; +} +int store_get_layer(EngramPagedStore* s, uint32_t layer_id, StoreLayer* out){ + if (!s || !out) return -1; + uint8_t buf[STORE_PAGE_SIZE]; + if (page_read(s, s->layer_registry_page, buf) != 0) return -1; + int nslot = slp_count(buf), found=0; + for (int i=0;ilayer_registry_page, buf) != 0) return -1; + StoreLayer* arr=NULL; size_t used=0, cap=0; + int nslot = slp_count(buf); + for (int i=0;itombstoned = 0; found = 1; + } else store_edge_free(&cand); + } + free(locs); + return found ? 1 : 0; +} + +/* ── hebb batch apply ──────────────────────────────────────────────────────── + * payload: [u32 count] then count × [u32 id_len][id][f64 hebb][i64 last_fired]. */ +static int apply_hebb_batch(EngramPagedStore* s, const uint8_t* p, uint32_t plen, uint64_t lsn){ + if (plen < 4) return -1; + uint32_t cnt = get_u32(p); size_t off = 4; + for (uint32_t i=0;i plen) break; + uint32_t idl = get_u32(p+off); off += 4; + if (off + idl + 16 > plen) break; + char* id = dup_str(p+off, idl); off += idl; + double hebb; { uint64_t u=get_u64(p+off); memcpy(&hebb,&u,8);} off += 8; + int64_t lf = (int64_t)get_u64(p+off); off += 8; + if (id){ + StoreEdge e; + if (store_get_edge(s, id, &e) == 1){ + e.hebb = hebb; e.last_fired = lf; + apply_edge_put(s, &e, lsn); + store_edge_free(&e); + } + free(id); + } + } + return 0; +} + +/* ── WAL replay dispatch ─────────────────────────────────────────────────────── */ +static void wal_dispatch(EngramPagedStore* s, uint8_t op, const uint8_t* p, uint32_t plen, uint64_t lsn){ + switch (op){ + case OP_NODE_PUT: { StoreNode n; node_parse(p,plen,&n); apply_node_put(s,&n,lsn); store_node_free(&n); break; } + case OP_EDGE_PUT: { StoreEdge e; edge_parse(p,plen,&e); apply_edge_put(s,&e,lsn); store_edge_free(&e); break; } + case OP_TOMBSTONE: + case OP_FORGET: { char* id=dup_str(p,plen); if(id){ tombstone_core(s,id,lsn); free(id);} break; } + case OP_SUPERSEDE:{ if(plen>=4){ uint32_t ol=get_u32(p); if(4+ol<=plen){ char* old=dup_str(p+4,ol); if(old){ tombstone_core(s,old,lsn); free(old);} } } break; } + case OP_LAYER_PUT:{ StoreLayer L; layer_parse(p,plen,&L); apply_layer_put(s,&L,lsn); store_layer_free(&L); break; } + case OP_LAYER_DEL:{ if(plen>=4) apply_layer_del(s, get_u32(p), lsn); break; } + case OP_HEBB_BATCH: apply_hebb_batch(s, p, plen, lsn); break; + case OP_CHECKPOINT: break; + default: break; + } +} +static int wal_replay(EngramPagedStore* s){ + EngramWal* w = s->wal; if (!w) return 0; + struct stat st; if (fstat(w->fd, &st) != 0) return -1; + off_t size = st.st_size, off = 0; + uint64_t maxlsn = 0; + for (;;){ + if (off + (off_t)WAL_HDR > size) break; /* torn/short tail */ + uint8_t hdr[WAL_HDR]; + if (pread(w->fd, hdr, WAL_HDR, off) != (ssize_t)WAL_HDR) break; + if (get_u32(hdr) != WAL_MAGIC) break; + uint32_t plen = get_u32(hdr + 4); + if (off + (off_t)WAL_HDR + plen > size) break; /* torn tail */ + uint32_t crc_stored = get_u32(hdr + 18); + uint8_t* fr = (uint8_t*)malloc(10 + plen); + if (!fr) return -1; + memcpy(fr, hdr + 8, 10); /* op|flags|lsn */ + if (plen && pread(w->fd, fr+10, plen, off+WAL_HDR) != (ssize_t)plen){ free(fr); break; } + if (crc32_buf(fr, 10+plen) != crc_stored){ free(fr); break; } /* bad crc → stop */ + uint8_t op = fr[0]; uint64_t lsn = get_u64(fr+2); + if (lsn > maxlsn) maxlsn = lsn; + if (lsn > s->last_checkpoint_lsn) wal_dispatch(s, op, fr+10, plen, lsn); + free(fr); + off += (off_t)WAL_HDR + plen; + } + if (maxlsn > s->next_lsn) s->next_lsn = maxlsn; + return 0; +} + +/* Heal torn DATA pages (bad crc) to empty pages of their type, so redo re-derives + * their records from the WAL. Index/superblock recovery uses the SB mirror + full + * replay-from-checkpoint; torn index pages are left for store_check to report. */ +static void heal_torn_pages(EngramPagedStore* s){ + /* Scan by the ACTUAL file size, not the SB page_count: a steal (dirty pages + * flushed without advancing the checkpoint SB) can leave valid/torn pages + * beyond the recorded page_count. Bump page_count so redo allocations never + * collide with those on-disk ghost pages. */ + struct stat fst; + uint64_t file_pages = (fstat(s->fd,&fst)==0) ? (uint64_t)(fst.st_size/STORE_PAGE_SIZE) : s->page_count; + if (file_pages > s->page_count) s->page_count = file_pages; + for (uint64_t id=2; idfd, buf, STORE_PAGE_SIZE, off) != (ssize_t)STORE_PAGE_SIZE) continue; + if (page_crc_ok(buf)) continue; + uint8_t t = buf[8]; + uint8_t nb[STORE_PAGE_SIZE]; memset(nb, 0, sizeof nb); + if (t==STORE_PT_NODE || t==STORE_PT_EDGE){ + nb[8]=t; put_u16(nb+10,0); put_u16(nb+12,(uint16_t)(STORE_PAGE_SIZE-STORE_HDR)); + } else if (t==STORE_PT_OVERFLOW){ + nb[8]=t; put_u64(nb+OVF_NEXT_OFF,0); put_u32(nb+OVF_LEN_OFF,0); + } else continue; /* don't heal index/free/garbage */ + put_u64(nb+0, id); put_u64(nb+16, 0); put_u32(nb+24, 0); + put_u32(nb+24, crc32_buf(nb, STORE_PAGE_SIZE)); + pc_put(s, id, nb, 1); /* dirty → persisted next checkpoint */ + } +} +static int wal_recover(EngramPagedStore* s){ + s->recovering = 1; + heal_torn_pages(s); + int rc = wal_replay(s); + s->recovering = 0; + return rc; +} + +/* ── checkpoint threshold trigger ────────────────────────────────────────────── */ +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); +} + +/* ── public mutation entry points (log-then-apply when a WAL is attached) ─────── */ +int store_put_node(EngramPagedStore* s, const StoreNode* n){ + if (!s || !n || !n->id) return -1; + uint64_t L = ++s->next_lsn; + if (s->wal){ + size_t blen; uint8_t* body = node_serialize(n, &blen); + if (!body) return -1; + int wr = wal_append(s, OP_NODE_PUT, body, (uint32_t)blen, L); + free(body); + if (wr != 0) return -1; + } + int r = apply_node_put(s, n, L); + ckpt_maybe(s); + return r; +} +int store_put_edge(EngramPagedStore* s, const StoreEdge* e){ + if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1; + uint64_t L = ++s->next_lsn; + if (s->wal){ + size_t blen; uint8_t* body = edge_serialize(e, &blen); + if (!body) return -1; + int wr = wal_append(s, OP_EDGE_PUT, body, (uint32_t)blen, L); + free(body); + if (wr != 0) return -1; + } + int r = apply_edge_put(s, e, L); + ckpt_maybe(s); + return r; +} +int store_tombstone(EngramPagedStore* s, const char* id){ + if (!s || !id) return -1; + uint64_t L = ++s->next_lsn; + if (s->wal){ + if (wal_append(s, OP_TOMBSTONE, (const uint8_t*)id, (uint32_t)strlen(id), L) != 0) return -1; + } + int r = tombstone_core(s, id, L); + ckpt_maybe(s); + return r; +} +int store_forget(EngramPagedStore* s, const char* id){ + if (!s || !id) return -1; + uint64_t L = ++s->next_lsn; + if (s->wal){ + if (wal_append(s, OP_FORGET, (const uint8_t*)id, (uint32_t)strlen(id), L) != 0) return -1; + } + int r = tombstone_core(s, id, L); + ckpt_maybe(s); + return r; +} +int store_supersede(EngramPagedStore* s, const char* old_id, const char* new_id){ + if (!s || !old_id) return -1; + uint64_t L = ++s->next_lsn; + if (s->wal){ + uint32_t ol=(uint32_t)strlen(old_id), nl=(uint32_t)(new_id?strlen(new_id):0); + size_t plen = 4+ol+4+nl; + uint8_t* p = (uint8_t*)malloc(plen); + if (!p) return -1; + put_u32(p, ol); memcpy(p+4, old_id, ol); + put_u32(p+4+ol, nl); if (nl) memcpy(p+8+ol, new_id, nl); + int wr = wal_append(s, OP_SUPERSEDE, p, (uint32_t)plen, L); + free(p); + if (wr != 0) return -1; + } + int r = tombstone_core(s, old_id, L); + ckpt_maybe(s); + return r; +} +int store_put_layer(EngramPagedStore* s, const StoreLayer* L){ + if (!s || !L) return -1; + uint64_t lsn = ++s->next_lsn; + if (s->wal){ + size_t blen; uint8_t* body = layer_serialize(L, &blen); + if (!body) return -1; + int wr = wal_append(s, OP_LAYER_PUT, body, (uint32_t)blen, lsn); + free(body); + if (wr != 0) return -1; + } + int r = apply_layer_put(s, L, lsn); + ckpt_maybe(s); + return r; +} +int store_del_layer(EngramPagedStore* s, uint32_t layer_id){ + if (!s) return -1; + uint64_t lsn = ++s->next_lsn; + if (s->wal){ + uint8_t p[4]; put_u32(p, layer_id); + if (wal_append(s, OP_LAYER_DEL, p, 4, lsn) != 0) return -1; + } + int r = apply_layer_del(s, layer_id, lsn); + ckpt_maybe(s); + return r; +} +int store_hebb_batch(EngramPagedStore* s, const StoreHebbDelta* d, size_t n){ + if (!s || (!d && n)) return -1; + uint64_t L = ++s->next_lsn; + /* build payload */ + Buf b = {0,0,0}; + { uint8_t c[4]; put_u32(c, (uint32_t)n); buf_raw(&b, c, 4); } + for (size_t i=0;iwal){ + if (wal_append(s, OP_HEBB_BATCH, b.p, (uint32_t)b.len, L) != 0){ free(b.p); return -1; } + } + int r = apply_hebb_batch(s, b.p, (uint32_t)b.len, L); + free(b.p); + ckpt_maybe(s); + return r; +} + +/* ── checkpoint (with test-only crash injection at each step) ─────────────────── */ +int store__checkpoint_crashat(EngramPagedStore* s, int phase){ + if (!s) return -1; + if (phase == 0){ store__crash(s); return 0; } + if (pc_flush(s) != 0) return -1; /* 1: dirty pages → disk */ + if (phase == 1){ store__crash(s); return 0; } + if (fsync(s->fd) != 0) return -1; /* 2: fsync store */ + if (phase == 2){ store__crash(s); return 0; } + uint64_t C = s->next_lsn; /* 3: advance ckpt LSN + SB */ + s->last_checkpoint_lsn = C; + s->sb_seq++; + if (sb_write_one(s, 0) != 0) return -1; + if (fsync(s->fd) != 0) return -1; + if (sb_write_one(s, 1) != 0) return -1; + if (fsync(s->fd) != 0) return -1; + 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; + if (phase == 4){ store__crash(s); return 0; } + return 0; +} +int engram_checkpoint(EngramPagedStore* s){ return store__checkpoint_crashat(s, -1); } + +/* ── crash / steal test hooks ────────────────────────────────────────────────── */ +void store__crash(EngramPagedStore* s){ + if (!s) return; /* abandon RAM: dirty pages lost, WAL as fsync'd */ + if (s->wal){ if (s->wal->fd>=0) close(s->wal->fd); free(s->wal); s->wal=NULL; } + if (s->cache){ pc_free(s->cache); s->cache=NULL; } + if (s->fd >= 0) close(s->fd); + free(s); +} +int store__flush_pages(EngramPagedStore* s){ + if (!s) return -1; + if (pc_flush(s) != 0) return -1; /* steal: dirty pages hit disk, no checkpoint */ + return fsync(s->fd); +} + +void engram_set_wal_sync(EngramPagedStore* s, EngramWalSync policy){ + if (s && s->wal) s->wal->sync = policy; +} +uint64_t engram_wal_next_lsn(const EngramPagedStore* s){ return s ? s->next_lsn : 0; } +uint64_t engram_last_checkpoint_lsn(const EngramPagedStore* s){ return s ? s->last_checkpoint_lsn : 0; } + +/* ══════════════════════════════════════════════════════════════════════════════ + * Minimal JSON reader for the ONE-TIME legacy snapshot.json import. + * Tolerant scanner over the `engram_save` schema ({nodes,edges,layers}); it is an + * IMPORT SOURCE only — after import + checkpoint the paged store is authoritative + * and this path is never taken again (engram_open imports only when the store is + * absent). JSON is never read as the ongoing store. + * ════════════════════════════════════════════════════════════════════════════ */ +static const char* js_ws(const char* p){ while (*p==' '||*p=='\t'||*p=='\n'||*p=='\r') p++; return p; } +static const char* js_skip_string(const char* p){ + p++; /* opening quote */ + while (*p){ if (*p=='\\'){ if(!p[1]) return p+1; p+=2; continue; } if (*p=='"') return p+1; p++; } + return p; +} +static const char* js_skip_value(const char* p){ + p = js_ws(p); + if (*p=='"') return js_skip_string(p); + if (*p=='{' || *p=='['){ + char open=*p, close=(open=='{')?'}':']'; int depth=0; + while (*p){ if (*p=='"'){ p=js_skip_string(p); continue; } + if (*p==open) depth++; else if (*p==close){ depth--; if(!depth) return p+1; } p++; } + return p; + } + while (*p && *p!=',' && *p!='}' && *p!=']') p++; + return p; +} +/* value of member `key` at the top level of the object at `obj` ('{'), or NULL. */ +static const char* js_member(const char* obj, const char* key){ + const char* p = js_ws(obj); + if (*p != '{') return NULL; + p++; + size_t keylen = strlen(key); + for (;;){ + p = js_ws(p); + if (*p=='}' || !*p) return NULL; + if (*p != '"') return NULL; + const char* kstart = p+1; + const char* kend = js_skip_string(p); /* points past closing quote */ + size_t klen = (size_t)(kend-1-kstart); + const char* colon = js_ws(kend); + if (*colon != ':') return NULL; + const char* val = js_ws(colon+1); + if (klen==keylen && strncmp(kstart,key,klen)==0) return val; + p = js_ws(js_skip_value(val)); + if (*p==','){ p++; continue; } + return NULL; + } +} +/* decode a JSON string value into a fresh C string (basic escapes + \uXXXX BMP). */ +static char* js_str(const char* v){ + if (!v || *v!='"') return NULL; + const char* p = v+1; + Buf b = {0,0,0}; + while (*p && *p!='"'){ + if (*p=='\\'){ + p++; + switch (*p){ + case 'n': { char c='\n'; buf_raw(&b,&c,1); } break; + case 't': { char c='\t'; buf_raw(&b,&c,1); } break; + case 'r': { char c='\r'; buf_raw(&b,&c,1); } break; + case 'b': { char c='\b'; buf_raw(&b,&c,1); } break; + case 'f': { char c='\f'; buf_raw(&b,&c,1); } break; + case '/': { char c='/'; buf_raw(&b,&c,1); } break; + case '\\':{ char c='\\'; buf_raw(&b,&c,1); } break; + case '"': { char c='"'; buf_raw(&b,&c,1); } break; + case 'u': { + unsigned h=0; for (int k=0;k<4 && p[1];k++){ char d=p[1]; int x=(d>='0'&&d<='9')?d-'0':(d>='a'&&d<='f')?d-'a'+10:(d>='A'&&d<='F')?d-'A'+10:0; h=h*16+x; p++; } + if (h<0x80){ char c=(char)h; buf_raw(&b,&c,1); } + else if (h<0x800){ char c[2]={(char)(0xC0|(h>>6)),(char)(0x80|(h&0x3F))}; buf_raw(&b,c,2); } + else { char c[3]={(char)(0xE0|(h>>12)),(char)(0x80|((h>>6)&0x3F)),(char)(0x80|(h&0x3F))}; buf_raw(&b,c,3); } + } break; + default: if (*p) buf_raw(&b, p, 1); break; + } + if (*p) p++; + } else { buf_raw(&b, p, 1); p++; } + } + char z=0; buf_raw(&b,&z,1); + return (char*)b.p; +} +static char* js_str_field(const char* obj, const char* key){ + const char* v = js_member(obj, key); + return (v && *v=='"') ? js_str(v) : NULL; +} +static double js_num_field(const char* obj, const char* key){ + const char* v = js_member(obj, key); + if (!v || *v=='"' || *v=='{' || *v=='[') return 0.0; + return strtod(v, NULL); +} +static int64_t js_int_field(const char* obj, const char* key){ + const char* v = js_member(obj, key); + if (!v || *v=='"' || *v=='{' || *v=='[') return 0; + return (int64_t)strtoll(v, NULL, 10); +} +/* iterate array elements: first element (or NULL), and next element from a ptr. */ +static const char* js_array_first(const char* arr){ + if (!arr) return NULL; + arr = js_ws(arr); + if (*arr != '[') return NULL; + arr = js_ws(arr+1); + return (*arr==']') ? NULL : arr; +} +static const char* js_array_next(const char* elem){ + const char* p = js_ws(js_skip_value(elem)); + return (*p==',') ? js_ws(p+1) : NULL; +} + +static int import_snapshot(EngramPagedStore* s, const char* path){ + FILE* f = fopen(path, "rb"); + if (!f) return -1; + fseek(f, 0, SEEK_END); long sz = ftell(f); fseek(f, 0, SEEK_SET); + if (sz < 0){ fclose(f); return -1; } + char* txt = (char*)malloc((size_t)sz + 1); + if (!txt){ fclose(f); return -1; } + size_t rd = fread(txt, 1, (size_t)sz, f); fclose(f); + txt[rd] = 0; + const char* root = js_ws(txt); + + /* nodes */ + const char* nodes = js_member(root, "nodes"); + for (const char* o = js_array_first(nodes); o; o = js_array_next(o)){ + StoreNode n; memset(&n, 0, sizeof n); + n.id = js_str_field(o, "id"); + if (!n.id){ continue; } + n.content = js_str_field(o, "content"); + n.node_type = js_str_field(o, "node_type"); + n.label = js_str_field(o, "label"); + n.tier = js_str_field(o, "tier"); + n.tags = js_str_field(o, "tags"); + n.metadata = js_str_field(o, "metadata"); + n.salience = js_num_field(o, "salience"); + n.importance = js_num_field(o, "importance"); + n.confidence = js_num_field(o, "confidence"); + n.temporal_decay_rate = js_num_field(o, "temporal_decay_rate"); + n.activation_count = js_int_field(o, "activation_count"); + n.last_activated = js_int_field(o, "last_activated"); + n.created_at = js_int_field(o, "created_at"); + n.updated_at = js_int_field(o, "updated_at"); + n.background_activation = js_num_field(o, "background_activation"); + n.working_memory_weight = js_num_field(o, "working_memory_weight"); + n.suppression_count = (int32_t)js_int_field(o, "suppression_count"); + n.layer_id = (uint32_t)js_int_field(o, "layer_id"); + n.wm_anchor = js_num_field(o, "wm_anchor"); + /* access_ts: chronological "t,t,t" → ring (head=count%K, filled=count) */ + char* ats = js_str_field(o, "access_ts"); + if (ats && *ats){ + int cnt=0; const char* q=ats; + while (*q && cnt0) */ + e.confidence = js_num_field(o, "confidence"); + e.created_at = js_int_field(o, "created_at"); + e.updated_at = js_int_field(o, "updated_at"); + e.last_fired = js_int_field(o, "last_fired"); + e.inhibitory = (int32_t)js_int_field(o, "inhibitory"); + e.layer_id = (uint32_t)js_int_field(o, "layer_id"); + store_put_edge(s, &e); + store_edge_free(&e); + } + /* layers */ + const char* layers = js_member(root, "layers"); + for (const char* o = js_array_first(layers); o; o = js_array_next(o)){ + StoreLayer L; memset(&L, 0, sizeof L); + L.layer_id = (uint32_t)js_int_field(o, "layer_id"); + L.name = js_str_field(o, "name"); + L.activation_priority = (uint32_t)js_int_field(o, "activation_priority"); + L.suppressible = (int32_t)js_int_field(o, "suppressible"); + L.transparent = (int32_t)js_int_field(o, "transparent"); + L.injectable = (int32_t)js_int_field(o, "injectable"); + store_put_layer(s, &L); + store_layer_free(&L); + } + free(txt); + return 0; +} + +/* ── durable-engram boot / close ─────────────────────────────────────────────── */ +EngramPagedStore* engram_open(const char* data_dir){ + if (!data_dir) return NULL; + char store_path[1200], wal_path[1200], snap_path[1200]; + snprintf(store_path, sizeof store_path, "%s/engram.store", data_dir); + snprintf(wal_path, sizeof wal_path, "%s/engram.wal", data_dir); + snprintf(snap_path, sizeof snap_path, "%s/snapshot.json", data_dir); + + EngramWalSync sync = ENGRAM_WAL_GROUP; + const char* env = getenv("ENGRAM_WAL_SYNC"); + if (env){ if(!strcmp(env,"always")) sync=ENGRAM_WAL_ALWAYS; + else if(!strcmp(env,"off")) sync=ENGRAM_WAL_OFF; + else sync=ENGRAM_WAL_GROUP; } + + struct stat st; + int store_exists = (stat(store_path, &st) == 0); + EngramPagedStore* s; + + if (store_exists){ + s = store_open(store_path); + if (!s) return NULL; + s->wal = wal_open(wal_path, sync); + if (!s->wal){ store_close(s); return NULL; } + s->ckpt_threshold = 100000; + wal_recover(s); /* replay post-checkpoint tail */ + return s; + } + /* Fresh: born on the final paged format. JSON (if any) is imported ONCE. */ + s = store_create(store_path); + if (!s) return NULL; + s->wal = wal_open(wal_path, sync); + if (!s->wal){ store_close(s); return NULL; } + s->ckpt_threshold = 100000; + if (stat(snap_path, &st) == 0) import_snapshot(s, snap_path); + engram_checkpoint(s); /* store is now authoritative */ + return s; +} +int engram_close(EngramPagedStore* s){ + if (!s) return -1; + engram_checkpoint(s); + return store_close(s); +} diff --git a/lang/runtime/engram_store.h b/lang/runtime/engram_store.h index c791da0..0bb0660 100644 --- a/lang/runtime/engram_store.h +++ b/lang/runtime/engram_store.h @@ -132,4 +132,79 @@ void store__set_btree_order(EngramPagedStore* s, int leaf_max, int internal_max) /* Introspection for tests/tools. */ uint64_t store_page_count(const EngramPagedStore* s); +/* ── M2: WAL + checkpoint + crash recovery + one-time legacy import ───────────── + * + * The durable engram is `engram.store` (paged) fronted by `engram.wal` + * (append-only). A mutation is durable once its WAL record is fsync'd + * (group-commit). Pages are held write-back in RAM (no-steal) and flushed to the + * store only at a checkpoint, so the store file on disk always reflects a + * consistent point (`last_checkpoint_lsn`) and the WAL owns everything since. + * Recovery = open store, replay WAL forward, redo a record only where the target + * record's home page LSN < record LSN (idempotent). JSON is ONLY an import + * source / export artifact — never the ongoing store. */ + +typedef enum { ENGRAM_WAL_ALWAYS = 0, ENGRAM_WAL_GROUP = 1, ENGRAM_WAL_OFF = 2 } EngramWalSync; + +/* Serializable layer-registry view (the `layers` array of the legacy snapshot). */ +typedef struct StoreLayer { + uint32_t layer_id; + char* name; + uint32_t activation_priority; + int32_t suppressible; + int32_t transparent; + int32_t injectable; + uint8_t* unknown; + size_t unknown_len; + int tombstoned; +} StoreLayer; + +/* Boot the durable engram in `data_dir` (holds engram.store + engram.wal). If the + * store is absent but a legacy snapshot.json exists, it is imported ONCE into a + * fresh store; thereafter the store is authoritative and JSON is never read again. + * On open, the WAL is replayed to recover any post-checkpoint mutations. */ +EngramPagedStore* engram_open(const char* data_dir); +int engram_close(EngramPagedStore* s); /* checkpoint + close */ + +/* Force a checkpoint: flush dirty pages → fsync store → advance checkpoint LSN → + * reclaim the WAL prefix. Also threshold-triggered automatically on the write path. */ +int engram_checkpoint(EngramPagedStore* s); + +/* WAL commit policy. engram_open honours env ENGRAM_WAL_SYNC=always|group|off. */ +void engram_set_wal_sync(EngramPagedStore* s, EngramWalSync policy); + +/* Layer registry. */ +int store_put_layer(EngramPagedStore* s, const StoreLayer* L); +int store_get_layer(EngramPagedStore* s, uint32_t layer_id, StoreLayer* out); +int store_del_layer(EngramPagedStore* s, uint32_t layer_id); +int store_list_layers(EngramPagedStore* s, StoreLayer** out, size_t* n); +void store_layer_free(StoreLayer* L); +void store_layers_free(StoreLayer* arr, size_t n); + +/* Edge lookup by id (for hebb updates + idempotency). 1 hit / 0 absent / <0 err. */ +int store_get_edge(EngramPagedStore* s, const char* id, StoreEdge* out); + +/* HEBB batch: one WAL record updating hebb (+ last_fired) on a set of edges. */ +typedef struct StoreHebbDelta { const char* edge_id; double hebb; int64_t last_fired; } StoreHebbDelta; +int store_hebb_batch(EngramPagedStore* s, const StoreHebbDelta* d, size_t n); + +/* Supersede: logs the (old,new) pair and tombstones old_id at the store; the new + * node + `supersedes` edge are logged separately (neuron-layer immutability). */ +int store_supersede(EngramPagedStore* s, const char* old_id, const char* new_id); + +/* Forget (GC): tombstone id at the store (hard-free deferred to compaction). */ +int store_forget(EngramPagedStore* s, const char* id); + +/* Introspection / test hooks. */ +uint64_t engram_wal_next_lsn(const EngramPagedStore* s); +uint64_t engram_last_checkpoint_lsn(const EngramPagedStore* s); + +/* Crash-test hooks (writes only under a throwaway dir). + * store__crash — abandon all RAM state without flush/fsync (power loss). + * store__flush_pages — pwrite dirty pages to disk WITHOUT a checkpoint (steal). + * store__checkpoint_crashat — run checkpoint but stop (then power-loss) after + * `phase` (0..4); phase<0 = full checkpoint. */ +void store__crash(EngramPagedStore* s); +int store__flush_pages(EngramPagedStore* s); +int store__checkpoint_crashat(EngramPagedStore* s, int phase); + #endif /* ENGRAM_STORE_H */