el/engram/test/test_vindex_concurrency.c

/* test_vindex_concurrency.c — regression test for the 2026-08-16 soul crash.
 *
 * WHAT BROKE: the soul daemon crash-looped (5 crashes in ~100s) with SIGSEGV in
 * search_layer <- vindex_insert <- eg_vindex_sync, a SIGABRT, and a fault inside
 * xzm_realloc's own freelist — i.e. heap corruption. The SIGSEGV address
 * 0x65646f4e6d617267 is little-endian ASCII "gramNode": string bytes being
 * dereferenced as an Elem vector pointer.
 *
 * ROOT CAUSE: VIndex owns its traversal scratch (visited[] + visit_epoch), and
 * search_layer mutates it via visited_reset(). So the index is unsafe for ANY
 * concurrent use — including two concurrent READS. soul.el starts http_serve_async
 * (a thread per connection) and then runs awareness_run() on the main thread, which
 * reaches the same global index through engram_activate; nothing serialized them.
 *
 * Neither hnswlib nor FAISS puts the visited set on the index: hnswlib checks one
 * out of a VisitedListPool per query, FAISS uses a thread_local VisitedTable.
 *
 * THE ORIGINAL `concurrent` HALF CONFLATED TWO DISTINCT HAZARDS (2026-08-16). It ran
 * a writer against a reader on one bare index, so it could not tell apart:
 *
 *   (a) READ/READ corruption — two searches stamping each other's visited epoch.
 *       A defect INSIDE engram_vindex.c, fixable there, and now fixed: the visited
 *       set moved to the call frame and vindex_search takes a `const VIndex*`.
 *
 *   (b) WRITE/READ corruption — vindex_insert rewires the neighbour lists of
 *       EXISTING elements and reallocs elems[], so an insert is a mutation of the
 *       whole structure. This is NOT fixable inside engram_vindex.c at any price:
 *       it is inherent to in-place HNSW. It requires a publication boundary ABOVE
 *       the data structure (el_runtime.c: eg_vindex_view / eg_vindex_maintain).
 *
 * Conflating them made the suite unfailable-then-unpassable: fixing (a) left (b)
 * still racing, which reads as "the fix did not work" when in fact a different,
 * correctly-located fix is what (b) needs. So the halves are now separate:
 *
 *   single          N clustered vectors, ONE thread, ASan. The CONTROL. Must always
 *                   be clean. When this passes and a concurrent half fails, the defect
 *                   is concurrency, not an out-of-bounds/logic error in the graph code.
 *                   (On 2026-08-16 this control cleared all 13,820 real dim-768 store
 *                   vectors under ASan, which DISPROVED an inspection-derived hypothesis
 *                   about an out-of-bounds reverse-link write at engram_vindex.c:340.)
 *
 *   readers         N reader threads, NO writer, one shared index, TSan. This is
 *                   hazard (a) in isolation. It RACED before the visited set moved off
 *                   the index struct and must be CLEAN now. Hard gate.
 *
 *   unsynchronized  writer + reader on a bare index, TSan. Hazard (b) in isolation.
 *                   EXPECTED TO RACE, permanently — it is the executable proof that
 *                   the index cannot be made safe from the inside, and therefore that
 *                   the publication boundary in el_runtime.c has to exist. If this
 *                   ever goes clean, the test stopped interleaving; do not celebrate.
 *
 *   published       writer + readers through a publication boundary that mirrors
 *                   eg_vindex_view / eg_vindex_maintain (rwlock: readers shared,
 *                   the single owner exclusive), TSan. Must be CLEAN. Hard gate.
 *                   This is what proves the shape of the runtime fix, in the same
 *                   process, rather than asserting it.
 *
 * Absence of a crash does NOT mean absence of a race — always read the sanitizer
 * verdict, never just the exit code.
 *
 * Build/run: engram/test/run_vindex_concurrency_tests.sh
 */
#include "engram_vindex.h"

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#define DIM   128
#define NVEC  3000
#define SEED_N 50

static VIndex* g_ix;
static float*  g_vecs;

/* Deterministic filler. Real embeddings are strongly correlated, not uniform noise;
 * clustering keeps many candidates near-equidistant, which exercises the diversity
 * heuristic and the visited set far harder than random vectors do. */
