nprasad2077/codebase-memory-mcp
1
0 Fork 0
mirror of https://github.com/DeusData/codebase-memory-mcp.git synced 2026-11-04 10:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(store): in-memory adjacency layer for trace_path and degree filters

Browse Source 
On a kernel-scale index (8.5M nodes, 16.9M edges) trace_path's recursive
CTE and search_graph's per-row degree subqueries spend seconds in B-tree
seeks. graph_mem reads the edges once, in one sequential scan of the
covering UNIQUE index, into a typed CSR (node id -> out/in neighbours with
interned edge types) and answers both from RAM:

- store_bfs (shortest-path mode): BFS over the layer with the CTE's exact
  semantics (minimal hop per node, the start only via a cycle back to it,
  (hop, id) order, probe row, JOIN-skips for ids without a node row); node
  rows and edge collection still come from SQL. Trail mode is unchanged.
- cbm_store_search with min/max_degree: the two correlated COUNT(*)
  subqueries become cbm_adj_degree(); filter, order, page and total are the
  same SQL.

Layers live in a process-wide registry (the daemon closes its stores after
every request), refcounted across request threads, allocated through the
memory core (CBM_MEM_CLASS_STORE) and bounded: all layers stay under a
quarter of the budget, idle ones are evicted oldest-first, and a layer that
does not fit is not built (SQL path, never a partial layer). A layer is
keyed by db path + store generation + the rollback-journal file state (the
header change counter every rollback commit bumps), read in one read
transaction with the scan; only read-only query stores of sealed DBs
qualify. The file state matters: the cross-repo pass writes CROSS_* edges
into other projects' published DBs in place without a new generation; such
a DB is in WAL mode and keeps the SQL path until the next publish.

Kernel, macOS, same index, fresh daemon: trace_path schedule 7.5 s -> 21 ms
warm (1.23 s first call incl. a 1.2 s build), search_graph Function
min_degree 50 4.2 s -> 170 ms warm (1.35 s first). Layer: 16,890,273 edges,
237 MB. Responses byte-identical to the baseline on django, redis and the
kernel (tools, Cypher and extra BFS/degree cases).

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
This commit is contained in:
Martin Vogel committed 2026-10-14 22:19:02 +02:00
1 parent 312fb36844
commit 8b4d5e8b5b
6 files changed
+1751 -65
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Makefile.cbm
+2 -1
View File
@@ -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 \
src/store/graph_mem.c
+772
View File
@@ -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 <sqlite3.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
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);
}
src/store/graph_mem.h
+95
View File
@@ -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 <stdbool.h>
#include <stddef.h>
#include <stdint.h>
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 */
src/store/store.c
+271 -64
View File
@@ -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 <stdatomic.h>
#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;
tests/test_graph_mem.c
+609
View File
@@ -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 <foundation/compat_thread.h>
#include <foundation/mem.h>
#include <store/graph_mem.h>
#include <store/store.h>
#include <sqlite3.h>
#include <stdio.h>
#include <string.h>
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);
}
tests/test_main.c
+2
View File
@@ -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);