Files
el/lang/runtime/engram_store.c
T
will.anderson 9e28defcab fix: bounds-check btree_insert + read_body to survive bloated store / WAL replay (live crash-loop root cause)
int_max_keys() computed an internal B+tree node's key capacity as
IDX_BODY/8 - 1 (2041 at a 16 KB page), dividing by 8 and ignoring that
each key also carries an 8-byte child pointer. The true capacity is
(IDX_BODY-8)/16 = 1020. An internal node was therefore allowed to grow to
~2x what a page holds; once it crossed 1020 keys, btree_insert's write-back
overran its STORE_PAGE_SIZE stack page buffer and smashed the stack canary
(__stack_chk_fail / SIGABRT). A clean/small store never grows an internal
node that large, so it never tripped; the ~8x-bloated live store (a day of
tombstone churn) plus a 44 MB un-checkpointed WAL replayed on open pushed a
node over the boundary during redo -> deterministic crash loop
(btree_insert <- apply_edge_put <- engram_open <- engram_store_boot).

Fixes:
- int_max_keys: use (IDX_BODY-8)/16 so internal nodes split at the real
  page capacity.
- btree_insert: reject any page whose on-disk nkeys exceeds physical
  capacity (fail loud, never smash the stack) -- overflow is now impossible
  regardless of on-disk content.
- read_body: bound the slot (off,len) and record length to the page before
  dereferencing; a stale/torn index entry could otherwise make store_get_node
  read off the stack (observed EXC_BAD_ACCESS on the bloated store). Fail safe.

Verified on a COPY of the live store: unfixed binary SIGABRTs in btree_insert
on open; fixed binary boots clean, recovers the store, checkpoints the WAL,
and M5 compaction shrinks 458 MB -> 57.7 MB with node/edge counts preserved.
2026-08-12 18:27:13 -05:00