static void fill_vectors(void) {
    g_vecs = (float*)malloc((size_t)NVEC * DIM * sizeof(float));
    if (!g_vecs) { fprintf(stderr, "OOM\n"); exit(1); }
    for (int i = 0; i < NVEC; i++) {
        int cluster = i % 8;
        for (int d = 0; d < DIM; d++)
            g_vecs[(size_t)i * DIM + d] =
                (float)(((d + cluster * 7) % 13) / 13.0) +
                (float)(((i * 2654435761u + (unsigned)d) % 97) / 9700.0);
    }
}

static void* writer_fn(void* arg) {
    (void)arg;
    for (int i = SEED_N; i < NVEC; i++)
        (void)vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM);
    return NULL;
}

static void* reader_fn(void* arg) {
    (void)arg;
    uint64_t ids[8]; float ds[8];
    for (int i = 0; i < 20000; i++)
        (void)vindex_search(g_ix, g_vecs + (size_t)(i % NVEC) * DIM, 8, 0, ids, ds);
    return NULL;
}

static int run_single(void) {
    printf("[single] inserting %d vectors on one thread (ASan control)\n", NVEC);
    g_ix = vindex_create(DIM, 0, 0);
    if (!g_ix) { fprintf(stderr, "[single] vindex_create failed\n"); return 1; }
    for (int i = 0; i < NVEC; i++) {
        if (vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM) != 0) {
            fprintf(stderr, "[single] insert %d failed\n", i); return 1;
        }
    }
    if (vindex_size(g_ix) != (size_t)NVEC) {
        fprintf(stderr, "[single] size %zu != %d\n", vindex_size(g_ix), NVEC); return 1;
    }
    uint64_t ids[16]; float ds[16];
    for (int q = 0; q < 200; q++) {
        int k = vindex_search(g_ix, g_vecs + (size_t)((q * 7) % NVEC) * DIM, 16, 0, ids, ds);
        if (k < 0) { fprintf(stderr, "[single] search failed at q=%d\n", q); return 1; }
    }
    vindex_free(g_ix); g_ix = NULL;
    printf("[single] PASS — no memory error (this must ALWAYS pass)\n");
    return 0;
}

/* Hazard (b) in isolation: writer + reader on a BARE index, no boundary. */
static int run_unsynchronized(void) {
    printf("[unsynchronized] 1 writer + 1 reader on a BARE index (TSan probe)\n");
    printf("[unsynchronized] a race here is EXPECTED and PERMANENT — in-place HNSW\n");
    printf("[unsynchronized] insert rewires existing elements. This is the proof that\n");
    printf("[unsynchronized] the publication boundary must live ABOVE engram_vindex.c.\n");
    g_ix = vindex_create(DIM, 0, 0);
    if (!g_ix) { fprintf(stderr, "[unsynchronized] vindex_create failed\n"); return 1; }
    for (int i = 0; i < SEED_N; i++)
        (void)vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM);

    pthread_t w, r;
    if (pthread_create(&w, NULL, writer_fn, NULL) ||
        pthread_create(&r, NULL, reader_fn, NULL)) {
        fprintf(stderr, "[unsynchronized] pthread_create failed\n"); return 1;
    }
    pthread_join(w, NULL);
    pthread_join(r, NULL);
    vindex_free(g_ix); g_ix = NULL;
    printf("[unsynchronized] completed — CHECK THE SANITIZER VERDICT, not this line.\n");
    return 0;
}

/* ── hazard (a) in isolation: concurrent READS only ───────────────────────────
 * This is what the frame-owned visited set fixes. Before that change, two
 * vindex_search calls on one index wrote each other's epoch stamp; TSan reported
 * the race at visited_reset and the traversal then walked bogus element indices. */
#define NREADERS 4

