/* 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 #include #include #include #include #include #include #include #include /* 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> 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=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;iaccess_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;iemb_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;kaccess_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;kemb[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){ /* keys*8 + (keys+1)*8 <= IDX_BODY → keys <= IDX_BODY/8 - 1 */ int nat = (int)(IDX_BODY / 8) - 1; 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]; 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;ici;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' 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 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; uint8_t rec_flags = buf[off + 3]; if (rec_flags & REC_OVERFLOW){ 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); 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;itombstoned = 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= 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;ipage_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;ileaf_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 by their 64-bit id-hash — the same key the * primary B+-tree uses (design §2.4) — 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. */ typedef struct { uint64_t* h; size_t n, cap; } U64Set; static int u64set_add(U64Set* s, uint64_t v){ /* 1 = newly added, 0 = present */ if ((s->n + 1) * 4 >= s->cap * 3){ size_t nc = s->cap ? s->cap * 2 : 1024; uint64_t* nh = (uint64_t*)calloc(nc, sizeof(uint64_t)); if (!nh) return 1; /* degrade rather than crash */ for (size_t i = 0; i < s->cap; i++){ uint64_t k = s->h[i]; if (k){ size_t j = k & (nc - 1); while (nh[j]) j = (j + 1) & (nc - 1); nh[j] = k; } } free(s->h); s->h = nh; s->cap = nc; } uint64_t k = v ? v : 1; /* 0 reserved as empty slot */ size_t j = k & (s->cap - 1); while (s->h[j]){ if (s->h[j] == k) return 0; j = (j + 1) & (s->cap - 1); } s->h[j] = k; s->n++; return 1; } int store_scan_nodes(EngramPagedStore* s, StoreNodeScanCb cb, void* ctx){ if (!s || !cb) return -1; U64Set 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 && u64set_add(&seen, id_hash(cand.id))){ StoreNode canon; if (store_get_node(s, cand.id, &canon) == 1){ cb(&canon, ctx); count++; store_node_free(&canon); } } store_node_free(&cand); } } free(seen.h); return count; } int store_scan_edges(EngramPagedStore* s, StoreEdgeScanCb cb, void* ctx){ if (!s || !cb) return -1; U64Set 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 && u64set_add(&seen, id_hash(cand.id))){ StoreEdge canon; if (store_get_edge(s, cand.id, &canon) == 1){ cb(&canon, ctx); count++; store_edge_free(&canon); } } store_edge_free(&cand); } } free(seen.h); 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 /* ══════════════════════════════════════════════════════════════════════════════ * 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;inbuckets;i++){ PgEnt* e = c->buckets[i]; while (e){ PgEnt* n=e->next; free(e->buf); free(e); e=n; } } for (size_t i=0;ilp_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;inbuckets;i++){ PgEnt* e = c->buckets[i]; while (e){ PgEnt* nx=e->next; size_t b=e->id%nn; e->next=nb[b]; nb[b]=e; e=nx; } } free(c->buckets); c->buckets=nb; c->nbuckets=nn; } /* 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;inbuckets;i++) for (PgEnt* e=c->buckets[i]; e; e=e->next) if (e->dirty){ if (page_write_raw(s, e->id, e->buf)!=0) return -1; e->dirty=0; } c->dirty_count = 0; /* all frames clean after flush */ /* Post-checkpoint the just-cleaned frames are now evictable; trim the pool * back to budget so a dirty-heavy burst that transiently overshot cap does * not leave the pool oversized. No-op at the default (unlimited-ish) cap. */ 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;inbuckets;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 best) best = l; } free(locs); return best; } /* Mark live record(s) for `id` DEAD where the home page LSN < lsn. `want_edge` * disambiguates node vs edge records (page type). Caller sets s->stamp_lsn. */ static void kill_live_id(EngramPagedStore* s, const char* id, int want_edge, uint64_t lsn){ uint8_t* locs; size_t n; if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return; for (size_t i=0;i= lsn) continue; if (buf[8] != (want_edge ? STORE_PT_EDGE : STORE_PT_NODE)) continue; if (slot >= slp_count(buf)) continue; uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl); if (fl != SLOT_LIVE) continue; uint8_t* body; size_t blen; int live; if (read_body(s, page, slot, &body, &blen, &live) != 0) continue; int match; if (want_edge){ StoreEdge e; edge_parse(body,blen,&e); match = e.id && !strcmp(e.id,id); store_edge_free(&e); } else { StoreNode nn; node_parse(body,blen,&nn); match = nn.id && !strcmp(nn.id,id); store_node_free(&nn); } free(body); if (!match) continue; slp_set_slot(buf, slot, off, len, SLOT_DEAD); page_write(s, page, buf); } free(locs); } /* ── apply routines (used by both the normal write path and WAL replay) ──────── */ static int apply_node_put(EngramPagedStore* s, const StoreNode* n, uint64_t lsn){ if (!s || !n || !n->id) return -1; if (max_page_lsn_for_id(s, n->id, 0) >= lsn) return 0; /* already durable */ uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; int r = node_place(s, n); /* re-put appends; reads dedup */ s->stamp_lsn = prev; return r; } static int apply_edge_put(EngramPagedStore* s, const StoreEdge* e, uint64_t lsn){ if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1; if (max_page_lsn_for_id(s, e->id, 1) >= lsn) return 0; /* already durable */ uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; kill_live_id(s, e->id, 1, lsn); /* supersede prior versions (no adjacency dup) */ int r = edge_place(s, e); s->stamp_lsn = prev; return r; } /* ── layer registry (single slotted page rooted at layer_registry_page) ──────── */ enum { LT_ID=1, LT_NAME, LT_PRIO, LT_SUPPRESS, LT_TRANSPARENT, LT_INJECTABLE }; static uint8_t* layer_serialize(const StoreLayer* L, size_t* out_len){ Buf b = {0,0,0}; tlv_u32(&b, LT_ID, L->layer_id); tlv_str(&b, LT_NAME, L->name); tlv_u32(&b, LT_PRIO, L->activation_priority); tlv_i32(&b, LT_SUPPRESS, L->suppressible); tlv_i32(&b, LT_TRANSPARENT, L->transparent); tlv_i32(&b, LT_INJECTABLE, L->injectable); if (L->unknown && L->unknown_len) buf_raw(&b, L->unknown, L->unknown_len); *out_len = b.len; return b.p; } static void layer_parse(const uint8_t* body, size_t len, StoreLayer* L){ memset(L, 0, sizeof *L); Buf unk = {0,0,0}; size_t i = 0; while (i + 5 <= len){ uint8_t tag = body[i]; uint32_t flen = get_u32(body+i+1); if (i + 5 + flen > len) break; const uint8_t* v = body + i + 5; switch (tag){ case LT_ID: L->layer_id = get_u32(v); break; case LT_NAME: L->name = dup_str(v, flen); break; case LT_PRIO: L->activation_priority = get_u32(v); break; case LT_SUPPRESS: L->suppressible = (int32_t)get_u32(v); break; case LT_TRANSPARENT: L->transparent = (int32_t)get_u32(v); break; case LT_INJECTABLE: L->injectable = (int32_t)get_u32(v); break; default: buf_raw(&unk, body+i, 5+flen); break; } i += 5 + flen; } L->unknown = unk.p; L->unknown_len = unk.len; } void store_layer_free(StoreLayer* L){ if (!L) return; free(L->name); free(L->unknown); memset(L, 0, sizeof *L); } void store_layers_free(StoreLayer* arr, size_t n){ if (!arr) return; for (size_t i=0;ilayer_registry_page; uint8_t buf[STORE_PAGE_SIZE]; if (page_read(s, pg, buf) != 0) return -1; /* already-applied? a live record with this layer_id and page LSN >= lsn. */ if (get_u64(buf + 16) >= lsn){ int nslot = slp_count(buf); for (int i=0;ilayer_id); store_layer_free(&c); if (hit) return 0; } } uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; /* supersede prior live record(s) with this layer_id */ int nslot = slp_count(buf); for (int i=0;ilayer_id); store_layer_free(&c); if (hit) slp_set_slot(buf, i, off, len, SLOT_DEAD); } /* append new layer record */ size_t blen; uint8_t* body = layer_serialize(L, &blen); if (!body){ s->stamp_lsn=prev; return -1; } uint16_t reclen = (uint16_t)(REC_HDR + blen); uint8_t* rec = (uint8_t*)malloc(reclen); if (!rec){ free(body); s->stamp_lsn=prev; return -1; } put_u16(rec, reclen); rec[2]=REC_VER; rec[3]=0; memcpy(rec + REC_HDR, body, blen); free(body); int slot = slp_put(buf, rec, reclen); free(rec); if (slot < 0){ s->stamp_lsn=prev; return -1; } /* registry page full (M2: single page) */ int rc = page_write(s, pg, buf); s->stamp_lsn = prev; return rc; } static int apply_layer_del(EngramPagedStore* s, uint32_t layer_id, uint64_t lsn){ uint64_t pg = s->layer_registry_page; uint8_t buf[STORE_PAGE_SIZE]; if (page_read(s, pg, buf) != 0) return -1; if (get_u64(buf + 16) >= lsn) return 0; uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn; int nslot = slp_count(buf), changed=0; for (int i=0;istamp_lsn = prev; return rc; } int store_get_layer(EngramPagedStore* s, uint32_t layer_id, StoreLayer* out){ if (!s || !out) return -1; uint8_t buf[STORE_PAGE_SIZE]; if (page_read(s, s->layer_registry_page, buf) != 0) return -1; int nslot = slp_count(buf), found=0; for (int i=0;ilayer_registry_page, buf) != 0) return -1; StoreLayer* arr=NULL; size_t used=0, cap=0; int nslot = slp_count(buf); for (int i=0;itombstoned = 0; found = 1; } else store_edge_free(&cand); } free(locs); return found ? 1 : 0; } /* ── hebb batch apply ──────────────────────────────────────────────────────── * payload: [u32 count] then count × [u32 id_len][id][f64 hebb][i64 last_fired]. */ static int apply_hebb_batch(EngramPagedStore* s, const uint8_t* p, uint32_t plen, uint64_t lsn){ if (plen < 4) return -1; uint32_t cnt = get_u32(p); size_t off = 4; for (uint32_t i=0;i plen) break; uint32_t idl = get_u32(p+off); off += 4; if (off + idl + 16 > plen) break; char* id = dup_str(p+off, idl); off += idl; double hebb; { uint64_t u=get_u64(p+off); memcpy(&hebb,&u,8);} off += 8; int64_t lf = (int64_t)get_u64(p+off); off += 8; if (id){ StoreEdge e; if (store_get_edge(s, id, &e) == 1){ e.hebb = hebb; e.last_fired = lf; apply_edge_put(s, &e, lsn); store_edge_free(&e); } free(id); } } return 0; } /* ── WAL replay dispatch ─────────────────────────────────────────────────────── */ static void wal_dispatch(EngramPagedStore* s, uint8_t op, const uint8_t* p, uint32_t plen, uint64_t lsn){ switch (op){ case OP_NODE_PUT: { StoreNode n; node_parse(p,plen,&n); apply_node_put(s,&n,lsn); store_node_free(&n); break; } case OP_EDGE_PUT: { StoreEdge e; edge_parse(p,plen,&e); apply_edge_put(s,&e,lsn); store_edge_free(&e); break; } case OP_TOMBSTONE: case OP_FORGET: { char* id=dup_str(p,plen); if(id){ tombstone_core(s,id,lsn); free(id);} break; } case OP_SUPERSEDE:{ if(plen>=4){ uint32_t ol=get_u32(p); if(4+ol<=plen){ char* old=dup_str(p+4,ol); if(old){ tombstone_core(s,old,lsn); free(old);} } } break; } case OP_LAYER_PUT:{ StoreLayer L; layer_parse(p,plen,&L); apply_layer_put(s,&L,lsn); store_layer_free(&L); break; } case OP_LAYER_DEL:{ if(plen>=4) apply_layer_del(s, get_u32(p), lsn); break; } case OP_HEBB_BATCH: apply_hebb_batch(s, p, plen, lsn); break; case OP_CHECKPOINT: break; default: break; } } static int wal_replay(EngramPagedStore* s){ EngramWal* w = s->wal; if (!w) return 0; struct stat st; if (fstat(w->fd, &st) != 0) return -1; off_t size = st.st_size, off = 0; uint64_t maxlsn = 0; for (;;){ if (off + (off_t)WAL_HDR > size) break; /* torn/short tail */ uint8_t hdr[WAL_HDR]; if (pread(w->fd, hdr, WAL_HDR, off) != (ssize_t)WAL_HDR) break; if (get_u32(hdr) != WAL_MAGIC) break; uint32_t plen = get_u32(hdr + 4); if (off + (off_t)WAL_HDR + plen > size) break; /* torn tail */ uint32_t crc_stored = get_u32(hdr + 18); uint8_t* fr = (uint8_t*)malloc(10 + plen); if (!fr) return -1; memcpy(fr, hdr + 8, 10); /* op|flags|lsn */ if (plen && pread(w->fd, fr+10, plen, off+WAL_HDR) != (ssize_t)plen){ free(fr); break; } if (crc32_buf(fr, 10+plen) != crc_stored){ free(fr); break; } /* bad crc → stop */ uint8_t op = fr[0]; uint64_t lsn = get_u64(fr+2); if (lsn > maxlsn) maxlsn = lsn; if (lsn > s->last_checkpoint_lsn) wal_dispatch(s, op, fr+10, plen, lsn); free(fr); off += (off_t)WAL_HDR + plen; } if (maxlsn > s->next_lsn) s->next_lsn = maxlsn; return 0; } /* Heal torn DATA pages (bad crc) to empty pages of their type, so redo re-derives * their records from the WAL. Index/superblock recovery uses the SB mirror + full * replay-from-checkpoint; torn index pages are left for store_check to report. */ static void heal_torn_pages(EngramPagedStore* s){ /* Scan by the ACTUAL file size, not the SB page_count: a steal (dirty pages * flushed without advancing the checkpoint SB) can leave valid/torn pages * beyond the recorded page_count. Bump page_count so redo allocations never * collide with those on-disk ghost pages. */ struct stat fst; uint64_t file_pages = (fstat(s->fd,&fst)==0) ? (uint64_t)(fst.st_size/STORE_PAGE_SIZE) : s->page_count; if (file_pages > s->page_count) s->page_count = file_pages; for (uint64_t id=2; idfd, buf, STORE_PAGE_SIZE, off) != (ssize_t)STORE_PAGE_SIZE) continue; if (page_crc_ok(buf)) continue; uint8_t t = buf[8]; uint8_t nb[STORE_PAGE_SIZE]; memset(nb, 0, sizeof nb); if (t==STORE_PT_NODE || t==STORE_PT_EDGE){ nb[8]=t; put_u16(nb+10,0); put_u16(nb+12,(uint16_t)(STORE_PAGE_SIZE-STORE_HDR)); } else if (t==STORE_PT_OVERFLOW){ nb[8]=t; put_u64(nb+OVF_NEXT_OFF,0); put_u32(nb+OVF_LEN_OFF,0); } else continue; /* don't heal index/free/garbage */ put_u64(nb+0, id); put_u64(nb+16, 0); put_u32(nb+24, 0); put_u32(nb+24, crc32_buf(nb, STORE_PAGE_SIZE)); pc_put(s, id, nb, 1); /* dirty → persisted next checkpoint */ } } static int wal_recover(EngramPagedStore* s){ s->recovering = 1; heal_torn_pages(s); int rc = wal_replay(s); s->recovering = 0; return rc; } /* ── 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;iwal){ if (wal_append(s, OP_HEBB_BATCH, b.p, (uint32_t)b.len, L) != 0){ free(b.p); return -1; } } int r = apply_hebb_batch(s, b.p, (uint32_t)b.len, L); free(b.p); ckpt_maybe(s); return r; } /* ── checkpoint (with test-only crash injection at each step) ─────────────────── */ int store__checkpoint_crashat(EngramPagedStore* s, int phase){ if (!s) return -1; if (phase == 0){ store__crash(s); return 0; } if (pc_flush(s) != 0) return -1; /* 1: dirty pages → disk */ if (phase == 1){ store__crash(s); return 0; } if (fsync(s->fd) != 0) return -1; /* 2: fsync store */ if (phase == 2){ store__crash(s); return 0; } uint64_t C = s->next_lsn; /* 3: advance ckpt LSN + SB */ s->last_checkpoint_lsn = C; s->sb_seq++; if (sb_write_one(s, 0) != 0) return -1; if (fsync(s->fd) != 0) return -1; if (sb_write_one(s, 1) != 0) return -1; if (fsync(s->fd) != 0) return -1; if (phase == 3){ store__crash(s); return 0; } if (wal_reclaim(s, C) != 0) return -1; /* 4: reclaim WAL prefix */ s->ops_since_ckpt = 0; 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 && cnt0) */ e.confidence = js_num_field(o, "confidence"); e.created_at = js_int_field(o, "created_at"); e.updated_at = js_int_field(o, "updated_at"); e.last_fired = js_int_field(o, "last_fired"); e.inhibitory = (int32_t)js_int_field(o, "inhibitory"); e.layer_id = (uint32_t)js_int_field(o, "layer_id"); store_put_edge(s, &e); store_edge_free(&e); } /* layers */ const char* layers = js_member(root, "layers"); for (const char* o = js_array_first(layers); o; o = js_array_next(o)){ StoreLayer L; memset(&L, 0, sizeof L); L.layer_id = (uint32_t)js_int_field(o, "layer_id"); L.name = js_str_field(o, "name"); L.activation_priority = (uint32_t)js_int_field(o, "activation_priority"); L.suppressible = (int32_t)js_int_field(o, "suppressible"); L.transparent = (int32_t)js_int_field(o, "transparent"); L.injectable = (int32_t)js_int_field(o, "injectable"); store_put_layer(s, &L); store_layer_free(&L); } free(txt); return 0; } /* 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 `.compact` holding ONLY the live records * — walked canonically (latest-live per id) and re-placed bit-exact into * fresh, densely packed pages with fresh id + adjacency B+-trees. Every page * is stamped with LSN = C and the new superblock records last_checkpoint_lsn * = C, so it is LSN-consistent with the (unchanged) WAL. fsync it. * C. Commit by rename(.compact → ) — POSIX-atomic: recovery sees * either the whole old file or the whole new file, never a torn mix. * D. Reopen in place: swap the fd, INVALIDATE every pool frame (old page ids now * hold different data — this is the M4 "remap relocated pages" step), reload * the superblock, re-autopin. * * Crash safety (proven by the test at phases 0/1/2): * • crash in A/B (before rename): old .egm is byte-for-byte intact and the WAL * still matches it → recovery = PRE-compaction (all live records present). * The half-built `.compact` temp is ignored by engram_open and unlinked at the * next compaction. * • crash after rename (C/D): the new .egm is fully fsync'd with last_checkpoint * = C and the WAL (CHECKPOINT{C}, nothing newer) matches it → recovery = * POST-compaction. No undo ever needed because relocation is copy-then-swap, * never in-place mutation of a still-referenced page. * * Online vs quiesce: the store is single-threaded, so "online" means it is safe * to interleave between mutations (each mutation is a synchronous call) — NOT that * it runs concurrently with one. It takes a checkpoint quiesce point at entry. * * M4 cooperation: the build writes into a SEPARATE store `d` whose own pool obeys * ENGRAM_POOL_FRAMES (so a small pool evicts/re-faults throughout the build, * no-steal + pins honoured there); the live store's pool is fully invalidated on * reopen, guaranteeing no stale frame maps a relocated page. * ════════════════════════════════════════════════════════════════════════════ */ /* Drop every resident frame (relocated pages are no longer valid) but keep the * pool object with its cap / prefetch / stats. Layer-pin bookkeeping is cleared * (those page ids belong to the old image). */ static void pc_invalidate_all(PgCache* c){ if (!c) return; for (size_t i=0;inbuckets;i++){ PgEnt* e = c->buckets[i]; while (e){ PgEnt* n=e->next; free(e->buf); free(e); e=n; } c->buckets[i] = NULL; } for (size_t i=0;ilp_n;i++) free(c->lp[i].pages); c->lp_n = 0; c->count = 0; c->dirty_count = 0; c->mru = c->lru = NULL; } /* Re-open the store file in place after an atomic swap: swap fd, invalidate the * pool, reload the superblock, resume the LSN watermark, re-autopin. Keeps the * attached WAL (it references the same checkpoint LSN the new file carries). */ static int store__reopen_swapped(EngramPagedStore* s){ if (s->fd >= 0) close(s->fd); s->fd = open(s->path, O_RDWR); if (s->fd < 0) return -1; pc_invalidate_all(s->cache); /* M4: no stale frame for a relocated page */ uint8_t b0[STORE_PAGE_SIZE], b1[STORE_PAGE_SIZE]; uint64_t s0=0, s1=0; int ok0 = sb_load_one(s, 0, b0, &s0) == 0; int ok1 = sb_load_one(s, 1, b1, &s1) == 0; if (!ok0 && !ok1) return -1; const uint8_t* pick = (ok0&&ok1) ? ((s0>=s1)?b0:b1) : (ok0?b0:b1); sb_apply(s, pick); s->next_lsn = (s->last_checkpoint_lsn > s->sb_seq) ? s->last_checkpoint_lsn : s->sb_seq; s->cur_node_page = 0; s->cur_edge_page = 0; s->ops_since_ckpt = 0; s->last_ckpt_ms = now_ms(); store__autopin(s); return 0; } /* Callback context for copying live records into the compacted store `d`. */ typedef struct { EngramPagedStore* d; int err; } CompactCtx; static void compact_node_cb(const StoreNode* n, void* ctx){ CompactCtx* c = (CompactCtx*)ctx; if (c->err) return; if (node_place(c->d, n) != 0) c->err = 1; /* bit-exact re-place; stamped d->stamp_lsn */ } static void compact_edge_cb(const StoreEdge* e, void* ctx){ CompactCtx* c = (CompactCtx*)ctx; if (c->err) return; if (edge_place(c->d, e) != 0) c->err = 1; } /* Build the compacted image (only live records, fresh indexes) into a new file. */ static int compact_build(EngramPagedStore* s, const char* tmp_path){ unlink(tmp_path); /* drop any temp from a crashed run */ EngramPagedStore* d = store_create(tmp_path); if (!d) return -1; uint64_t W = s->next_lsn; /* LSN watermark == checkpoint LSN */ memcpy(d->uuid, s->uuid, 16); /* preserve store identity */ d->last_checkpoint_lsn = W; d->next_lsn = W; d->stamp_lsn = W; /* every compacted page → LSN W */ int rc = 0; /* live layers */ StoreLayer* layers = NULL; size_t nlay = 0; if (store_list_layers(s, &layers, &nlay) == 0){ for (size_t i=0;istamp_lsn = 0; if (rc == 0 && store_sync(d) != 0) rc = -1; /* flush + fsync + both superblocks */ store_close(d); return rc; } int store__compact_crashat(EngramPagedStore* s, int phase){ if (!s) return -1; /* A. quiesce → on-disk store consistent, WAL reduced to its checkpoint marker */ if (s->wal){ if (engram_checkpoint(s) != 0) return -1; } else { if (store_sync(s) != 0) return -1; } if (phase == 0){ store__crash(s); return 0; } /* → recovers pre-compaction */ char tmp[1200]; snprintf(tmp, sizeof tmp, "%s.compact", s->path); if (compact_build(s, tmp) != 0){ unlink(tmp); return -1; } if (phase == 1){ store__crash(s); return 0; } /* built, not renamed → pre-compaction */ /* C. atomic commit */ if (rename(tmp, s->path) != 0){ unlink(tmp); return -1; } if (phase == 2){ store__crash(s); return 0; } /* renamed, not reopened → post-compaction */ /* D. reopen RAM state against the compacted file */ return store__reopen_swapped(s); } int store_compact(EngramPagedStore* s){ return store__compact_crashat(s, -1); } void store_set_checkpoint_policy(EngramPagedStore* s, uint64_t ops, size_t dirty_pages, uint64_t wal_bytes, long long interval_ms){ if (!s) return; s->ckpt_threshold = ops; s->ckpt_dirty_threshold = dirty_pages; s->ckpt_wal_threshold = wal_bytes; s->ckpt_interval_ms = interval_ms; s->last_ckpt_ms = now_ms(); } uint64_t store_free_page_count(const EngramPagedStore* s){ if (!s) return 0; uint64_t n = 0, id = s->free_list_head; uint8_t buf[STORE_PAGE_SIZE]; while (id){ if (page_read((EngramPagedStore*)s, id, buf) != 0) break; if (buf[8] != STORE_PT_FREE) break; n++; id = get_u64(buf + OVF_NEXT_OFF); } return n; }