diff --git a/Makefile.cbm b/Makefile.cbm index f1402881b..a1839b52b 100644 --- a/Makefile.cbm +++ b/Makefile.cbm @@ -449,7 +449,7 @@ PREPROCESSOR_SRC = $(CBM_DIR)/preprocessor.cpp SQLITE_WRITER_SRC = $(CBM_DIR)/sqlite_writer.c # Store module (new) -STORE_SRCS = src/store/store.c src/store/architecture_projection.c src/store/store_impact.c src/store/store_graph_digest.c src/store/store_content_digest.c +STORE_SRCS = src/store/store.c src/store/graph_mem.c src/store/architecture_projection.c src/store/store_impact.c src/store/store_graph_digest.c src/store/store_content_digest.c # Cypher module (new) CYPHER_SRCS = src/cypher/cypher.c @@ -744,6 +744,7 @@ TEST_STORE_SRCS = \ tests/test_store_impact.c \ tests/test_store_scope.c \ tests/test_store_graph_digest.c \ + tests/test_graph_mem.c \ tests/test_store_search.c \ tests/test_store_arch.c \ tests/test_architecture_projection.c \ diff --git a/src/store/graph_mem.c b/src/store/graph_mem.c new file mode 100644 index 000000000..02d5d6cd9 --- /dev/null +++ b/src/store/graph_mem.c @@ -0,0 +1,772 @@ +/* + * graph_mem.c — read-only in-memory adjacency over a published store's edges. + * See graph_mem.h for the identity, lifetime and budget contract. + */ +#include "store/graph_mem.h" +#include "foundation/constants.h" +#include "foundation/log.h" +#include "foundation/mem.h" +#include "foundation/mem_core.h" +#include "foundation/platform.h" + +#include +#include +#include +#include + +enum { + GM_MAX_TYPES = 64, /* one bit each in cbm_graph_mem_types_t */ + GM_VERSION_MAX = 96, /* generation + file state, see store_graph_mem */ + GM_CAP_DIVISOR = 4, /* all layers together: at most budget / 4 */ + GM_SET_MIN_CAP = 64, /* BFS visited set: initial slots (power of two) */ + GM_SET_LOAD_NUM = 7, /* grow the visited set past 7/10 load */ + GM_SET_LOAD_DEN = 10, + GM_GROWTH = 2, + GM_COL_TARGET = 1, + GM_COL_TYPE = 2, +}; + +/* Ids and edge counts must fit the uint32 CSR with room for "+1" offsets. */ +#define GM_ID_LIMIT ((int64_t)INT32_MAX - 1) +#define GM_HASH_MUL 2654435761U + +struct cbm_graph_mem { + /* Registry state, guarded by the registry mutex. */ + struct cbm_graph_mem *next; + char *path; + char version[GM_VERSION_MAX]; + int refs; + bool ready; /* built; false while its builder runs */ + bool detached; /* unlinked from the registry; freed at the last release */ + uint64_t last_use; + size_t bytes; /* reserved estimate while building, actual once ready */ + + /* The graph, immutable once ready. Offsets cover node ids 0..n_ids-1: + * the out-edges of id u are out_nbr/out_type[out_off[u] .. out_off[u+1]). */ + uint32_t n_ids; + uint32_t n_edges; + uint32_t *out_off; + uint32_t *out_nbr; /* target of each out-edge */ + uint8_t *out_type; + uint32_t *in_off; + uint32_t *in_nbr; /* source of each in-edge */ + uint8_t *in_type; + int type_count; + char *types[GM_MAX_TYPES]; + size_t type_len[GM_MAX_TYPES]; +}; + +static cbm_graph_mem_t *g_layers; /* linked (not detached) layers */ +static size_t g_total_bytes; /* every live layer, linked or detached */ +static uint64_t g_use_clock; /* logical LRU clock */ +static uint64_t g_builds; + +/* A static SQLite mutex needs no init and works on every platform SQLite + * builds on (SQLITE_THREADSAFE=1). store.c owns APP1. */ +static sqlite3_mutex *gm_registry_mutex(void) { + return sqlite3_mutex_alloc(SQLITE_MUTEX_STATIC_APP2); +} + +static void *gm_alloc(size_t bytes) { + return cbm_alloc(CBM_MEM_CLASS_STORE, bytes ? bytes : 1); +} + +static void gm_free_ptr(void *p) { + cbm_free(CBM_MEM_CLASS_STORE, p); +} + +static void gm_free_graph(cbm_graph_mem_t *g) { + gm_free_ptr(g->out_off); + gm_free_ptr(g->out_nbr); + gm_free_ptr(g->out_type); + gm_free_ptr(g->in_off); + gm_free_ptr(g->in_nbr); + gm_free_ptr(g->in_type); + g->out_off = g->out_nbr = g->in_off = g->in_nbr = NULL; + g->out_type = g->in_type = NULL; + for (int i = 0; i < g->type_count; i++) { + gm_free_ptr(g->types[i]); + g->types[i] = NULL; + } + g->type_count = 0; +} + +static void gm_destroy(cbm_graph_mem_t *g) { + gm_free_graph(g); + gm_free_ptr(g->path); + gm_free_ptr(g); +} + +/* ── Registry (caller holds the mutex) ───────────────────────────── */ + +static cbm_graph_mem_t *gm_find(const char *path, const char *version) { + for (cbm_graph_mem_t *g = g_layers; g; g = g->next) { + if (strcmp(g->path, path) == 0 && strcmp(g->version, version) == 0) { + return g; + } + } + return NULL; +} + +static void gm_unlink(cbm_graph_mem_t *g) { + for (cbm_graph_mem_t **p = &g_layers; *p; p = &(*p)->next) { + if (*p == g) { + *p = g->next; + g->next = NULL; + g->detached = true; + return; + } + } +} + +/* Unlink a layer; free it now when nobody holds it, else at its last release. */ +static void gm_retire(cbm_graph_mem_t *g) { + gm_unlink(g); + if (g->refs == 0) { + g_total_bytes -= g->bytes; + gm_destroy(g); + } +} + +/* Older versions of a path are never asked for again by a new store (one + * opened later reads the current version), so they go now. */ +static void gm_retire_other_versions(const char *path, const char *version) { + cbm_graph_mem_t *g = g_layers; + while (g) { + cbm_graph_mem_t *next = g->next; + if (strcmp(g->path, path) == 0 && strcmp(g->version, version) != 0) { + gm_retire(g); + } + g = next; + } +} + +/* Make room for `bytes` under the cap by evicting idle layers, oldest first. + * False when it cannot fit: the cap is unset (no budget) or every remaining + * byte is held by a request. */ +static bool gm_make_room(size_t bytes) { + size_t budget = cbm_mem_budget(); + size_t cap = budget / GM_CAP_DIVISOR; + if (cap == 0 || bytes > cap) { + return false; + } + while (g_total_bytes + bytes > cap) { + cbm_graph_mem_t *victim = NULL; + for (cbm_graph_mem_t *g = g_layers; g; g = g->next) { + if (g->ready && g->refs == 0 && (!victim || g->last_use < victim->last_use)) { + victim = g; + } + } + if (!victim) { + return false; + } + gm_retire(victim); + } + /* The cap bounds the layers; the process as a whole must still fit. */ + return cbm_mem_charged() + bytes <= budget; +} + +/* ── Build ──────────────────────────────────────────────────────── */ + +static bool gm_bounds(sqlite3 *db, int64_t *max_node, int64_t *max_edge) { + sqlite3_stmt *st = NULL; + if (sqlite3_prepare_v2(db, "SELECT (SELECT max(id) FROM nodes), (SELECT max(id) FROM edges);", + -1, &st, NULL) != SQLITE_OK) { + sqlite3_finalize(st); + return false; + } + bool ok = sqlite3_step(st) == SQLITE_ROW; + if (ok) { + *max_node = sqlite3_column_int64(st, 0); + *max_edge = sqlite3_column_int64(st, 1); + } + sqlite3_finalize(st); + return ok && *max_node >= 0 && *max_edge >= 0 && *max_node < GM_ID_LIMIT && + *max_edge < GM_ID_LIMIT; +} + +/* Upper bound of a layer's bytes: edge ids are unique positive rowids, so the + * edge count never exceeds max(edges.id). */ +static size_t gm_estimate(int64_t max_node, int64_t max_edge) { + size_t offsets = ((size_t)max_node + 2) * sizeof(uint32_t); + size_t per_edge = sizeof(uint32_t) + sizeof(uint8_t); + return GM_GROWTH * offsets + GM_GROWTH * (size_t)max_edge * per_edge; +} + +static int gm_intern(cbm_graph_mem_t *g, const unsigned char *text, size_t len, int *last) { + if (*last >= 0 && g->type_len[*last] == len && memcmp(g->types[*last], text, len) == 0) { + return *last; + } + for (int i = 0; i < g->type_count; i++) { + if (g->type_len[i] == len && memcmp(g->types[i], text, len) == 0) { + *last = i; + return i; + } + } + if (g->type_count >= GM_MAX_TYPES) { + return -1; + } + char *copy = gm_alloc(len + 1); + if (!copy) { + return -1; + } + memcpy(copy, text, len); + copy[len] = '\0'; + g->types[g->type_count] = copy; + g->type_len[g->type_count] = len; + *last = g->type_count; + return g->type_count++; +} + +static bool gm_grow_u32(uint32_t **arr, size_t *cap, size_t need) { + if (need <= *cap) { + return true; + } + size_t next = *cap ? *cap : 1; + while (next < need) { + next *= GM_GROWTH; + } + uint32_t *grown = cbm_realloc(CBM_MEM_CLASS_STORE, *arr, next * sizeof(uint32_t)); + if (!grown) { + return false; + } + *arr = grown; + *cap = next; + return true; +} + +/* The out-CSR, filled straight from the scan: rows arrive ordered by + * source_id, so each node's list is contiguous and its start offset is the + * edge count when its first row (or a later id's) shows up. */ +typedef struct { + size_t off_cap; + size_t edge_cap; + uint32_t next_start; /* smallest id whose start offset is not set yet */ + uint32_t max_id; /* largest endpoint seen */ + int64_t prev_src; + int last_type; +} gm_scan_t; + +static bool gm_scan_row(cbm_graph_mem_t *g, gm_scan_t *sc, sqlite3_stmt *st) { + int64_t src = sqlite3_column_int64(st, 0); + int64_t dst = sqlite3_column_int64(st, GM_COL_TARGET); + const unsigned char *type = sqlite3_column_text(st, GM_COL_TYPE); + if (src < 0 || dst < 0 || src >= GM_ID_LIMIT || dst >= GM_ID_LIMIT || src < sc->prev_src || + !type || g->n_edges >= (uint32_t)GM_ID_LIMIT) { + return false; + } + int type_id = gm_intern(g, type, (size_t)sqlite3_column_bytes(st, GM_COL_TYPE), &sc->last_type); + if (type_id < 0 || !gm_grow_u32(&g->out_off, &sc->off_cap, (size_t)src + GM_GROWTH)) { + return false; + } + while (sc->next_start <= (uint32_t)src) { + g->out_off[sc->next_start++] = g->n_edges; + } + if (g->n_edges >= sc->edge_cap) { + size_t next = sc->edge_cap ? sc->edge_cap * GM_GROWTH : 1; + uint32_t *nbr = cbm_realloc(CBM_MEM_CLASS_STORE, g->out_nbr, next * sizeof(uint32_t)); + if (!nbr) { + return false; + } + g->out_nbr = nbr; + uint8_t *types = cbm_realloc(CBM_MEM_CLASS_STORE, g->out_type, next); + if (!types) { + return false; + } + g->out_type = types; + sc->edge_cap = next; + } + g->out_nbr[g->n_edges] = (uint32_t)dst; + g->out_type[g->n_edges] = (uint8_t)type_id; + g->n_edges++; + sc->prev_src = src; + if ((uint32_t)src > sc->max_id) { + sc->max_id = (uint32_t)src; + } + if ((uint32_t)dst > sc->max_id) { + sc->max_id = (uint32_t)dst; + } + return true; +} + +/* The in-CSR from the out-CSR: count, prefix-sum, then place each edge using + * in_off[v] as v's cursor, which leaves in_off shifted by one id — shift it + * back. Iterating sources in id order keeps every in-list sorted by source. */ +static bool gm_build_inbound(cbm_graph_mem_t *g) { + size_t n_ids = g->n_ids; + g->in_off = cbm_calloc(CBM_MEM_CLASS_STORE, (n_ids + 1) * sizeof(uint32_t)); + g->in_nbr = gm_alloc((size_t)g->n_edges * sizeof(uint32_t)); + g->in_type = gm_alloc(g->n_edges); + if (!g->in_off || !g->in_nbr || !g->in_type) { + return false; + } + for (uint32_t e = 0; e < g->n_edges; e++) { + g->in_off[g->out_nbr[e] + 1]++; + } + for (size_t v = 0; v < n_ids; v++) { + g->in_off[v + 1] += g->in_off[v]; + } + for (uint32_t u = 0; u < g->n_ids; u++) { + for (uint32_t e = g->out_off[u]; e < g->out_off[u + 1]; e++) { + uint32_t pos = g->in_off[g->out_nbr[e]]++; + g->in_nbr[pos] = u; + g->in_type[pos] = g->out_type[e]; + } + } + for (size_t v = n_ids; v > 0; v--) { + g->in_off[v] = g->in_off[v - 1]; + } + g->in_off[0] = 0; + return true; +} + +static bool gm_build(cbm_graph_mem_t *g, sqlite3 *db, int64_t max_node, int64_t max_edge) { + gm_scan_t sc = {.prev_src = 0, .last_type = -1, .max_id = (uint32_t)max_node}; + sc.edge_cap = (size_t)max_edge; + g->out_nbr = gm_alloc(sc.edge_cap * sizeof(uint32_t)); + g->out_type = gm_alloc(sc.edge_cap); + if (!g->out_nbr || !g->out_type || + !gm_grow_u32(&g->out_off, &sc.off_cap, (size_t)max_node + GM_GROWTH)) { + return false; + } + /* ORDER BY the covering UNIQUE index's leading columns: a sequential walk + * of sqlite_autoindex_edges_1, no sort and no table access. */ + sqlite3_stmt *st = NULL; + if (sqlite3_prepare_v2(db, + "SELECT source_id, target_id, type FROM edges " + "ORDER BY source_id, target_id, type;", + -1, &st, NULL) != SQLITE_OK) { + sqlite3_finalize(st); + return false; + } + int rc; + bool ok = true; + while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + if (!gm_scan_row(g, &sc, st)) { + ok = false; + break; + } + } + sqlite3_finalize(st); + if (!ok || rc != SQLITE_DONE) { + return false; + } + g->n_ids = sc.max_id + 1; + if (!gm_grow_u32(&g->out_off, &sc.off_cap, (size_t)g->n_ids + 1)) { + return false; + } + while (sc.next_start <= g->n_ids) { + g->out_off[sc.next_start++] = g->n_edges; + } + /* max(edges.id) over-counts after deletions; give the slack back. */ + if (g->n_edges < sc.edge_cap) { + size_t keep = g->n_edges ? g->n_edges : 1; + uint32_t *nbr = cbm_realloc(CBM_MEM_CLASS_STORE, g->out_nbr, keep * sizeof(uint32_t)); + uint8_t *types = nbr ? cbm_realloc(CBM_MEM_CLASS_STORE, g->out_type, keep) : NULL; + if (nbr) { + g->out_nbr = nbr; + } + if (types) { + g->out_type = types; + } + } + return gm_build_inbound(g); +} + +static size_t gm_actual_bytes(const cbm_graph_mem_t *g) { + size_t offsets = ((size_t)g->n_ids + 1) * sizeof(uint32_t); + size_t per_edge = sizeof(uint32_t) + sizeof(uint8_t); + size_t bytes = GM_GROWTH * offsets + GM_GROWTH * (size_t)g->n_edges * per_edge; + for (int i = 0; i < g->type_count; i++) { + bytes += g->type_len[i] + 1; + } + return bytes + sizeof(*g) + strlen(g->path) + 1; +} + +/* ── Acquire / release ──────────────────────────────────────────── */ + +/* A ready hit takes a reference; a hit still being built answers NULL (the + * caller uses SQL rather than waiting on another thread's build). */ +static cbm_graph_mem_t *gm_take(cbm_graph_mem_t *hit) { + if (!hit->ready) { + return NULL; + } + hit->refs++; + hit->last_use = ++g_use_clock; + return hit; +} + +static cbm_graph_mem_t *gm_new_entry(const char *db_path, const char *version, size_t est) { + cbm_graph_mem_t *g = cbm_calloc(CBM_MEM_CLASS_STORE, sizeof(*g)); + if (!g) { + return NULL; + } + g->path = cbm_mem_strdup(CBM_MEM_CLASS_STORE, db_path); + if (!g->path) { + gm_free_ptr(g); + return NULL; + } + memcpy(g->version, version, strlen(version) + 1); + g->refs = 1; + g->bytes = est; + return g; +} + +cbm_graph_mem_t *cbm_graph_mem_acquire(sqlite3 *db, const char *db_path, const char *version) { + if (!db || !db_path || !version || !version[0] || strlen(version) >= GM_VERSION_MAX) { + return NULL; + } + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + cbm_graph_mem_t *hit = gm_find(db_path, version); + cbm_graph_mem_t *taken = hit ? gm_take(hit) : NULL; + sqlite3_mutex_leave(mu); + if (hit) { + return taken; + } + + int64_t max_node = 0; + int64_t max_edge = 0; + if (!gm_bounds(db, &max_node, &max_edge)) { + return NULL; + } + size_t est = gm_estimate(max_node, max_edge); + + sqlite3_mutex_enter(mu); + hit = gm_find(db_path, version); + if (hit) { + taken = gm_take(hit); + sqlite3_mutex_leave(mu); + return taken; + } + gm_retire_other_versions(db_path, version); + bool fits = gm_make_room(est); + cbm_graph_mem_t *g = fits ? gm_new_entry(db_path, version, est) : NULL; + if (g) { + g->next = g_layers; + g_layers = g; + g_total_bytes += est; + } + sqlite3_mutex_leave(mu); + if (!g) { + if (!fits) { + char need[CBM_SZ_32]; + (void)snprintf(need, sizeof(need), "%zu", est); + cbm_log_info("graph_mem.skip", "reason", "budget", "bytes", need); + } + return NULL; + } + + /* The build runs outside the mutex: other keys stay servable, and a + * request for this key meanwhile takes the SQL path (gm_take). */ + uint64_t started_ms = cbm_now_ms(); + bool ok = gm_build(g, db, max_node, max_edge); + if (!ok) { + gm_free_graph(g); /* all-or-nothing: no partial layer survives */ + } + size_t actual = ok ? gm_actual_bytes(g) : 0; + if (ok) { + char edges[CBM_SZ_32]; + char bytes[CBM_SZ_32]; + char elapsed[CBM_SZ_32]; + (void)snprintf(edges, sizeof(edges), "%u", g->n_edges); + (void)snprintf(bytes, sizeof(bytes), "%zu", actual); + (void)snprintf(elapsed, sizeof(elapsed), "%llu", + (unsigned long long)(cbm_now_ms() - started_ms)); + cbm_log_info("graph_mem.build", "edges", edges, "bytes", bytes, "elapsed_ms", elapsed); + } + + sqlite3_mutex_enter(mu); + g_total_bytes = g_total_bytes - g->bytes + actual; + g->bytes = actual; + if (ok) { + g->ready = true; + g->last_use = ++g_use_clock; + g_builds++; + } else { + g->refs = 0; + gm_retire(g); + g = NULL; + } + sqlite3_mutex_leave(mu); + return g; +} + +void cbm_graph_mem_release(cbm_graph_mem_t *g) { + if (!g) { + return; + } + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + g->refs--; + if (g->refs == 0 && g->detached) { + g_total_bytes -= g->bytes; + gm_destroy(g); + } + sqlite3_mutex_leave(mu); +} + +/* ── Queries ────────────────────────────────────────────────────── */ + +cbm_graph_mem_types_t cbm_graph_mem_type_mask(const cbm_graph_mem_t *g, const char *const *types, + int count) { + cbm_graph_mem_types_t mask = 0; + for (int i = 0; g && types && i < count; i++) { + if (!types[i]) { + continue; + } + size_t len = strlen(types[i]); + for (int t = 0; t < g->type_count; t++) { + if (g->type_len[t] == len && memcmp(g->types[t], types[i], len) == 0) { + mask |= (cbm_graph_mem_types_t)1 << t; + break; + } + } + } + return mask; +} + +int cbm_graph_mem_degree(const cbm_graph_mem_t *g, int64_t id, bool inbound, + cbm_graph_mem_types_t mask) { + if (!g || id < 0 || id >= (int64_t)g->n_ids) { + return 0; + } + const uint32_t *off = inbound ? g->in_off : g->out_off; + const uint8_t *type = inbound ? g->in_type : g->out_type; + int n = 0; + for (uint32_t e = off[id]; e < off[id + 1]; e++) { + n += (int)((mask >> type[e]) & 1U); + } + return n; +} + +size_t cbm_graph_mem_bytes(const cbm_graph_mem_t *g) { + return g ? g->bytes : 0; +} + +/* Visited set for one BFS: open addressing over id + 1 (0 marks a free slot). */ +typedef struct { + uint32_t *slots; + size_t cap; + size_t count; +} gm_set_t; + +static size_t gm_set_home(uint32_t id, size_t mask) { + return (size_t)(id * GM_HASH_MUL) & mask; +} + +static bool gm_set_grow(gm_set_t *s) { + size_t cap = s->cap ? s->cap * GM_GROWTH : GM_SET_MIN_CAP; + uint32_t *slots = cbm_calloc(CBM_MEM_CLASS_STORE, cap * sizeof(uint32_t)); + if (!slots) { + return false; + } + for (size_t i = 0; i < s->cap; i++) { + uint32_t key = s->slots[i]; + if (key) { + size_t j = gm_set_home(key - 1, cap - 1); + while (slots[j]) { + j = (j + 1) & (cap - 1); + } + slots[j] = key; + } + } + gm_free_ptr(s->slots); + s->slots = slots; + s->cap = cap; + return true; +} + +/* 1 = added, 0 = already present, -1 = out of memory. */ +static int gm_set_add(gm_set_t *s, uint32_t id) { + if ((s->count + 1) * GM_SET_LOAD_DEN > s->cap * GM_SET_LOAD_NUM && !gm_set_grow(s)) { + return -1; + } + size_t mask = s->cap - 1; + for (size_t i = gm_set_home(id, mask);; i = (i + 1) & mask) { + if (s->slots[i] == 0) { + s->slots[i] = id + 1; + s->count++; + return 1; + } + if (s->slots[i] == id + 1) { + return 0; + } + } +} + +typedef struct { + uint32_t *ids; + size_t count; + size_t cap; +} gm_list_t; + +static bool gm_list_push(gm_list_t *l, uint32_t id) { + if (!gm_grow_u32(&l->ids, &l->cap, l->count + 1)) { + return false; + } + l->ids[l->count++] = id; + return true; +} + +typedef struct { + cbm_graph_mem_hop_t *rows; + size_t count; + size_t cap; +} gm_hops_t; + +static bool gm_hops_push(gm_hops_t *h, uint32_t id, int hop) { + if (h->count == h->cap) { + size_t next = h->cap ? h->cap * GM_GROWTH : GM_SET_MIN_CAP; + cbm_graph_mem_hop_t *rows = + cbm_realloc(CBM_MEM_CLASS_STORE, h->rows, next * sizeof(cbm_graph_mem_hop_t)); + if (!rows) { + return false; + } + h->rows = rows; + h->cap = next; + } + h->rows[h->count].id = id; + h->rows[h->count].hop = hop; + h->count++; + return true; +} + +static int gm_hop_id_cmp(const void *pa, const void *pb) { + const cbm_graph_mem_hop_t *a = pa; + const cbm_graph_mem_hop_t *b = pb; + return (a->id > b->id) - (a->id < b->id); +} + +/* One BFS level: expand every frontier node over the masked edges. A node is + * first reached at its shortest distance; the start is never re-entered (its + * neighbours were all reached at hop 1), only recorded once at the length of + * the shortest cycle back to it — the CTE's (start, hop>0) row. */ +typedef struct { + const uint32_t *off; + const uint32_t *nbr; + const uint8_t *type; + cbm_graph_mem_types_t mask; + uint32_t start; + bool start_back; + gm_set_t seen; + gm_hops_t out; +} gm_walk_t; + +static bool gm_expand_level(gm_walk_t *w, const gm_list_t *frontier, gm_list_t *next, int hop) { + for (size_t f = 0; f < frontier->count; f++) { + uint32_t u = frontier->ids[f]; + for (uint32_t e = w->off[u]; e < w->off[u + 1]; e++) { + if (((w->mask >> w->type[e]) & 1U) == 0) { + continue; + } + uint32_t v = w->nbr[e]; + if (v == w->start) { + if (!w->start_back) { + w->start_back = true; + if (!gm_hops_push(&w->out, v, hop)) { + return false; + } + } + continue; + } + int added = gm_set_add(&w->seen, v); + if (added < 0) { + return false; + } + if (added > 0 && (!gm_hops_push(&w->out, v, hop) || !gm_list_push(next, v))) { + return false; + } + } + } + return true; +} + +bool cbm_graph_mem_bfs(const cbm_graph_mem_t *g, int64_t start, bool inbound, + cbm_graph_mem_types_t mask, int max_depth, cbm_graph_mem_hop_t **out, + int *count) { + *out = NULL; + *count = 0; + if (!g || max_depth < 1 || mask == 0 || start < 0 || start >= (int64_t)g->n_ids) { + return true; /* nothing reachable: the CTE yields only the anchor */ + } + gm_walk_t w = { + .off = inbound ? g->in_off : g->out_off, + .nbr = inbound ? g->in_nbr : g->out_nbr, + .type = inbound ? g->in_type : g->out_type, + .mask = mask, + .start = (uint32_t)start, + }; + gm_list_t frontier = {0}; + gm_list_t next = {0}; + bool ok = gm_list_push(&frontier, w.start); + for (int hop = 1; ok && hop <= max_depth && frontier.count > 0; hop++) { + size_t level_begin = w.out.count; + next.count = 0; + ok = gm_expand_level(&w, &frontier, &next, hop); + if (ok && w.out.count - level_begin > 1) { + qsort(w.out.rows + level_begin, w.out.count - level_begin, sizeof(cbm_graph_mem_hop_t), + gm_hop_id_cmp); + } + gm_list_t swap = frontier; + frontier = next; + next = swap; + } + gm_free_ptr(frontier.ids); + gm_free_ptr(next.ids); + gm_free_ptr(w.seen.slots); + if (!ok || w.out.count > (size_t)INT32_MAX) { + gm_free_ptr(w.out.rows); + return false; + } + *out = w.out.rows; + *count = (int)w.out.count; + return true; +} + +void cbm_graph_mem_hops_free(cbm_graph_mem_hop_t *hops) { + gm_free_ptr(hops); +} + +/* ── Registry introspection ─────────────────────────────────────── */ + +int cbm_graph_mem_cached_count(void) { + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + int n = 0; + for (cbm_graph_mem_t *g = g_layers; g; g = g->next) { + n++; + } + sqlite3_mutex_leave(mu); + return n; +} + +size_t cbm_graph_mem_cached_bytes(void) { + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + size_t bytes = g_total_bytes; + sqlite3_mutex_leave(mu); + return bytes; +} + +uint64_t cbm_graph_mem_build_count(void) { + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + uint64_t n = g_builds; + sqlite3_mutex_leave(mu); + return n; +} + +void cbm_graph_mem_drop_idle(void) { + sqlite3_mutex *mu = gm_registry_mutex(); + sqlite3_mutex_enter(mu); + cbm_graph_mem_t *g = g_layers; + while (g) { + cbm_graph_mem_t *next = g->next; + if (g->ready && g->refs == 0) { + gm_retire(g); + } + g = next; + } + sqlite3_mutex_leave(mu); +} diff --git a/src/store/graph_mem.h b/src/store/graph_mem.h new file mode 100644 index 000000000..575a1e148 --- /dev/null +++ b/src/store/graph_mem.h @@ -0,0 +1,95 @@ +/* + * graph_mem.h — read-only in-memory adjacency over a published store's edges. + * + * A graph walk on a kernel-scale index (8.5M nodes, 16.9M edges) spends its + * time in SQLite B-tree seeks: trace_path's recursive CTE took 6.6 s and a + * degree-filtered search_graph 3.9 s, where the same questions on django take + * 20-100 ms. This layer reads the edges ONCE, in one sequential scan of the + * covering UNIQUE(source_id, target_id, type, ...) index, into a typed CSR + * (node id -> out/in neighbours, each with its interned edge type) and answers + * those walks from RAM. Node rows still come from SQLite; only the edge + * structure is held. + * + * Identity: a layer is keyed by (db path, version), where the caller promises + * that one version names one edge set. store.c builds the version from the + * store generation (every publish stamps a new one) plus the rollback-journal + * file state (the header change counter every commit bumps — it also catches + * writers that mutate a published DB in place without a new generation), and + * admits only read-only query stores of a sealed, non-legacy DB. + * + * Lifetime: the daemon closes its query stores after every request, so layers + * live in a process-wide registry, shared by refcount across request threads. + * Every byte is allocated through the memory core (CBM_MEM_CLASS_STORE) and + * the registry is bounded: the layers together stay under a quarter of the + * memory budget, idle ones are evicted oldest-first to make room, and a layer + * that still would not fit (or that would push the process over its budget) is + * not built — the caller keeps the SQL path. A build is all-or-nothing: a + * failed or refused build leaves no partial layer behind. + */ +#ifndef CBM_GRAPH_MEM_H +#define CBM_GRAPH_MEM_H + +#include +#include +#include + +struct sqlite3; + +typedef struct cbm_graph_mem cbm_graph_mem_t; + +/* A set of interned edge types: bit i = the layer's type i. */ +typedef uint64_t cbm_graph_mem_types_t; + +/* One reached node: its id and its shortest distance from the start. */ +typedef struct { + int64_t id; + int hop; +} cbm_graph_mem_hop_t; + +/* The layer for (db_path, version): the cached one, or one built now from + * `db`'s edges table. The caller must hold one read transaction across the + * version read and this call, so the scan sees that version's edges. + * NULL means "use the SQL path": no budget, over the cap, another thread is + * building this key right now, a schema or id range the layer cannot hold, or + * a failed scan. A non-NULL layer is held until cbm_graph_mem_release. */ +cbm_graph_mem_t *cbm_graph_mem_acquire(struct sqlite3 *db, const char *db_path, + const char *version); +void cbm_graph_mem_release(cbm_graph_mem_t *g); + +/* The type set naming `types` (exact byte match, like SQL `type IN (...)`). + * NULL entries and types absent from the graph contribute nothing. */ +cbm_graph_mem_types_t cbm_graph_mem_type_mask(const cbm_graph_mem_t *g, const char *const *types, + int count); + +/* Shortest-walk BFS with the exact semantics of store_bfs's recursive CTE: + * every node reached by a walk of 1..max_depth edges whose types are in + * `mask`, at its MINIMAL such length. The start itself appears only when a + * cycle leads back to it (at the cycle's length). Rows come sorted by + * (hop, id); ids absent from the nodes table are NOT filtered here. + * Returns false only on allocation failure. *out is freed with + * cbm_graph_mem_hops_free. */ +bool cbm_graph_mem_bfs(const cbm_graph_mem_t *g, int64_t start, bool inbound, + cbm_graph_mem_types_t mask, int max_depth, cbm_graph_mem_hop_t **out, + int *count); +void cbm_graph_mem_hops_free(cbm_graph_mem_hop_t *hops); + +/* Number of edges with a type in `mask` ending at (inbound) or leaving + * (outbound) node `id` — SQL's COUNT(*) over target_id/source_id = id. */ +int cbm_graph_mem_degree(const cbm_graph_mem_t *g, int64_t id, bool inbound, + cbm_graph_mem_types_t mask); + +/* Bytes this layer holds (all through the memory core). */ +size_t cbm_graph_mem_bytes(const cbm_graph_mem_t *g); + +/* ── Registry introspection (tests, diagnostics) ─────────────────── */ + +/* Layers currently held by the registry, and their total bytes. */ +int cbm_graph_mem_cached_count(void); +size_t cbm_graph_mem_cached_bytes(void); +/* Builds completed since process start: a test's proof that a call hit the + * cache (unchanged) or built (incremented). */ +uint64_t cbm_graph_mem_build_count(void); +/* Drop every idle layer (refs == 0). Held layers stay until released. */ +void cbm_graph_mem_drop_idle(void); + +#endif /* CBM_GRAPH_MEM_H */ diff --git a/src/store/store.c b/src/store/store.c index d98c07e84..d19e52890 100644 --- a/src/store/store.c +++ b/src/store/store.c @@ -74,10 +74,21 @@ enum { ST_HANDLER_PROP_LEN = 9, ST_BFS_CTE_ROW_MULTIPLIER = 8, ST_BFS_MAX_CTE_ROWS = 4096, + /* SQLite database header (fileformat2 §1.3) */ + ST_DB_HEADER_LEN = 100, + ST_HDR_WRITE_VERSION = 18, /* 1 = rollback journal, 2 = WAL */ + ST_HDR_READ_VERSION = 19, + ST_HDR_ROLLBACK = 1, + ST_HDR_CHANGE_COUNTER = 24, + ST_HDR_PAGE_COUNT = 28, + ST_BE_SHIFT_8 = 8, + ST_BE_SHIFT_16 = 16, + ST_BE_SHIFT_24 = 24, }; #define SLEN(s) (sizeof(s) - 1) #include "store/store.h" +#include "store/graph_mem.h" #include #include "foundation/compat_fs.h" @@ -128,6 +139,14 @@ struct cbm_store { char errbuf[CBM_SZ_512]; bool read_batch; /* a read batch's transaction is open */ uint64_t read_batch_started_ms; /* when it was last (re)begun */ + /* In-memory adjacency (graph_mem.h). Only a read-only query store of a + * sealed (rollback-journal) DB qualifies, keyed by its generation and + * file state (store_rollback_file_state). Acquired on the first walk, + * released at close. */ + bool query_store; + bool graph_mem_tried; + cbm_graph_mem_t *graph_mem; + cbm_graph_mem_types_t graph_mem_degree_mask; /* search's in_deg/out_deg types */ /* Prepared statements (lazily initialized, cached for lifetime) */ sqlite3_stmt *stmt_upsert_node; @@ -1189,6 +1208,7 @@ cbm_store_t *cbm_store_open_path_query(const char *db_path) { } s->db_path = heap_strdup(db_path); + s->query_store = true; /* Security: block ATTACH/DETACH to prevent file creation via SQL injection. */ sqlite3_set_authorizer(s->db, store_authorizer, NULL); @@ -1467,6 +1487,7 @@ void cbm_store_close(cbm_store_t *s) { /* Use sqlite3_close_v2 — auto-deallocates when last statement finalizes. * Prevents ASan false-positive leaks from sqlite3 internal state. */ sqlite3_close_v2(s->db); + cbm_graph_mem_release(s->graph_mem); safe_str_free(&s->db_path); free(s); } @@ -3114,11 +3135,12 @@ static void scan_node(sqlite3_stmt *stmt, cbm_node_t *n) { n->properties_json = heap_strdup((const char *)sqlite3_column_text(stmt, ST_COL_8)); } +static const char st_find_node_by_id_sql[] = + "SELECT id, project, label, name, qualified_name, file_path, " + "start_line, end_line, properties FROM nodes WHERE id = ?1;"; + int cbm_store_find_node_by_id(cbm_store_t *s, int64_t id, cbm_node_t *out) { - sqlite3_stmt *stmt = - prepare_cached(s, &s->stmt_find_node_by_id, - "SELECT id, project, label, name, qualified_name, file_path, " - "start_line, end_line, properties FROM nodes WHERE id = ?1;"); + sqlite3_stmt *stmt = prepare_cached(s, &s->stmt_find_node_by_id, st_find_node_by_id_sql); if (!stmt) { return CBM_STORE_ERR; } @@ -6634,6 +6656,100 @@ int cbm_store_restore_from(cbm_store_t *dst, cbm_store_t *src) { return CBM_STORE_OK; } +/* ── In-memory adjacency ─────────────────────────────────────────── */ + +/* The edge types search_graph's in_deg/out_deg count (cbm_store_search). */ +static const char *const st_degree_types[] = {"CALLS", "USAGE", "CALL_REFERENCE", "INHERITS", + "IMPLEMENTS"}; + +/* cbm_adj_degree(id, inbound): the in-memory twin of search's two + * correlated COUNT(*) subqueries over st_degree_types. */ +static void sqlite_graph_mem_degree(sqlite3_context *ctx, int argc, sqlite3_value **argv) { + (void)argc; + const cbm_store_t *s = sqlite3_user_data(ctx); + sqlite3_result_int(ctx, cbm_graph_mem_degree(s->graph_mem, sqlite3_value_int64(argv[0]), + sqlite3_value_int(argv[SKIP_ONE]) != 0, + s->graph_mem_degree_mask)); +} + +static uint32_t st_be32(const unsigned char *p) { + return ((uint32_t)p[0] << ST_BE_SHIFT_24) | ((uint32_t)p[SKIP_ONE] << ST_BE_SHIFT_16) | + ((uint32_t)p[ST_COL_2] << ST_BE_SHIFT_8) | (uint32_t)p[ST_COL_3]; +} + +/* The committed state of a rollback-journal DB: its header's change counter + * and page count, read through SQLite's own file handle while this + * connection's read transaction holds the SHARED lock, so no commit can land + * mid-read. Every rollback-mode commit bumps the counter (it is how SQLite + * tells other connections their page cache is stale), and switching into or + * out of WAL is itself such a commit. That makes the state exact where the + * store generation is not: the cross-repo pass writes CROSS_* edges straight + * into other projects' published DBs without stamping a generation. A + * WAL-mode file is refused — a WAL commit need not touch the header — which + * is the state of a published DB after any in-place writer switched it to + * WAL; it keeps the SQL path until the next publish seals it again. */ +static bool store_rollback_file_state(cbm_store_t *s, char *out, size_t out_sz) { + sqlite3_file *file = NULL; + unsigned char hdr[ST_DB_HEADER_LEN]; + if (sqlite3_file_control(s->db, "main", SQLITE_FCNTL_FILE_POINTER, (void *)&file) != + SQLITE_OK || + !file || !file->pMethods || + file->pMethods->xRead(file, hdr, (int)sizeof(hdr), 0) != SQLITE_OK) { + return false; + } + if (hdr[ST_HDR_WRITE_VERSION] != ST_HDR_ROLLBACK || + hdr[ST_HDR_READ_VERSION] != ST_HDR_ROLLBACK) { + return false; + } + snprintf(out, out_sz, "c%u.p%u", st_be32(hdr + ST_HDR_CHANGE_COUNTER), + st_be32(hdr + ST_HDR_PAGE_COUNT)); + return true; +} + +/* This store's adjacency layer, or NULL for the SQL path; tried once per + * store. The layer's version is the store generation plus the file state, and + * both — like a miss's edge scan — are read inside one read transaction, so a + * layer holds exactly the edges of the version it is keyed to. */ +static cbm_graph_mem_t *store_graph_mem(cbm_store_t *s) { + if (s->graph_mem_tried) { + return s->graph_mem; + } + s->graph_mem_tried = true; + if (!s->query_store || !s->db || !s->db_path) { + return NULL; + } + bool own_txn = sqlite3_get_autocommit(s->db) != 0; + if (own_txn && sqlite3_exec(s->db, "BEGIN DEFERRED;", NULL, NULL, NULL) != SQLITE_OK) { + return NULL; + } + char generation[CBM_SZ_128]; + char state[CBM_SZ_64]; + char version[CBM_SZ_256]; + /* The generation read is the transaction's first read: the SHARED lock + * the file-state read relies on is held from here on. */ + if (cbm_store_generation(s, generation, sizeof(generation)) == CBM_STORE_OK && + strcmp(generation, "legacy") != 0 && store_rollback_file_state(s, state, sizeof(state))) { + snprintf(version, sizeof(version), "%s/%s", generation, state); + s->graph_mem = cbm_graph_mem_acquire(s->db, s->db_path, version); + } + if (own_txn) { + (void)sqlite3_exec(s->db, "COMMIT;", NULL, NULL, NULL); + } + if (s->graph_mem && + sqlite3_create_function(s->db, "cbm_adj_degree", ST_COL_2, + SQLITE_UTF8 | SQLITE_DETERMINISTIC | SQLITE_DIRECTONLY, s, + sqlite_graph_mem_degree, NULL, NULL) != SQLITE_OK) { + cbm_graph_mem_release(s->graph_mem); + s->graph_mem = NULL; + } + if (s->graph_mem) { + s->graph_mem_degree_mask = + cbm_graph_mem_type_mask(s->graph_mem, st_degree_types, + (int)(sizeof(st_degree_types) / sizeof(st_degree_types[0]))); + } + return s->graph_mem; +} + /* ── Search ─────────────────────────────────────────────────────── */ /* Convert a glob pattern to SQL LIKE pattern. */ @@ -7226,6 +7342,16 @@ int cbm_store_search(cbm_store_t *s, const cbm_search_params_t *params, cbm_sear "(SELECT COUNT(*) FROM edges e WHERE e.source_id = n.id AND " "e.type IN ('CALLS', 'USAGE', 'CALL_REFERENCE', 'INHERITS', " "'IMPLEMENTS')) AS out_deg "; + /* A degree filter evaluates both counts for EVERY candidate row (787k + * Functions on the kernel: two B-tree seeks each, 1.9 s per pass). The + * adjacency layer answers them from RAM; the query is otherwise the same, + * so filter, order, page and total are unchanged. */ + bool has_degree_filter = (params->min_degree >= 0 || params->max_degree >= 0); + if (has_degree_filter && store_graph_mem(s)) { + select_cols = "SELECT n.id, n.project, n.label, n.name, n.qualified_name, " + "n.file_path, n.start_line, n.end_line, n.properties, " + "cbm_adj_degree(n.id, 1) AS in_deg, cbm_adj_degree(n.id, 0) AS out_deg "; + } char where[CBM_SZ_2K] = ""; search_bind_t binds[ST_SEARCH_MAX_BINDS]; @@ -7242,7 +7368,6 @@ int cbm_store_search(cbm_store_t *s, const cbm_search_params_t *params, cbm_sear } /* Degree filters */ - bool has_degree_filter = (params->min_degree >= 0 || params->max_degree >= 0); search_apply_degree_filter(sql, sizeof(sql), params); /* Count query — stripped of per-row edge subqueries for the common (no-degree-filter) @@ -7531,43 +7656,47 @@ static int bfs_cte_row_limit_for_depth(int max_results, int max_depth) { return (int)scaled; } -static int store_bfs(cbm_store_t *s, int64_t start_id, const char *direction, - const char **edge_types, int edge_type_count, int max_depth, int max_results, - bool trail, int max_edges, cbm_traverse_result_t *out) { - memset(out, 0, sizeof(*out)); - if (max_results < 0) { - max_results = 0; +/* One store_bfs walk, shared by its SQL and in-memory paths. */ +typedef struct { + int64_t start_id; + bool inbound; + const char *types_clause; + const char **edge_types; + int edge_type_count; + int max_depth; + int max_results; + int probe_limit; /* max_results + 1: saturation stays observable */ + bool trail; +} bfs_query_t; + +typedef struct { + cbm_node_hop_t *visited; + int count; + int cap; +} bfs_rows_t; + +static void bfs_rows_init(bfs_rows_t *rows) { + rows->cap = ST_INIT_CAP_16; + rows->count = 0; + rows->visited = malloc(rows->cap * sizeof(cbm_node_hop_t)); +} + +static cbm_node_hop_t *bfs_rows_next(bfs_rows_t *rows) { + if (rows->count >= rows->cap) { + rows->cap *= ST_GROWTH; + rows->visited = safe_realloc(rows->visited, rows->cap * sizeof(cbm_node_hop_t)); } + return &rows->visited[rows->count++]; +} - cbm_node_t root = {0}; - int rc = cbm_store_find_node_by_id(s, start_id, &root); - if (rc != CBM_STORE_OK) { - return rc; - } - out->root = root; - - char types_clause[CBM_SZ_512]; - bfs_build_types_clause(edge_type_count, types_clause, (int)sizeof(types_clause)); - - /* Build recursive CTE for BFS */ +/* The recursive CTE: trail (relationship-unique, for Cypher) or shortest-path. */ +static int bfs_visit_sql(cbm_store_t *s, const bfs_query_t *q, bfs_rows_t *rows, int *cte_rows, + int *cte_row_limit) { char sql[CBM_SZ_4K]; - const char *join_cond; - const char *next_id; - bool is_inbound = (direction != NULL) && (strcmp(direction, "inbound") == 0); - - if (is_inbound) { - join_cond = "e.target_id = bfs.node_id"; - next_id = "e.source_id"; - } else { - join_cond = "e.source_id = bfs.node_id"; - next_id = "e.target_id"; - } - - /* One extra row makes max_results saturation observable on both branches. */ - int probe_limit = max_results < INT_MAX ? max_results + SKIP_ONE : max_results; - int cte_row_limit = 0; - if (trail) { - cte_row_limit = bfs_cte_row_limit_for_depth(max_results, max_depth); + const char *join_cond = q->inbound ? "e.target_id = bfs.node_id" : "e.source_id = bfs.node_id"; + const char *next_id = q->inbound ? "e.source_id" : "e.target_id"; + if (q->trail) { + *cte_row_limit = bfs_cte_row_limit_for_depth(q->max_results, q->max_depth); snprintf(sql, sizeof(sql), "WITH RECURSIVE bfs(node_id, hop, edge_path) AS (" " SELECT %lld, 0, ''" @@ -7592,8 +7721,8 @@ static int store_bfs(cbm_store_t *s, int64_t start_id, const char *direction, "(SELECT count(*) FROM bfs) " "FROM bfs JOIN nodes n ON n.id = bfs.node_id " "WHERE bfs.hop > 0 ORDER BY bfs.hop, n.id LIMIT %d;", - (long long)start_id, next_id, join_cond, types_clause, max_depth, - cte_row_limit + SKIP_ONE, probe_limit); + (long long)q->start_id, next_id, join_cond, q->types_clause, q->max_depth, + *cte_row_limit + SKIP_ONE, q->probe_limit); } else { snprintf(sql, sizeof(sql), /* SHORTEST-PATH semantics: the UNION dedupes (node, hop) pairs. @@ -7616,52 +7745,130 @@ static int store_bfs(cbm_store_t *s, int64_t start_id, const char *direction, "GROUP BY n.id " "ORDER BY hop, n.id " "LIMIT %d;", - (long long)start_id, next_id, join_cond, types_clause, max_depth, probe_limit); + (long long)q->start_id, next_id, join_cond, q->types_clause, q->max_depth, + q->probe_limit); } sqlite3_stmt *stmt = NULL; - rc = sqlite3_prepare_v2(s->db, sql, CBM_NOT_FOUND, &stmt, NULL); - if (rc != SQLITE_OK) { + if (sqlite3_prepare_v2(s->db, sql, CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) { store_set_error_sqlite(s, "bfs prepare"); return CBM_STORE_ERR; } /* Bind edge type parameters */ - if (edge_type_count > 0) { - for (int i = 0; i < edge_type_count; i++) { - bind_text(stmt, i + SKIP_ONE, edge_types[i]); + if (q->edge_type_count > 0) { + for (int i = 0; i < q->edge_type_count; i++) { + bind_text(stmt, i + SKIP_ONE, q->edge_types[i]); } } else { bind_text(stmt, SKIP_ONE, "CALLS"); } - int cap = ST_INIT_CAP_16; - int n = 0; - int cte_rows = 0; - cbm_node_hop_t *visited = malloc(cap * sizeof(cbm_node_hop_t)); - + bfs_rows_init(rows); int scan_rc15; while ((scan_rc15 = sqlite3_step(stmt)) == SQLITE_ROW) { - if (n >= cap) { - cap *= ST_GROWTH; - visited = safe_realloc(visited, cap * sizeof(cbm_node_hop_t)); + cbm_node_hop_t *row = bfs_rows_next(rows); + scan_node(stmt, &row->node); + row->hop = sqlite3_column_int(stmt, ST_COL_9); + if (q->trail) { + *cte_rows = sqlite3_column_int(stmt, ST_COL_10); } - scan_node(stmt, &visited[n].node); - visited[n].hop = sqlite3_column_int(stmt, ST_COL_9); - if (trail) { - cte_rows = sqlite3_column_int(stmt, ST_COL_10); - } - n++; } if (scan_rc15 != SQLITE_DONE) { /* SCANCHK:15:stmt */ store_set_error_sqlite(s, "row scan aborted"); sqlite3_finalize(stmt); - out->visited = visited; - out->visited_count = n; return CBM_STORE_ERR; } - sqlite3_finalize(stmt); + return CBM_STORE_OK; +} + +/* The shortest-path walk from the adjacency layer: the same rows as the CTE's + * JOIN nodes + ORDER BY hop, n.id + LIMIT, because cbm_graph_mem_bfs yields + * every reached id at its minimal hop in (hop, id) order and the node rows are + * read here in that order. An id with no node row is skipped, as the JOIN + * skips it. */ +static int bfs_visit_mem(cbm_store_t *s, const cbm_graph_mem_t *gm, const bfs_query_t *q, + bfs_rows_t *rows) { + static const char *const calls_only[] = {"CALLS"}; + bool typed = q->edge_type_count > 0; + cbm_graph_mem_types_t mask = cbm_graph_mem_type_mask(gm, typed ? q->edge_types : calls_only, + typed ? q->edge_type_count : SKIP_ONE); + cbm_graph_mem_hop_t *hops = NULL; + int hop_count = 0; + if (!cbm_graph_mem_bfs(gm, q->start_id, q->inbound, mask, q->max_depth, &hops, &hop_count)) { + store_set_error(s, "bfs: adjacency walk out of memory"); + return CBM_STORE_ERR; + } + bfs_rows_init(rows); + int rc = CBM_STORE_OK; + for (int i = 0; i < hop_count && rows->count < q->probe_limit && rc == CBM_STORE_OK; i++) { + sqlite3_stmt *stmt = prepare_cached(s, &s->stmt_find_node_by_id, st_find_node_by_id_sql); + if (!stmt) { + rc = CBM_STORE_ERR; + break; + } + sqlite3_bind_int64(stmt, SKIP_ONE, hops[i].id); + int step = sqlite3_step(stmt); + if (step == SQLITE_ROW) { + cbm_node_hop_t *row = bfs_rows_next(rows); + scan_node(stmt, &row->node); + row->hop = hops[i].hop; + } else if (step != SQLITE_DONE) { + store_set_error_sqlite(s, "bfs node row"); + rc = CBM_STORE_ERR; + } + sqlite3_reset(stmt); + } + cbm_graph_mem_hops_free(hops); + return rc; +} + +static int store_bfs(cbm_store_t *s, int64_t start_id, const char *direction, + const char **edge_types, int edge_type_count, int max_depth, int max_results, + bool trail, int max_edges, cbm_traverse_result_t *out) { + memset(out, 0, sizeof(*out)); + if (max_results < 0) { + max_results = 0; + } + + cbm_node_t root = {0}; + int rc = cbm_store_find_node_by_id(s, start_id, &root); + if (rc != CBM_STORE_OK) { + return rc; + } + out->root = root; + + char types_clause[CBM_SZ_512]; + bfs_build_types_clause(edge_type_count, types_clause, (int)sizeof(types_clause)); + + bfs_query_t q = { + .start_id = start_id, + .inbound = (direction != NULL) && (strcmp(direction, "inbound") == 0), + .types_clause = types_clause, + .edge_types = edge_types, + .edge_type_count = edge_type_count, + .max_depth = max_depth, + .max_results = max_results, + /* One extra row makes max_results saturation observable on both branches. */ + .probe_limit = max_results < INT_MAX ? max_results + SKIP_ONE : max_results, + .trail = trail, + }; + bfs_rows_t rows = {0}; + int cte_rows = 0; + int cte_row_limit = 0; + /* A shortest-path walk is answered from the adjacency layer when this + * store has one; trail mode keeps its relationship-unique CTE. */ + cbm_graph_mem_t *gm = trail ? NULL : store_graph_mem(s); + rc = gm ? bfs_visit_mem(s, gm, &q, &rows) + : bfs_visit_sql(s, &q, &rows, &cte_rows, &cte_row_limit); + cbm_node_hop_t *visited = rows.visited; + int n = rows.count; + if (rc != CBM_STORE_OK) { + out->visited = visited; + out->visited_count = n; + return rc; + } if (trail && cte_rows > cte_row_limit) { out->truncated = true; diff --git a/tests/test_graph_mem.c b/tests/test_graph_mem.c new file mode 100644 index 000000000..7be3c6d14 --- /dev/null +++ b/tests/test_graph_mem.c @@ -0,0 +1,609 @@ +/* + * test_graph_mem.c — the in-memory adjacency layer (store/graph_mem.h). + * + * The contract is byte-identical output: a read-only query store answers + * BFS and degree filters from the layer, a writable store of the same file + * keeps the SQL path, and every row, order, hop, degree, total and flag must + * match. The fixture mixes a seeded random graph with the shapes that break a + * naive BFS: ties at one hop, a cycle back to the start, a self-loop, a + * diamond, duplicate IMPORTS edges, a cross-project edge and edgeless nodes. + */ +#include "test_framework.h" +#include "test_helpers.h" +#include +#include +#include +#include +#include +#include +#include + +enum { + GMT_NODES = 40, + GMT_RANDOM_EDGES = 140, + GMT_THREADS = 4, + GMT_SEED = 20261014, + GMT_PATH_MAX = 512, +}; + +typedef struct { + char dir[GMT_PATH_MAX]; + char db[GMT_PATH_MAX]; + int64_t ids[GMT_NODES]; + int64_t other_id; /* the cross-project node */ +} gmt_fixture_t; + +static const char *const gmt_types[] = {"CALLS", "USAGE", "DATA_FLOWS", "IMPORTS"}; + +static uint32_t gmt_next(uint32_t *state) { + *state = *state * 1103515245U + 12345U; + return *state >> 8; +} + +static int64_t gmt_node(cbm_store_t *s, const char *project, int i, const char *name) { + char qn[64]; + char file[32]; + snprintf(qn, sizeof(qn), "%s.mod%d.n%02d", project, i % 5, i); + snprintf(file, sizeof(file), "src/m%d.c", i % 5); + cbm_node_t n = {.project = project, + .label = (i % 7 == 3) ? "Method" : "Function", + .name = name, + .qualified_name = qn, + .file_path = file, + .start_line = i, + .end_line = i + 3}; + return cbm_store_upsert_node(s, &n); +} + +static void gmt_edge(cbm_store_t *s, int64_t src, int64_t dst, const char *type, + const char *props) { + cbm_edge_t e = { + .project = "p", .source_id = src, .target_id = dst, .type = type, .properties_json = props}; + (void)cbm_store_insert_edge(s, &e); +} + +/* Build the fixture file and stamp a generation, the way a publish does. */ +static bool gmt_fixture_build(gmt_fixture_t *f) { + char *tmp = th_mktempdir("cbm_graph_mem"); + if (!tmp) { + return false; + } + snprintf(f->dir, sizeof(f->dir), "%s", tmp); + snprintf(f->db, sizeof(f->db), "%s/graph.db", f->dir); + cbm_store_t *s = cbm_store_open_path(f->db); + if (!s || cbm_store_upsert_project(s, "p", "/p") != CBM_STORE_OK || + cbm_store_upsert_project(s, "q", "/q") != CBM_STORE_OK) { + cbm_store_close(s); + return false; + } + for (int i = 0; i < GMT_NODES; i++) { + char name[16]; + /* Every fifth node shares one name: search's (name, id) order has ties. */ + snprintf(name, sizeof(name), (i % 5 == 4) ? "dup" : "n%02d", i); + f->ids[i] = gmt_node(s, "p", i, name); + } + f->other_id = gmt_node(s, "q", 99, "other"); + const int64_t *id = f->ids; + /* chain with a cycle back to the start (0 at hop 3), a self-loop, a diamond */ + gmt_edge(s, id[0], id[1], "CALLS", NULL); + gmt_edge(s, id[1], id[2], "CALLS", NULL); + gmt_edge(s, id[2], id[0], "CALLS", NULL); + gmt_edge(s, id[2], id[3], "CALLS", NULL); + gmt_edge(s, id[5], id[5], "CALLS", NULL); + gmt_edge(s, id[0], id[6], "CALLS", NULL); + gmt_edge(s, id[0], id[7], "CALLS", NULL); + gmt_edge(s, id[6], id[8], "CALLS", NULL); + gmt_edge(s, id[7], id[8], "CALLS", NULL); + /* two IMPORTS rows for one (source, target): distinct local names */ + gmt_edge(s, id[3], id[4], "IMPORTS", "{\"local_name\":\"a\"}"); + gmt_edge(s, id[3], id[4], "IMPORTS", "{\"local_name\":\"b\"}"); + /* a cross-project hop: BFS does not filter by project */ + gmt_edge(s, id[8], f->other_id, "CALLS", NULL); + gmt_edge(s, f->other_id, id[9], "CALLS", NULL); + /* a hub for the degree filters */ + for (int i = 10; i < 30; i++) { + gmt_edge(s, id[i], id[1], (i % 2) ? "CALLS" : "USAGE", NULL); + } + /* seeded random edges over the first 34 nodes; 34..39 stay edgeless */ + uint32_t state = GMT_SEED; + for (int k = 0; k < GMT_RANDOM_EDGES; k++) { + int a = (int)(gmt_next(&state) % 34U); + int b = (int)(gmt_next(&state) % 34U); + const char *type = gmt_types[gmt_next(&state) % 4U]; + gmt_edge(s, id[a], id[b], type, "{\"w\":1}"); + } + /* Stamp and seal, as a publish does: a sealed rollback-journal file is + * the only kind of DB the layer serves. */ + bool ok = cbm_store_generation_advance(s) == CBM_STORE_OK && + cbm_store_prepare_for_publish(s) == CBM_STORE_OK; + cbm_store_close(s); + return ok; +} + +static bool gmt_reseal(const gmt_fixture_t *f) { + cbm_store_t *w = cbm_store_open_path(f->db); + bool ok = w && cbm_store_prepare_for_publish(w) == CBM_STORE_OK; + cbm_store_close(w); + return ok; +} + +static void gmt_fixture_free(gmt_fixture_t *f) { + cbm_graph_mem_drop_idle(); + th_rmtree(f->dir); +} + +static bool gmt_str_eq(const char *a, const char *b) { + return (a == NULL && b == NULL) || (a && b && strcmp(a, b) == 0); +} + +static bool gmt_node_eq(const cbm_node_t *a, const cbm_node_t *b) { + return a->id == b->id && gmt_str_eq(a->project, b->project) && gmt_str_eq(a->label, b->label) && + gmt_str_eq(a->name, b->name) && gmt_str_eq(a->qualified_name, b->qualified_name) && + gmt_str_eq(a->file_path, b->file_path) && a->start_line == b->start_line && + a->end_line == b->end_line && gmt_str_eq(a->properties_json, b->properties_json); +} + +static bool gmt_edge_eq(const cbm_edge_info_t *a, const cbm_edge_info_t *b) { + return gmt_str_eq(a->from_name, b->from_name) && gmt_str_eq(a->to_name, b->to_name) && + gmt_str_eq(a->type, b->type) && a->confidence == b->confidence && + a->source_id == b->source_id && a->target_id == b->target_id && + gmt_str_eq(a->properties_json, b->properties_json); +} + +static bool gmt_traverse_eq(const cbm_traverse_result_t *a, const cbm_traverse_result_t *b) { + if (!gmt_node_eq(&a->root, &b->root) || a->visited_count != b->visited_count || + a->edge_count != b->edge_count || a->truncated != b->truncated || + a->edges_truncated != b->edges_truncated) { + return false; + } + for (int i = 0; i < a->visited_count; i++) { + if (a->visited[i].hop != b->visited[i].hop || + !gmt_node_eq(&a->visited[i].node, &b->visited[i].node)) { + return false; + } + } + for (int i = 0; i < a->edge_count; i++) { + if (!gmt_edge_eq(&a->edges[i], &b->edges[i])) { + return false; + } + } + return true; +} + +typedef struct { + const char *types[3]; + int count; +} gmt_typeset_t; + +static const gmt_typeset_t gmt_typesets[] = { + {{NULL}, 0}, /* the default: CALLS */ + {{"CALLS", "USAGE"}, 2}, + {{"DATA_FLOWS"}, 1}, + {{"IMPORTS"}, 1}, + {{"NO_SUCH_TYPE", "CALLS"}, 2}, +}; + +/* One BFS shape on both stores; prints the shape that diverged. */ +static bool gmt_bfs_same(cbm_store_t *sql, cbm_store_t *mem, int64_t start, const char *dir, + const gmt_typeset_t *ts, int depth, int max_results, int max_edges) { + cbm_traverse_result_t a = {0}; + cbm_traverse_result_t b = {0}; + const char **types = (const char **)ts->types; + int rca = cbm_store_bfs_with_edge_limit(sql, start, dir, types, ts->count, depth, max_results, + max_edges, &a); + int rcb = cbm_store_bfs_with_edge_limit(mem, start, dir, types, ts->count, depth, max_results, + max_edges, &b); + bool same = rca == rcb && gmt_traverse_eq(&a, &b); + if (!same) { + (void)fprintf(stderr, + " graph_mem divergence: start=%lld dir=%s types=%d depth=%d max=%d " + "edges=%d rc=%d/%d visited=%d/%d\n", + (long long)start, dir, ts->count, depth, max_results, max_edges, rca, rcb, + a.visited_count, b.visited_count); + } + cbm_store_traverse_free(&a); + cbm_store_traverse_free(&b); + return same; +} + +/* The runner initializes no memory budget unless the mem suite ran first, and + * without one no layer is ever built. Each test pins a budget the 3 KB fixture + * layer fits, collects its observations, and restores the budget BEFORE + * asserting (the hook's contract: a failed assert must not leak the budget). */ +static size_t gmt_budget_pin(size_t bytes) { + size_t saved = cbm_mem_budget(); + cbm_mem_set_budget_for_tests(bytes); + return saved; +} + +static size_t gmt_budget_roomy(void) { + return cbm_mem_charged() + ((size_t)1 << 30); +} + +static int gmt_visited(cbm_store_t *s, int64_t start) { + cbm_traverse_result_t tr = {0}; + int n = cbm_store_bfs(s, start, "outbound", NULL, 0, 3, 100, &tr) == CBM_STORE_OK + ? tr.visited_count + : -1; + cbm_store_traverse_free(&tr); + return n; +} + +/* One request: open a query store, walk from `start`, close. */ +static int gmt_request(const char *db, int64_t start) { + cbm_store_t *q = cbm_store_open_path_query(db); + int n = q ? gmt_visited(q, start) : -1; + cbm_store_close(q); + return n; +} + +/* The SQL reference: a read-only store like the one under test (a writable + * open would switch the sealed file to WAL), pinned to the SQL path because + * its one layer attempt runs under a budget no layer fits. */ +static cbm_store_t *gmt_open_sql_reference(const gmt_fixture_t *f) { + size_t saved = gmt_budget_pin(64); + cbm_store_t *s = cbm_store_open_path_query(f->db); + if (s) { + (void)gmt_visited(s, f->ids[0]); + } + cbm_mem_set_budget_for_tests(saved); + return s; +} + +typedef struct { + int shapes; + int diverged; + uint64_t builds; + bool start_back; /* the cycle 0->1->2->0 returns node 0 at hop 3 */ +} gmt_bfs_run_t; + +static void gmt_bfs_all_shapes(const gmt_fixture_t *f, cbm_store_t *sql, cbm_store_t *mem, + gmt_bfs_run_t *run) { + static const char *const dirs[] = {"outbound", "inbound"}; + static const int depths[] = {0, 1, 2, 3, 6}; + for (int n = 0; n <= GMT_NODES; n++) { + int64_t start = n < GMT_NODES ? f->ids[n] : f->other_id; + for (int d = 0; d < 2; d++) { + for (size_t t = 0; t < sizeof(gmt_typesets) / sizeof(gmt_typesets[0]); t++) { + for (size_t k = 0; k < sizeof(depths) / sizeof(depths[0]); k++) { + const gmt_typeset_t *ts = &gmt_typesets[t]; + /* lean, with edge data, and saturated (probe row + edge ceiling) */ + run->diverged += !gmt_bfs_same(sql, mem, start, dirs[d], ts, depths[k], 100, 0); + run->diverged += + !gmt_bfs_same(sql, mem, start, dirs[d], ts, depths[k], 100, 1000); + run->diverged += !gmt_bfs_same(sql, mem, start, dirs[d], ts, depths[k], 3, 4); + run->shapes += 3; + } + } + } + } + cbm_traverse_result_t tr = {0}; + if (cbm_store_bfs(mem, f->ids[0], "outbound", NULL, 0, 3, 100, &tr) == CBM_STORE_OK) { + for (int i = 0; i < tr.visited_count; i++) { + if (tr.visited[i].node.id == f->ids[0]) { + run->start_back = tr.visited[i].hop == 3; + } + } + } + cbm_store_traverse_free(&tr); +} + +TEST(graph_mem_bfs_matches_sql_on_every_shape) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + cbm_store_t *sql = gmt_open_sql_reference(&f); + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + cbm_store_t *mem = cbm_store_open_path_query(f.db); + gmt_bfs_run_t run = {0}; + if (sql && mem) { + gmt_bfs_all_shapes(&f, sql, mem, &run); + } + run.builds = cbm_graph_mem_build_count() - builds; + cbm_mem_set_budget_for_tests(saved); + cbm_store_close(sql); + cbm_store_close(mem); + gmt_fixture_free(&f); + + ASSERT_GT(run.shapes, 0); + ASSERT_EQ(run.diverged, 0); + /* The layer really answered: one build, for the store under test only. */ + ASSERT_EQ(run.builds, 1); + ASSERT_TRUE(run.start_back); + PASS(); +} + +static bool gmt_search_same(cbm_store_t *sql, cbm_store_t *mem, cbm_search_params_t p) { + cbm_search_output_t a = {0}; + cbm_search_output_t b = {0}; + int rca = cbm_store_search(sql, &p, &a); + int rcb = cbm_store_search(mem, &p, &b); + bool same = rca == rcb && a.total == b.total && a.count == b.count; + for (int i = 0; same && i < a.count; i++) { + same = gmt_node_eq(&a.results[i].node, &b.results[i].node) && + a.results[i].in_degree == b.results[i].in_degree && + a.results[i].out_degree == b.results[i].out_degree; + } + if (!same) { + (void)fprintf(stderr, + " graph_mem search divergence: min=%d max=%d limit=%d offset=%d " + "total=%d/%d count=%d/%d\n", + p.min_degree, p.max_degree, p.limit, p.offset, a.total, b.total, a.count, + b.count); + } + cbm_store_search_free(&a); + cbm_store_search_free(&b); + return same; +} + +static int gmt_search_all_shapes(cbm_store_t *sql, cbm_store_t *mem) { + static const int ranges[][2] = {{1, -1}, {3, -1}, {10, -1}, {0, 0}, {-1, 0}, + {-1, 2}, {2, 5}, {0, -1}, {50, -1}}; + static const char *const labels[] = {NULL, "Function"}; + int diverged = 0; + for (size_t r = 0; r < sizeof(ranges) / sizeof(ranges[0]); r++) { + for (int l = 0; l < 2; l++) { + for (int page = 0; page < 3; page++) { + cbm_search_params_t p = {.project = "p", + .label = labels[l], + .min_degree = ranges[r][0], + .max_degree = ranges[r][1], + .limit = page == 2 ? 100 : 4, + .offset = page == 1 ? 3 : 0}; + diverged += !gmt_search_same(sql, mem, p); + } + } + } + cbm_search_params_t named = { + .project = "p", .name_pattern = "dup", .min_degree = 1, .max_degree = -1, .limit = 10}; + diverged += !gmt_search_same(sql, mem, named); + return diverged; +} + +TEST(graph_mem_degree_search_matches_sql) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + cbm_store_t *sql = gmt_open_sql_reference(&f); + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + + /* No degree filter: the layer is neither needed nor built. */ + cbm_store_t *plain = cbm_store_open_path_query(f.db); + cbm_search_params_t nofilter = {.project = "p", .min_degree = -1, .max_degree = -1}; + cbm_search_output_t out = {0}; + int plain_rc = plain ? cbm_store_search(plain, &nofilter, &out) : CBM_STORE_ERR; + cbm_store_search_free(&out); + cbm_store_close(plain); + uint64_t plain_builds = cbm_graph_mem_build_count() - builds; + + cbm_store_t *mem = cbm_store_open_path_query(f.db); + int diverged = sql && mem ? gmt_search_all_shapes(sql, mem) : -1; + uint64_t filter_builds = cbm_graph_mem_build_count() - builds; + cbm_mem_set_budget_for_tests(saved); + cbm_store_close(sql); + cbm_store_close(mem); + gmt_fixture_free(&f); + + ASSERT_EQ(plain_rc, CBM_STORE_OK); + ASSERT_EQ(plain_builds, 0); + ASSERT_EQ(diverged, 0); + ASSERT_EQ(filter_builds, 1); + PASS(); +} + +TEST(graph_mem_generation_change_invalidates) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + int64_t leaf = f.ids[36]; /* edgeless */ + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + + int first = gmt_request(f.db, leaf); + uint64_t first_builds = cbm_graph_mem_build_count() - builds; + /* A later request store on the same version reuses the layer. */ + int second = gmt_request(f.db, leaf); + uint64_t second_builds = cbm_graph_mem_build_count() - builds; + + /* A mutation stamped with a new generation and sealed: a publish. */ + cbm_store_t *w = cbm_store_open_path(f.db); + if (w) { + gmt_edge(w, leaf, f.ids[37], "CALLS", NULL); + } + bool published = w && cbm_store_generation_advance(w) == CBM_STORE_OK && + cbm_store_prepare_for_publish(w) == CBM_STORE_OK; + cbm_store_close(w); + + cbm_store_t *q3 = cbm_store_open_path_query(f.db); + int third = q3 ? gmt_visited(q3, leaf) : -1; + cbm_store_t *sql = gmt_open_sql_reference(&f); + int third_sql = sql ? gmt_visited(sql, leaf) : -1; + bool same = + sql && q3 && gmt_bfs_same(sql, q3, f.ids[0], "outbound", &gmt_typesets[1], 6, 100, 1000); + uint64_t third_builds = cbm_graph_mem_build_count() - builds; + /* The old version's layer is retired: one layer for this file. */ + int cached = cbm_graph_mem_cached_count(); + cbm_store_close(sql); + cbm_store_close(q3); + cbm_mem_set_budget_for_tests(saved); + gmt_fixture_free(&f); + int cached_after = cbm_graph_mem_cached_count(); + + ASSERT_EQ(first, 0); + ASSERT_EQ(first_builds, 1); + ASSERT_EQ(second, 0); + ASSERT_EQ(second_builds, 1); + ASSERT_TRUE(published); + ASSERT_EQ(third, 1); + ASSERT_EQ(third_sql, 1); + ASSERT_TRUE(same); + ASSERT_EQ(third_builds, 2); + ASSERT_EQ(cached, 1); + ASSERT_EQ(cached_after, 0); + PASS(); +} + +/* The cross-repo shape: edges written straight into a published file, with + * NO new generation. A layer of the old content must never answer for it — + * not while the file is in WAL mode, and not after it is sealed again. */ +TEST(graph_mem_in_place_write_never_serves_stale) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + int64_t leaf = f.ids[36]; /* edgeless */ + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + + int before = gmt_request(f.db, leaf); + uint64_t before_builds = cbm_graph_mem_build_count() - builds; + + /* The writer stays open across two commits, so no checkpoint lands + * between them: the main file — header included — is the same for both. + * Only refusing WAL mode keeps the second request from a stale answer. */ + cbm_store_t *w = cbm_store_open_path(f.db); + if (w) { + gmt_edge(w, leaf, f.ids[37], "CALLS", NULL); + } + int wal_one = gmt_request(f.db, leaf); + if (w) { + gmt_edge(w, leaf, f.ids[38], "CALLS", NULL); + } + int wal_two = gmt_request(f.db, leaf); + uint64_t wal_builds = cbm_graph_mem_build_count() - builds; + cbm_store_close(w); + + bool resealed = gmt_reseal(&f); + int after = gmt_request(f.db, leaf); + uint64_t after_builds = cbm_graph_mem_build_count() - builds; + int cached = cbm_graph_mem_cached_count(); + cbm_mem_set_budget_for_tests(saved); + gmt_fixture_free(&f); + + ASSERT_NOT_NULL(w); + ASSERT_EQ(before, 0); + ASSERT_EQ(before_builds, 1); + /* WAL mode: the SQL path answers, with each new edge. */ + ASSERT_EQ(wal_one, 1); + ASSERT_EQ(wal_two, 2); + ASSERT_EQ(wal_builds, 1); + /* Sealed again under the SAME generation: a new version, a new layer. */ + ASSERT_TRUE(resealed); + ASSERT_EQ(after, 2); + ASSERT_EQ(after_builds, 2); + ASSERT_EQ(cached, 1); + PASS(); +} + +TEST(graph_mem_over_budget_keeps_sql_path) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + cbm_store_t *sql = gmt_open_sql_reference(&f); + /* A budget whose quarter cannot hold the layer: no build, SQL answers. The + * file stays sealed, so the budget is the only reason to refuse. */ + size_t saved = gmt_budget_pin(64); + cbm_store_t *mem = cbm_store_open_path_query(f.db); + bool same = + sql && mem && gmt_bfs_same(sql, mem, f.ids[0], "outbound", &gmt_typesets[1], 3, 100, 1000); + cbm_search_params_t p = {.project = "p", .min_degree = 3, .max_degree = -1, .limit = 50}; + bool same_search = sql && mem && gmt_search_same(sql, mem, p); + uint64_t builds_after = cbm_graph_mem_build_count(); + int cached = cbm_graph_mem_cached_count(); + cbm_mem_set_budget_for_tests(saved); + cbm_store_close(sql); + cbm_store_close(mem); + gmt_fixture_free(&f); + + ASSERT_TRUE(same); + ASSERT_TRUE(same_search); + ASSERT_EQ(builds_after, builds); + ASSERT_EQ(cached, 0); + PASS(); +} + +TEST(graph_mem_legacy_generation_keeps_sql_path) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + /* A DB that predates generation tracking cannot detect staleness. The raw + * connection leaves the file in rollback mode: legacy is the only reason + * to refuse. */ + sqlite3 *raw = NULL; + ASSERT_EQ(sqlite3_open(f.db, &raw), SQLITE_OK); + ASSERT_EQ(sqlite3_exec(raw, "DROP TABLE store_meta;", NULL, NULL, NULL), SQLITE_OK); + sqlite3_close(raw); + uint64_t builds = cbm_graph_mem_build_count(); + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + int visited = gmt_request(f.db, f.ids[0]); + uint64_t builds_after = cbm_graph_mem_build_count(); + cbm_mem_set_budget_for_tests(saved); + gmt_fixture_free(&f); + + ASSERT_GT(visited, 0); + ASSERT_EQ(builds_after, builds); + PASS(); +} + +typedef struct { + const char *db; + int64_t start; + const cbm_traverse_result_t *expect; + bool same; +} gmt_thread_arg_t; + +static void *gmt_thread_main(void *arg) { + gmt_thread_arg_t *t = arg; + cbm_store_t *s = cbm_store_open_path_query(t->db); + cbm_traverse_result_t got = {0}; + t->same = s && cbm_store_bfs(s, t->start, "inbound", NULL, 0, 6, 100, &got) == CBM_STORE_OK && + gmt_traverse_eq(t->expect, &got); + cbm_store_traverse_free(&got); + cbm_store_close(s); + return NULL; +} + +TEST(graph_mem_concurrent_requests_build_once) { + gmt_fixture_t f; + ASSERT_TRUE(gmt_fixture_build(&f)); + uint64_t builds = cbm_graph_mem_build_count(); + cbm_store_t *sql = gmt_open_sql_reference(&f); + cbm_traverse_result_t expect = {0}; + int expect_rc = + sql ? cbm_store_bfs(sql, f.ids[1], "inbound", NULL, 0, 6, 100, &expect) : CBM_STORE_ERR; + size_t saved = gmt_budget_pin(gmt_budget_roomy()); + cbm_thread_t threads[GMT_THREADS]; + gmt_thread_arg_t args[GMT_THREADS]; + int started = 0; + for (int i = 0; i < GMT_THREADS; i++) { + args[i] = (gmt_thread_arg_t){.db = f.db, .start = f.ids[1], .expect = &expect}; + started += cbm_thread_create(&threads[i], 0, gmt_thread_main, &args[i]) == 0; + } + int joined = 0; + for (int i = 0; i < started; i++) { + joined += cbm_thread_join(&threads[i]) == 0; + } + uint64_t built = cbm_graph_mem_build_count() - builds; + cbm_mem_set_budget_for_tests(saved); + int same = 0; + for (int i = 0; i < started; i++) { + same += args[i].same; + } + cbm_store_traverse_free(&expect); + cbm_store_close(sql); + gmt_fixture_free(&f); + + ASSERT_EQ(expect_rc, CBM_STORE_OK); + ASSERT_EQ(started, GMT_THREADS); + ASSERT_EQ(joined, GMT_THREADS); + ASSERT_EQ(same, GMT_THREADS); + /* A request meeting another's build takes SQL; nobody builds twice. */ + ASSERT_EQ(built, 1); + PASS(); +} + +SUITE(graph_mem) { + RUN_TEST(graph_mem_bfs_matches_sql_on_every_shape); + RUN_TEST(graph_mem_degree_search_matches_sql); + RUN_TEST(graph_mem_generation_change_invalidates); + RUN_TEST(graph_mem_in_place_write_never_serves_stale); + RUN_TEST(graph_mem_over_budget_keeps_sql_path); + RUN_TEST(graph_mem_legacy_generation_keeps_sql_path); + RUN_TEST(graph_mem_concurrent_requests_build_once); +} diff --git a/tests/test_main.c b/tests/test_main.c index 962fc1482..5ff6bdb9f 100644 --- a/tests/test_main.c +++ b/tests/test_main.c @@ -1732,6 +1732,7 @@ extern void suite_store_edges(void); extern void suite_store_impact(void); extern void suite_store_scope(void); extern void suite_store_graph_digest(void); +extern void suite_graph_mem(void); extern void suite_test_impact(void); extern void suite_test_impact_profiles(void); extern void suite_test_impact_changes(void); @@ -2194,6 +2195,7 @@ int main(int argc, char **argv) { RUN_SELECTED_SUITE(store_impact); RUN_SELECTED_SUITE(store_scope); RUN_SELECTED_SUITE(store_graph_digest); + RUN_SELECTED_SUITE(graph_mem); RUN_SELECTED_SUITE(test_impact); RUN_SELECTED_SUITE(test_impact_profiles); RUN_SELECTED_SUITE(test_impact_changes);