2686 lines
123 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/* engram_store.c — M1 paged store: writer/reader for the FINAL on-disk format.
*
* Self-contained. Standard libs only. See engram_store.h and the design doc
* (engram-tiered-storage-engine.md §2/§4/§7). Write-through page I/O via
* pread/pwrite (no buffer pool yet — that is M4). Every mutation immediately
* lands on disk; store_sync/close rewrite both superblocks and fsync.
*
* On-disk layout (all integers little-endian):
* page 0,1 : superblock + mirror (recovery picks higher valid sb_seq)
* page N : generic 32-byte page header, then type-specific body
*
* Generic page header (32 bytes):
* +0 u64 page_id
* +8 u8 page_type (NODE|EDGE|INDEX|OVERFLOW|FREE)
* +9 u8 reserved
* +10 u16 slot_count (slotted pages) / nkeys (index pages)
* +12 u16 free_bytes (slotted pages: contiguous heap gap)
* +14 u16 reserved
* +16 u64 lsn (LSN of last write — ARIES redo idempotency, M2)
* +24 u32 crc32 (over the whole page with these 4 bytes zeroed)
* +28 u32 reserved
*
* Slotted page (NODE/EDGE): slot directory grows from +32 (6 bytes each:
* [u16 off][u16 len][u16 flags]); records grow down from PAGE_SIZE.
*
* On-page record: [u16 rec_len][u8 rec_ver][u8 rec_flags] then either the TLV
* field body (inline) or, if REC_OVERFLOW, a stub [u64 head_page][u64 total].
*
* TLV field: [u8 tag][u32 len][bytes]. NOTE: the design §2 sketch says u16 len;
* node content can exceed 64 KiB (tests use 100 KiB), so the field length MUST
* be u32. This is the one format decision beyond §2 — flagged in the report.
*
* Overflow page: header, then +32 [u64 next_page][u32 chunk_len][payload...].
*
* Index page (B+-tree, u64 keys, duplicates allowed, fixed payload per tree):
* +32 u8 is_leaf; +36 u64 next_leaf; +44 entries.
* leaf entry = [u64 key][payload]; payload = primary(10) or adjacency(11).
* internal = nkeys u64 keys, then (nkeys+1) u64 child page ids.
*/
#include "engram_store.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
#include <sys/stat.h>
/* STORE_BLL_K must track ENGRAM_BLL_K in el_runtime.c. */
typedef char store__bll_k_check[(STORE_BLL_K == 10) ? 1 : -1];
#define STORE_HDR 32u /* generic page header size */
#define SLOT_SIZE 6u /* [u16 off][u16 len][u16 flags] */
#define SLOT_LIVE 1u
#define SLOT_DEAD 2u
#define REC_HDR 4u /* [u16 rec_len][u8 ver][u8 flags] */
#define REC_VER 1u
#define REC_OVERFLOW 1u /* rec_flags bit0: body is in an overflow chain */
#define MAX_INLINE_BODY (STORE_PAGE_SIZE - STORE_HDR - SLOT_SIZE - REC_HDR)
/* Index page geometry. */
#define IDX_LEAF_OFF 32u
#define IDX_NEXT_OFF 36u
#define IDX_ENT_OFF 44u
#define IDX_BODY (STORE_PAGE_SIZE - IDX_ENT_OFF)
/* Overflow page geometry. */
#define OVF_NEXT_OFF 32u
#define OVF_LEN_OFF 40u
#define OVF_DATA_OFF 44u
#define OVF_CHUNK (STORE_PAGE_SIZE - OVF_DATA_OFF)
/* Superblock layout (page 0 and mirror page 1). */
#define SB_MAGIC_OFF 0u
#define SB_FMT_OFF 8u
#define SB_PGSZ_OFF 12u
#define SB_PGCNT_OFF 16u
#define SB_FREE_OFF 24u
#define SB_ROOT_OFF 32u
#define SB_ADJ_OFF 40u
#define SB_LAYER_OFF 48u
#define SB_CKPT_OFF 56u
#define SB_SEQ_OFF 64u
#define SB_UUID_OFF 72u /* 16 bytes */
#define SB_CRC_OFF 88u
/* Which index tree. */
enum { TREE_PRIMARY = 0, TREE_ADJ = 1 };
#define PRIMARY_PAYLOAD 10u /* [u64 page_id][u16 slot] */
#define ADJ_PAYLOAD 11u /* [u8 dir][u64 page_id][u16 slot] */
#define ADJ_DIR_FROM 0u
#define ADJ_DIR_TO 1u
#define MAX_PAYLOAD 16u
/* Node/Edge TLV tag namespaces (per-kind; page type also disambiguates). */
enum {
NT_ID = 1, NT_CONTENT, NT_NODE_TYPE, NT_LABEL, NT_TIER, NT_TAGS, NT_METADATA,
NT_SALIENCE, NT_IMPORTANCE, NT_CONFIDENCE, NT_DECAY,
NT_ACT_COUNT, NT_LAST_ACT, NT_CREATED, NT_UPDATED,
NT_BG_ACT, NT_WM_WEIGHT, NT_SUPPRESS, NT_LAYER,
NT_ACCESS_TS, NT_ACCESS_HEAD, NT_ACCESS_FILLED, NT_WM_ANCHOR,
NT_EMB, NT_EMB_DIM
};
enum {
ET_ID = 1, ET_FROM, ET_TO, ET_RELATION, ET_METADATA,
ET_WEIGHT, ET_HEBB, ET_CONFIDENCE, ET_CREATED, ET_UPDATED, ET_LAST_FIRED,
ET_INHIBITORY, ET_LAYER
};
/* ── M2 forward decls (buffer pool + WAL live in the M2 section below) ─────── */
typedef struct PgCache PgCache;
typedef struct EngramWal EngramWal;
struct EngramPagedStore {
int fd;
char path[1024];
uint32_t format_version;
uint32_t page_size;
uint64_t page_count;
uint64_t free_list_head; /* page id of first FREE page, 0 = none */
uint64_t root_index_page; /* primary id->loc B+-tree root */
uint64_t adj_index_page; /* from/to adjacency B+-tree root */
uint64_t layer_registry_page;
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; /* 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) */
/* ── 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).
* M2 shipped a write-back, no-steal cache (dirty→disk only at checkpoint). M4
* turns it into a bounded, demand-paged buffer pool: a fixed frame budget, LRU
* eviction of CLEAN unpinned frames (no-steal preserved — dirty frames are never
* stolen), pinning of hot/structural pages, and bounded read-ahead. `lru_*`
* thread every resident frame onto an MRU→LRU list; `pin` is an explicit pin
* count (0 = unpinned). */
typedef struct PgEnt {
uint64_t id; uint8_t* buf; uint64_t lsn; int dirty; struct PgEnt* next;
int pin; /* explicit pin count (0 = unpinned) */
struct PgEnt* lru_prev; /* MRU→LRU doubly-linked list */
struct PgEnt* lru_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 */
static void pc_prefetch(EngramPagedStore* s, uint64_t from_id, unsigned window);
static void store__autopin(EngramPagedStore* s); /* pin superblocks + index roots */
/* Per-layer pin record: the set of pages pinned on behalf of a hot layer, kept
* so store_unpin_layer can release exactly what store_pin_layer pinned. */
typedef struct { uint32_t layer; uint64_t* pages; size_t n; } LayerPin;
/* The bounded, demand-paged frame table (M4). Defined here (not in the pool
* section) so page_read / the scan loops can read its stats + prefetch window. */
struct PgCache {
PgEnt** buckets; size_t nbuckets; size_t count;
size_t cap; /* max resident frames; 0 = unlimited */
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;
};
/* ── 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)); }
static void put_u64(uint8_t* p, uint64_t v){ for(int i=0;i<8;i++) p[i]=(uint8_t)(v>>(8*i)); }
static uint16_t get_u16(const uint8_t* p){ return (uint16_t)(p[0] | (p[1]<<8)); }
static uint32_t get_u32(const uint8_t* p){ uint32_t v=0; for(int i=0;i<4;i++) v|=(uint32_t)p[i]<<(8*i); return v; }
static uint64_t get_u64(const uint8_t* p){ uint64_t v=0; for(int i=0;i<8;i++) v|=(uint64_t)p[i]<<(8*i); return v; }
static void put_f64(uint8_t* p, double d){ uint64_t u; memcpy(&u,&d,8); put_u64(p,u); }
static double get_f64(const uint8_t* p){ uint64_t u=get_u64(p); double d; memcpy(&d,&u,8); return d; }
/* ── crc32 (IEEE 802.3, reflected, poly 0xEDB88320); matches eg_crc32 in
* el_runtime.c so integrity checks are consistent cross-module.
* crc32("") == 0, crc32("123456789") == 0xCBF43926. ─────────────────────── */
static uint32_t s_crc_table[256];
static int s_crc_ready = 0;
static void crc_init(void){
for (uint32_t i=0;i<256;i++){
uint32_t c=i;
for (int k=0;k<8;k++) c = (c&1) ? (0xEDB88320u ^ (c>>1)) : (c>>1);
s_crc_table[i]=c;
}
s_crc_ready=1;
}
static uint32_t crc32_buf(const void* data, size_t n){
if (!s_crc_ready) crc_init();
const uint8_t* p = (const uint8_t*)data;
uint32_t c = 0xFFFFFFFFu;
for (size_t i=0;i<n;i++) c = s_crc_table[(c ^ p[i]) & 0xFF] ^ (c >> 8);
return c ^ 0xFFFFFFFFu;
}
/* FNV-1a 64-bit over a NUL-terminated id; the B+-tree key. Exact id is verified
* against the record on read, so hash collisions are correctness-safe. */
static uint64_t id_hash(const char* s){
uint64_t h = 1469598103934665603ULL;
for (; *s; ++s){ h ^= (uint8_t)*s; h *= 1099511628211ULL; }
return h;
}
/* ── 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); s->cache->hits++; return 0; }
}
off_t off = (off_t)id * STORE_PAGE_SIZE; /* demand fault: not resident */
ssize_t r = pread(s->fd, buf, STORE_PAGE_SIZE, off);
if (r != (ssize_t)STORE_PAGE_SIZE) return -1;
if (s->cache){ s->cache->misses++; pc_put(s, id, buf, 0); } /* cache clean */
return 0;
}
static int page_write_raw(EngramPagedStore* s, uint64_t id, const uint8_t* buf){
off_t off = (off_t)id * STORE_PAGE_SIZE;
ssize_t w = pwrite(s->fd, buf, STORE_PAGE_SIZE, off);
if (w != (ssize_t)STORE_PAGE_SIZE) return -1;
return 0;
}
/* 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);
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);
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];
memcpy(tmp, buf, STORE_PAGE_SIZE);
uint32_t stored = get_u32(tmp + 24);
put_u32(tmp + 24, 0);
return crc32_buf(tmp, STORE_PAGE_SIZE) == stored;
}
/* Allocate a page: reuse a FREE page if available, else extend the file. */
static uint64_t page_alloc(EngramPagedStore* s, uint8_t type){
uint8_t buf[STORE_PAGE_SIZE];
uint64_t id;
if (s->free_list_head){
id = s->free_list_head;
if (page_read(s, id, buf) != 0) return 0;
s->free_list_head = get_u64(buf + OVF_NEXT_OFF); /* next_free stashed here */
} else {
id = s->page_count++;
}
memset(buf, 0, STORE_PAGE_SIZE);
buf[8] = type;
if (type == STORE_PT_NODE || type == STORE_PT_EDGE){
put_u16(buf + 10, 0);
put_u16(buf + 12, (uint16_t)(STORE_PAGE_SIZE - STORE_HDR)); /* all heap free */
}
if (page_write(s, id, buf) != 0) return 0;
return id;
}
/* Return a now-empty page to the free list. */
static int page_free(EngramPagedStore* s, uint64_t id){
uint8_t buf[STORE_PAGE_SIZE];
memset(buf, 0, STORE_PAGE_SIZE);
buf[8] = STORE_PT_FREE;
put_u64(buf + OVF_NEXT_OFF, s->free_list_head);
if (page_write(s, id, buf) != 0) return -1;
s->free_list_head = id;
return 0;
}
/* ── slotted page helpers (operate on a page already in `buf`) ────────────── */
static int slp_count(const uint8_t* buf){ return get_u16(buf + 10); }
static int slp_free(const uint8_t* buf){ return get_u16(buf + 12); }
static void slp_slot(const uint8_t* buf, int i, uint16_t* off, uint16_t* len, uint16_t* fl){
const uint8_t* p = buf + STORE_HDR + (size_t)i*SLOT_SIZE;
*off = get_u16(p); *len = get_u16(p+2); *fl = get_u16(p+4);
}
static void slp_set_slot(uint8_t* buf, int i, uint16_t off, uint16_t len, uint16_t fl){
uint8_t* p = buf + STORE_HDR + (size_t)i*SLOT_SIZE;
put_u16(p, off); put_u16(p+2, len); put_u16(p+4, fl);
}
static int slp_live_count(const uint8_t* buf){
int n = slp_count(buf), live = 0;
for (int i=0;i<n;i++){ uint16_t o,l,f; slp_slot(buf,i,&o,&l,&f); if (f==SLOT_LIVE) live++; }
return live;
}
/* Place `rec` (reclen bytes) into the page. Returns slot index, or -1 if no room.
* Prefers reusing a DEAD slot whose span fits (reclaims tombstoned space). */
static int slp_put(uint8_t* buf, const uint8_t* rec, uint16_t reclen){
int n = slp_count(buf);
for (int i=0;i<n;i++){
uint16_t o,l,f; slp_slot(buf,i,&o,&l,&f);
if (f==SLOT_DEAD && l>=reclen){
memcpy(buf + o, rec, reclen);
slp_set_slot(buf, i, o, reclen, SLOT_LIVE);
return i;
}
}
int freeb = slp_free(buf);
if ((int)reclen + (int)SLOT_SIZE > freeb) return -1;
uint16_t heap_top = (uint16_t)(STORE_HDR + n*SLOT_SIZE + freeb);
uint16_t off = (uint16_t)(heap_top - reclen);
memcpy(buf + off, rec, reclen);
slp_set_slot(buf, n, off, reclen, SLOT_LIVE);
put_u16(buf + 10, (uint16_t)(n+1));
put_u16(buf + 12, (uint16_t)(freeb - reclen - SLOT_SIZE));
return n;
}
/* ── overflow chains ──────────────────────────────────────────────────────── */
static uint64_t ovf_write_chain(EngramPagedStore* s, const uint8_t* blob, size_t len){
uint64_t head = 0, prev = 0;
size_t off = 0;
do {
uint64_t id = page_alloc(s, STORE_PT_OVERFLOW);
if (!id) return 0;
if (!head) head = id;
if (prev){
uint8_t pbuf[STORE_PAGE_SIZE];
if (page_read(s, prev, pbuf)!=0) return 0;
put_u64(pbuf + OVF_NEXT_OFF, id);
if (page_write(s, prev, pbuf)!=0) return 0;
}
uint8_t buf[STORE_PAGE_SIZE];
memset(buf, 0, STORE_PAGE_SIZE);
buf[8] = STORE_PT_OVERFLOW;
size_t chunk = len - off; if (chunk > OVF_CHUNK) chunk = OVF_CHUNK;
put_u64(buf + OVF_NEXT_OFF, 0);
put_u32(buf + OVF_LEN_OFF, (uint32_t)chunk);
memcpy(buf + OVF_DATA_OFF, blob + off, chunk);
if (page_write(s, id, buf)!=0) return 0;
prev = id; off += chunk;
} while (off < len);
return head;
}
static uint8_t* ovf_read_chain(EngramPagedStore* s, uint64_t head, size_t total){
uint8_t* out = (uint8_t*)malloc(total ? total : 1);
if (!out) return NULL;
size_t off = 0; uint64_t id = head;
while (id){
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, id, buf)!=0){ free(out); return NULL; }
uint32_t chunk = get_u32(buf + OVF_LEN_OFF);
if (off + chunk > total){ free(out); return NULL; }
memcpy(out + off, buf + OVF_DATA_OFF, chunk);
off += chunk;
id = get_u64(buf + OVF_NEXT_OFF);
}
if (off != total){ free(out); return NULL; }
return out;
}
/* ── growable byte buffer for TLV assembly ────────────────────────────────── */
typedef struct { uint8_t* p; size_t len, cap; } Buf;
static int buf_reserve(Buf* b, size_t extra){
if (b->len + extra <= b->cap) return 0;
size_t nc = b->cap ? b->cap*2 : 256;
while (nc < b->len + extra) nc *= 2;
uint8_t* np = (uint8_t*)realloc(b->p, nc);
if (!np) return -1;
b->p = np; b->cap = nc; return 0;
}
static int buf_raw(Buf* b, const void* data, size_t n){
if (buf_reserve(b, n)) return -1;
memcpy(b->p + b->len, data, n); b->len += n; return 0;
}
static int tlv_field(Buf* b, uint8_t tag, const void* data, uint32_t len){
if (buf_reserve(b, 5 + len)) return -1;
b->p[b->len++] = tag;
put_u32(b->p + b->len, len); b->len += 4;
if (len){ memcpy(b->p + b->len, data, len); b->len += len; }
return 0;
}
static int tlv_str(Buf* b, uint8_t tag, const char* s){
if (!s) return 0; /* NULL string omitted → reader defaults NULL */
return tlv_field(b, tag, s, (uint32_t)strlen(s));
}
static int tlv_f64(Buf* b, uint8_t tag, double v){ uint8_t t[8]; put_f64(t,v); return tlv_field(b,tag,t,8); }
static int tlv_i64(Buf* b, uint8_t tag, int64_t v){ uint8_t t[8]; put_u64(t,(uint64_t)v); return tlv_field(b,tag,t,8); }
static int tlv_i32(Buf* b, uint8_t tag, int32_t v){ uint8_t t[4]; put_u32(t,(uint32_t)v); return tlv_field(b,tag,t,4); }
static int tlv_u32(Buf* b, uint8_t tag, uint32_t v){ uint8_t t[4]; put_u32(t,v); return tlv_field(b,tag,t,4); }
/* ── node/edge serialization to a TLV body ────────────────────────────────── */
static uint8_t* node_serialize(const StoreNode* n, size_t* out_len){
Buf b = {0,0,0};
tlv_str(&b, NT_ID, n->id);
tlv_str(&b, NT_CONTENT, n->content);
tlv_str(&b, NT_NODE_TYPE, n->node_type);
tlv_str(&b, NT_LABEL, n->label);
tlv_str(&b, NT_TIER, n->tier);
tlv_str(&b, NT_TAGS, n->tags);
tlv_str(&b, NT_METADATA, n->metadata);
tlv_f64(&b, NT_SALIENCE, n->salience);
tlv_f64(&b, NT_IMPORTANCE, n->importance);
tlv_f64(&b, NT_CONFIDENCE, n->confidence);
tlv_f64(&b, NT_DECAY, n->temporal_decay_rate);
tlv_i64(&b, NT_ACT_COUNT, n->activation_count);
tlv_i64(&b, NT_LAST_ACT, n->last_activated);
tlv_i64(&b, NT_CREATED, n->created_at);
tlv_i64(&b, NT_UPDATED, n->updated_at);
tlv_f64(&b, NT_BG_ACT, n->background_activation);
tlv_f64(&b, NT_WM_WEIGHT, n->working_memory_weight);
tlv_i32(&b, NT_SUPPRESS, n->suppression_count);
tlv_u32(&b, NT_LAYER, n->layer_id);
{ uint8_t at[STORE_BLL_K*8];
for (int i=0;i<STORE_BLL_K;i++) put_u64(at + i*8, (uint64_t)n->access_ts[i]);
tlv_field(&b, NT_ACCESS_TS, at, sizeof at); }
tlv_i32(&b, NT_ACCESS_HEAD, n->access_head);
tlv_i32(&b, NT_ACCESS_FILLED, n->access_filled);
tlv_f64(&b, NT_WM_ANCHOR, n->wm_anchor);
if (n->emb && n->emb_dim > 0){
uint8_t* eb = (uint8_t*)malloc((size_t)n->emb_dim*4);
for (int i=0;i<n->emb_dim;i++){ uint32_t u; memcpy(&u,&n->emb[i],4); put_u32(eb+i*4,u); }
tlv_field(&b, NT_EMB, eb, (uint32_t)n->emb_dim*4);
free(eb);
tlv_i32(&b, NT_EMB_DIM, n->emb_dim);
}
if (n->unknown && n->unknown_len) buf_raw(&b, n->unknown, n->unknown_len);
*out_len = b.len;
return b.p;
}
static uint8_t* edge_serialize(const StoreEdge* e, size_t* out_len){
Buf b = {0,0,0};
tlv_str(&b, ET_ID, e->id);
tlv_str(&b, ET_FROM, e->from_id);
tlv_str(&b, ET_TO, e->to_id);
tlv_str(&b, ET_RELATION, e->relation);
tlv_str(&b, ET_METADATA, e->metadata);
tlv_f64(&b, ET_WEIGHT, e->weight);
tlv_f64(&b, ET_HEBB, e->hebb);
tlv_f64(&b, ET_CONFIDENCE, e->confidence);
tlv_i64(&b, ET_CREATED, e->created_at);
tlv_i64(&b, ET_UPDATED, e->updated_at);
tlv_i64(&b, ET_LAST_FIRED, e->last_fired);
tlv_i32(&b, ET_INHIBITORY, e->inhibitory);
tlv_u32(&b, ET_LAYER, e->layer_id);
if (e->unknown && e->unknown_len) buf_raw(&b, e->unknown, e->unknown_len);
*out_len = b.len;
return b.p;
}
/* ── TLV parse helpers ────────────────────────────────────────────────────── */
static char* dup_str(const uint8_t* p, uint32_t len){
char* s = (char*)malloc(len + 1);
if (!s) return NULL;
memcpy(s, p, len); s[len] = 0; return s;
}
static void node_parse(const uint8_t* body, size_t len, StoreNode* n){
memset(n, 0, sizeof *n);
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 NT_ID: n->id = dup_str(v, flen); break;
case NT_CONTENT: n->content = dup_str(v, flen); break;
case NT_NODE_TYPE: n->node_type = dup_str(v, flen); break;
case NT_LABEL: n->label = dup_str(v, flen); break;
case NT_TIER: n->tier = dup_str(v, flen); break;
case NT_TAGS: n->tags = dup_str(v, flen); break;
case NT_METADATA: n->metadata = dup_str(v, flen); break;
case NT_SALIENCE: n->salience = get_f64(v); break;
case NT_IMPORTANCE: n->importance = get_f64(v); break;
case NT_CONFIDENCE: n->confidence = get_f64(v); break;
case NT_DECAY: n->temporal_decay_rate = get_f64(v); break;
case NT_ACT_COUNT: n->activation_count = (int64_t)get_u64(v); break;
case NT_LAST_ACT: n->last_activated = (int64_t)get_u64(v); break;
case NT_CREATED: n->created_at = (int64_t)get_u64(v); break;
case NT_UPDATED: n->updated_at = (int64_t)get_u64(v); break;
case NT_BG_ACT: n->background_activation = get_f64(v); break;
case NT_WM_WEIGHT: n->working_memory_weight = get_f64(v); break;
case NT_SUPPRESS: n->suppression_count = (int32_t)get_u32(v); break;
case NT_LAYER: n->layer_id = get_u32(v); break;
case NT_ACCESS_TS:
for (int k=0;k<STORE_BLL_K && (uint32_t)(k*8+8)<=flen;k++)
n->access_ts[k] = (int64_t)get_u64(v + k*8);
break;
case NT_ACCESS_HEAD: n->access_head = (int32_t)get_u32(v); break;
case NT_ACCESS_FILLED: n->access_filled = (int32_t)get_u32(v); break;
case NT_WM_ANCHOR: n->wm_anchor = get_f64(v); break;
case NT_EMB: {
int32_t dim = (int32_t)(flen/4);
n->emb = (float*)malloc(flen ? flen : 4);
for (int32_t k=0;k<dim;k++){ uint32_t u=get_u32(v+k*4); memcpy(&n->emb[k],&u,4); }
if (n->emb_dim == 0) n->emb_dim = dim; /* honour explicit dim if seen */
break; }
case NT_EMB_DIM: n->emb_dim = (int32_t)get_u32(v); break;
default:
/* unknown tag: preserve verbatim */
buf_raw(&unk, body + i, 5 + flen);
break;
}
i += 5 + flen;
}
n->unknown = unk.p; n->unknown_len = unk.len;
}
static void edge_parse(const uint8_t* body, size_t len, StoreEdge* e){
memset(e, 0, sizeof *e);
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 ET_ID: e->id = dup_str(v, flen); break;
case ET_FROM: e->from_id = dup_str(v, flen); break;
case ET_TO: e->to_id = dup_str(v, flen); break;
case ET_RELATION: e->relation = dup_str(v, flen); break;
case ET_METADATA: e->metadata = dup_str(v, flen); break;
case ET_WEIGHT: e->weight = get_f64(v); break;
case ET_HEBB: e->hebb = get_f64(v); break;
case ET_CONFIDENCE: e->confidence = get_f64(v); break;
case ET_CREATED: e->created_at = (int64_t)get_u64(v); break;
case ET_UPDATED: e->updated_at = (int64_t)get_u64(v); break;
case ET_LAST_FIRED: e->last_fired = (int64_t)get_u64(v); break;
case ET_INHIBITORY: e->inhibitory = (int32_t)get_u32(v); break;
case ET_LAYER: e->layer_id = get_u32(v); break;
default:
buf_raw(&unk, body + i, 5 + flen);
break;
}
i += 5 + flen;
}
e->unknown = unk.p; e->unknown_len = unk.len;
}
/* ── B+-tree ──────────────────────────────────────────────────────────────── */
static uint32_t tree_payload(int tree){ return tree==TREE_PRIMARY ? PRIMARY_PAYLOAD : ADJ_PAYLOAD; }
static uint64_t tree_root(EngramPagedStore* s, int tree){
return tree==TREE_PRIMARY ? s->root_index_page : s->adj_index_page;
}
static void tree_set_root(EngramPagedStore* s, int tree, uint64_t id){
if (tree==TREE_PRIMARY) s->root_index_page = id; else s->adj_index_page = id;
}
static int leaf_max_entries(EngramPagedStore* s, uint32_t payload){
int nat = (int)(IDX_BODY / (8 + payload));
if (s->leaf_max > 0 && s->leaf_max < nat) return s->leaf_max;
return nat;
}
static int int_max_keys(EngramPagedStore* s){
/* An internal node stores `keys` u64 keys FOLLOWED BY (keys+1) u64 child
* pointers, so both arrays must fit the page body:
* keys*8 + (keys+1)*8 = 16*keys + 8 <= IDX_BODY → keys <= (IDX_BODY-8)/16.
* The prior form `IDX_BODY/8 - 1` divided by 8 instead of 16 — it counted
* only the key array and ignored the child array's 8 bytes/key — so it
* returned ~2x the real capacity (2041 vs 1020 at a 16 KB page). An internal
* node was then allowed to grow past what a page holds, and btree_insert's
* write-back overran its STORE_PAGE_SIZE stack page buffer, smashing the
* stack canary (__stack_chk_fail). That was the live crash-loop root cause. */
int nat = (int)((IDX_BODY - 8) / 16);
if (s->int_max > 0 && s->int_max < nat) return s->int_max;
return nat;
}
/* Recursive insert. Returns 0 (no split) or 1 (split; sep_key and new_page set). */
static int btree_insert(EngramPagedStore* s, int tree, uint64_t page_id,
uint64_t key, const uint8_t* payload,
uint64_t* sep_key, uint64_t* new_page){
uint32_t pl = tree_payload(tree);
uint32_t esz = 8 + pl;
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, page_id, buf)!=0) return -1;
int nkeys = get_u16(buf + 10);
int is_leaf = buf[IDX_LEAF_OFF];
/* Defensive bound: never trust an on-disk entry count enough to overflow the
* fixed STORE_PAGE_SIZE stack buffer below. A leaf holds at most IDX_BODY/esz
* entries; an internal node at most (IDX_BODY-8)/16 keys (keys + child ptrs).
* A page claiming more than its physical capacity is torn/corrupt (or was
* written by a pre-fix build) — fail LOUD and abort this insert rather than
* smash the stack or silently truncate. With this guard the memmove/memcpy/
* put_u64 write-backs are provably in-bounds regardless of on-disk content. */
int _cap = is_leaf ? (int)(IDX_BODY / esz) : (int)((IDX_BODY - 8) / 16);
if (nkeys < 0 || nkeys > _cap){
fprintf(stderr, "engram_store: corrupt %s index page %llu: nkeys=%d "
"exceeds page capacity %d — refusing insert (fail-safe)\n",
is_leaf ? "leaf" : "internal",
(unsigned long long)page_id, nkeys, _cap);
return -1;
}
if (is_leaf){
/* find insert position (after equal keys → stable duplicates) */
int pos = 0;
while (pos < nkeys){
uint64_t k = get_u64(buf + IDX_ENT_OFF + (size_t)pos*esz);
if (k > key) break;
pos++;
}
int lmax = leaf_max_entries(s, pl);
if (nkeys + 1 <= lmax){
uint8_t* base = buf + IDX_ENT_OFF;
memmove(base + (size_t)(pos+1)*esz, base + (size_t)pos*esz,
(size_t)(nkeys - pos)*esz);
put_u64(base + (size_t)pos*esz, key);
memcpy(base + (size_t)pos*esz + 8, payload, pl);
put_u16(buf + 10, (uint16_t)(nkeys+1));
return page_write(s, page_id, buf)==0 ? 0 : -1;
}
/* split: build the full sorted list in a temp, halve it */
int total = nkeys + 1;
uint8_t* all = (uint8_t*)malloc((size_t)total*esz);
if (!all) return -1;
memcpy(all, buf + IDX_ENT_OFF, (size_t)pos*esz);
put_u64(all + (size_t)pos*esz, key);
memcpy(all + (size_t)pos*esz + 8, payload, pl);
memcpy(all + (size_t)(pos+1)*esz, buf + IDX_ENT_OFF + (size_t)pos*esz,
(size_t)(nkeys - pos)*esz);
int left = total/2, right = total - left;
uint64_t old_next = get_u64(buf + IDX_NEXT_OFF);
uint64_t rid = page_alloc(s, STORE_PT_INDEX);
if (!rid){ free(all); return -1; }
uint8_t rbuf[STORE_PAGE_SIZE];
if (page_read(s, rid, rbuf)!=0){ free(all); return -1; }
rbuf[IDX_LEAF_OFF] = 1;
put_u16(rbuf + 10, (uint16_t)right);
put_u64(rbuf + IDX_NEXT_OFF, old_next);
memcpy(rbuf + IDX_ENT_OFF, all + (size_t)left*esz, (size_t)right*esz);
/* left page keeps first `left` entries, points to right */
buf[IDX_LEAF_OFF] = 1;
put_u16(buf + 10, (uint16_t)left);
put_u64(buf + IDX_NEXT_OFF, rid);
memcpy(buf + IDX_ENT_OFF, all, (size_t)left*esz);
*sep_key = get_u64(all + (size_t)left*esz);
free(all);
if (page_write(s, rid, rbuf)!=0) return -1;
if (page_write(s, page_id, buf)!=0) return -1;
*new_page = rid;
return 1;
}
/* internal: keys at IDX_ENT_OFF (nkeys u64), children after (nkeys+1 u64) */
uint8_t* K = buf + IDX_ENT_OFF;
uint8_t* C = buf + IDX_ENT_OFF + (size_t)nkeys*8;
int ci = 0;
while (ci < nkeys && key >= get_u64(K + (size_t)ci*8)) ci++;
uint64_t child = get_u64(C + (size_t)ci*8);
uint64_t csep, cnew;
int sp = btree_insert(s, tree, child, key, payload, &csep, &cnew);
if (sp < 0) return -1;
if (sp == 0) return 0;
/* child split → insert (csep, cnew) at position ci / ci+1 */
int nk = nkeys + 1;
uint64_t* keys = (uint64_t*)malloc((size_t)nk*8);
uint64_t* kids = (uint64_t*)malloc((size_t)(nk+1)*8);
if (!keys || !kids){ free(keys); free(kids); return -1; }
for (int i=0;i<nkeys;i++) keys[i] = get_u64(K + (size_t)i*8);
for (int i=0;i<=nkeys;i++) kids[i] = get_u64(C + (size_t)i*8);
for (int i=nkeys;i>ci;i--) keys[i] = keys[i-1];
keys[ci] = csep;
for (int i=nkeys+1;i>ci+1;i--) kids[i] = kids[i-1];
kids[ci+1] = cnew;
int imax = int_max_keys(s);
if (nk <= imax){
put_u16(buf + 10, (uint16_t)nk);
uint8_t* K2 = buf + IDX_ENT_OFF;
uint8_t* C2 = buf + IDX_ENT_OFF + (size_t)nk*8;
for (int i=0;i<nk;i++) put_u64(K2 + (size_t)i*8, keys[i]);
for (int i=0;i<=nk;i++) put_u64(C2 + (size_t)i*8, kids[i]);
free(keys); free(kids);
return page_write(s, page_id, buf)==0 ? 0 : -1;
}
/* split internal: middle key moves up */
int mid = nk/2;
uint64_t up = keys[mid];
int lk = mid, rk = nk - mid - 1;
uint64_t rid = page_alloc(s, STORE_PT_INDEX);
if (!rid){ free(keys); free(kids); return -1; }
uint8_t rbuf[STORE_PAGE_SIZE];
if (page_read(s, rid, rbuf)!=0){ free(keys); free(kids); return -1; }
rbuf[IDX_LEAF_OFF] = 0;
put_u16(rbuf + 10, (uint16_t)rk);
{ uint8_t* RK = rbuf + IDX_ENT_OFF; uint8_t* RC = rbuf + IDX_ENT_OFF + (size_t)rk*8;
for (int i=0;i<rk;i++) put_u64(RK + (size_t)i*8, keys[mid+1+i]);
for (int i=0;i<=rk;i++) put_u64(RC + (size_t)i*8, kids[mid+1+i]); }
put_u16(buf + 10, (uint16_t)lk);
{ uint8_t* LK = buf + IDX_ENT_OFF; uint8_t* LC = buf + IDX_ENT_OFF + (size_t)lk*8;
for (int i=0;i<lk;i++) put_u64(LK + (size_t)i*8, keys[i]);
for (int i=0;i<=lk;i++) put_u64(LC + (size_t)i*8, kids[i]); }
free(keys); free(kids);
if (page_write(s, rid, rbuf)!=0) return -1;
if (page_write(s, page_id, buf)!=0) return -1;
*sep_key = up; *new_page = rid;
return 1;
}
static int btree_put(EngramPagedStore* s, int tree, uint64_t key, const uint8_t* payload){
uint64_t root = tree_root(s, tree);
uint64_t sep, np;
int sp = btree_insert(s, tree, root, key, payload, &sep, &np);
if (sp < 0) return -1;
if (sp == 0) return 0;
/* root split → new internal root */
uint64_t nr = page_alloc(s, STORE_PT_INDEX);
if (!nr) return -1;
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, nr, buf)!=0) return -1;
buf[IDX_LEAF_OFF] = 0;
put_u16(buf + 10, 1);
put_u64(buf + IDX_ENT_OFF, sep);
put_u64(buf + IDX_ENT_OFF + 8, root);
put_u64(buf + IDX_ENT_OFF + 16, np);
if (page_write(s, nr, buf)!=0) return -1;
tree_set_root(s, tree, nr);
return 0;
}
/* Collect every payload whose key == target. Payloads appended to *out (esz-pl
* bytes each, i.e. just the payload, pl bytes). Caller frees *out. */
static int btree_lookup(EngramPagedStore* s, int tree, uint64_t target,
uint8_t** out, size_t* count){
uint32_t pl = tree_payload(tree);
uint32_t esz = 8 + pl;
*out = NULL; *count = 0;
uint64_t id = tree_root(s, tree);
uint8_t buf[STORE_PAGE_SIZE];
/* descend to the leftmost leaf that could hold target */
for (;;){
if (page_read(s, id, buf)!=0) return -1;
if (buf[IDX_LEAF_OFF]) break;
int nkeys = get_u16(buf + 10);
uint8_t* K = buf + IDX_ENT_OFF;
uint8_t* C = buf + IDX_ENT_OFF + (size_t)nkeys*8;
/* Descend to the LEFTMOST child that may hold `target`. Duplicate keys
* span a contiguous run of leaves (a leaf split can leave copies of the
* separator value on both sides), so we must use strict '>' here — not
* the '>=' of insert routing — then scan forward across next_leaf. */
int ci = 0;
while (ci < nkeys && target > get_u64(K + (size_t)ci*8)) ci++;
id = get_u64(C + (size_t)ci*8);
}
Buf acc = {0,0,0};
for (;;){
int nkeys = get_u16(buf + 10);
int past = 0;
for (int i=0;i<nkeys;i++){
uint64_t k = get_u64(buf + IDX_ENT_OFF + (size_t)i*esz);
if (k < target) continue;
if (k > target){ past = 1; break; }
if (buf_raw(&acc, buf + IDX_ENT_OFF + (size_t)i*esz + 8, pl)){ free(acc.p); return -1; }
}
if (past) break;
uint64_t nxt = get_u64(buf + IDX_NEXT_OFF);
if (!nxt) break;
if (page_read(s, nxt, buf)!=0){ free(acc.p); return -1; }
}
*out = acc.p; *count = acc.len / pl;
return 0;
}
/* ── superblock ───────────────────────────────────────────────────────────── */
static int sb_write_one(EngramPagedStore* s, uint64_t page){
uint8_t buf[STORE_PAGE_SIZE];
memset(buf, 0, STORE_PAGE_SIZE);
memcpy(buf + SB_MAGIC_OFF, STORE_MAGIC, 8); /* 7 chars + NUL */
put_u32(buf + SB_FMT_OFF, s->format_version);
put_u32(buf + SB_PGSZ_OFF, s->page_size);
put_u64(buf + SB_PGCNT_OFF, s->page_count);
put_u64(buf + SB_FREE_OFF, s->free_list_head);
put_u64(buf + SB_ROOT_OFF, s->root_index_page);
put_u64(buf + SB_ADJ_OFF, s->adj_index_page);
put_u64(buf + SB_LAYER_OFF, s->layer_registry_page);
put_u64(buf + SB_CKPT_OFF, s->last_checkpoint_lsn);
put_u64(buf + SB_SEQ_OFF, s->sb_seq);
memcpy(buf + SB_UUID_OFF, s->uuid, 16);
put_u32(buf + SB_CRC_OFF, 0);
uint32_t crc = crc32_buf(buf, STORE_PAGE_SIZE);
put_u32(buf + SB_CRC_OFF, crc);
return page_write_raw(s, page, buf);
}
static int sb_load_one(EngramPagedStore* s, uint64_t page, uint8_t* buf, uint64_t* seq_out){
if (page_read(s, page, buf)!=0) return -1;
if (memcmp(buf + SB_MAGIC_OFF, STORE_MAGIC, 8)!=0) return -1;
uint32_t stored = get_u32(buf + SB_CRC_OFF);
uint8_t tmp[STORE_PAGE_SIZE];
memcpy(tmp, buf, STORE_PAGE_SIZE);
put_u32(tmp + SB_CRC_OFF, 0);
if (crc32_buf(tmp, STORE_PAGE_SIZE) != stored) return -1;
*seq_out = get_u64(buf + SB_SEQ_OFF);
return 0;
}
static void sb_apply(EngramPagedStore* s, const uint8_t* buf){
s->format_version = get_u32(buf + SB_FMT_OFF);
s->page_size = get_u32(buf + SB_PGSZ_OFF);
s->page_count = get_u64(buf + SB_PGCNT_OFF);
s->free_list_head = get_u64(buf + SB_FREE_OFF);
s->root_index_page = get_u64(buf + SB_ROOT_OFF);
s->adj_index_page = get_u64(buf + SB_ADJ_OFF);
s->layer_registry_page= get_u64(buf + SB_LAYER_OFF);
s->last_checkpoint_lsn= get_u64(buf + SB_CKPT_OFF);
s->sb_seq = get_u64(buf + SB_SEQ_OFF);
memcpy(s->uuid, buf + SB_UUID_OFF, 16);
}
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. */
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;
return 0;
}
/* ── lifecycle ────────────────────────────────────────────────────────────── */
static void gen_uuid(uint8_t out[16]){
int fd = open("/dev/urandom", O_RDONLY);
if (fd >= 0){ ssize_t r = read(fd, out, 16); close(fd); if (r == 16) return; }
uint64_t t = (uint64_t)time(NULL) ^ ((uint64_t)getpid() << 32);
for (int i=0;i<16;i++){ t = t*6364136223846793005ULL + 1442695040888963407ULL; out[i] = (uint8_t)(t>>56); }
}
static uint64_t idx_alloc_leaf(EngramPagedStore* s){
uint64_t id = page_alloc(s, STORE_PT_INDEX);
if (!id) return 0;
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, id, buf)!=0) return 0;
buf[IDX_LEAF_OFF] = 1;
put_u16(buf + 10, 0);
put_u64(buf + IDX_NEXT_OFF, 0);
if (page_write(s, id, buf)!=0) return 0;
return id;
}
EngramPagedStore* store_create(const char* path){
if (!s_crc_ready) crc_init();
struct stat st;
if (stat(path, &st) == 0){ errno = EEXIST; return NULL; }
EngramPagedStore* s = (EngramPagedStore*)calloc(1, sizeof *s);
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;
s->page_count = 2; /* pages 0,1 are the superblock + mirror */
s->free_list_head = 0;
s->next_lsn = 0;
s->sb_seq = 0;
gen_uuid(s->uuid);
/* reserve pages 0,1 on disk */
uint8_t zero[STORE_PAGE_SIZE]; memset(zero, 0, sizeof zero);
if (page_write_raw(s, 0, zero)!=0 || page_write_raw(s, 1, zero)!=0){ close(s->fd); free(s); return NULL; }
s->root_index_page = idx_alloc_leaf(s);
s->adj_index_page = idx_alloc_leaf(s);
s->layer_registry_page = page_alloc(s, STORE_PT_NODE); /* reserved; unused in M1 */
if (!s->root_index_page || !s->adj_index_page || !s->layer_registry_page){
close(s->fd); free(s); return NULL;
}
if (store_sync(s)!=0){ close(s->fd); free(s); return NULL; }
store__autopin(s); /* keep superblocks + index roots resident */
return s;
}
EngramPagedStore* store_open(const char* path){
if (!s_crc_ready) crc_init();
EngramPagedStore* s = (EngramPagedStore*)calloc(1, sizeof *s);
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;
int ok0 = sb_load_one(s, 0, b0, &s0) == 0;
int ok1 = sb_load_one(s, 1, b1, &s1) == 0;
if (!ok0 && !ok1){ close(s->fd); free(s); return NULL; }
const uint8_t* pick;
if (ok0 && ok1) pick = (s0 >= s1) ? b0 : b1;
else pick = ok0 ? b0 : b1;
sb_apply(s, pick);
if (s->page_size != STORE_PAGE_SIZE){ close(s->fd); free(s); return NULL; }
/* 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;
store__autopin(s); /* keep superblocks + index roots resident */
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;
}
/* ── put node/edge ────────────────────────────────────────────────────────── */
/* Build the on-page record (inline or overflow stub) from a TLV body. */
static int build_record(EngramPagedStore* s, const uint8_t* body, size_t blen,
uint8_t** rec_out, uint16_t* reclen_out){
if (blen <= MAX_INLINE_BODY){
uint16_t reclen = (uint16_t)(REC_HDR + blen);
uint8_t* rec = (uint8_t*)malloc(reclen);
if (!rec) return -1;
put_u16(rec, reclen); rec[2] = REC_VER; rec[3] = 0;
memcpy(rec + REC_HDR, body, blen);
*rec_out = rec; *reclen_out = reclen;
return 0;
}
uint64_t head = ovf_write_chain(s, body, blen);
if (!head) return -1;
uint16_t reclen = REC_HDR + 16;
uint8_t* rec = (uint8_t*)malloc(reclen);
if (!rec) return -1;
put_u16(rec, reclen); rec[2] = REC_VER; rec[3] = REC_OVERFLOW;
put_u64(rec + REC_HDR, head);
put_u64(rec + REC_HDR + 8, (uint64_t)blen);
*rec_out = rec; *reclen_out = reclen;
return 0;
}
/* Place a record into a NODE or EDGE page; returns (page_id,slot). Uses/creates
* the current page of that type; on-page dead-slot reuse handled by slp_put. */
static int place_record(EngramPagedStore* s, uint8_t ptype, const uint8_t* rec,
uint16_t reclen, uint64_t* page_id_out, uint16_t* slot_out){
uint64_t* cur = (ptype==STORE_PT_NODE) ? &s->cur_node_page : &s->cur_edge_page;
uint8_t buf[STORE_PAGE_SIZE];
if (*cur){
if (page_read(s, *cur, buf)==0){
int slot = slp_put(buf, rec, reclen);
if (slot >= 0){
if (page_write(s, *cur, buf)!=0) return -1;
*page_id_out = *cur; *slot_out = (uint16_t)slot; return 0;
}
}
}
uint64_t pid = page_alloc(s, ptype);
if (!pid) return -1;
if (page_read(s, pid, buf)!=0) return -1;
int slot = slp_put(buf, rec, reclen);
if (slot < 0) return -1; /* record too big for an empty page */
if (page_write(s, pid, buf)!=0) return -1;
*cur = pid;
*page_id_out = pid; *slot_out = (uint16_t)slot;
return 0;
}
/* 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;
uint8_t* rec; uint16_t reclen;
if (build_record(s, body, blen, &rec, &reclen)){ free(body); return -1; }
free(body);
uint64_t pid; uint16_t slot;
if (place_record(s, STORE_PT_NODE, rec, reclen, &pid, &slot)){ free(rec); return -1; }
free(rec);
uint8_t payload[PRIMARY_PAYLOAD];
put_u64(payload, pid); put_u16(payload + 8, slot);
return btree_put(s, TREE_PRIMARY, id_hash(n->id), payload);
}
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;
uint8_t* rec; uint16_t reclen;
if (build_record(s, body, blen, &rec, &reclen)){ free(body); return -1; }
free(body);
uint64_t pid; uint16_t slot;
if (place_record(s, STORE_PT_EDGE, rec, reclen, &pid, &slot)){ free(rec); return -1; }
free(rec);
/* primary index: edge id → loc */
uint8_t pp[PRIMARY_PAYLOAD];
put_u64(pp, pid); put_u16(pp + 8, slot);
if (btree_put(s, TREE_PRIMARY, id_hash(e->id), pp)!=0) return -1;
/* adjacency: from_id and to_id → edge loc */
uint8_t af[ADJ_PAYLOAD]; af[0]=ADJ_DIR_FROM; put_u64(af+1,pid); put_u16(af+9,slot);
if (btree_put(s, TREE_ADJ, id_hash(e->from_id), af)!=0) return -1;
uint8_t at[ADJ_PAYLOAD]; at[0]=ADJ_DIR_TO; put_u64(at+1,pid); put_u16(at+9,slot);
if (btree_put(s, TREE_ADJ, id_hash(e->to_id), at)!=0) return -1;
return 0;
}
/* ── read a record body (resolving overflow) at (page,slot). Returns malloc'd
* body + length, and whether the slot was live. ─────────────────────────── */
static int read_body(EngramPagedStore* s, uint64_t page, uint16_t slot,
uint8_t** body_out, size_t* blen_out, int* live_out){
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, page, buf)!=0) return -1;
if (slot >= slp_count(buf)) return -1;
uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl);
*live_out = (fl == SLOT_LIVE);
if (len < REC_HDR) return -1;
/* Defensive: the slot's (off,len) come from on-disk bytes. A stale primary
* index entry (churn/crash can leave one pointing at a page later repurposed)
* or a torn slot dir can yield an off/len that runs past this 16 KB stack page
* buffer — buf[off+..] would then read off the stack (observed EXC_BAD_ACCESS
* via store_get_node on the bloated store). Bound the record to the page and
* fail safe rather than over-read. */
if ((size_t)off + REC_HDR > STORE_PAGE_SIZE || (size_t)off + len > STORE_PAGE_SIZE)
return -1;
uint8_t rec_flags = buf[off + 3];
if (rec_flags & REC_OVERFLOW){
if ((size_t)off + REC_HDR + 16 > STORE_PAGE_SIZE) return -1; /* head+total u64s */
uint64_t head = get_u64(buf + off + REC_HDR);
uint64_t total = get_u64(buf + off + REC_HDR + 8);
uint8_t* body = ovf_read_chain(s, head, (size_t)total);
if (!body) return -1;
*body_out = body; *blen_out = (size_t)total;
} else {
uint16_t reclen = get_u16(buf + off);
if (reclen < REC_HDR || (size_t)off + reclen > STORE_PAGE_SIZE) return -1;
size_t blen = reclen - REC_HDR;
uint8_t* body = (uint8_t*)malloc(blen ? blen : 1);
if (!body) return -1;
memcpy(body, buf + off + REC_HDR, blen);
*body_out = body; *blen_out = blen;
}
return 0;
}
int store_get_node(EngramPagedStore* s, const char* id, StoreNode* out){
if (!s || !id || !out) return -1;
uint8_t* locs; size_t n;
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n)!=0) return -1;
int found = 0;
for (size_t i=0;i<n;i++){
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
uint8_t* body; size_t blen; int live;
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
StoreNode cand; node_parse(body, blen, &cand); free(body);
if (cand.id && strcmp(cand.id, id)==0){
if (live){ /* latest live wins (later entries) */
if (found) store_node_free(out);
*out = cand; out->tombstoned = 0; found = 1;
} else store_node_free(&cand);
} else store_node_free(&cand);
}
free(locs);
return found ? 1 : 0;
}
/* 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;
uint64_t prev_stamp = s->stamp_lsn;
s->stamp_lsn = lsn;
for (size_t i=0;i<n;i++){
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, page, buf)!=0) continue;
if (get_u64(buf + 16) >= 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;
/* verify id matches this record before killing it */
uint8_t* body; size_t blen; int live;
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
StoreNode cand; node_parse(body, blen, &cand); free(body);
int match = cand.id && strcmp(cand.id, id)==0;
store_node_free(&cand);
if (!match) continue;
slp_set_slot(buf, slot, off, len, SLOT_DEAD);
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;
if (s->cur_edge_page == page) s->cur_edge_page = 0;
page_free(s, page);
}
}
s->stamp_lsn = prev_stamp;
free(locs);
return 0; /* absent id is a no-op success */
}
static int get_edges_dir(EngramPagedStore* s, const char* id, uint8_t want_dir,
StoreEdge** out, size_t* n){
*out = NULL; *n = 0;
uint8_t* locs; size_t cnt;
if (btree_lookup(s, TREE_ADJ, id_hash(id), &locs, &cnt)!=0) return -1;
StoreEdge* arr = NULL; size_t used = 0, cap = 0;
for (size_t i=0;i<cnt;i++){
uint8_t dir = locs[i*ADJ_PAYLOAD];
if (dir != want_dir) continue;
uint64_t page = get_u64(locs + i*ADJ_PAYLOAD + 1);
uint16_t slot = get_u16(locs + i*ADJ_PAYLOAD + 9);
uint8_t* body; size_t blen; int live;
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
if (!live){ free(body); continue; }
StoreEdge e; edge_parse(body, blen, &e); free(body);
const char* side = (want_dir==ADJ_DIR_FROM) ? e.from_id : e.to_id;
if (!side || strcmp(side, id)!=0){ store_edge_free(&e); continue; }
if (used == cap){
cap = cap ? cap*2 : 8;
StoreEdge* na = (StoreEdge*)realloc(arr, cap*sizeof *na);
if (!na){ store_edge_free(&e); break; }
arr = na;
}
arr[used++] = e;
}
free(locs);
*out = arr; *n = used;
return 0;
}
int store_get_edges_from(EngramPagedStore* s, const char* from_id, StoreEdge** out, size_t* n){
if (!s || !from_id) return -1;
return get_edges_dir(s, from_id, ADJ_DIR_FROM, out, n);
}
int store_get_edges_to(EngramPagedStore* s, const char* to_id, StoreEdge** out, size_t* n){
if (!s || !to_id) return -1;
return get_edges_dir(s, to_id, ADJ_DIR_TO, out, n);
}
/* ── integrity ────────────────────────────────────────────────────────────── */
int store_check(EngramPagedStore* s, unsigned flags){
if (!s) return -1;
int bad = 0;
uint8_t buf[STORE_PAGE_SIZE];
for (uint64_t id = 0; id < s->page_count; id++){
if (page_read(s, id, buf)!=0) return -1;
if (id == 0 || id == 1){
uint32_t stored = get_u32(buf + SB_CRC_OFF);
uint8_t tmp[STORE_PAGE_SIZE]; memcpy(tmp, buf, STORE_PAGE_SIZE);
put_u32(tmp + SB_CRC_OFF, 0);
int okmagic = memcmp(buf + SB_MAGIC_OFF, STORE_MAGIC, 8)==0;
if (!okmagic || crc32_buf(tmp, STORE_PAGE_SIZE) != stored) bad++;
} else {
if (flags & STORE_CHECK_CRC){ if (!page_crc_ok(buf)) bad++; }
}
}
return bad;
}
/* ── ownership helpers ────────────────────────────────────────────────────── */
void store_node_free(StoreNode* n){
if (!n) return;
free(n->id); free(n->content); free(n->node_type); free(n->label);
free(n->tier); free(n->tags); free(n->metadata); free(n->emb); free(n->unknown);
memset(n, 0, sizeof *n);
}
void store_edge_free(StoreEdge* e){
if (!e) return;
free(e->id); free(e->from_id); free(e->to_id); free(e->relation);
free(e->metadata); free(e->unknown);
memset(e, 0, sizeof *e);
}
void store_edges_free(StoreEdge* arr, size_t n){
if (!arr) return;
for (size_t i=0;i<n;i++) store_edge_free(&arr[i]);
free(arr);
}
void store__set_btree_order(EngramPagedStore* s, int leaf_max, int internal_max){
if (!s) return;
s->leaf_max = leaf_max; s->int_max = internal_max;
}
uint64_t store_page_count(const EngramPagedStore* s){ return s ? s->page_count : 0; }
/* ── M3: full live enumeration (boundary-clean; StoreNode/StoreEdge out only) ──
* Page-walk every NODE/EDGE page, emitting each DISTINCT live record. A re-put
* leaves several live records for one id (apply_node_put appends; reads dedup),
* so we track ids already emitted and fetch the canonical latest-live via the
* point-read path so a scan and a get agree exactly. Used by the caller
* (el_runtime) to load the whole store resident at boot and to export JSON.
*
* DEDUP IS BY FULL ID STRING, NOT BY id_hash. (2026-08-12 self-review — the
* "saved but not findable" bug.) The dedup set formerly keyed on the 64-bit
* id_hash alone; two DISTINCT ids that collide under FNV-1a-64 therefore
* emitted only the first, and the second — durably on a live page and findable
* by store_get_node, which disambiguates by strcmp — was SILENTLY DROPPED from
* the resident boot-load. After any reopen it was unretrievable by id, absent
* from lexical search, and missing from the recent list. The primary B+-tree
* keys on id_hash too, but every reader there re-reads the record and strcmp's
* the id; the scan's dedup must apply the same full-id discipline. Keyed on the
* hash for O(1) bucketing, compared by strcmp for correctness. */
typedef struct { char* key; } StrSlot;
typedef struct { StrSlot* t; size_t n, cap; } StrSet;
static int strset_add(StrSet* s, const char* id){ /* 1 = newly added, 0 = present */
if (!id) return 1;
if ((s->n + 1) * 4 >= s->cap * 3){
size_t nc = s->cap ? s->cap * 2 : 1024;
StrSlot* nt = (StrSlot*)calloc(nc, sizeof(StrSlot));
if (!nt) return 1; /* degrade rather than crash */
for (size_t i = 0; i < s->cap; i++){
char* k = s->t[i].key;
if (k){ size_t j = id_hash(k) & (nc - 1); while (nt[j].key) j = (j + 1) & (nc - 1); nt[j].key = k; }
}
free(s->t); s->t = nt; s->cap = nc;
}
size_t j = id_hash(id) & (s->cap - 1);
while (s->t[j].key){ if (strcmp(s->t[j].key, id) == 0) return 0; j = (j + 1) & (s->cap - 1); }
s->t[j].key = strdup(id);
if (!s->t[j].key) return 1; /* OOM: don't dedup, never drop */
s->n++; return 1;
}
static void strset_free(StrSet* s){
for (size_t i = 0; i < s->cap; i++) free(s->t[i].key);
free(s->t); s->t = NULL; s->n = s->cap = 0;
}
int store_scan_nodes(EngramPagedStore* s, StoreNodeScanCb cb, void* ctx){
if (!s || !cb) return -1;
StrSet seen = {0, 0, 0};
uint8_t buf[STORE_PAGE_SIZE];
int count = 0;
for (uint64_t pg = 2; pg < s->page_count; pg++){
if (page_read(s, pg, buf) != 0) continue;
if (s->cache) pc_prefetch(s, pg, s->cache->prefetch); /* sequential read-ahead */
if (buf[8] != STORE_PT_NODE) continue;
int ns = slp_count(buf);
for (int i = 0; i < ns; i++){
uint16_t off, len, fl; slp_slot(buf, i, &off, &len, &fl);
if (fl != SLOT_LIVE) continue;
uint8_t* body; size_t blen; int live;
if (read_body(s, pg, (uint16_t)i, &body, &blen, &live) != 0) continue;
StoreNode cand; node_parse(body, blen, &cand); free(body);
if (cand.id && *cand.id && strset_add(&seen, cand.id)){
StoreNode canon;
if (store_get_node(s, cand.id, &canon) == 1){
cb(&canon, ctx); count++; /* canonical latest-live */
store_node_free(&canon);
}
}
store_node_free(&cand);
}
}
strset_free(&seen);
return count;
}
int store_scan_edges(EngramPagedStore* s, StoreEdgeScanCb cb, void* ctx){
if (!s || !cb) return -1;
StrSet seen = {0, 0, 0};
uint8_t buf[STORE_PAGE_SIZE];
int count = 0;
for (uint64_t pg = 2; pg < s->page_count; pg++){
if (page_read(s, pg, buf) != 0) continue;
if (s->cache) pc_prefetch(s, pg, s->cache->prefetch); /* sequential read-ahead */
if (buf[8] != STORE_PT_EDGE) continue;
int ns = slp_count(buf);
for (int i = 0; i < ns; i++){
uint16_t off, len, fl; slp_slot(buf, i, &off, &len, &fl);
if (fl != SLOT_LIVE) continue;
uint8_t* body; size_t blen; int live;
if (read_body(s, pg, (uint16_t)i, &body, &blen, &live) != 0) continue;
StoreEdge cand; edge_parse(body, blen, &cand); free(body);
if (cand.id && *cand.id && strset_add(&seen, cand.id)){
StoreEdge canon;
if (store_get_edge(s, cand.id, &canon) == 1){
cb(&canon, ctx); count++; /* canonical latest-live */
store_edge_free(&canon);
}
}
store_edge_free(&cand);
}
}
strset_free(&seen);
return count;
}
/* ══════════════════════════════════════════════════════════════════════════════
* 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 neuron.egm 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 <sys/time.h>
/* ══════════════════════════════════════════════════════════════════════════════
* M4 — demand-paging BUFFER POOL (bounded, LRU, pinned, read-ahead)
*
* A frame table (id→frame hash) capped at `cap` resident frames. On a page
* access that is not resident, page_read faults it in from neuron.egm; if the
* pool is full, the LRU eviction path reclaims a CLEAN, unpinned frame. This is
* purely additive residency — the on-disk format is unchanged, and with the
* DEFAULT cap (large) no eviction ever fires, so behaviour is byte-for-byte the
* Phase-1 resident store.
*
* Invariants preserved from M2 (write-back, NO-STEAL):
* • A DIRTY frame is NEVER evicted (never stolen) — its only durable copy is
* the fsync'd WAL, and the store page reaches disk solely at a checkpoint.
* pc_flush (checkpoint) is what turns dirty→clean and thus evictable.
* • A PINNED frame is never evicted. Structural pages are auto-pinned: the two
* superblocks (pages 0,1) and every index ROOT/INTERIOR page (type INDEX,
* leaf-flag 0). Leaves are pageable. Explicit pins (pin count) cover hot
* layers and any caller-designated page.
* Correctness under a pool SMALLER than the store rests on: every caller copies
* page bytes into a local stack buffer (memcpy in page_read / out in page_write)
* and never retains a frame pointer across another page access, so a frame may
* be evicted and later re-faulted with no aliasing hazard. A clean frame always
* matches disk, so a re-fault reproduces identical bytes.
* ════════════════════════════════════════════════════════════════════════════ */
/* default frame budget: large enough that today's whole store stays resident
* (== Phase 1). Override with env ENGRAM_POOL_FRAMES (0 = unlimited). */
#ifndef ENGRAM_POOL_FRAMES_DEFAULT
#define ENGRAM_POOL_FRAMES_DEFAULT (1u<<20) /* ~1M frames × 16KiB = 16 GiB */
#endif
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; }
c->cap = ENGRAM_POOL_FRAMES_DEFAULT;
c->prefetch = 8;
const char* pf = getenv("ENGRAM_POOL_FRAMES");
if (pf && *pf){ char* end=NULL; unsigned long long v = strtoull(pf,&end,10); c->cap = (size_t)v; }
const char* pw = getenv("ENGRAM_PREFETCH");
if (pw && *pw){ char* end=NULL; unsigned long v = strtoul(pw,&end,10); c->prefetch = (unsigned)v; }
return c;
}
static void pc_free(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; }
}
for (size_t i=0;i<c->lp_n;i++) free(c->lp[i].pages);
free(c->lp);
free(c->buckets); free(c);
}
/* ── LRU recency list (front = MRU, back = LRU) ─────────────────────────────── */
static void lru_unlink(PgCache* c, PgEnt* e){
if (e->lru_prev) e->lru_prev->lru_next = e->lru_next; else c->mru = e->lru_next;
if (e->lru_next) e->lru_next->lru_prev = e->lru_prev; else c->lru = e->lru_prev;
e->lru_prev = e->lru_next = NULL;
}
static void lru_push_front(PgCache* c, PgEnt* e){
e->lru_prev = NULL; e->lru_next = c->mru;
if (c->mru) c->mru->lru_prev = e; c->mru = e;
if (!c->lru) c->lru = e;
}
static void lru_touch(PgCache* c, PgEnt* e){
if (c->mru == e) return;
lru_unlink(c, e); lru_push_front(c, e);
}
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;i<c->nbuckets;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;
}
/* A frame is EVICTABLE iff it is clean, unpinned, not a superblock, and not an
* index root/interior page. This is the sole place the no-steal + structural-pin
* policy is enforced. */
static int pc_evictable(const PgEnt* e){
if (e->dirty) return 0; /* no-steal: dirty pages are pinned to RAM */
if (e->pin > 0) return 0; /* explicit / hot-layer pin */
if (e->id == 0 || e->id == 1) return 0; /* superblock + mirror */
if (e->buf[8] == STORE_PT_INDEX && e->buf[IDX_LEAF_OFF] == 0) return 0; /* root/interior */
return 1;
}
/* Detach `e` from both the hash chain and the recency list, and free it. */
static void pc_remove(PgCache* c, PgEnt* e){
size_t b = e->id % c->nbuckets;
PgEnt** pp = &c->buckets[b];
while (*pp && *pp != e) pp = &(*pp)->next;
if (*pp == e) *pp = e->next;
lru_unlink(c, e);
free(e->buf); free(e);
c->count--;
}
/* Reclaim clean unpinned frames from the LRU end until under budget, or until no
* evictable frame remains (a dirty/pinned-heavy pool may transiently exceed cap —
* that is the no-steal guarantee, not a bug: the next checkpoint frees them). */
static void pc_evict_to_budget(PgCache* c){
if (!c->cap) return; /* unlimited */
while (c->count > c->cap){
PgEnt* e = c->lru; int freed = 0;
while (e){
PgEnt* prev = e->lru_prev; /* walk LRU→MRU */
if (pc_evictable(e)){ pc_remove(c, e); c->evictions++; freed = 1; break; }
e = prev;
}
if (!freed) break; /* nothing evictable — allowed to exceed cap */
}
}
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){ lru_touch(c, e); return e; } e=e->next; }
return NULL;
}
/* Insert-or-update a frame. New frames go to MRU; then evict down to budget.
* The just-touched frame is at MRU and can never be the eviction victim. */
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); /* pc_get also bumps it to MRU on a hit */
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++;
lru_push_front(c, e);
pc_maybe_grow(c);
}
memcpy(e->buf, buf, STORE_PAGE_SIZE);
e->lsn = get_u64(buf + 16);
if (dirty){ if (!e->dirty) c->dirty_count++; e->dirty = 1; } /* clean→dirty transition */
pc_evict_to_budget(c);
return 0;
}
static int pc_flush(EngramPagedStore* s){
if (!s->cache) return 0;
PgCache* c = s->cache;
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. */
pc_evict_to_budget(c);
return 0;
}
/* Bounded sequential read-ahead: fault the next `window` pages after `from_id`
* into any spare capacity, so a forward scan/leaf-walk hits them instead of
* faulting one-by-one. Never forces an eviction (fills slack only), never
* re-reads a resident page. Prefetch reads are counted separately from demand
* faults so a scan's fault count reflects on-demand misses only. */
static void pc_prefetch(EngramPagedStore* s, uint64_t from_id, unsigned window){
PgCache* c = s->cache;
if (!c || !window) return;
for (unsigned k=1; k<=window; k++){
uint64_t id = from_id + k;
if (id >= s->page_count) break;
if (c->cap && c->count + 1 > c->cap) break; /* no eviction for read-ahead */
if (c->buckets[id % c->nbuckets]){
PgEnt* e = c->buckets[id % c->nbuckets];
int resident = 0; while (e){ if (e->id==id){ resident=1; break; } e=e->next; }
if (resident) continue;
}
uint8_t buf[STORE_PAGE_SIZE];
off_t off = (off_t)id * STORE_PAGE_SIZE;
if (pread(s->fd, buf, STORE_PAGE_SIZE, off) != (ssize_t)STORE_PAGE_SIZE) break;
pc_put(s, id, buf, 0);
c->prefetch_reads++;
}
}
/* Non-LRU-touching frame lookup (for pin bookkeeping that must not reorder). */
static PgEnt* pc_find(PgCache* c, uint64_t id){
PgEnt* e = c->buckets[id % c->nbuckets];
while (e){ if (e->id==id) return e; e=e->next; }
return NULL;
}
/* ── public pin / prefetch / stats API (M4) ─────────────────────────────────── */
int store_pin_page(EngramPagedStore* s, uint64_t page_id){
if (!s || !s->cache) return -1;
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, page_id, buf) != 0) return -1; /* fault in + make resident */
PgEnt* e = pc_find(s->cache, page_id);
if (!e) return -1;
e->pin++;
return 0;
}
int store_unpin_page(EngramPagedStore* s, uint64_t page_id){
if (!s || !s->cache) return -1;
PgEnt* e = pc_find(s->cache, page_id);
if (e && e->pin > 0) e->pin--;
return 0;
}
/* Pin every page currently holding a live record of `layer` (hot-layer residency).
* Records the pinned pages so store_unpin_layer releases exactly this set. Pages
* are pinned BEFORE their bodies are read so a small pool cannot evict them mid-scan. */
int store_pin_layer(EngramPagedStore* s, uint32_t layer){
if (!s || !s->cache) return -1;
uint64_t* pages = NULL; size_t np = 0, cap = 0;
uint8_t buf[STORE_PAGE_SIZE];
for (uint64_t pg = 2; pg < s->page_count; pg++){
if (page_read(s, pg, buf) != 0) continue;
int t = buf[8];
if (t != STORE_PT_NODE && t != STORE_PT_EDGE) continue;
PgEnt* pe = pc_find(s->cache, pg);
if (!pe) continue;
pe->pin++; /* provisional pin: keeps pg resident */
int ns = slp_count(buf), match = 0;
for (int i = 0; i < ns && !match; i++){
uint16_t off, len, fl; slp_slot(buf, i, &off, &len, &fl);
if (fl != SLOT_LIVE) continue;
uint8_t* body; size_t blen; int live;
if (read_body(s, pg, (uint16_t)i, &body, &blen, &live) != 0) continue;
uint32_t lid = 0;
if (t == STORE_PT_NODE){ StoreNode c; node_parse(body, blen, &c); lid = c.layer_id; store_node_free(&c); }
else { StoreEdge c; edge_parse(body, blen, &c); lid = c.layer_id; store_edge_free(&c); }
free(body);
if (lid == layer) match = 1;
}
if (match){
if (np == cap){ cap = cap ? cap*2 : 16; uint64_t* np2 = (uint64_t*)realloc(pages, cap*sizeof *pages); if (!np2){ free(pages); return -1; } pages = np2; }
pages[np++] = pg; /* keep the pin */
} else {
pe->pin--; /* no match on this page: drop provisional pin */
}
}
PgCache* c = s->cache;
if (c->lp_n == c->lp_cap){ c->lp_cap = c->lp_cap ? c->lp_cap*2 : 8; c->lp = (LayerPin*)realloc(c->lp, c->lp_cap*sizeof *c->lp); }
c->lp[c->lp_n].layer = layer; c->lp[c->lp_n].pages = pages; c->lp[c->lp_n].n = np; c->lp_n++;
return (int)np;
}
int store_unpin_layer(EngramPagedStore* s, uint32_t layer){
if (!s || !s->cache) return -1;
PgCache* c = s->cache;
for (size_t i = 0; i < c->lp_n; i++){
if (c->lp[i].layer != layer) continue;
for (size_t j = 0; j < c->lp[i].n; j++){
PgEnt* e = pc_find(c, c->lp[i].pages[j]);
if (e && e->pin > 0) e->pin--;
}
free(c->lp[i].pages);
c->lp[i] = c->lp[--c->lp_n]; /* swap-remove */
return 0;
}
return 0;
}
/* Auto-pin the structural pages: both superblocks and the two index roots (plus
* the layer registry). A SHALLOW index root is a LEAF, so it is not covered by
* the "index interior" eviction rule — pinning it explicitly guarantees the root
* is never evicted even for a tiny tree. Deeper roots/interiors are additionally
* covered by pc_evictable's INDEX-non-leaf rule. Best-effort (ignores errors on
* a not-yet-built store). */
static void store__autopin(EngramPagedStore* s){
if (!s || !s->cache) return;
store_pin_page(s, 0);
store_pin_page(s, 1);
if (s->root_index_page) store_pin_page(s, s->root_index_page);
if (s->adj_index_page) store_pin_page(s, s->adj_index_page);
if (s->layer_registry_page) store_pin_page(s, s->layer_registry_page);
}
/* Introspection + test hooks. */
void store_pool_stats(const EngramPagedStore* s, StorePoolStats* out){
if (!out) return;
memset(out, 0, sizeof *out);
if (!s || !s->cache) return;
const PgCache* c = s->cache;
out->cap = c->cap; out->resident = c->count; out->prefetch = c->prefetch;
out->hits = c->hits; out->misses = c->misses;
out->evictions = c->evictions; out->prefetch_reads = c->prefetch_reads;
size_t pinned = 0, dirty = 0;
for (size_t i=0;i<c->nbuckets;i++)
for (PgEnt* e=c->buckets[i]; e; e=e->next){
if (!pc_evictable(e)) pinned++;
if (e->dirty) dirty++;
}
out->pinned = pinned; out->dirty = dirty;
}
int store_pool_resident(const EngramPagedStore* s, uint64_t page_id){
if (!s || !s->cache) return -1;
return pc_find(s->cache, page_id) ? 1 : 0;
}
void store__set_pool_frames(EngramPagedStore* s, size_t frames){
if (!s || !s->cache) return;
s->cache->cap = frames;
pc_evict_to_budget(s->cache); /* apply the new budget now */
}
void store__set_prefetch(EngramPagedStore* s, unsigned window){
if (s && s->cache) s->cache->prefetch = window;
}
/* ── 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;
uint64_t bytes_since_reclaim; /* WAL bytes appended since last reclaim (M5) */
};
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++;
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();
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<n;i++){
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
uint8_t pg[STORE_PAGE_SIZE];
if (page_read(s, page, pg) != 0) continue;
uint8_t pt = pg[8];
if (want_edge && pt != STORE_PT_EDGE) continue;
if (!want_edge && pt != STORE_PT_NODE) 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;
uint64_t l = get_u64(pg + 16);
if (l > 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<n;i++){
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, page, buf) != 0) continue;
if (get_u64(buf + 16) >= 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;i<n;i++) store_layer_free(&arr[i]); free(arr);
}
/* Idempotency for layers: whole-registry-page LSN (all layers share one page). */
static int apply_layer_put(EngramPagedStore* s, const StoreLayer* L, uint64_t lsn){
if (!s || !L) return -1;
uint64_t pg = s->layer_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;i<nslot;i++){
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
if (fl != SLOT_LIVE) continue;
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
int hit = (c.layer_id == L->layer_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;i<nslot;i++){
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
if (fl != SLOT_LIVE) continue;
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
int hit = (c.layer_id == L->layer_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;i<nslot;i++){
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
if (fl != SLOT_LIVE) continue;
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
int hit = (c.layer_id == layer_id); store_layer_free(&c);
if (hit){ slp_set_slot(buf, i, off, len, SLOT_DEAD); changed=1; }
}
int rc = changed ? page_write(s, pg, buf) : 0;
s->stamp_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;i<nslot;i++){
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
if (fl != SLOT_LIVE) continue;
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
if (c.layer_id == layer_id){ if (found) store_layer_free(out); *out=c; found=1; }
else store_layer_free(&c);
}
return found ? 1 : 0;
}
int store_list_layers(EngramPagedStore* s, StoreLayer** out, size_t* n){
if (!s || !out || !n) return -1;
*out = NULL; *n = 0;
uint8_t buf[STORE_PAGE_SIZE];
if (page_read(s, s->layer_registry_page, buf) != 0) return -1;
StoreLayer* arr=NULL; size_t used=0, cap=0;
int nslot = slp_count(buf);
for (int i=0;i<nslot;i++){
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
if (fl != SLOT_LIVE) continue;
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
/* keep only the latest live per layer_id (supersede leaves one live) */
int dup=0;
for (size_t k=0;k<used;k++) if (arr[k].layer_id==c.layer_id){ dup=1; break; }
if (dup){ store_layer_free(&c); continue; }
if (used==cap){ cap=cap?cap*2:8; StoreLayer* na=(StoreLayer*)realloc(arr,cap*sizeof*na); if(!na){store_layer_free(&c);break;} arr=na; }
arr[used++]=c;
}
*out=arr; *n=used;
return 0;
}
/* ── edge lookup by id (latest live) ─────────────────────────────────────────── */
int store_get_edge(EngramPagedStore* s, const char* id, StoreEdge* out){
if (!s || !id || !out) return -1;
uint8_t* locs; size_t n;
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return -1;
int found = 0;
for (size_t i=0;i<n;i++){
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
uint8_t pg[STORE_PAGE_SIZE];
if (page_read(s, page, pg) != 0) continue;
if (pg[8] != STORE_PT_EDGE) continue;
uint8_t* body; size_t blen; int live;
if (read_body(s, page, slot, &body, &blen, &live) != 0) continue;
StoreEdge cand; edge_parse(body, blen, &cand); free(body);
if (cand.id && strcmp(cand.id, id)==0 && live){
if (found) store_edge_free(out);
*out = cand; out->tombstoned = 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<cnt;i++){
if (off + 4 > 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; id<file_pages; id++){
uint8_t buf[STORE_PAGE_SIZE];
off_t off = (off_t)id*STORE_PAGE_SIZE;
if (pread(s->fd, 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;
}
/* ── 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++;
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) ─────── */
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;i<n;i++){
uint32_t idl = (uint32_t)strlen(d[i].edge_id);
uint8_t t4[4]; put_u32(t4, idl); buf_raw(&b, t4, 4);
buf_raw(&b, d[i].edge_id, idl);
uint8_t t8[8]; uint64_t u; memcpy(&u,&d[i].hebb,8); put_u64(t8,u); buf_raw(&b, t8, 8);
put_u64(t8, (uint64_t)d[i].last_fired); buf_raw(&b, t8, 8);
}
if (s->wal){
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;
s->last_ckpt_ms = now_ms(); /* arm the interval trigger */
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 && cnt<STORE_BLL_K){ n.access_ts[cnt++] = (int64_t)strtoll(q,NULL,10);
const char* c=strchr(q,','); if(!c) break; q=c+1; }
n.access_filled = cnt; n.access_head = cnt % STORE_BLL_K;
}
free(ats);
/* emb: ".4g,.4g,..." → float vector (bit-exact vs strtof of the tokens) */
char* es = js_str_field(o, "emb");
if (es && *es){
int cap=16, cnt=0; float* v=(float*)malloc((size_t)cap*sizeof(float));
const char* q=es;
while (*q && v){ if(cnt==cap){ cap*=2; float* nv=(float*)realloc(v,(size_t)cap*sizeof(float)); if(!nv){free(v);v=NULL;break;} v=nv; }
v[cnt++] = strtof(q, NULL);
const char* c=strchr(q,','); if(!c) break; q=c+1; }
n.emb = v; n.emb_dim = v ? cnt : 0;
}
free(es);
store_put_node(s, &n);
store_node_free(&n);
}
/* edges */
const char* edges = js_member(root, "edges");
for (const char* o = js_array_first(edges); o; o = js_array_next(o)){
StoreEdge e; memset(&e, 0, sizeof e);
e.id = js_str_field(o, "id");
e.from_id = js_str_field(o, "from_id");
e.to_id = js_str_field(o, "to_id");
if (!e.id || !e.from_id || !e.to_id){ store_edge_free(&e); continue; }
e.relation = js_str_field(o, "relation");
e.metadata = js_str_field(o, "metadata");
e.weight = js_num_field(o, "weight");
e.hebb = js_num_field(o, "hebb"); /* absent ⇒ 0 (legacy emits only if >0) */
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;
}
/* 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;
char store_path[1200], wal_path[1200], snap_path[1200];
snprintf(store_path, sizeof store_path, "%s/neuron.egm", data_dir);
snprintf(wal_path, sizeof wal_path, "%s/neuron.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; }
engram__default_ckpt_policy(s);
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; }
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;
}
int engram_close(EngramPagedStore* s){
if (!s) return -1;
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;
}