static int run_readers(void) {
    printf("[readers] %d concurrent readers, NO writer, one shared index (TSan)\n", NREADERS);
    printf("[readers] this is the visited-set regression gate — must be CLEAN.\n");
    g_ix = vindex_create(DIM, 0, 0);
    if (!g_ix) { fprintf(stderr, "[readers] vindex_create failed\n"); return 1; }
    for (int i = 0; i < NVEC; i++)
        (void)vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM);

    pthread_t t[NREADERS];
    for (int i = 0; i < NREADERS; i++)
        if (pthread_create(&t[i], NULL, reader_fn, NULL)) {
            fprintf(stderr, "[readers] pthread_create failed\n"); return 1;
        }
    for (int i = 0; i < NREADERS; i++) pthread_join(t[i], NULL);
    vindex_free(g_ix); g_ix = NULL;
    printf("[readers] completed — CHECK THE SANITIZER VERDICT, not this line.\n");
    return 0;
}

/* ── the publication boundary, mirroring el_runtime.c ─────────────────────────
 * Readers take the boundary SHARED and hold it across the whole search; the one
 * owner takes it EXCLUSIVE to extend. Same shape as eg_vindex_view /
 * eg_vindex_maintain. Note the reader's index pointer is `const VIndex*` — the
 * compiler, not this comment, is what stops a reader inserting. */
static pthread_rwlock_t g_pub = PTHREAD_RWLOCK_INITIALIZER;

static void* pub_writer_fn(void* arg) {
    (void)arg;
    for (int i = SEED_N; i < NVEC; i++) {
        pthread_rwlock_wrlock(&g_pub);
        (void)vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM);
        pthread_rwlock_unlock(&g_pub);
    }
    return NULL;
}

static void* pub_reader_fn(void* arg) {
    (void)arg;
    uint64_t ids[8]; float ds[8];
    for (int i = 0; i < 5000; i++) {
        pthread_rwlock_rdlock(&g_pub);
        const VIndex* view = g_ix;                 /* immutable view */
        (void)vindex_search(view, g_vecs + (size_t)(i % NVEC) * DIM, 8, 0, ids, ds);
        pthread_rwlock_unlock(&g_pub);
    }
    return NULL;
}

static int run_published(void) {
    printf("[published] 1 owner + %d readers through a publication boundary (TSan)\n", NREADERS);
    printf("[published] this is the eg_vindex_view/eg_vindex_maintain gate — must be CLEAN.\n");
    g_ix = vindex_create(DIM, 0, 0);
    if (!g_ix) { fprintf(stderr, "[published] vindex_create failed\n"); return 1; }
    for (int i = 0; i < SEED_N; i++)
        (void)vindex_insert(g_ix, (uint64_t)i, g_vecs + (size_t)i * DIM);

    pthread_t w, r[NREADERS];
    if (pthread_create(&w, NULL, pub_writer_fn, NULL)) {
        fprintf(stderr, "[published] pthread_create failed\n"); return 1;
    }
    for (int i = 0; i < NREADERS; i++)
        if (pthread_create(&r[i], NULL, pub_reader_fn, NULL)) {
            fprintf(stderr, "[published] pthread_create failed\n"); return 1;
        }
    pthread_join(w, NULL);
    for (int i = 0; i < NREADERS; i++) pthread_join(r[i], NULL);
    if (vindex_size(g_ix) != (size_t)NVEC) {
        fprintf(stderr, "[published] size %zu != %d — the owner lost inserts\n",
                vindex_size(g_ix), NVEC);
        vindex_free(g_ix); g_ix = NULL; return 1;
    }
    vindex_free(g_ix); g_ix = NULL;
    printf("[published] all %d inserts landed; CHECK THE SANITIZER VERDICT too.\n", NVEC);
    return 0;
}

int main(int argc, char** argv) {
    const char* mode = (argc > 1) ? argv[1] : "single";
    fill_vectors();
    int rc;
    if      (!strcmp(mode, "single"))         rc = run_single();
    else if (!strcmp(mode, "readers"))        rc = run_readers();
    else if (!strcmp(mode, "unsynchronized")) rc = run_unsynchronized();
    else if (!strcmp(mode, "published"))      rc = run_published();
    /* back-compat: the pre-split name meant the bare writer+reader probe. */
    else if (!strcmp(mode, "concurrent"))     rc = run_unsynchronized();
    else {
        fprintf(stderr, "usage: %s [single|readers|unsynchronized|published]\n", argv[0]);
        rc = 2;
    }
    free(g_vecs);
    return rc;
}