will.anderson
8affb1d6e0
Write-back no-steal buffer pool makes the fsync'd WAL load-bearing (M1 was write-through). Logical WAL with record-granularity page-LSN redo idempotency. Checkpoint = flush dirty pages, fsync store, advance last_checkpoint_lsn, reclaim WAL prefix. One-time snapshot.json import only when store absent; JSON never read as the ongoing store thereafter. Gates: 33/33 M1 (no regression) + 36/36 M2 — replay parity, torn-tail fuzz (every byte offset), checkpoint-crash at all 5 phases, torn-page+WAL redo, legacy-import parity, hebb-survives-crash.
2059 lines
91 KiB
C
2059 lines
91 KiB
C
/* engram_store.c — M1 paged store: writer/reader for the FINAL on-disk format.
|
||
*
|
||
* Self-contained. Standard libs only. See engram_store.h and the design doc
|
||
* (engram-tiered-storage-engine.md §2/§4/§7). Write-through page I/O via
|
||
* pread/pwrite (no buffer pool yet — that is M4). Every mutation immediately
|
||
* lands on disk; store_sync/close rewrite both superblocks and fsync.
|
||
*
|
||
* On-disk layout (all integers little-endian):
|
||
* page 0,1 : superblock + mirror (recovery picks higher valid sb_seq)
|
||
* page N : generic 32-byte page header, then type-specific body
|
||
*
|
||
* Generic page header (32 bytes):
|
||
* +0 u64 page_id
|
||
* +8 u8 page_type (NODE|EDGE|INDEX|OVERFLOW|FREE)
|
||
* +9 u8 reserved
|
||
* +10 u16 slot_count (slotted pages) / nkeys (index pages)
|
||
* +12 u16 free_bytes (slotted pages: contiguous heap gap)
|
||
* +14 u16 reserved
|
||
* +16 u64 lsn (LSN of last write — ARIES redo idempotency, M2)
|
||
* +24 u32 crc32 (over the whole page with these 4 bytes zeroed)
|
||
* +28 u32 reserved
|
||
*
|
||
* Slotted page (NODE/EDGE): slot directory grows from +32 (6 bytes each:
|
||
* [u16 off][u16 len][u16 flags]); records grow down from PAGE_SIZE.
|
||
*
|
||
* On-page record: [u16 rec_len][u8 rec_ver][u8 rec_flags] then either the TLV
|
||
* field body (inline) or, if REC_OVERFLOW, a stub [u64 head_page][u64 total].
|
||
*
|
||
* TLV field: [u8 tag][u32 len][bytes]. NOTE: the design §2 sketch says u16 len;
|
||
* node content can exceed 64 KiB (tests use 100 KiB), so the field length MUST
|
||
* be u32. This is the one format decision beyond §2 — flagged in the report.
|
||
*
|
||
* Overflow page: header, then +32 [u64 next_page][u32 chunk_len][payload...].
|
||
*
|
||
* Index page (B+-tree, u64 keys, duplicates allowed, fixed payload per tree):
|
||
* +32 u8 is_leaf; +36 u64 next_leaf; +44 entries.
|
||
* leaf entry = [u64 key][payload]; payload = primary(10) or adjacency(11).
|
||
* internal = nkeys u64 keys, then (nkeys+1) u64 child page ids.
|
||
*/
|
||
#include "engram_store.h"
|
||
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include <stdint.h>
|
||
#include <unistd.h>
|
||
#include <fcntl.h>
|
||
#include <errno.h>
|
||
#include <time.h>
|
||
#include <sys/stat.h>
|
||
|
||
/* STORE_BLL_K must track ENGRAM_BLL_K in el_runtime.c. */
|
||
typedef char store__bll_k_check[(STORE_BLL_K == 10) ? 1 : -1];
|
||
|
||
#define STORE_HDR 32u /* generic page header size */
|
||
#define SLOT_SIZE 6u /* [u16 off][u16 len][u16 flags] */
|
||
#define SLOT_LIVE 1u
|
||
#define SLOT_DEAD 2u
|
||
|
||
#define REC_HDR 4u /* [u16 rec_len][u8 ver][u8 flags] */
|
||
#define REC_VER 1u
|
||
#define REC_OVERFLOW 1u /* rec_flags bit0: body is in an overflow chain */
|
||
|
||
#define MAX_INLINE_BODY (STORE_PAGE_SIZE - STORE_HDR - SLOT_SIZE - REC_HDR)
|
||
|
||
/* Index page geometry. */
|
||
#define IDX_LEAF_OFF 32u
|
||
#define IDX_NEXT_OFF 36u
|
||
#define IDX_ENT_OFF 44u
|
||
#define IDX_BODY (STORE_PAGE_SIZE - IDX_ENT_OFF)
|
||
|
||
/* Overflow page geometry. */
|
||
#define OVF_NEXT_OFF 32u
|
||
#define OVF_LEN_OFF 40u
|
||
#define OVF_DATA_OFF 44u
|
||
#define OVF_CHUNK (STORE_PAGE_SIZE - OVF_DATA_OFF)
|
||
|
||
/* Superblock layout (page 0 and mirror page 1). */
|
||
#define SB_MAGIC_OFF 0u
|
||
#define SB_FMT_OFF 8u
|
||
#define SB_PGSZ_OFF 12u
|
||
#define SB_PGCNT_OFF 16u
|
||
#define SB_FREE_OFF 24u
|
||
#define SB_ROOT_OFF 32u
|
||
#define SB_ADJ_OFF 40u
|
||
#define SB_LAYER_OFF 48u
|
||
#define SB_CKPT_OFF 56u
|
||
#define SB_SEQ_OFF 64u
|
||
#define SB_UUID_OFF 72u /* 16 bytes */
|
||
#define SB_CRC_OFF 88u
|
||
|
||
/* Which index tree. */
|
||
enum { TREE_PRIMARY = 0, TREE_ADJ = 1 };
|
||
#define PRIMARY_PAYLOAD 10u /* [u64 page_id][u16 slot] */
|
||
#define ADJ_PAYLOAD 11u /* [u8 dir][u64 page_id][u16 slot] */
|
||
#define ADJ_DIR_FROM 0u
|
||
#define ADJ_DIR_TO 1u
|
||
#define MAX_PAYLOAD 16u
|
||
|
||
/* Node/Edge TLV tag namespaces (per-kind; page type also disambiguates). */
|
||
enum {
|
||
NT_ID = 1, NT_CONTENT, NT_NODE_TYPE, NT_LABEL, NT_TIER, NT_TAGS, NT_METADATA,
|
||
NT_SALIENCE, NT_IMPORTANCE, NT_CONFIDENCE, NT_DECAY,
|
||
NT_ACT_COUNT, NT_LAST_ACT, NT_CREATED, NT_UPDATED,
|
||
NT_BG_ACT, NT_WM_WEIGHT, NT_SUPPRESS, NT_LAYER,
|
||
NT_ACCESS_TS, NT_ACCESS_HEAD, NT_ACCESS_FILLED, NT_WM_ANCHOR,
|
||
NT_EMB, NT_EMB_DIM
|
||
};
|
||
enum {
|
||
ET_ID = 1, ET_FROM, ET_TO, ET_RELATION, ET_METADATA,
|
||
ET_WEIGHT, ET_HEBB, ET_CONFIDENCE, ET_CREATED, ET_UPDATED, ET_LAST_FIRED,
|
||
ET_INHIBITORY, ET_LAYER
|
||
};
|
||
|
||
/* ── M2 forward decls (buffer pool + WAL live in the M2 section below) ─────── */
|
||
typedef struct PgCache PgCache;
|
||
typedef struct EngramWal EngramWal;
|
||
|
||
struct EngramPagedStore {
|
||
int fd;
|
||
char path[1024];
|
||
uint32_t format_version;
|
||
uint32_t page_size;
|
||
uint64_t page_count;
|
||
uint64_t free_list_head; /* page id of first FREE page, 0 = none */
|
||
uint64_t root_index_page; /* primary id->loc B+-tree root */
|
||
uint64_t adj_index_page; /* from/to adjacency B+-tree root */
|
||
uint64_t layer_registry_page;
|
||
uint64_t last_checkpoint_lsn; /* store on disk reflects all ops with lsn<=this */
|
||
uint64_t sb_seq;
|
||
uint8_t uuid[16];
|
||
uint64_t next_lsn; /* monotonic LSN counter (WAL + page stamp share it) */
|
||
uint64_t cur_node_page; /* last NODE page with room, 0 = none */
|
||
uint64_t cur_edge_page; /* last EDGE page with room, 0 = none */
|
||
int leaf_max; /* test hook; 0 = natural */
|
||
int int_max;
|
||
/* ── M2 additions ─────────────────────────────────────────────────────── */
|
||
PgCache* cache; /* write-back buffer pool (no-steal) */
|
||
EngramWal* wal; /* attached WAL, or NULL (M1 direct mode) */
|
||
uint64_t stamp_lsn; /* LSN to stamp on the next page_write (0 = auto) */
|
||
int recovering; /* set during WAL replay */
|
||
uint64_t ops_since_ckpt; /* checkpoint threshold counter */
|
||
uint64_t ckpt_threshold; /* auto-checkpoint after this many ops (0 = never) */
|
||
};
|
||
|
||
/* M2 buffer-pool hooks (defined in the M2 section at the bottom of this file). */
|
||
typedef struct PgEnt { uint64_t id; uint8_t* buf; uint64_t lsn; int dirty; struct PgEnt* next; } PgEnt;
|
||
static PgCache* pc_new(void);
|
||
static void pc_free(PgCache* c);
|
||
static PgEnt* pc_get(EngramPagedStore* s, uint64_t id);
|
||
static int pc_put(EngramPagedStore* s, uint64_t id, const uint8_t* buf, int dirty);
|
||
static int pc_flush(EngramPagedStore* s); /* pwrite all dirty → clean */
|
||
|
||
/* ── little-endian scalar codecs ──────────────────────────────────────────── */
|
||
static void put_u16(uint8_t* p, uint16_t v){ p[0]=(uint8_t)v; p[1]=(uint8_t)(v>>8); }
|
||
static void put_u32(uint8_t* p, uint32_t v){ for(int i=0;i<4;i++) p[i]=(uint8_t)(v>>(8*i)); }
|
||
static void put_u64(uint8_t* p, uint64_t v){ for(int i=0;i<8;i++) p[i]=(uint8_t)(v>>(8*i)); }
|
||
static uint16_t get_u16(const uint8_t* p){ return (uint16_t)(p[0] | (p[1]<<8)); }
|
||
static uint32_t get_u32(const uint8_t* p){ uint32_t v=0; for(int i=0;i<4;i++) v|=(uint32_t)p[i]<<(8*i); return v; }
|
||
static uint64_t get_u64(const uint8_t* p){ uint64_t v=0; for(int i=0;i<8;i++) v|=(uint64_t)p[i]<<(8*i); return v; }
|
||
static void put_f64(uint8_t* p, double d){ uint64_t u; memcpy(&u,&d,8); put_u64(p,u); }
|
||
static double get_f64(const uint8_t* p){ uint64_t u=get_u64(p); double d; memcpy(&d,&u,8); return d; }
|
||
|
||
/* ── crc32 (IEEE 802.3, reflected, poly 0xEDB88320); matches eg_crc32 in
|
||
* el_runtime.c so integrity checks are consistent cross-module.
|
||
* crc32("") == 0, crc32("123456789") == 0xCBF43926. ─────────────────────── */
|
||
static uint32_t s_crc_table[256];
|
||
static int s_crc_ready = 0;
|
||
static void crc_init(void){
|
||
for (uint32_t i=0;i<256;i++){
|
||
uint32_t c=i;
|
||
for (int k=0;k<8;k++) c = (c&1) ? (0xEDB88320u ^ (c>>1)) : (c>>1);
|
||
s_crc_table[i]=c;
|
||
}
|
||
s_crc_ready=1;
|
||
}
|
||
static uint32_t crc32_buf(const void* data, size_t n){
|
||
if (!s_crc_ready) crc_init();
|
||
const uint8_t* p = (const uint8_t*)data;
|
||
uint32_t c = 0xFFFFFFFFu;
|
||
for (size_t i=0;i<n;i++) c = s_crc_table[(c ^ p[i]) & 0xFF] ^ (c >> 8);
|
||
return c ^ 0xFFFFFFFFu;
|
||
}
|
||
|
||
/* FNV-1a 64-bit over a NUL-terminated id; the B+-tree key. Exact id is verified
|
||
* against the record on read, so hash collisions are correctness-safe. */
|
||
static uint64_t id_hash(const char* s){
|
||
uint64_t h = 1469598103934665603ULL;
|
||
for (; *s; ++s){ h ^= (uint8_t)*s; h *= 1099511628211ULL; }
|
||
return h;
|
||
}
|
||
|
||
/* ── raw page I/O ─────────────────────────────────────────────────────────── */
|
||
/* Read a page: served from the write-back cache if resident, else from disk (and
|
||
* cached clean). This is the ONLY page-read path, so a dirty (not-yet-flushed)
|
||
* page is always observed with its in-RAM contents. */
|
||
static int page_read(EngramPagedStore* s, uint64_t id, uint8_t* buf){
|
||
if (s->cache){
|
||
PgEnt* e = pc_get(s, id);
|
||
if (e){ memcpy(buf, e->buf, STORE_PAGE_SIZE); return 0; }
|
||
}
|
||
off_t off = (off_t)id * STORE_PAGE_SIZE;
|
||
ssize_t r = pread(s->fd, buf, STORE_PAGE_SIZE, off);
|
||
if (r != (ssize_t)STORE_PAGE_SIZE) return -1;
|
||
if (s->cache) pc_put(s, id, buf, 0); /* cache clean */
|
||
return 0;
|
||
}
|
||
static int page_write_raw(EngramPagedStore* s, uint64_t id, const uint8_t* buf){
|
||
off_t off = (off_t)id * STORE_PAGE_SIZE;
|
||
ssize_t w = pwrite(s->fd, buf, STORE_PAGE_SIZE, off);
|
||
if (w != (ssize_t)STORE_PAGE_SIZE) return -1;
|
||
return 0;
|
||
}
|
||
/* Stamp page_id + lsn + crc into the generic header, then write BACK to the cache
|
||
* (dirty). No-steal: the page reaches disk only at the next checkpoint. The stamped
|
||
* LSN is the current op's WAL LSN (s->stamp_lsn) when set, else a fresh counter
|
||
* value — this is the page LSN used for ARIES-style redo idempotency. */
|
||
static int page_write(EngramPagedStore* s, uint64_t id, uint8_t* buf){
|
||
put_u64(buf + 0, id);
|
||
uint64_t lsn = s->stamp_lsn ? s->stamp_lsn : ++s->next_lsn;
|
||
put_u64(buf + 16, lsn);
|
||
put_u32(buf + 24, 0);
|
||
uint32_t crc = crc32_buf(buf, STORE_PAGE_SIZE);
|
||
put_u32(buf + 24, crc);
|
||
if (s->cache) return pc_put(s, id, buf, 1);
|
||
return page_write_raw(s, id, buf); /* pre-cache bootstrap (should not happen) */
|
||
}
|
||
static int page_crc_ok(const uint8_t* buf){
|
||
uint8_t tmp[STORE_PAGE_SIZE];
|
||
memcpy(tmp, buf, STORE_PAGE_SIZE);
|
||
uint32_t stored = get_u32(tmp + 24);
|
||
put_u32(tmp + 24, 0);
|
||
return crc32_buf(tmp, STORE_PAGE_SIZE) == stored;
|
||
}
|
||
|
||
/* Allocate a page: reuse a FREE page if available, else extend the file. */
|
||
static uint64_t page_alloc(EngramPagedStore* s, uint8_t type){
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
uint64_t id;
|
||
if (s->free_list_head){
|
||
id = s->free_list_head;
|
||
if (page_read(s, id, buf) != 0) return 0;
|
||
s->free_list_head = get_u64(buf + OVF_NEXT_OFF); /* next_free stashed here */
|
||
} else {
|
||
id = s->page_count++;
|
||
}
|
||
memset(buf, 0, STORE_PAGE_SIZE);
|
||
buf[8] = type;
|
||
if (type == STORE_PT_NODE || type == STORE_PT_EDGE){
|
||
put_u16(buf + 10, 0);
|
||
put_u16(buf + 12, (uint16_t)(STORE_PAGE_SIZE - STORE_HDR)); /* all heap free */
|
||
}
|
||
if (page_write(s, id, buf) != 0) return 0;
|
||
return id;
|
||
}
|
||
/* Return a now-empty page to the free list. */
|
||
static int page_free(EngramPagedStore* s, uint64_t id){
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
memset(buf, 0, STORE_PAGE_SIZE);
|
||
buf[8] = STORE_PT_FREE;
|
||
put_u64(buf + OVF_NEXT_OFF, s->free_list_head);
|
||
if (page_write(s, id, buf) != 0) return -1;
|
||
s->free_list_head = id;
|
||
return 0;
|
||
}
|
||
|
||
/* ── slotted page helpers (operate on a page already in `buf`) ────────────── */
|
||
static int slp_count(const uint8_t* buf){ return get_u16(buf + 10); }
|
||
static int slp_free(const uint8_t* buf){ return get_u16(buf + 12); }
|
||
static void slp_slot(const uint8_t* buf, int i, uint16_t* off, uint16_t* len, uint16_t* fl){
|
||
const uint8_t* p = buf + STORE_HDR + (size_t)i*SLOT_SIZE;
|
||
*off = get_u16(p); *len = get_u16(p+2); *fl = get_u16(p+4);
|
||
}
|
||
static void slp_set_slot(uint8_t* buf, int i, uint16_t off, uint16_t len, uint16_t fl){
|
||
uint8_t* p = buf + STORE_HDR + (size_t)i*SLOT_SIZE;
|
||
put_u16(p, off); put_u16(p+2, len); put_u16(p+4, fl);
|
||
}
|
||
static int slp_live_count(const uint8_t* buf){
|
||
int n = slp_count(buf), live = 0;
|
||
for (int i=0;i<n;i++){ uint16_t o,l,f; slp_slot(buf,i,&o,&l,&f); if (f==SLOT_LIVE) live++; }
|
||
return live;
|
||
}
|
||
/* Place `rec` (reclen bytes) into the page. Returns slot index, or -1 if no room.
|
||
* Prefers reusing a DEAD slot whose span fits (reclaims tombstoned space). */
|
||
static int slp_put(uint8_t* buf, const uint8_t* rec, uint16_t reclen){
|
||
int n = slp_count(buf);
|
||
for (int i=0;i<n;i++){
|
||
uint16_t o,l,f; slp_slot(buf,i,&o,&l,&f);
|
||
if (f==SLOT_DEAD && l>=reclen){
|
||
memcpy(buf + o, rec, reclen);
|
||
slp_set_slot(buf, i, o, reclen, SLOT_LIVE);
|
||
return i;
|
||
}
|
||
}
|
||
int freeb = slp_free(buf);
|
||
if ((int)reclen + (int)SLOT_SIZE > freeb) return -1;
|
||
uint16_t heap_top = (uint16_t)(STORE_HDR + n*SLOT_SIZE + freeb);
|
||
uint16_t off = (uint16_t)(heap_top - reclen);
|
||
memcpy(buf + off, rec, reclen);
|
||
slp_set_slot(buf, n, off, reclen, SLOT_LIVE);
|
||
put_u16(buf + 10, (uint16_t)(n+1));
|
||
put_u16(buf + 12, (uint16_t)(freeb - reclen - SLOT_SIZE));
|
||
return n;
|
||
}
|
||
|
||
/* ── overflow chains ──────────────────────────────────────────────────────── */
|
||
static uint64_t ovf_write_chain(EngramPagedStore* s, const uint8_t* blob, size_t len){
|
||
uint64_t head = 0, prev = 0;
|
||
size_t off = 0;
|
||
do {
|
||
uint64_t id = page_alloc(s, STORE_PT_OVERFLOW);
|
||
if (!id) return 0;
|
||
if (!head) head = id;
|
||
if (prev){
|
||
uint8_t pbuf[STORE_PAGE_SIZE];
|
||
if (page_read(s, prev, pbuf)!=0) return 0;
|
||
put_u64(pbuf + OVF_NEXT_OFF, id);
|
||
if (page_write(s, prev, pbuf)!=0) return 0;
|
||
}
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
memset(buf, 0, STORE_PAGE_SIZE);
|
||
buf[8] = STORE_PT_OVERFLOW;
|
||
size_t chunk = len - off; if (chunk > OVF_CHUNK) chunk = OVF_CHUNK;
|
||
put_u64(buf + OVF_NEXT_OFF, 0);
|
||
put_u32(buf + OVF_LEN_OFF, (uint32_t)chunk);
|
||
memcpy(buf + OVF_DATA_OFF, blob + off, chunk);
|
||
if (page_write(s, id, buf)!=0) return 0;
|
||
prev = id; off += chunk;
|
||
} while (off < len);
|
||
return head;
|
||
}
|
||
static uint8_t* ovf_read_chain(EngramPagedStore* s, uint64_t head, size_t total){
|
||
uint8_t* out = (uint8_t*)malloc(total ? total : 1);
|
||
if (!out) return NULL;
|
||
size_t off = 0; uint64_t id = head;
|
||
while (id){
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, id, buf)!=0){ free(out); return NULL; }
|
||
uint32_t chunk = get_u32(buf + OVF_LEN_OFF);
|
||
if (off + chunk > total){ free(out); return NULL; }
|
||
memcpy(out + off, buf + OVF_DATA_OFF, chunk);
|
||
off += chunk;
|
||
id = get_u64(buf + OVF_NEXT_OFF);
|
||
}
|
||
if (off != total){ free(out); return NULL; }
|
||
return out;
|
||
}
|
||
|
||
/* ── growable byte buffer for TLV assembly ────────────────────────────────── */
|
||
typedef struct { uint8_t* p; size_t len, cap; } Buf;
|
||
static int buf_reserve(Buf* b, size_t extra){
|
||
if (b->len + extra <= b->cap) return 0;
|
||
size_t nc = b->cap ? b->cap*2 : 256;
|
||
while (nc < b->len + extra) nc *= 2;
|
||
uint8_t* np = (uint8_t*)realloc(b->p, nc);
|
||
if (!np) return -1;
|
||
b->p = np; b->cap = nc; return 0;
|
||
}
|
||
static int buf_raw(Buf* b, const void* data, size_t n){
|
||
if (buf_reserve(b, n)) return -1;
|
||
memcpy(b->p + b->len, data, n); b->len += n; return 0;
|
||
}
|
||
static int tlv_field(Buf* b, uint8_t tag, const void* data, uint32_t len){
|
||
if (buf_reserve(b, 5 + len)) return -1;
|
||
b->p[b->len++] = tag;
|
||
put_u32(b->p + b->len, len); b->len += 4;
|
||
if (len){ memcpy(b->p + b->len, data, len); b->len += len; }
|
||
return 0;
|
||
}
|
||
static int tlv_str(Buf* b, uint8_t tag, const char* s){
|
||
if (!s) return 0; /* NULL string omitted → reader defaults NULL */
|
||
return tlv_field(b, tag, s, (uint32_t)strlen(s));
|
||
}
|
||
static int tlv_f64(Buf* b, uint8_t tag, double v){ uint8_t t[8]; put_f64(t,v); return tlv_field(b,tag,t,8); }
|
||
static int tlv_i64(Buf* b, uint8_t tag, int64_t v){ uint8_t t[8]; put_u64(t,(uint64_t)v); return tlv_field(b,tag,t,8); }
|
||
static int tlv_i32(Buf* b, uint8_t tag, int32_t v){ uint8_t t[4]; put_u32(t,(uint32_t)v); return tlv_field(b,tag,t,4); }
|
||
static int tlv_u32(Buf* b, uint8_t tag, uint32_t v){ uint8_t t[4]; put_u32(t,v); return tlv_field(b,tag,t,4); }
|
||
|
||
/* ── node/edge serialization to a TLV body ────────────────────────────────── */
|
||
static uint8_t* node_serialize(const StoreNode* n, size_t* out_len){
|
||
Buf b = {0,0,0};
|
||
tlv_str(&b, NT_ID, n->id);
|
||
tlv_str(&b, NT_CONTENT, n->content);
|
||
tlv_str(&b, NT_NODE_TYPE, n->node_type);
|
||
tlv_str(&b, NT_LABEL, n->label);
|
||
tlv_str(&b, NT_TIER, n->tier);
|
||
tlv_str(&b, NT_TAGS, n->tags);
|
||
tlv_str(&b, NT_METADATA, n->metadata);
|
||
tlv_f64(&b, NT_SALIENCE, n->salience);
|
||
tlv_f64(&b, NT_IMPORTANCE, n->importance);
|
||
tlv_f64(&b, NT_CONFIDENCE, n->confidence);
|
||
tlv_f64(&b, NT_DECAY, n->temporal_decay_rate);
|
||
tlv_i64(&b, NT_ACT_COUNT, n->activation_count);
|
||
tlv_i64(&b, NT_LAST_ACT, n->last_activated);
|
||
tlv_i64(&b, NT_CREATED, n->created_at);
|
||
tlv_i64(&b, NT_UPDATED, n->updated_at);
|
||
tlv_f64(&b, NT_BG_ACT, n->background_activation);
|
||
tlv_f64(&b, NT_WM_WEIGHT, n->working_memory_weight);
|
||
tlv_i32(&b, NT_SUPPRESS, n->suppression_count);
|
||
tlv_u32(&b, NT_LAYER, n->layer_id);
|
||
{ uint8_t at[STORE_BLL_K*8];
|
||
for (int i=0;i<STORE_BLL_K;i++) put_u64(at + i*8, (uint64_t)n->access_ts[i]);
|
||
tlv_field(&b, NT_ACCESS_TS, at, sizeof at); }
|
||
tlv_i32(&b, NT_ACCESS_HEAD, n->access_head);
|
||
tlv_i32(&b, NT_ACCESS_FILLED, n->access_filled);
|
||
tlv_f64(&b, NT_WM_ANCHOR, n->wm_anchor);
|
||
if (n->emb && n->emb_dim > 0){
|
||
uint8_t* eb = (uint8_t*)malloc((size_t)n->emb_dim*4);
|
||
for (int i=0;i<n->emb_dim;i++){ uint32_t u; memcpy(&u,&n->emb[i],4); put_u32(eb+i*4,u); }
|
||
tlv_field(&b, NT_EMB, eb, (uint32_t)n->emb_dim*4);
|
||
free(eb);
|
||
tlv_i32(&b, NT_EMB_DIM, n->emb_dim);
|
||
}
|
||
if (n->unknown && n->unknown_len) buf_raw(&b, n->unknown, n->unknown_len);
|
||
*out_len = b.len;
|
||
return b.p;
|
||
}
|
||
static uint8_t* edge_serialize(const StoreEdge* e, size_t* out_len){
|
||
Buf b = {0,0,0};
|
||
tlv_str(&b, ET_ID, e->id);
|
||
tlv_str(&b, ET_FROM, e->from_id);
|
||
tlv_str(&b, ET_TO, e->to_id);
|
||
tlv_str(&b, ET_RELATION, e->relation);
|
||
tlv_str(&b, ET_METADATA, e->metadata);
|
||
tlv_f64(&b, ET_WEIGHT, e->weight);
|
||
tlv_f64(&b, ET_HEBB, e->hebb);
|
||
tlv_f64(&b, ET_CONFIDENCE, e->confidence);
|
||
tlv_i64(&b, ET_CREATED, e->created_at);
|
||
tlv_i64(&b, ET_UPDATED, e->updated_at);
|
||
tlv_i64(&b, ET_LAST_FIRED, e->last_fired);
|
||
tlv_i32(&b, ET_INHIBITORY, e->inhibitory);
|
||
tlv_u32(&b, ET_LAYER, e->layer_id);
|
||
if (e->unknown && e->unknown_len) buf_raw(&b, e->unknown, e->unknown_len);
|
||
*out_len = b.len;
|
||
return b.p;
|
||
}
|
||
|
||
/* ── TLV parse helpers ────────────────────────────────────────────────────── */
|
||
static char* dup_str(const uint8_t* p, uint32_t len){
|
||
char* s = (char*)malloc(len + 1);
|
||
if (!s) return NULL;
|
||
memcpy(s, p, len); s[len] = 0; return s;
|
||
}
|
||
static void node_parse(const uint8_t* body, size_t len, StoreNode* n){
|
||
memset(n, 0, sizeof *n);
|
||
Buf unk = {0,0,0};
|
||
size_t i = 0;
|
||
while (i + 5 <= len){
|
||
uint8_t tag = body[i];
|
||
uint32_t flen = get_u32(body + i + 1);
|
||
if (i + 5 + flen > len) break;
|
||
const uint8_t* v = body + i + 5;
|
||
switch (tag){
|
||
case NT_ID: n->id = dup_str(v, flen); break;
|
||
case NT_CONTENT: n->content = dup_str(v, flen); break;
|
||
case NT_NODE_TYPE: n->node_type = dup_str(v, flen); break;
|
||
case NT_LABEL: n->label = dup_str(v, flen); break;
|
||
case NT_TIER: n->tier = dup_str(v, flen); break;
|
||
case NT_TAGS: n->tags = dup_str(v, flen); break;
|
||
case NT_METADATA: n->metadata = dup_str(v, flen); break;
|
||
case NT_SALIENCE: n->salience = get_f64(v); break;
|
||
case NT_IMPORTANCE: n->importance = get_f64(v); break;
|
||
case NT_CONFIDENCE: n->confidence = get_f64(v); break;
|
||
case NT_DECAY: n->temporal_decay_rate = get_f64(v); break;
|
||
case NT_ACT_COUNT: n->activation_count = (int64_t)get_u64(v); break;
|
||
case NT_LAST_ACT: n->last_activated = (int64_t)get_u64(v); break;
|
||
case NT_CREATED: n->created_at = (int64_t)get_u64(v); break;
|
||
case NT_UPDATED: n->updated_at = (int64_t)get_u64(v); break;
|
||
case NT_BG_ACT: n->background_activation = get_f64(v); break;
|
||
case NT_WM_WEIGHT: n->working_memory_weight = get_f64(v); break;
|
||
case NT_SUPPRESS: n->suppression_count = (int32_t)get_u32(v); break;
|
||
case NT_LAYER: n->layer_id = get_u32(v); break;
|
||
case NT_ACCESS_TS:
|
||
for (int k=0;k<STORE_BLL_K && (uint32_t)(k*8+8)<=flen;k++)
|
||
n->access_ts[k] = (int64_t)get_u64(v + k*8);
|
||
break;
|
||
case NT_ACCESS_HEAD: n->access_head = (int32_t)get_u32(v); break;
|
||
case NT_ACCESS_FILLED: n->access_filled = (int32_t)get_u32(v); break;
|
||
case NT_WM_ANCHOR: n->wm_anchor = get_f64(v); break;
|
||
case NT_EMB: {
|
||
int32_t dim = (int32_t)(flen/4);
|
||
n->emb = (float*)malloc(flen ? flen : 4);
|
||
for (int32_t k=0;k<dim;k++){ uint32_t u=get_u32(v+k*4); memcpy(&n->emb[k],&u,4); }
|
||
if (n->emb_dim == 0) n->emb_dim = dim; /* honour explicit dim if seen */
|
||
break; }
|
||
case NT_EMB_DIM: n->emb_dim = (int32_t)get_u32(v); break;
|
||
default:
|
||
/* unknown tag: preserve verbatim */
|
||
buf_raw(&unk, body + i, 5 + flen);
|
||
break;
|
||
}
|
||
i += 5 + flen;
|
||
}
|
||
n->unknown = unk.p; n->unknown_len = unk.len;
|
||
}
|
||
static void edge_parse(const uint8_t* body, size_t len, StoreEdge* e){
|
||
memset(e, 0, sizeof *e);
|
||
Buf unk = {0,0,0};
|
||
size_t i = 0;
|
||
while (i + 5 <= len){
|
||
uint8_t tag = body[i];
|
||
uint32_t flen = get_u32(body + i + 1);
|
||
if (i + 5 + flen > len) break;
|
||
const uint8_t* v = body + i + 5;
|
||
switch (tag){
|
||
case ET_ID: e->id = dup_str(v, flen); break;
|
||
case ET_FROM: e->from_id = dup_str(v, flen); break;
|
||
case ET_TO: e->to_id = dup_str(v, flen); break;
|
||
case ET_RELATION: e->relation = dup_str(v, flen); break;
|
||
case ET_METADATA: e->metadata = dup_str(v, flen); break;
|
||
case ET_WEIGHT: e->weight = get_f64(v); break;
|
||
case ET_HEBB: e->hebb = get_f64(v); break;
|
||
case ET_CONFIDENCE: e->confidence = get_f64(v); break;
|
||
case ET_CREATED: e->created_at = (int64_t)get_u64(v); break;
|
||
case ET_UPDATED: e->updated_at = (int64_t)get_u64(v); break;
|
||
case ET_LAST_FIRED: e->last_fired = (int64_t)get_u64(v); break;
|
||
case ET_INHIBITORY: e->inhibitory = (int32_t)get_u32(v); break;
|
||
case ET_LAYER: e->layer_id = get_u32(v); break;
|
||
default:
|
||
buf_raw(&unk, body + i, 5 + flen);
|
||
break;
|
||
}
|
||
i += 5 + flen;
|
||
}
|
||
e->unknown = unk.p; e->unknown_len = unk.len;
|
||
}
|
||
|
||
/* ── B+-tree ──────────────────────────────────────────────────────────────── */
|
||
static uint32_t tree_payload(int tree){ return tree==TREE_PRIMARY ? PRIMARY_PAYLOAD : ADJ_PAYLOAD; }
|
||
static uint64_t tree_root(EngramPagedStore* s, int tree){
|
||
return tree==TREE_PRIMARY ? s->root_index_page : s->adj_index_page;
|
||
}
|
||
static void tree_set_root(EngramPagedStore* s, int tree, uint64_t id){
|
||
if (tree==TREE_PRIMARY) s->root_index_page = id; else s->adj_index_page = id;
|
||
}
|
||
static int leaf_max_entries(EngramPagedStore* s, uint32_t payload){
|
||
int nat = (int)(IDX_BODY / (8 + payload));
|
||
if (s->leaf_max > 0 && s->leaf_max < nat) return s->leaf_max;
|
||
return nat;
|
||
}
|
||
static int int_max_keys(EngramPagedStore* s){
|
||
/* 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;i<nkeys;i++) keys[i] = get_u64(K + (size_t)i*8);
|
||
for (int i=0;i<=nkeys;i++) kids[i] = get_u64(C + (size_t)i*8);
|
||
for (int i=nkeys;i>ci;i--) keys[i] = keys[i-1];
|
||
keys[ci] = csep;
|
||
for (int i=nkeys+1;i>ci+1;i--) kids[i] = kids[i-1];
|
||
kids[ci+1] = cnew;
|
||
|
||
int imax = int_max_keys(s);
|
||
if (nk <= imax){
|
||
put_u16(buf + 10, (uint16_t)nk);
|
||
uint8_t* K2 = buf + IDX_ENT_OFF;
|
||
uint8_t* C2 = buf + IDX_ENT_OFF + (size_t)nk*8;
|
||
for (int i=0;i<nk;i++) put_u64(K2 + (size_t)i*8, keys[i]);
|
||
for (int i=0;i<=nk;i++) put_u64(C2 + (size_t)i*8, kids[i]);
|
||
free(keys); free(kids);
|
||
return page_write(s, page_id, buf)==0 ? 0 : -1;
|
||
}
|
||
/* split internal: middle key moves up */
|
||
int mid = nk/2;
|
||
uint64_t up = keys[mid];
|
||
int lk = mid, rk = nk - mid - 1;
|
||
uint64_t rid = page_alloc(s, STORE_PT_INDEX);
|
||
if (!rid){ free(keys); free(kids); return -1; }
|
||
uint8_t rbuf[STORE_PAGE_SIZE];
|
||
if (page_read(s, rid, rbuf)!=0){ free(keys); free(kids); return -1; }
|
||
rbuf[IDX_LEAF_OFF] = 0;
|
||
put_u16(rbuf + 10, (uint16_t)rk);
|
||
{ uint8_t* RK = rbuf + IDX_ENT_OFF; uint8_t* RC = rbuf + IDX_ENT_OFF + (size_t)rk*8;
|
||
for (int i=0;i<rk;i++) put_u64(RK + (size_t)i*8, keys[mid+1+i]);
|
||
for (int i=0;i<=rk;i++) put_u64(RC + (size_t)i*8, kids[mid+1+i]); }
|
||
put_u16(buf + 10, (uint16_t)lk);
|
||
{ uint8_t* LK = buf + IDX_ENT_OFF; uint8_t* LC = buf + IDX_ENT_OFF + (size_t)lk*8;
|
||
for (int i=0;i<lk;i++) put_u64(LK + (size_t)i*8, keys[i]);
|
||
for (int i=0;i<=lk;i++) put_u64(LC + (size_t)i*8, kids[i]); }
|
||
free(keys); free(kids);
|
||
if (page_write(s, rid, rbuf)!=0) return -1;
|
||
if (page_write(s, page_id, buf)!=0) return -1;
|
||
*sep_key = up; *new_page = rid;
|
||
return 1;
|
||
}
|
||
|
||
static int btree_put(EngramPagedStore* s, int tree, uint64_t key, const uint8_t* payload){
|
||
uint64_t root = tree_root(s, tree);
|
||
uint64_t sep, np;
|
||
int sp = btree_insert(s, tree, root, key, payload, &sep, &np);
|
||
if (sp < 0) return -1;
|
||
if (sp == 0) return 0;
|
||
/* root split → new internal root */
|
||
uint64_t nr = page_alloc(s, STORE_PT_INDEX);
|
||
if (!nr) return -1;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, nr, buf)!=0) return -1;
|
||
buf[IDX_LEAF_OFF] = 0;
|
||
put_u16(buf + 10, 1);
|
||
put_u64(buf + IDX_ENT_OFF, sep);
|
||
put_u64(buf + IDX_ENT_OFF + 8, root);
|
||
put_u64(buf + IDX_ENT_OFF + 16, np);
|
||
if (page_write(s, nr, buf)!=0) return -1;
|
||
tree_set_root(s, tree, nr);
|
||
return 0;
|
||
}
|
||
|
||
/* Collect every payload whose key == target. Payloads appended to *out (esz-pl
|
||
* bytes each, i.e. just the payload, pl bytes). Caller frees *out. */
|
||
static int btree_lookup(EngramPagedStore* s, int tree, uint64_t target,
|
||
uint8_t** out, size_t* count){
|
||
uint32_t pl = tree_payload(tree);
|
||
uint32_t esz = 8 + pl;
|
||
*out = NULL; *count = 0;
|
||
uint64_t id = tree_root(s, tree);
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
/* descend to the leftmost leaf that could hold target */
|
||
for (;;){
|
||
if (page_read(s, id, buf)!=0) return -1;
|
||
if (buf[IDX_LEAF_OFF]) break;
|
||
int nkeys = get_u16(buf + 10);
|
||
uint8_t* K = buf + IDX_ENT_OFF;
|
||
uint8_t* C = buf + IDX_ENT_OFF + (size_t)nkeys*8;
|
||
/* Descend to the LEFTMOST child that may hold `target`. Duplicate keys
|
||
* span a contiguous run of leaves (a leaf split can leave copies of the
|
||
* separator value on both sides), so we must use strict '>' here — not
|
||
* the '>=' of insert routing — then scan forward across next_leaf. */
|
||
int ci = 0;
|
||
while (ci < nkeys && target > get_u64(K + (size_t)ci*8)) ci++;
|
||
id = get_u64(C + (size_t)ci*8);
|
||
}
|
||
Buf acc = {0,0,0};
|
||
for (;;){
|
||
int nkeys = get_u16(buf + 10);
|
||
int past = 0;
|
||
for (int i=0;i<nkeys;i++){
|
||
uint64_t k = get_u64(buf + IDX_ENT_OFF + (size_t)i*esz);
|
||
if (k < target) continue;
|
||
if (k > target){ past = 1; break; }
|
||
if (buf_raw(&acc, buf + IDX_ENT_OFF + (size_t)i*esz + 8, pl)){ free(acc.p); return -1; }
|
||
}
|
||
if (past) break;
|
||
uint64_t nxt = get_u64(buf + IDX_NEXT_OFF);
|
||
if (!nxt) break;
|
||
if (page_read(s, nxt, buf)!=0){ free(acc.p); return -1; }
|
||
}
|
||
*out = acc.p; *count = acc.len / pl;
|
||
return 0;
|
||
}
|
||
|
||
/* ── superblock ───────────────────────────────────────────────────────────── */
|
||
static int sb_write_one(EngramPagedStore* s, uint64_t page){
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
memset(buf, 0, STORE_PAGE_SIZE);
|
||
memcpy(buf + SB_MAGIC_OFF, STORE_MAGIC, 8); /* 7 chars + NUL */
|
||
put_u32(buf + SB_FMT_OFF, s->format_version);
|
||
put_u32(buf + SB_PGSZ_OFF, s->page_size);
|
||
put_u64(buf + SB_PGCNT_OFF, s->page_count);
|
||
put_u64(buf + SB_FREE_OFF, s->free_list_head);
|
||
put_u64(buf + SB_ROOT_OFF, s->root_index_page);
|
||
put_u64(buf + SB_ADJ_OFF, s->adj_index_page);
|
||
put_u64(buf + SB_LAYER_OFF, s->layer_registry_page);
|
||
put_u64(buf + SB_CKPT_OFF, s->last_checkpoint_lsn);
|
||
put_u64(buf + SB_SEQ_OFF, s->sb_seq);
|
||
memcpy(buf + SB_UUID_OFF, s->uuid, 16);
|
||
put_u32(buf + SB_CRC_OFF, 0);
|
||
uint32_t crc = crc32_buf(buf, STORE_PAGE_SIZE);
|
||
put_u32(buf + SB_CRC_OFF, crc);
|
||
return page_write_raw(s, page, buf);
|
||
}
|
||
static int sb_load_one(EngramPagedStore* s, uint64_t page, uint8_t* buf, uint64_t* seq_out){
|
||
if (page_read(s, page, buf)!=0) return -1;
|
||
if (memcmp(buf + SB_MAGIC_OFF, STORE_MAGIC, 8)!=0) return -1;
|
||
uint32_t stored = get_u32(buf + SB_CRC_OFF);
|
||
uint8_t tmp[STORE_PAGE_SIZE];
|
||
memcpy(tmp, buf, STORE_PAGE_SIZE);
|
||
put_u32(tmp + SB_CRC_OFF, 0);
|
||
if (crc32_buf(tmp, STORE_PAGE_SIZE) != stored) return -1;
|
||
*seq_out = get_u64(buf + SB_SEQ_OFF);
|
||
return 0;
|
||
}
|
||
static void sb_apply(EngramPagedStore* s, const uint8_t* buf){
|
||
s->format_version = get_u32(buf + SB_FMT_OFF);
|
||
s->page_size = get_u32(buf + SB_PGSZ_OFF);
|
||
s->page_count = get_u64(buf + SB_PGCNT_OFF);
|
||
s->free_list_head = get_u64(buf + SB_FREE_OFF);
|
||
s->root_index_page = get_u64(buf + SB_ROOT_OFF);
|
||
s->adj_index_page = get_u64(buf + SB_ADJ_OFF);
|
||
s->layer_registry_page= get_u64(buf + SB_LAYER_OFF);
|
||
s->last_checkpoint_lsn= get_u64(buf + SB_CKPT_OFF);
|
||
s->sb_seq = get_u64(buf + SB_SEQ_OFF);
|
||
memcpy(s->uuid, buf + SB_UUID_OFF, 16);
|
||
}
|
||
|
||
int store_sync(EngramPagedStore* s){
|
||
if (!s) return -1;
|
||
/* flush the write-back cache so the store file reflects the SB we are about
|
||
* to stamp (this is the checkpoint page-flush + fsync). */
|
||
if (pc_flush(s) != 0) return -1;
|
||
if (fsync(s->fd) != 0) return -1;
|
||
s->sb_seq++;
|
||
/* write primary, fsync, then mirror, fsync — so a torn write of one leaves
|
||
* the other valid; recovery prefers the higher valid sb_seq. */
|
||
if (sb_write_one(s, 0)!=0) return -1;
|
||
if (fsync(s->fd)!=0) return -1;
|
||
if (sb_write_one(s, 1)!=0) return -1;
|
||
if (fsync(s->fd)!=0) return -1;
|
||
return 0;
|
||
}
|
||
|
||
/* ── lifecycle ────────────────────────────────────────────────────────────── */
|
||
static void gen_uuid(uint8_t out[16]){
|
||
int fd = open("/dev/urandom", O_RDONLY);
|
||
if (fd >= 0){ ssize_t r = read(fd, out, 16); close(fd); if (r == 16) return; }
|
||
uint64_t t = (uint64_t)time(NULL) ^ ((uint64_t)getpid() << 32);
|
||
for (int i=0;i<16;i++){ t = t*6364136223846793005ULL + 1442695040888963407ULL; out[i] = (uint8_t)(t>>56); }
|
||
}
|
||
static uint64_t idx_alloc_leaf(EngramPagedStore* s){
|
||
uint64_t id = page_alloc(s, STORE_PT_INDEX);
|
||
if (!id) return 0;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, id, buf)!=0) return 0;
|
||
buf[IDX_LEAF_OFF] = 1;
|
||
put_u16(buf + 10, 0);
|
||
put_u64(buf + IDX_NEXT_OFF, 0);
|
||
if (page_write(s, id, buf)!=0) return 0;
|
||
return id;
|
||
}
|
||
|
||
EngramPagedStore* store_create(const char* path){
|
||
if (!s_crc_ready) crc_init();
|
||
struct stat st;
|
||
if (stat(path, &st) == 0){ errno = EEXIST; return NULL; }
|
||
EngramPagedStore* s = (EngramPagedStore*)calloc(1, sizeof *s);
|
||
if (!s) return NULL;
|
||
s->fd = open(path, O_RDWR | O_CREAT | O_EXCL, 0600);
|
||
if (s->fd < 0){ free(s); return NULL; }
|
||
s->cache = pc_new();
|
||
if (!s->cache){ close(s->fd); free(s); return NULL; }
|
||
snprintf(s->path, sizeof s->path, "%s", path);
|
||
s->format_version = STORE_FORMAT_VERSION;
|
||
s->page_size = STORE_PAGE_SIZE;
|
||
s->page_count = 2; /* pages 0,1 are the superblock + mirror */
|
||
s->free_list_head = 0;
|
||
s->next_lsn = 0;
|
||
s->sb_seq = 0;
|
||
gen_uuid(s->uuid);
|
||
/* reserve pages 0,1 on disk */
|
||
uint8_t zero[STORE_PAGE_SIZE]; memset(zero, 0, sizeof zero);
|
||
if (page_write_raw(s, 0, zero)!=0 || page_write_raw(s, 1, zero)!=0){ close(s->fd); free(s); return NULL; }
|
||
s->root_index_page = idx_alloc_leaf(s);
|
||
s->adj_index_page = idx_alloc_leaf(s);
|
||
s->layer_registry_page = page_alloc(s, STORE_PT_NODE); /* reserved; unused in M1 */
|
||
if (!s->root_index_page || !s->adj_index_page || !s->layer_registry_page){
|
||
close(s->fd); free(s); return NULL;
|
||
}
|
||
if (store_sync(s)!=0){ close(s->fd); free(s); return NULL; }
|
||
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;
|
||
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;i<n;i++){
|
||
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
|
||
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
|
||
StoreNode cand; node_parse(body, blen, &cand); free(body);
|
||
if (cand.id && strcmp(cand.id, id)==0){
|
||
if (live){ /* latest live wins (later entries) */
|
||
if (found) store_node_free(out);
|
||
*out = cand; out->tombstoned = 0; found = 1;
|
||
} else store_node_free(&cand);
|
||
} else store_node_free(&cand);
|
||
}
|
||
free(locs);
|
||
return found ? 1 : 0;
|
||
}
|
||
|
||
/* Tombstone the live record(s) for `id`, but only where the record's home page
|
||
* LSN < `lsn` (ARIES redo idempotency: a replayed tombstone whose effect is
|
||
* already on the page is skipped). Stamps killed pages with `lsn`. */
|
||
static int tombstone_core(EngramPagedStore* s, const char* id, uint64_t lsn){
|
||
if (!s || !id) return -1;
|
||
uint8_t* locs; size_t n;
|
||
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n)!=0) return -1;
|
||
uint64_t prev_stamp = s->stamp_lsn;
|
||
s->stamp_lsn = lsn;
|
||
for (size_t i=0;i<n;i++){
|
||
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
|
||
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, page, buf)!=0) continue;
|
||
if (get_u64(buf + 16) >= lsn) continue; /* already covered by >= this LSN */
|
||
if (slot >= slp_count(buf)) continue;
|
||
uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
/* verify id matches this record before killing it */
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
|
||
StoreNode cand; node_parse(body, blen, &cand); free(body);
|
||
int match = cand.id && strcmp(cand.id, id)==0;
|
||
store_node_free(&cand);
|
||
if (!match) continue;
|
||
slp_set_slot(buf, slot, off, len, SLOT_DEAD);
|
||
if (page_write(s, page, buf)!=0){ s->stamp_lsn = prev_stamp; free(locs); return -1; }
|
||
/* if the page is now empty, reclaim it to the free list */
|
||
if (page_read(s, page, buf)==0 && slp_live_count(buf)==0){
|
||
if (s->cur_node_page == page) s->cur_node_page = 0;
|
||
if (s->cur_edge_page == page) s->cur_edge_page = 0;
|
||
page_free(s, page);
|
||
}
|
||
}
|
||
s->stamp_lsn = prev_stamp;
|
||
free(locs);
|
||
return 0; /* absent id is a no-op success */
|
||
}
|
||
|
||
static int get_edges_dir(EngramPagedStore* s, const char* id, uint8_t want_dir,
|
||
StoreEdge** out, size_t* n){
|
||
*out = NULL; *n = 0;
|
||
uint8_t* locs; size_t cnt;
|
||
if (btree_lookup(s, TREE_ADJ, id_hash(id), &locs, &cnt)!=0) return -1;
|
||
StoreEdge* arr = NULL; size_t used = 0, cap = 0;
|
||
for (size_t i=0;i<cnt;i++){
|
||
uint8_t dir = locs[i*ADJ_PAYLOAD];
|
||
if (dir != want_dir) continue;
|
||
uint64_t page = get_u64(locs + i*ADJ_PAYLOAD + 1);
|
||
uint16_t slot = get_u16(locs + i*ADJ_PAYLOAD + 9);
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live)!=0) continue;
|
||
if (!live){ free(body); continue; }
|
||
StoreEdge e; edge_parse(body, blen, &e); free(body);
|
||
const char* side = (want_dir==ADJ_DIR_FROM) ? e.from_id : e.to_id;
|
||
if (!side || strcmp(side, id)!=0){ store_edge_free(&e); continue; }
|
||
if (used == cap){
|
||
cap = cap ? cap*2 : 8;
|
||
StoreEdge* na = (StoreEdge*)realloc(arr, cap*sizeof *na);
|
||
if (!na){ store_edge_free(&e); break; }
|
||
arr = na;
|
||
}
|
||
arr[used++] = e;
|
||
}
|
||
free(locs);
|
||
*out = arr; *n = used;
|
||
return 0;
|
||
}
|
||
int store_get_edges_from(EngramPagedStore* s, const char* from_id, StoreEdge** out, size_t* n){
|
||
if (!s || !from_id) return -1;
|
||
return get_edges_dir(s, from_id, ADJ_DIR_FROM, out, n);
|
||
}
|
||
int store_get_edges_to(EngramPagedStore* s, const char* to_id, StoreEdge** out, size_t* n){
|
||
if (!s || !to_id) return -1;
|
||
return get_edges_dir(s, to_id, ADJ_DIR_TO, out, n);
|
||
}
|
||
|
||
/* ── integrity ────────────────────────────────────────────────────────────── */
|
||
int store_check(EngramPagedStore* s, unsigned flags){
|
||
if (!s) return -1;
|
||
int bad = 0;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
for (uint64_t id = 0; id < s->page_count; id++){
|
||
if (page_read(s, id, buf)!=0) return -1;
|
||
if (id == 0 || id == 1){
|
||
uint32_t stored = get_u32(buf + SB_CRC_OFF);
|
||
uint8_t tmp[STORE_PAGE_SIZE]; memcpy(tmp, buf, STORE_PAGE_SIZE);
|
||
put_u32(tmp + SB_CRC_OFF, 0);
|
||
int okmagic = memcmp(buf + SB_MAGIC_OFF, STORE_MAGIC, 8)==0;
|
||
if (!okmagic || crc32_buf(tmp, STORE_PAGE_SIZE) != stored) bad++;
|
||
} else {
|
||
if (flags & STORE_CHECK_CRC){ if (!page_crc_ok(buf)) bad++; }
|
||
}
|
||
}
|
||
return bad;
|
||
}
|
||
|
||
/* ── ownership helpers ────────────────────────────────────────────────────── */
|
||
void store_node_free(StoreNode* n){
|
||
if (!n) return;
|
||
free(n->id); free(n->content); free(n->node_type); free(n->label);
|
||
free(n->tier); free(n->tags); free(n->metadata); free(n->emb); free(n->unknown);
|
||
memset(n, 0, sizeof *n);
|
||
}
|
||
void store_edge_free(StoreEdge* e){
|
||
if (!e) return;
|
||
free(e->id); free(e->from_id); free(e->to_id); free(e->relation);
|
||
free(e->metadata); free(e->unknown);
|
||
memset(e, 0, sizeof *e);
|
||
}
|
||
void store_edges_free(StoreEdge* arr, size_t n){
|
||
if (!arr) return;
|
||
for (size_t i=0;i<n;i++) store_edge_free(&arr[i]);
|
||
free(arr);
|
||
}
|
||
void store__set_btree_order(EngramPagedStore* s, int leaf_max, int internal_max){
|
||
if (!s) return;
|
||
s->leaf_max = leaf_max; s->int_max = internal_max;
|
||
}
|
||
uint64_t store_page_count(const EngramPagedStore* s){ return s ? s->page_count : 0; }
|
||
|
||
/* ══════════════════════════════════════════════════════════════════════════════
|
||
* M2 — WAL + write-back buffer pool + checkpoint + crash recovery + legacy import
|
||
*
|
||
* Durability model (design §2.2/§4, ARIES-lite):
|
||
* • Buffer pool is WRITE-BACK, no-steal: a mutation dirties a page in RAM; the
|
||
* page reaches engram.store ONLY at a checkpoint. So after a crash the store
|
||
* file reflects exactly `last_checkpoint_lsn`, and everything since lives in
|
||
* the WAL. This is what makes the WAL load-bearing (durability = fsync'd WAL,
|
||
* not the page).
|
||
* • Each mutation: assign LSN → append WAL record (fsync per policy) → apply to
|
||
* the page(s), stamping page-LSN = record LSN.
|
||
* • Recovery: open store (at checkpoint), heal torn data pages, replay WAL
|
||
* forward; a record is REDONE only where the target record's home-page LSN is
|
||
* < the record LSN (idempotent — safe to replay any number of times, and safe
|
||
* across torn checkpoints where some dirty pages reached disk).
|
||
*
|
||
* DELTA vs the design sketch (flagged, becoming permanent): the WAL is LOGICAL
|
||
* (op + node/edge/layer payload), not physical page images, so the "redo only if
|
||
* rec.lsn > page.lsn" test is applied at RECORD granularity — against the LSN of
|
||
* the page currently holding the record for that id — rather than against a
|
||
* single physical target page. Records are never relocated once placed, so a
|
||
* record placed by op L keeps its page (whose LSN only rises), making
|
||
* "page-LSN ≥ L ⇒ op L already durable" a sound, precise idempotency test.
|
||
* ════════════════════════════════════════════════════════════════════════════ */
|
||
|
||
#include <sys/time.h>
|
||
|
||
/* ── write-back buffer pool ────────────────────────────────────────────────── */
|
||
struct PgCache { PgEnt** buckets; size_t nbuckets; size_t count; };
|
||
|
||
static PgCache* pc_new(void){
|
||
PgCache* c = (PgCache*)calloc(1, sizeof *c);
|
||
if (!c) return NULL;
|
||
c->nbuckets = 1024;
|
||
c->buckets = (PgEnt**)calloc(c->nbuckets, sizeof(PgEnt*));
|
||
if (!c->buckets){ free(c); return NULL; }
|
||
return c;
|
||
}
|
||
static void pc_free(PgCache* c){
|
||
if (!c) return;
|
||
for (size_t i=0;i<c->nbuckets;i++){
|
||
PgEnt* e = c->buckets[i];
|
||
while (e){ PgEnt* n=e->next; free(e->buf); free(e); e=n; }
|
||
}
|
||
free(c->buckets); free(c);
|
||
}
|
||
static PgEnt* pc_get(EngramPagedStore* s, uint64_t id){
|
||
PgCache* c = s->cache;
|
||
PgEnt* e = c->buckets[id % c->nbuckets];
|
||
while (e){ if (e->id==id) return e; e=e->next; }
|
||
return NULL;
|
||
}
|
||
static void pc_maybe_grow(PgCache* c){
|
||
if (c->count <= c->nbuckets*4) return;
|
||
size_t nn = c->nbuckets*2;
|
||
PgEnt** nb = (PgEnt**)calloc(nn, sizeof(PgEnt*));
|
||
if (!nb) return;
|
||
for (size_t i=0;i<c->nbuckets;i++){
|
||
PgEnt* e = c->buckets[i];
|
||
while (e){ PgEnt* nx=e->next; size_t b=e->id%nn; e->next=nb[b]; nb[b]=e; e=nx; }
|
||
}
|
||
free(c->buckets); c->buckets=nb; c->nbuckets=nn;
|
||
}
|
||
static int pc_put(EngramPagedStore* s, uint64_t id, const uint8_t* buf, int dirty){
|
||
PgCache* c = s->cache;
|
||
PgEnt* e = pc_get(s, id);
|
||
if (!e){
|
||
e = (PgEnt*)calloc(1, sizeof *e);
|
||
if (!e) return -1;
|
||
e->buf = (uint8_t*)malloc(STORE_PAGE_SIZE);
|
||
if (!e->buf){ free(e); return -1; }
|
||
e->id = id;
|
||
size_t b = id % c->nbuckets;
|
||
e->next = c->buckets[b]; c->buckets[b] = e; c->count++;
|
||
pc_maybe_grow(c);
|
||
}
|
||
memcpy(e->buf, buf, STORE_PAGE_SIZE);
|
||
e->lsn = get_u64(buf + 16);
|
||
if (dirty) e->dirty = 1;
|
||
return 0;
|
||
}
|
||
static int pc_flush(EngramPagedStore* s){
|
||
if (!s->cache) return 0;
|
||
PgCache* c = s->cache;
|
||
for (size_t i=0;i<c->nbuckets;i++)
|
||
for (PgEnt* e=c->buckets[i]; e; e=e->next)
|
||
if (e->dirty){ if (page_write_raw(s, e->id, e->buf)!=0) return -1; e->dirty=0; }
|
||
return 0;
|
||
}
|
||
/* ── WAL log ───────────────────────────────────────────────────────────────── */
|
||
enum { OP_NODE_PUT=1, OP_EDGE_PUT, OP_TOMBSTONE, OP_SUPERSEDE,
|
||
OP_LAYER_PUT, OP_LAYER_DEL, OP_FORGET, OP_HEBB_BATCH, OP_CHECKPOINT };
|
||
|
||
#define WAL_MAGIC 0x314C5745u /* 'E''W''L''1' little-endian */
|
||
#define WAL_HDR 22u /* magic4 len4 op1 flags1 lsn8 crc4 */
|
||
|
||
struct EngramWal {
|
||
int fd;
|
||
char path[1024];
|
||
EngramWalSync sync;
|
||
uint64_t last_fsync_lsn;
|
||
long long last_fsync_ms;
|
||
uint64_t appended_since_fsync;
|
||
};
|
||
|
||
static long long now_ms(void){
|
||
struct timeval tv; gettimeofday(&tv, NULL);
|
||
return (long long)tv.tv_sec*1000 + tv.tv_usec/1000;
|
||
}
|
||
static EngramWal* wal_open(const char* path, EngramWalSync sync){
|
||
EngramWal* w = (EngramWal*)calloc(1, sizeof *w);
|
||
if (!w) return NULL;
|
||
w->fd = open(path, O_RDWR | O_CREAT | O_APPEND, 0600);
|
||
if (w->fd < 0){ free(w); return NULL; }
|
||
snprintf(w->path, sizeof w->path, "%s", path);
|
||
w->sync = sync; w->last_fsync_ms = now_ms();
|
||
return w;
|
||
}
|
||
static void wal_close(EngramWal* w){
|
||
if (!w) return;
|
||
if (w->fd >= 0){ fsync(w->fd); close(w->fd); }
|
||
free(w);
|
||
}
|
||
static void wal_maybe_fsync(EngramPagedStore* s, uint64_t lsn){
|
||
EngramWal* w = s->wal; if (!w) return;
|
||
if (w->sync == ENGRAM_WAL_OFF) return;
|
||
if (w->sync == ENGRAM_WAL_ALWAYS){
|
||
fsync(w->fd); w->last_fsync_lsn=lsn; w->last_fsync_ms=now_ms(); w->appended_since_fsync=0; return;
|
||
}
|
||
/* GROUP: fsync at most every ~50 ms or every 256 records. */
|
||
long long now = now_ms();
|
||
if (now - w->last_fsync_ms >= 50 || w->appended_since_fsync >= 256){
|
||
fsync(w->fd); w->last_fsync_lsn=lsn; w->last_fsync_ms=now; w->appended_since_fsync=0;
|
||
}
|
||
}
|
||
static int wal_append(EngramPagedStore* s, uint8_t op, const uint8_t* payload,
|
||
uint32_t plen, uint64_t lsn){
|
||
EngramWal* w = s->wal; if (!w) return 0;
|
||
size_t fl = WAL_HDR + plen;
|
||
uint8_t* fr = (uint8_t*)malloc(fl);
|
||
if (!fr) return -1;
|
||
put_u32(fr + 0, WAL_MAGIC);
|
||
put_u32(fr + 4, plen);
|
||
fr[8] = op; fr[9] = 0;
|
||
put_u64(fr + 10, lsn);
|
||
put_u32(fr + 18, 0);
|
||
if (plen) memcpy(fr + WAL_HDR, payload, plen);
|
||
/* crc over op|flags|lsn|payload — these are NOT contiguous in the frame (the
|
||
* crc field sits between), so hash a scratch copy that omits it. */
|
||
uint8_t* cb = (uint8_t*)malloc(10 + plen);
|
||
if (!cb){ free(fr); return -1; }
|
||
memcpy(cb, fr + 8, 10);
|
||
if (plen) memcpy(cb + 10, payload, plen);
|
||
uint32_t crc = crc32_buf(cb, 10 + plen);
|
||
free(cb);
|
||
put_u32(fr + 18, crc);
|
||
ssize_t wr = write(w->fd, fr, fl);
|
||
free(fr);
|
||
if (wr != (ssize_t)fl) return -1;
|
||
w->appended_since_fsync++;
|
||
wal_maybe_fsync(s, lsn);
|
||
return 0;
|
||
}
|
||
static int wal_reclaim(EngramPagedStore* s, uint64_t ckpt_lsn){
|
||
EngramWal* w = s->wal; if (!w) return 0;
|
||
if (ftruncate(w->fd, 0) != 0) return -1; /* prefix <= ckpt reclaimed */
|
||
uint8_t p[8]; put_u64(p, ckpt_lsn);
|
||
if (wal_append(s, OP_CHECKPOINT, p, 8, ckpt_lsn) != 0) return -1;
|
||
fsync(w->fd); w->last_fsync_ms = now_ms();
|
||
return 0;
|
||
}
|
||
|
||
/* ── idempotency: max home-page LSN of the live/dead record(s) for `id` ──────── */
|
||
static uint64_t max_page_lsn_for_id(EngramPagedStore* s, const char* id, int want_edge){
|
||
uint8_t* locs; size_t n;
|
||
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return 0;
|
||
uint64_t best = 0;
|
||
for (size_t i=0;i<n;i++){
|
||
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
|
||
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
|
||
uint8_t pg[STORE_PAGE_SIZE];
|
||
if (page_read(s, page, pg) != 0) continue;
|
||
uint8_t pt = pg[8];
|
||
if (want_edge && pt != STORE_PT_EDGE) continue;
|
||
if (!want_edge && pt != STORE_PT_NODE) continue;
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live) != 0) continue;
|
||
int match;
|
||
if (want_edge){ StoreEdge e; edge_parse(body,blen,&e); match = e.id && !strcmp(e.id,id); store_edge_free(&e); }
|
||
else { StoreNode nn; node_parse(body,blen,&nn); match = nn.id && !strcmp(nn.id,id); store_node_free(&nn); }
|
||
free(body);
|
||
if (!match) continue;
|
||
uint64_t l = get_u64(pg + 16);
|
||
if (l > best) best = l;
|
||
}
|
||
free(locs);
|
||
return best;
|
||
}
|
||
|
||
/* Mark live record(s) for `id` DEAD where the home page LSN < lsn. `want_edge`
|
||
* disambiguates node vs edge records (page type). Caller sets s->stamp_lsn. */
|
||
static void kill_live_id(EngramPagedStore* s, const char* id, int want_edge, uint64_t lsn){
|
||
uint8_t* locs; size_t n;
|
||
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return;
|
||
for (size_t i=0;i<n;i++){
|
||
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
|
||
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, page, buf) != 0) continue;
|
||
if (get_u64(buf + 16) >= lsn) continue;
|
||
if (buf[8] != (want_edge ? STORE_PT_EDGE : STORE_PT_NODE)) continue;
|
||
if (slot >= slp_count(buf)) continue;
|
||
uint16_t off,len,fl; slp_slot(buf, slot, &off, &len, &fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live) != 0) continue;
|
||
int match;
|
||
if (want_edge){ StoreEdge e; edge_parse(body,blen,&e); match = e.id && !strcmp(e.id,id); store_edge_free(&e); }
|
||
else { StoreNode nn; node_parse(body,blen,&nn); match = nn.id && !strcmp(nn.id,id); store_node_free(&nn); }
|
||
free(body);
|
||
if (!match) continue;
|
||
slp_set_slot(buf, slot, off, len, SLOT_DEAD);
|
||
page_write(s, page, buf);
|
||
}
|
||
free(locs);
|
||
}
|
||
|
||
/* ── apply routines (used by both the normal write path and WAL replay) ──────── */
|
||
static int apply_node_put(EngramPagedStore* s, const StoreNode* n, uint64_t lsn){
|
||
if (!s || !n || !n->id) return -1;
|
||
if (max_page_lsn_for_id(s, n->id, 0) >= lsn) return 0; /* already durable */
|
||
uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn;
|
||
int r = node_place(s, n); /* re-put appends; reads dedup */
|
||
s->stamp_lsn = prev;
|
||
return r;
|
||
}
|
||
static int apply_edge_put(EngramPagedStore* s, const StoreEdge* e, uint64_t lsn){
|
||
if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1;
|
||
if (max_page_lsn_for_id(s, e->id, 1) >= lsn) return 0; /* already durable */
|
||
uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn;
|
||
kill_live_id(s, e->id, 1, lsn); /* supersede prior versions (no adjacency dup) */
|
||
int r = edge_place(s, e);
|
||
s->stamp_lsn = prev;
|
||
return r;
|
||
}
|
||
|
||
/* ── layer registry (single slotted page rooted at layer_registry_page) ──────── */
|
||
enum { LT_ID=1, LT_NAME, LT_PRIO, LT_SUPPRESS, LT_TRANSPARENT, LT_INJECTABLE };
|
||
|
||
static uint8_t* layer_serialize(const StoreLayer* L, size_t* out_len){
|
||
Buf b = {0,0,0};
|
||
tlv_u32(&b, LT_ID, L->layer_id);
|
||
tlv_str(&b, LT_NAME, L->name);
|
||
tlv_u32(&b, LT_PRIO, L->activation_priority);
|
||
tlv_i32(&b, LT_SUPPRESS, L->suppressible);
|
||
tlv_i32(&b, LT_TRANSPARENT, L->transparent);
|
||
tlv_i32(&b, LT_INJECTABLE, L->injectable);
|
||
if (L->unknown && L->unknown_len) buf_raw(&b, L->unknown, L->unknown_len);
|
||
*out_len = b.len; return b.p;
|
||
}
|
||
static void layer_parse(const uint8_t* body, size_t len, StoreLayer* L){
|
||
memset(L, 0, sizeof *L);
|
||
Buf unk = {0,0,0};
|
||
size_t i = 0;
|
||
while (i + 5 <= len){
|
||
uint8_t tag = body[i]; uint32_t flen = get_u32(body+i+1);
|
||
if (i + 5 + flen > len) break;
|
||
const uint8_t* v = body + i + 5;
|
||
switch (tag){
|
||
case LT_ID: L->layer_id = get_u32(v); break;
|
||
case LT_NAME: L->name = dup_str(v, flen); break;
|
||
case LT_PRIO: L->activation_priority = get_u32(v); break;
|
||
case LT_SUPPRESS: L->suppressible = (int32_t)get_u32(v); break;
|
||
case LT_TRANSPARENT: L->transparent = (int32_t)get_u32(v); break;
|
||
case LT_INJECTABLE: L->injectable = (int32_t)get_u32(v); break;
|
||
default: buf_raw(&unk, body+i, 5+flen); break;
|
||
}
|
||
i += 5 + flen;
|
||
}
|
||
L->unknown = unk.p; L->unknown_len = unk.len;
|
||
}
|
||
void store_layer_free(StoreLayer* L){
|
||
if (!L) return; free(L->name); free(L->unknown); memset(L, 0, sizeof *L);
|
||
}
|
||
void store_layers_free(StoreLayer* arr, size_t n){
|
||
if (!arr) return; for (size_t i=0;i<n;i++) store_layer_free(&arr[i]); free(arr);
|
||
}
|
||
/* Idempotency for layers: whole-registry-page LSN (all layers share one page). */
|
||
static int apply_layer_put(EngramPagedStore* s, const StoreLayer* L, uint64_t lsn){
|
||
if (!s || !L) return -1;
|
||
uint64_t pg = s->layer_registry_page;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, pg, buf) != 0) return -1;
|
||
/* already-applied? a live record with this layer_id and page LSN >= lsn. */
|
||
if (get_u64(buf + 16) >= lsn){
|
||
int nslot = slp_count(buf);
|
||
for (int i=0;i<nslot;i++){
|
||
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
|
||
int hit = (c.layer_id == L->layer_id); store_layer_free(&c);
|
||
if (hit) return 0;
|
||
}
|
||
}
|
||
uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn;
|
||
/* supersede prior live record(s) with this layer_id */
|
||
int nslot = slp_count(buf);
|
||
for (int i=0;i<nslot;i++){
|
||
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
|
||
int hit = (c.layer_id == L->layer_id); store_layer_free(&c);
|
||
if (hit) slp_set_slot(buf, i, off, len, SLOT_DEAD);
|
||
}
|
||
/* append new layer record */
|
||
size_t blen; uint8_t* body = layer_serialize(L, &blen);
|
||
if (!body){ s->stamp_lsn=prev; return -1; }
|
||
uint16_t reclen = (uint16_t)(REC_HDR + blen);
|
||
uint8_t* rec = (uint8_t*)malloc(reclen);
|
||
if (!rec){ free(body); s->stamp_lsn=prev; return -1; }
|
||
put_u16(rec, reclen); rec[2]=REC_VER; rec[3]=0;
|
||
memcpy(rec + REC_HDR, body, blen); free(body);
|
||
int slot = slp_put(buf, rec, reclen); free(rec);
|
||
if (slot < 0){ s->stamp_lsn=prev; return -1; } /* registry page full (M2: single page) */
|
||
int rc = page_write(s, pg, buf);
|
||
s->stamp_lsn = prev;
|
||
return rc;
|
||
}
|
||
static int apply_layer_del(EngramPagedStore* s, uint32_t layer_id, uint64_t lsn){
|
||
uint64_t pg = s->layer_registry_page;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, pg, buf) != 0) return -1;
|
||
if (get_u64(buf + 16) >= lsn) return 0;
|
||
uint64_t prev = s->stamp_lsn; s->stamp_lsn = lsn;
|
||
int nslot = slp_count(buf), changed=0;
|
||
for (int i=0;i<nslot;i++){
|
||
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
|
||
int hit = (c.layer_id == layer_id); store_layer_free(&c);
|
||
if (hit){ slp_set_slot(buf, i, off, len, SLOT_DEAD); changed=1; }
|
||
}
|
||
int rc = changed ? page_write(s, pg, buf) : 0;
|
||
s->stamp_lsn = prev;
|
||
return rc;
|
||
}
|
||
int store_get_layer(EngramPagedStore* s, uint32_t layer_id, StoreLayer* out){
|
||
if (!s || !out) return -1;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, s->layer_registry_page, buf) != 0) return -1;
|
||
int nslot = slp_count(buf), found=0;
|
||
for (int i=0;i<nslot;i++){
|
||
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
|
||
if (c.layer_id == layer_id){ if (found) store_layer_free(out); *out=c; found=1; }
|
||
else store_layer_free(&c);
|
||
}
|
||
return found ? 1 : 0;
|
||
}
|
||
int store_list_layers(EngramPagedStore* s, StoreLayer** out, size_t* n){
|
||
if (!s || !out || !n) return -1;
|
||
*out = NULL; *n = 0;
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
if (page_read(s, s->layer_registry_page, buf) != 0) return -1;
|
||
StoreLayer* arr=NULL; size_t used=0, cap=0;
|
||
int nslot = slp_count(buf);
|
||
for (int i=0;i<nslot;i++){
|
||
uint16_t off,len,fl; slp_slot(buf,i,&off,&len,&fl);
|
||
if (fl != SLOT_LIVE) continue;
|
||
StoreLayer c; layer_parse(buf+off+REC_HDR, (size_t)(len-REC_HDR), &c);
|
||
/* keep only the latest live per layer_id (supersede leaves one live) */
|
||
int dup=0;
|
||
for (size_t k=0;k<used;k++) if (arr[k].layer_id==c.layer_id){ dup=1; break; }
|
||
if (dup){ store_layer_free(&c); continue; }
|
||
if (used==cap){ cap=cap?cap*2:8; StoreLayer* na=(StoreLayer*)realloc(arr,cap*sizeof*na); if(!na){store_layer_free(&c);break;} arr=na; }
|
||
arr[used++]=c;
|
||
}
|
||
*out=arr; *n=used;
|
||
return 0;
|
||
}
|
||
|
||
/* ── edge lookup by id (latest live) ─────────────────────────────────────────── */
|
||
int store_get_edge(EngramPagedStore* s, const char* id, StoreEdge* out){
|
||
if (!s || !id || !out) return -1;
|
||
uint8_t* locs; size_t n;
|
||
if (btree_lookup(s, TREE_PRIMARY, id_hash(id), &locs, &n) != 0) return -1;
|
||
int found = 0;
|
||
for (size_t i=0;i<n;i++){
|
||
uint64_t page = get_u64(locs + i*PRIMARY_PAYLOAD);
|
||
uint16_t slot = get_u16(locs + i*PRIMARY_PAYLOAD + 8);
|
||
uint8_t pg[STORE_PAGE_SIZE];
|
||
if (page_read(s, page, pg) != 0) continue;
|
||
if (pg[8] != STORE_PT_EDGE) continue;
|
||
uint8_t* body; size_t blen; int live;
|
||
if (read_body(s, page, slot, &body, &blen, &live) != 0) continue;
|
||
StoreEdge cand; edge_parse(body, blen, &cand); free(body);
|
||
if (cand.id && strcmp(cand.id, id)==0 && live){
|
||
if (found) store_edge_free(out);
|
||
*out = cand; out->tombstoned = 0; found = 1;
|
||
} else store_edge_free(&cand);
|
||
}
|
||
free(locs);
|
||
return found ? 1 : 0;
|
||
}
|
||
|
||
/* ── hebb batch apply ────────────────────────────────────────────────────────
|
||
* payload: [u32 count] then count × [u32 id_len][id][f64 hebb][i64 last_fired]. */
|
||
static int apply_hebb_batch(EngramPagedStore* s, const uint8_t* p, uint32_t plen, uint64_t lsn){
|
||
if (plen < 4) return -1;
|
||
uint32_t cnt = get_u32(p); size_t off = 4;
|
||
for (uint32_t i=0;i<cnt;i++){
|
||
if (off + 4 > plen) break;
|
||
uint32_t idl = get_u32(p+off); off += 4;
|
||
if (off + idl + 16 > plen) break;
|
||
char* id = dup_str(p+off, idl); off += idl;
|
||
double hebb; { uint64_t u=get_u64(p+off); memcpy(&hebb,&u,8);} off += 8;
|
||
int64_t lf = (int64_t)get_u64(p+off); off += 8;
|
||
if (id){
|
||
StoreEdge e;
|
||
if (store_get_edge(s, id, &e) == 1){
|
||
e.hebb = hebb; e.last_fired = lf;
|
||
apply_edge_put(s, &e, lsn);
|
||
store_edge_free(&e);
|
||
}
|
||
free(id);
|
||
}
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
/* ── WAL replay dispatch ─────────────────────────────────────────────────────── */
|
||
static void wal_dispatch(EngramPagedStore* s, uint8_t op, const uint8_t* p, uint32_t plen, uint64_t lsn){
|
||
switch (op){
|
||
case OP_NODE_PUT: { StoreNode n; node_parse(p,plen,&n); apply_node_put(s,&n,lsn); store_node_free(&n); break; }
|
||
case OP_EDGE_PUT: { StoreEdge e; edge_parse(p,plen,&e); apply_edge_put(s,&e,lsn); store_edge_free(&e); break; }
|
||
case OP_TOMBSTONE:
|
||
case OP_FORGET: { char* id=dup_str(p,plen); if(id){ tombstone_core(s,id,lsn); free(id);} break; }
|
||
case OP_SUPERSEDE:{ if(plen>=4){ uint32_t ol=get_u32(p); if(4+ol<=plen){ char* old=dup_str(p+4,ol); if(old){ tombstone_core(s,old,lsn); free(old);} } } break; }
|
||
case OP_LAYER_PUT:{ StoreLayer L; layer_parse(p,plen,&L); apply_layer_put(s,&L,lsn); store_layer_free(&L); break; }
|
||
case OP_LAYER_DEL:{ if(plen>=4) apply_layer_del(s, get_u32(p), lsn); break; }
|
||
case OP_HEBB_BATCH: apply_hebb_batch(s, p, plen, lsn); break;
|
||
case OP_CHECKPOINT: break;
|
||
default: break;
|
||
}
|
||
}
|
||
static int wal_replay(EngramPagedStore* s){
|
||
EngramWal* w = s->wal; if (!w) return 0;
|
||
struct stat st; if (fstat(w->fd, &st) != 0) return -1;
|
||
off_t size = st.st_size, off = 0;
|
||
uint64_t maxlsn = 0;
|
||
for (;;){
|
||
if (off + (off_t)WAL_HDR > size) break; /* torn/short tail */
|
||
uint8_t hdr[WAL_HDR];
|
||
if (pread(w->fd, hdr, WAL_HDR, off) != (ssize_t)WAL_HDR) break;
|
||
if (get_u32(hdr) != WAL_MAGIC) break;
|
||
uint32_t plen = get_u32(hdr + 4);
|
||
if (off + (off_t)WAL_HDR + plen > size) break; /* torn tail */
|
||
uint32_t crc_stored = get_u32(hdr + 18);
|
||
uint8_t* fr = (uint8_t*)malloc(10 + plen);
|
||
if (!fr) return -1;
|
||
memcpy(fr, hdr + 8, 10); /* op|flags|lsn */
|
||
if (plen && pread(w->fd, fr+10, plen, off+WAL_HDR) != (ssize_t)plen){ free(fr); break; }
|
||
if (crc32_buf(fr, 10+plen) != crc_stored){ free(fr); break; } /* bad crc → stop */
|
||
uint8_t op = fr[0]; uint64_t lsn = get_u64(fr+2);
|
||
if (lsn > maxlsn) maxlsn = lsn;
|
||
if (lsn > s->last_checkpoint_lsn) wal_dispatch(s, op, fr+10, plen, lsn);
|
||
free(fr);
|
||
off += (off_t)WAL_HDR + plen;
|
||
}
|
||
if (maxlsn > s->next_lsn) s->next_lsn = maxlsn;
|
||
return 0;
|
||
}
|
||
|
||
/* Heal torn DATA pages (bad crc) to empty pages of their type, so redo re-derives
|
||
* their records from the WAL. Index/superblock recovery uses the SB mirror + full
|
||
* replay-from-checkpoint; torn index pages are left for store_check to report. */
|
||
static void heal_torn_pages(EngramPagedStore* s){
|
||
/* Scan by the ACTUAL file size, not the SB page_count: a steal (dirty pages
|
||
* flushed without advancing the checkpoint SB) can leave valid/torn pages
|
||
* beyond the recorded page_count. Bump page_count so redo allocations never
|
||
* collide with those on-disk ghost pages. */
|
||
struct stat fst;
|
||
uint64_t file_pages = (fstat(s->fd,&fst)==0) ? (uint64_t)(fst.st_size/STORE_PAGE_SIZE) : s->page_count;
|
||
if (file_pages > s->page_count) s->page_count = file_pages;
|
||
for (uint64_t id=2; id<file_pages; id++){
|
||
uint8_t buf[STORE_PAGE_SIZE];
|
||
off_t off = (off_t)id*STORE_PAGE_SIZE;
|
||
if (pread(s->fd, buf, STORE_PAGE_SIZE, off) != (ssize_t)STORE_PAGE_SIZE) continue;
|
||
if (page_crc_ok(buf)) continue;
|
||
uint8_t t = buf[8];
|
||
uint8_t nb[STORE_PAGE_SIZE]; memset(nb, 0, sizeof nb);
|
||
if (t==STORE_PT_NODE || t==STORE_PT_EDGE){
|
||
nb[8]=t; put_u16(nb+10,0); put_u16(nb+12,(uint16_t)(STORE_PAGE_SIZE-STORE_HDR));
|
||
} else if (t==STORE_PT_OVERFLOW){
|
||
nb[8]=t; put_u64(nb+OVF_NEXT_OFF,0); put_u32(nb+OVF_LEN_OFF,0);
|
||
} else continue; /* don't heal index/free/garbage */
|
||
put_u64(nb+0, id); put_u64(nb+16, 0); put_u32(nb+24, 0);
|
||
put_u32(nb+24, crc32_buf(nb, STORE_PAGE_SIZE));
|
||
pc_put(s, id, nb, 1); /* dirty → persisted next checkpoint */
|
||
}
|
||
}
|
||
static int wal_recover(EngramPagedStore* s){
|
||
s->recovering = 1;
|
||
heal_torn_pages(s);
|
||
int rc = wal_replay(s);
|
||
s->recovering = 0;
|
||
return rc;
|
||
}
|
||
|
||
/* ── checkpoint threshold trigger ────────────────────────────────────────────── */
|
||
static void ckpt_maybe(EngramPagedStore* s){
|
||
if (s->recovering) return;
|
||
s->ops_since_ckpt++;
|
||
if (s->ckpt_threshold && s->ops_since_ckpt >= s->ckpt_threshold)
|
||
engram_checkpoint(s);
|
||
}
|
||
|
||
/* ── public mutation entry points (log-then-apply when a WAL is attached) ─────── */
|
||
int store_put_node(EngramPagedStore* s, const StoreNode* n){
|
||
if (!s || !n || !n->id) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
if (s->wal){
|
||
size_t blen; uint8_t* body = node_serialize(n, &blen);
|
||
if (!body) return -1;
|
||
int wr = wal_append(s, OP_NODE_PUT, body, (uint32_t)blen, L);
|
||
free(body);
|
||
if (wr != 0) return -1;
|
||
}
|
||
int r = apply_node_put(s, n, L);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_put_edge(EngramPagedStore* s, const StoreEdge* e){
|
||
if (!s || !e || !e->id || !e->from_id || !e->to_id) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
if (s->wal){
|
||
size_t blen; uint8_t* body = edge_serialize(e, &blen);
|
||
if (!body) return -1;
|
||
int wr = wal_append(s, OP_EDGE_PUT, body, (uint32_t)blen, L);
|
||
free(body);
|
||
if (wr != 0) return -1;
|
||
}
|
||
int r = apply_edge_put(s, e, L);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_tombstone(EngramPagedStore* s, const char* id){
|
||
if (!s || !id) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
if (s->wal){
|
||
if (wal_append(s, OP_TOMBSTONE, (const uint8_t*)id, (uint32_t)strlen(id), L) != 0) return -1;
|
||
}
|
||
int r = tombstone_core(s, id, L);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_forget(EngramPagedStore* s, const char* id){
|
||
if (!s || !id) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
if (s->wal){
|
||
if (wal_append(s, OP_FORGET, (const uint8_t*)id, (uint32_t)strlen(id), L) != 0) return -1;
|
||
}
|
||
int r = tombstone_core(s, id, L);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_supersede(EngramPagedStore* s, const char* old_id, const char* new_id){
|
||
if (!s || !old_id) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
if (s->wal){
|
||
uint32_t ol=(uint32_t)strlen(old_id), nl=(uint32_t)(new_id?strlen(new_id):0);
|
||
size_t plen = 4+ol+4+nl;
|
||
uint8_t* p = (uint8_t*)malloc(plen);
|
||
if (!p) return -1;
|
||
put_u32(p, ol); memcpy(p+4, old_id, ol);
|
||
put_u32(p+4+ol, nl); if (nl) memcpy(p+8+ol, new_id, nl);
|
||
int wr = wal_append(s, OP_SUPERSEDE, p, (uint32_t)plen, L);
|
||
free(p);
|
||
if (wr != 0) return -1;
|
||
}
|
||
int r = tombstone_core(s, old_id, L);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_put_layer(EngramPagedStore* s, const StoreLayer* L){
|
||
if (!s || !L) return -1;
|
||
uint64_t lsn = ++s->next_lsn;
|
||
if (s->wal){
|
||
size_t blen; uint8_t* body = layer_serialize(L, &blen);
|
||
if (!body) return -1;
|
||
int wr = wal_append(s, OP_LAYER_PUT, body, (uint32_t)blen, lsn);
|
||
free(body);
|
||
if (wr != 0) return -1;
|
||
}
|
||
int r = apply_layer_put(s, L, lsn);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_del_layer(EngramPagedStore* s, uint32_t layer_id){
|
||
if (!s) return -1;
|
||
uint64_t lsn = ++s->next_lsn;
|
||
if (s->wal){
|
||
uint8_t p[4]; put_u32(p, layer_id);
|
||
if (wal_append(s, OP_LAYER_DEL, p, 4, lsn) != 0) return -1;
|
||
}
|
||
int r = apply_layer_del(s, layer_id, lsn);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
int store_hebb_batch(EngramPagedStore* s, const StoreHebbDelta* d, size_t n){
|
||
if (!s || (!d && n)) return -1;
|
||
uint64_t L = ++s->next_lsn;
|
||
/* build payload */
|
||
Buf b = {0,0,0};
|
||
{ uint8_t c[4]; put_u32(c, (uint32_t)n); buf_raw(&b, c, 4); }
|
||
for (size_t i=0;i<n;i++){
|
||
uint32_t idl = (uint32_t)strlen(d[i].edge_id);
|
||
uint8_t t4[4]; put_u32(t4, idl); buf_raw(&b, t4, 4);
|
||
buf_raw(&b, d[i].edge_id, idl);
|
||
uint8_t t8[8]; uint64_t u; memcpy(&u,&d[i].hebb,8); put_u64(t8,u); buf_raw(&b, t8, 8);
|
||
put_u64(t8, (uint64_t)d[i].last_fired); buf_raw(&b, t8, 8);
|
||
}
|
||
if (s->wal){
|
||
if (wal_append(s, OP_HEBB_BATCH, b.p, (uint32_t)b.len, L) != 0){ free(b.p); return -1; }
|
||
}
|
||
int r = apply_hebb_batch(s, b.p, (uint32_t)b.len, L);
|
||
free(b.p);
|
||
ckpt_maybe(s);
|
||
return r;
|
||
}
|
||
|
||
/* ── checkpoint (with test-only crash injection at each step) ─────────────────── */
|
||
int store__checkpoint_crashat(EngramPagedStore* s, int phase){
|
||
if (!s) return -1;
|
||
if (phase == 0){ store__crash(s); return 0; }
|
||
if (pc_flush(s) != 0) return -1; /* 1: dirty pages → disk */
|
||
if (phase == 1){ store__crash(s); return 0; }
|
||
if (fsync(s->fd) != 0) return -1; /* 2: fsync store */
|
||
if (phase == 2){ store__crash(s); return 0; }
|
||
uint64_t C = s->next_lsn; /* 3: advance ckpt LSN + SB */
|
||
s->last_checkpoint_lsn = C;
|
||
s->sb_seq++;
|
||
if (sb_write_one(s, 0) != 0) return -1;
|
||
if (fsync(s->fd) != 0) return -1;
|
||
if (sb_write_one(s, 1) != 0) return -1;
|
||
if (fsync(s->fd) != 0) return -1;
|
||
if (phase == 3){ store__crash(s); return 0; }
|
||
if (wal_reclaim(s, C) != 0) return -1; /* 4: reclaim WAL prefix */
|
||
s->ops_since_ckpt = 0;
|
||
if (phase == 4){ store__crash(s); return 0; }
|
||
return 0;
|
||
}
|
||
int engram_checkpoint(EngramPagedStore* s){ return store__checkpoint_crashat(s, -1); }
|
||
|
||
/* ── crash / steal test hooks ────────────────────────────────────────────────── */
|
||
void store__crash(EngramPagedStore* s){
|
||
if (!s) return; /* abandon RAM: dirty pages lost, WAL as fsync'd */
|
||
if (s->wal){ if (s->wal->fd>=0) close(s->wal->fd); free(s->wal); s->wal=NULL; }
|
||
if (s->cache){ pc_free(s->cache); s->cache=NULL; }
|
||
if (s->fd >= 0) close(s->fd);
|
||
free(s);
|
||
}
|
||
int store__flush_pages(EngramPagedStore* s){
|
||
if (!s) return -1;
|
||
if (pc_flush(s) != 0) return -1; /* steal: dirty pages hit disk, no checkpoint */
|
||
return fsync(s->fd);
|
||
}
|
||
|
||
void engram_set_wal_sync(EngramPagedStore* s, EngramWalSync policy){
|
||
if (s && s->wal) s->wal->sync = policy;
|
||
}
|
||
uint64_t engram_wal_next_lsn(const EngramPagedStore* s){ return s ? s->next_lsn : 0; }
|
||
uint64_t engram_last_checkpoint_lsn(const EngramPagedStore* s){ return s ? s->last_checkpoint_lsn : 0; }
|
||
|
||
/* ══════════════════════════════════════════════════════════════════════════════
|
||
* Minimal JSON reader for the ONE-TIME legacy snapshot.json import.
|
||
* Tolerant scanner over the `engram_save` schema ({nodes,edges,layers}); it is an
|
||
* IMPORT SOURCE only — after import + checkpoint the paged store is authoritative
|
||
* and this path is never taken again (engram_open imports only when the store is
|
||
* absent). JSON is never read as the ongoing store.
|
||
* ════════════════════════════════════════════════════════════════════════════ */
|
||
static const char* js_ws(const char* p){ while (*p==' '||*p=='\t'||*p=='\n'||*p=='\r') p++; return p; }
|
||
static const char* js_skip_string(const char* p){
|
||
p++; /* opening quote */
|
||
while (*p){ if (*p=='\\'){ if(!p[1]) return p+1; p+=2; continue; } if (*p=='"') return p+1; p++; }
|
||
return p;
|
||
}
|
||
static const char* js_skip_value(const char* p){
|
||
p = js_ws(p);
|
||
if (*p=='"') return js_skip_string(p);
|
||
if (*p=='{' || *p=='['){
|
||
char open=*p, close=(open=='{')?'}':']'; int depth=0;
|
||
while (*p){ if (*p=='"'){ p=js_skip_string(p); continue; }
|
||
if (*p==open) depth++; else if (*p==close){ depth--; if(!depth) return p+1; } p++; }
|
||
return p;
|
||
}
|
||
while (*p && *p!=',' && *p!='}' && *p!=']') p++;
|
||
return p;
|
||
}
|
||
/* value of member `key` at the top level of the object at `obj` ('{'), or NULL. */
|
||
static const char* js_member(const char* obj, const char* key){
|
||
const char* p = js_ws(obj);
|
||
if (*p != '{') return NULL;
|
||
p++;
|
||
size_t keylen = strlen(key);
|
||
for (;;){
|
||
p = js_ws(p);
|
||
if (*p=='}' || !*p) return NULL;
|
||
if (*p != '"') return NULL;
|
||
const char* kstart = p+1;
|
||
const char* kend = js_skip_string(p); /* points past closing quote */
|
||
size_t klen = (size_t)(kend-1-kstart);
|
||
const char* colon = js_ws(kend);
|
||
if (*colon != ':') return NULL;
|
||
const char* val = js_ws(colon+1);
|
||
if (klen==keylen && strncmp(kstart,key,klen)==0) return val;
|
||
p = js_ws(js_skip_value(val));
|
||
if (*p==','){ p++; continue; }
|
||
return NULL;
|
||
}
|
||
}
|
||
/* decode a JSON string value into a fresh C string (basic escapes + \uXXXX BMP). */
|
||
static char* js_str(const char* v){
|
||
if (!v || *v!='"') return NULL;
|
||
const char* p = v+1;
|
||
Buf b = {0,0,0};
|
||
while (*p && *p!='"'){
|
||
if (*p=='\\'){
|
||
p++;
|
||
switch (*p){
|
||
case 'n': { char c='\n'; buf_raw(&b,&c,1); } break;
|
||
case 't': { char c='\t'; buf_raw(&b,&c,1); } break;
|
||
case 'r': { char c='\r'; buf_raw(&b,&c,1); } break;
|
||
case 'b': { char c='\b'; buf_raw(&b,&c,1); } break;
|
||
case 'f': { char c='\f'; buf_raw(&b,&c,1); } break;
|
||
case '/': { char c='/'; buf_raw(&b,&c,1); } break;
|
||
case '\\':{ char c='\\'; buf_raw(&b,&c,1); } break;
|
||
case '"': { char c='"'; buf_raw(&b,&c,1); } break;
|
||
case 'u': {
|
||
unsigned h=0; for (int k=0;k<4 && p[1];k++){ char d=p[1]; int x=(d>='0'&&d<='9')?d-'0':(d>='a'&&d<='f')?d-'a'+10:(d>='A'&&d<='F')?d-'A'+10:0; h=h*16+x; p++; }
|
||
if (h<0x80){ char c=(char)h; buf_raw(&b,&c,1); }
|
||
else if (h<0x800){ char c[2]={(char)(0xC0|(h>>6)),(char)(0x80|(h&0x3F))}; buf_raw(&b,c,2); }
|
||
else { char c[3]={(char)(0xE0|(h>>12)),(char)(0x80|((h>>6)&0x3F)),(char)(0x80|(h&0x3F))}; buf_raw(&b,c,3); }
|
||
} break;
|
||
default: if (*p) buf_raw(&b, p, 1); break;
|
||
}
|
||
if (*p) p++;
|
||
} else { buf_raw(&b, p, 1); p++; }
|
||
}
|
||
char z=0; buf_raw(&b,&z,1);
|
||
return (char*)b.p;
|
||
}
|
||
static char* js_str_field(const char* obj, const char* key){
|
||
const char* v = js_member(obj, key);
|
||
return (v && *v=='"') ? js_str(v) : NULL;
|
||
}
|
||
static double js_num_field(const char* obj, const char* key){
|
||
const char* v = js_member(obj, key);
|
||
if (!v || *v=='"' || *v=='{' || *v=='[') return 0.0;
|
||
return strtod(v, NULL);
|
||
}
|
||
static int64_t js_int_field(const char* obj, const char* key){
|
||
const char* v = js_member(obj, key);
|
||
if (!v || *v=='"' || *v=='{' || *v=='[') return 0;
|
||
return (int64_t)strtoll(v, NULL, 10);
|
||
}
|
||
/* iterate array elements: first element (or NULL), and next element from a ptr. */
|
||
static const char* js_array_first(const char* arr){
|
||
if (!arr) return NULL;
|
||
arr = js_ws(arr);
|
||
if (*arr != '[') return NULL;
|
||
arr = js_ws(arr+1);
|
||
return (*arr==']') ? NULL : arr;
|
||
}
|
||
static const char* js_array_next(const char* elem){
|
||
const char* p = js_ws(js_skip_value(elem));
|
||
return (*p==',') ? js_ws(p+1) : NULL;
|
||
}
|
||
|
||
static int import_snapshot(EngramPagedStore* s, const char* path){
|
||
FILE* f = fopen(path, "rb");
|
||
if (!f) return -1;
|
||
fseek(f, 0, SEEK_END); long sz = ftell(f); fseek(f, 0, SEEK_SET);
|
||
if (sz < 0){ fclose(f); return -1; }
|
||
char* txt = (char*)malloc((size_t)sz + 1);
|
||
if (!txt){ fclose(f); return -1; }
|
||
size_t rd = fread(txt, 1, (size_t)sz, f); fclose(f);
|
||
txt[rd] = 0;
|
||
const char* root = js_ws(txt);
|
||
|
||
/* nodes */
|
||
const char* nodes = js_member(root, "nodes");
|
||
for (const char* o = js_array_first(nodes); o; o = js_array_next(o)){
|
||
StoreNode n; memset(&n, 0, sizeof n);
|
||
n.id = js_str_field(o, "id");
|
||
if (!n.id){ continue; }
|
||
n.content = js_str_field(o, "content");
|
||
n.node_type = js_str_field(o, "node_type");
|
||
n.label = js_str_field(o, "label");
|
||
n.tier = js_str_field(o, "tier");
|
||
n.tags = js_str_field(o, "tags");
|
||
n.metadata = js_str_field(o, "metadata");
|
||
n.salience = js_num_field(o, "salience");
|
||
n.importance = js_num_field(o, "importance");
|
||
n.confidence = js_num_field(o, "confidence");
|
||
n.temporal_decay_rate = js_num_field(o, "temporal_decay_rate");
|
||
n.activation_count = js_int_field(o, "activation_count");
|
||
n.last_activated = js_int_field(o, "last_activated");
|
||
n.created_at = js_int_field(o, "created_at");
|
||
n.updated_at = js_int_field(o, "updated_at");
|
||
n.background_activation = js_num_field(o, "background_activation");
|
||
n.working_memory_weight = js_num_field(o, "working_memory_weight");
|
||
n.suppression_count = (int32_t)js_int_field(o, "suppression_count");
|
||
n.layer_id = (uint32_t)js_int_field(o, "layer_id");
|
||
n.wm_anchor = js_num_field(o, "wm_anchor");
|
||
/* access_ts: chronological "t,t,t" → ring (head=count%K, filled=count) */
|
||
char* ats = js_str_field(o, "access_ts");
|
||
if (ats && *ats){
|
||
int cnt=0; const char* q=ats;
|
||
while (*q && cnt<STORE_BLL_K){ n.access_ts[cnt++] = (int64_t)strtoll(q,NULL,10);
|
||
const char* c=strchr(q,','); if(!c) break; q=c+1; }
|
||
n.access_filled = cnt; n.access_head = cnt % STORE_BLL_K;
|
||
}
|
||
free(ats);
|
||
/* emb: ".4g,.4g,..." → float vector (bit-exact vs strtof of the tokens) */
|
||
char* es = js_str_field(o, "emb");
|
||
if (es && *es){
|
||
int cap=16, cnt=0; float* v=(float*)malloc((size_t)cap*sizeof(float));
|
||
const char* q=es;
|
||
while (*q && v){ if(cnt==cap){ cap*=2; float* nv=(float*)realloc(v,(size_t)cap*sizeof(float)); if(!nv){free(v);v=NULL;break;} v=nv; }
|
||
v[cnt++] = strtof(q, NULL);
|
||
const char* c=strchr(q,','); if(!c) break; q=c+1; }
|
||
n.emb = v; n.emb_dim = v ? cnt : 0;
|
||
}
|
||
free(es);
|
||
store_put_node(s, &n);
|
||
store_node_free(&n);
|
||
}
|
||
/* edges */
|
||
const char* edges = js_member(root, "edges");
|
||
for (const char* o = js_array_first(edges); o; o = js_array_next(o)){
|
||
StoreEdge e; memset(&e, 0, sizeof e);
|
||
e.id = js_str_field(o, "id");
|
||
e.from_id = js_str_field(o, "from_id");
|
||
e.to_id = js_str_field(o, "to_id");
|
||
if (!e.id || !e.from_id || !e.to_id){ store_edge_free(&e); continue; }
|
||
e.relation = js_str_field(o, "relation");
|
||
e.metadata = js_str_field(o, "metadata");
|
||
e.weight = js_num_field(o, "weight");
|
||
e.hebb = js_num_field(o, "hebb"); /* absent ⇒ 0 (legacy emits only if >0) */
|
||
e.confidence = js_num_field(o, "confidence");
|
||
e.created_at = js_int_field(o, "created_at");
|
||
e.updated_at = js_int_field(o, "updated_at");
|
||
e.last_fired = js_int_field(o, "last_fired");
|
||
e.inhibitory = (int32_t)js_int_field(o, "inhibitory");
|
||
e.layer_id = (uint32_t)js_int_field(o, "layer_id");
|
||
store_put_edge(s, &e);
|
||
store_edge_free(&e);
|
||
}
|
||
/* layers */
|
||
const char* layers = js_member(root, "layers");
|
||
for (const char* o = js_array_first(layers); o; o = js_array_next(o)){
|
||
StoreLayer L; memset(&L, 0, sizeof L);
|
||
L.layer_id = (uint32_t)js_int_field(o, "layer_id");
|
||
L.name = js_str_field(o, "name");
|
||
L.activation_priority = (uint32_t)js_int_field(o, "activation_priority");
|
||
L.suppressible = (int32_t)js_int_field(o, "suppressible");
|
||
L.transparent = (int32_t)js_int_field(o, "transparent");
|
||
L.injectable = (int32_t)js_int_field(o, "injectable");
|
||
store_put_layer(s, &L);
|
||
store_layer_free(&L);
|
||
}
|
||
free(txt);
|
||
return 0;
|
||
}
|
||
|
||
/* ── durable-engram boot / close ─────────────────────────────────────────────── */
|
||
EngramPagedStore* engram_open(const char* data_dir){
|
||
if (!data_dir) return NULL;
|
||
char store_path[1200], wal_path[1200], snap_path[1200];
|
||
snprintf(store_path, sizeof store_path, "%s/engram.store", data_dir);
|
||
snprintf(wal_path, sizeof wal_path, "%s/engram.wal", data_dir);
|
||
snprintf(snap_path, sizeof snap_path, "%s/snapshot.json", data_dir);
|
||
|
||
EngramWalSync sync = ENGRAM_WAL_GROUP;
|
||
const char* env = getenv("ENGRAM_WAL_SYNC");
|
||
if (env){ if(!strcmp(env,"always")) sync=ENGRAM_WAL_ALWAYS;
|
||
else if(!strcmp(env,"off")) sync=ENGRAM_WAL_OFF;
|
||
else sync=ENGRAM_WAL_GROUP; }
|
||
|
||
struct stat st;
|
||
int store_exists = (stat(store_path, &st) == 0);
|
||
EngramPagedStore* s;
|
||
|
||
if (store_exists){
|
||
s = store_open(store_path);
|
||
if (!s) return NULL;
|
||
s->wal = wal_open(wal_path, sync);
|
||
if (!s->wal){ store_close(s); return NULL; }
|
||
s->ckpt_threshold = 100000;
|
||
wal_recover(s); /* replay post-checkpoint tail */
|
||
return s;
|
||
}
|
||
/* Fresh: born on the final paged format. JSON (if any) is imported ONCE. */
|
||
s = store_create(store_path);
|
||
if (!s) return NULL;
|
||
s->wal = wal_open(wal_path, sync);
|
||
if (!s->wal){ store_close(s); return NULL; }
|
||
s->ckpt_threshold = 100000;
|
||
if (stat(snap_path, &st) == 0) import_snapshot(s, snap_path);
|
||
engram_checkpoint(s); /* store is now authoritative */
|
||
return s;
|
||
}
|
||
int engram_close(EngramPagedStore* s){
|
||
if (!s) return -1;
|
||
engram_checkpoint(s);
|
||
return store_close(s);
|
||
}
|