diff --git a/src/services/startup.ts b/src/services/startup.ts index ff08ae6..6a127ed 100644 --- a/src/services/startup.ts +++ b/src/services/startup.ts @@ -15,7 +15,8 @@ import * as fs from "fs"; import * as path from "path"; -import { isConfigured, loadScarsWithEmbeddings } from "./supabase-client.js"; +import { isConfigured, loadScarsWithEmbeddings, getLearningsFingerprint, getSupabaseUrl } from "./supabase-client.js"; +import { computeStoreKey, readVectorCache, writeVectorCache, tryAcquireDownloadLock, waitForLeaderCache } from "./vector-disk-cache.js"; import { getTableName } from "./tier.js"; import { getGitmemDir } from "./gitmem-dir.js"; import { @@ -106,13 +107,51 @@ interface ScarWithEmbedding { * NOTE: Now loads all learning types (scars, patterns, wins, anti-patterns), * not just scars. This fixes the issue where ~64 patterns were being ignored. */ -async function loadScarsFromSupabase(): Promise<{ +async function loadScarsFromSupabase(options: { bypassDiskCache?: boolean } = {}): Promise<{ scars: ScarWithEmbedding[]; latestUpdatedAt: string | null; + source: "disk" | "network"; }> { - console.error(`[startup] Loading ALL learnings with embeddings from Supabase (cross-project)`); const startTime = Date.now(); + // GIT-98: ask the store what it holds (one row, a few hundred bytes) before + // paying for the bulk download. Observed BEFORE the download on purpose: if + // the store changes mid-load the cache is merely stale and the next start + // reloads, which is the safe direction. + const storeKey = computeStoreKey(getSupabaseUrl(), getTableName("learnings")); + const fingerprint = await getLearningsFingerprint(); + + if (!options.bypassDiskCache) { + const cached = readVectorCache(storeKey, fingerprint); + if (cached.hit) { + console.error( + `[startup] Vector cache HIT: ${cached.rows.length} learnings from disk in ${Date.now() - startTime}ms ` + + `(${cached.bytesOnDisk} bytes on disk, bulk download skipped)` + ); + return { scars: cached.rows, latestUpdatedAt: fingerprint.latestUpdatedAt, source: "disk" }; + } + console.error(`[startup] Vector cache MISS: ${cached.reason}`); + } + + // Single-flight the download across processes. A fan-out starts N workers at + // once and a single store write invalidates the cache for all of them, so + // without this a miss costs N identical downloads. + let releaseLock = tryAcquireDownloadLock(storeKey); + if (!releaseLock && !options.bypassDiskCache && fingerprint.count >= 0) { + console.error("[startup] Another process is downloading the index — waiting for its cache"); + const fromLeader = await waitForLeaderCache(storeKey, fingerprint); + if (fromLeader) { + console.error( + `[startup] Vector cache HIT (via leader): ${fromLeader.rows.length} learnings in ${Date.now() - startTime}ms, bulk download skipped` + ); + return { scars: fromLeader.rows, latestUpdatedAt: fingerprint.latestUpdatedAt, source: "disk" }; + } + console.error("[startup] Leader did not produce a usable cache — downloading"); + } + releaseLock = releaseLock ?? (() => {}); + + console.error(`[startup] Loading ALL learnings with embeddings from Supabase (cross-project)`); + try { // Load ALL learnings across projects — semantic similarity handles relevance const learnings = await loadScarsWithEmbeddings(undefined, 500); @@ -127,10 +166,19 @@ async function loadScarsFromSupabase(): Promise<{ // Get latest updated_at for staleness tracking const latestUpdatedAt = learnings.length > 0 ? learnings[0].updated_at || null : null; - return { scars: learnings, latestUpdatedAt }; + // Never cache an empty result: an empty download is far more often a + // transient failure than a genuinely empty store, and caching it would make + // the failure sticky. + if (learnings.length > 0) { + writeVectorCache(storeKey, fingerprint, learnings); + } + + return { scars: learnings, latestUpdatedAt, source: "network" }; } catch (error) { console.error("[startup] Failed to load learnings:", error); - return { scars: [], latestUpdatedAt: null }; + return { scars: [], latestUpdatedAt: null, source: "network" }; + } finally { + releaseLock(); } } @@ -576,8 +624,9 @@ export async function flushCache(_project?: Project): Promise const previousCount = getLocalVectorSearch().getScarCount(); try { - // Load ALL fresh scars (cross-project) - const { scars, latestUpdatedAt } = await loadScarsFromSupabase(); + // Load ALL fresh scars (cross-project). A flush is the user saying "I do not + // trust what you have" — bypass the disk cache (GIT-98) and rewrite it. + const { scars, latestUpdatedAt } = await loadScarsFromSupabase({ bypassDiskCache: true }); // Reinitialize the unified index await reinitializeLocalSearch(scars, undefined, latestUpdatedAt || undefined); diff --git a/src/services/supabase-client.ts b/src/services/supabase-client.ts index 171629c..1e52576 100644 --- a/src/services/supabase-client.ts +++ b/src/services/supabase-client.ts @@ -579,6 +579,66 @@ export async function loadScarsWithEmbeddings( } } +/** + * Cheap fingerprint of the learnings the local index is built from (GIT-98). + * + * One request, one row: `Prefer: count=exact` puts the total in Content-Range + * ("0-0/273") and the single row carries the newest updated_at. A few hundred + * bytes, versus megabytes for the bulk load it lets a warm start skip. + * + * The filters MUST stay identical to loadScarsWithEmbeddings, or the + * fingerprint describes a different set from the one that was cached. + * + * Returns count -1 on any failure; callers treat that as "unknown", never as a match. + */ +export async function getLearningsFingerprint(): Promise<{ count: number; latestUpdatedAt: string | null }> { + if (!isConfigured()) return { count: -1, latestUpdatedAt: null }; + + try { + const url = new URL(`${SUPABASE_REST_URL}/${getTableName("learnings")}`); + url.searchParams.set("select", "updated_at"); + url.searchParams.set("learning_type", "in.(scar,pattern,win,anti_pattern)"); + url.searchParams.set("is_active", "eq.true"); + url.searchParams.set("order", "updated_at.desc"); + url.searchParams.set("limit", "1"); + + const response = await fetch(url.toString(), { + method: "GET", + headers: { + apikey: SUPABASE_KEY, + Authorization: `Bearer ${SUPABASE_KEY}`, + "Accept-Profile": "public", + Prefer: "count=exact", + }, + signal: AbortSignal.timeout(10_000), + }); + + if (!response.ok) { + console.error(`[supabase-direct] Fingerprint request failed: ${response.status}`); + return { count: -1, latestUpdatedAt: null }; + } + + // Content-Range: "0-0/273", or "*/0" for an empty set. + const range = response.headers.get("content-range") || ""; + const total = Number(range.split("/")[1]); + if (!Number.isInteger(total) || total < 0) { + console.error(`[supabase-direct] Fingerprint: unparseable Content-Range "${range}"`); + return { count: -1, latestUpdatedAt: null }; + } + + const rows = (await response.json()) as Array<{ updated_at?: string }>; + return { count: total, latestUpdatedAt: rows[0]?.updated_at ?? null }; + } catch (error) { + console.error("[supabase-direct] Fingerprint request error:", error instanceof Error ? error.message : error); + return { count: -1, latestUpdatedAt: null }; + } +} + +/** The configured Supabase URL, for keying per-store caches. Empty when unconfigured. */ +export function getSupabaseUrl(): string { + return SUPABASE_URL || ""; +} + /** * Scar search with severity weighting * diff --git a/src/services/vector-disk-cache.ts b/src/services/vector-disk-cache.ts new file mode 100644 index 0000000..91853b9 --- /dev/null +++ b/src/services/vector-disk-cache.ts @@ -0,0 +1,235 @@ +/** + * Vector Disk Cache (GIT-98) + * + * In local search mode every server process used to download every learning + * with its 1536-dim embedding at startup and throw it away at exit. That is a + * fair trade for one long interactive session and the worst possible trade for + * short-lived processes — sub-agents, fan-outs, CI, cron — which are now the + * common case. A customer exhausted a Supabase egress allowance this way. + * + * This module makes a cold process cheap by default: + * + * 1. Ask the store for a FINGERPRINT — row count + newest updated_at. One + * request, one row, a few hundred bytes (PostgREST `Prefer: count=exact`). + * 2. If a cache file with the same fingerprint exists, load vectors from disk. + * 3. Otherwise download as before and rewrite the cache. + * + * ## What the fingerprint can and cannot see + * + * It detects inserts, deactivations (count changes) and any edit that bumps + * updated_at. It does NOT detect an in-place edit that leaves updated_at and + * the row count unchanged. setup.sql maintains updated_at by trigger, so that + * requires a store whose trigger is missing. `cache-flush` always bypasses the + * disk cache, so there is a manual way out. + * + * ## Failure posture + * + * Every failure here degrades to the old behaviour (download), never to stale + * or empty results. An unreadable, corrupt, mismatched or foreign cache file is + * a miss. A failed fingerprint request is a miss. A failed write is logged and + * ignored. The cache is an optimisation and must never be load-bearing. + * + * ## Concurrency + * + * Writes go to a temp file and are renamed into place, so a reader sees either + * the old complete file or the new complete file, never a partial one. + * + * A fan-out starts N processes at the same instant, and any write to the store + * invalidates the cache for all of them at once. Without coordination that is N + * identical downloads — the exact cost this module exists to remove. So a miss + * is single-flighted across processes: one LEADER takes an O_EXCL lock file and + * downloads; FOLLOWERS wait for the cache to appear and read it. A follower + * that waits too long, or finds the leader's lock stale, downloads for itself. + * Waiting is async (never blocks the event loop) and bounded, so the worst case + * is today's behaviour plus a bounded delay — never a hang. + */ + +import * as fs from "fs"; +import * as path from "path"; +import * as crypto from "crypto"; +import { getGitmemDir } from "./gitmem-dir.js"; + +/** Bump when the on-disk shape changes; an older file is then a miss. */ +export const VECTOR_CACHE_FORMAT = 1; + +export interface StoreFingerprint { + /** Active learnings of the indexed types, as counted by the store. */ + count: number; + /** Newest updated_at among them, or null for an empty store. */ + latestUpdatedAt: string | null; +} + +interface VectorCacheFile { + format: number; + /** Identifies the store + table, so switching stores can never serve foreign rows. */ + storeKey: string; + fingerprint: StoreFingerprint; + savedAt: string; + rows: T[]; +} + +export type CacheReadResult = + | { hit: true; rows: T[]; bytesOnDisk: number } + | { hit: false; reason: string }; + +/** Opt-out: GITMEM_VECTOR_DISK_CACHE=0 (or "false") restores download-every-start. */ +export function isVectorDiskCacheEnabled(): boolean { + const v = process.env.GITMEM_VECTOR_DISK_CACHE; + return !(v === "0" || v === "false"); +} + +/** + * A stable, non-reversible key for (store URL, table). Hashed so the cache file + * name and contents never carry the project URL in clear. + */ +export function computeStoreKey(supabaseUrl: string, table: string): string { + return crypto.createHash("sha256").update(`${supabaseUrl}\n${table}`).digest("hex").slice(0, 16); +} + +export function getVectorCachePath(storeKey: string): string { + return path.join(getGitmemDir(), "cache", `learnings-vectors-${storeKey}.json`); +} + +export function fingerprintsMatch(a: StoreFingerprint, b: StoreFingerprint): boolean { + return a.count === b.count && a.latestUpdatedAt === b.latestUpdatedAt; +} + +/** + * Read the cache if, and only if, it was written for this store and this + * fingerprint. Anything else is a miss with a reason (logged by the caller). + */ +export function readVectorCache(storeKey: string, expected: StoreFingerprint): CacheReadResult { + if (!isVectorDiskCacheEnabled()) return { hit: false, reason: "disabled by GITMEM_VECTOR_DISK_CACHE" }; + + // A failed fingerprint request reports count -1. Never match on it: two + // failures would otherwise "agree" and serve whatever is on disk. + if (expected.count < 0) return { hit: false, reason: "store fingerprint unavailable" }; + + const file = getVectorCachePath(storeKey); + let raw: string; + try { + if (!fs.existsSync(file)) return { hit: false, reason: "no cache file" }; + raw = fs.readFileSync(file, "utf-8"); + } catch (error) { + return { hit: false, reason: `unreadable: ${error instanceof Error ? error.message : String(error)}` }; + } + + let parsed: VectorCacheFile; + try { + parsed = JSON.parse(raw) as VectorCacheFile; + } catch { + return { hit: false, reason: "corrupt JSON" }; + } + + if (!parsed || typeof parsed !== "object") return { hit: false, reason: "corrupt shape" }; + if (parsed.format !== VECTOR_CACHE_FORMAT) return { hit: false, reason: `format ${parsed.format} != ${VECTOR_CACHE_FORMAT}` }; + if (parsed.storeKey !== storeKey) return { hit: false, reason: "written for a different store" }; + if (!Array.isArray(parsed.rows)) return { hit: false, reason: "corrupt rows" }; + if (!parsed.fingerprint || !fingerprintsMatch(parsed.fingerprint, expected)) { + return { hit: false, reason: "store changed since cache was written" }; + } + + return { hit: true, rows: parsed.rows, bytesOnDisk: Buffer.byteLength(raw) }; +} + +/** + * Persist rows under the fingerprint that was observed BEFORE they were + * downloaded. If the store changed mid-download the file is merely stale and + * the next start misses and reloads — the safe direction. + */ +export function writeVectorCache(storeKey: string, fingerprint: StoreFingerprint, rows: T[]): boolean { + if (!isVectorDiskCacheEnabled()) return false; + if (fingerprint.count < 0) return false; // never persist under an unknown fingerprint + + const file = getVectorCachePath(storeKey); + const tmp = `${file}.${process.pid}.${Date.now()}.tmp`; + try { + fs.mkdirSync(path.dirname(file), { recursive: true }); + const payload: VectorCacheFile = { + format: VECTOR_CACHE_FORMAT, + storeKey, + fingerprint, + savedAt: new Date().toISOString(), + rows, + }; + fs.writeFileSync(tmp, JSON.stringify(payload), { mode: 0o600 }); + fs.renameSync(tmp, file); + return true; + } catch (error) { + console.error("[vector-cache] Failed to write cache (non-fatal):", error instanceof Error ? error.message : error); + try { if (fs.existsSync(tmp)) fs.unlinkSync(tmp); } catch { /* best effort */ } + return false; + } +} + +// --------------------------------------------------------------------------- +// Cross-process single-flight for cache misses +// --------------------------------------------------------------------------- + +/** A leader that has held the lock this long is presumed dead. A bulk load has a 15-30s budget. */ +const LOCK_STALE_MS = 60_000; +/** How long a follower waits for the leader before downloading for itself. */ +const FOLLOWER_WAIT_MS = 25_000; +const FOLLOWER_POLL_MS = 150; + +function lockPathFor(storeKey: string): string { + return `${getVectorCachePath(storeKey)}.lock`; +} + +/** Try to become the downloader. Returns a release function, or null if someone else is. */ +export function tryAcquireDownloadLock(storeKey: string): (() => void) | null { + if (!isVectorDiskCacheEnabled()) return () => {}; + const lock = lockPathFor(storeKey); + try { + fs.mkdirSync(path.dirname(lock), { recursive: true }); + } catch { + return () => {}; // cannot coordinate — behave as a lone downloader + } + + for (let attempt = 0; attempt < 2; attempt++) { + try { + const fd = fs.openSync(lock, fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_WRONLY, 0o600); + fs.writeSync(fd, JSON.stringify({ pid: process.pid, acquired_at: new Date().toISOString() })); + fs.closeSync(fd); + return () => { try { fs.unlinkSync(lock); } catch { /* already gone */ } }; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") return () => {}; // odd fs — lone downloader + // Held by someone. Break it only if its holder is presumed dead, then retry once. + try { + if (Date.now() - fs.statSync(lock).mtimeMs > LOCK_STALE_MS) { + fs.unlinkSync(lock); + continue; + } + } catch { /* vanished between calls — retry */ continue; } + return null; + } + } + return null; +} + +/** + * Follower path: wait for the leader's cache to land. Resolves with the rows on + * a hit, or null when the caller should download for itself (leader gone, + * leader failed, or the wait budget ran out). + */ +export async function waitForLeaderCache( + storeKey: string, + expected: StoreFingerprint, + waitMs: number = FOLLOWER_WAIT_MS +): Promise & { hit: true } | null> { + const lock = lockPathFor(storeKey); + const deadline = Date.now() + waitMs; + + while (Date.now() < deadline) { + await new Promise((r) => setTimeout(r, FOLLOWER_POLL_MS)); + + const lockHeld = fs.existsSync(lock); + const cached = readVectorCache(storeKey, expected); + if (cached.hit) return cached; + + // Lock released and still no usable cache: the leader failed, or it cached a + // newer fingerprint than the one we observed. Either way, stop waiting. + if (!lockHeld) return null; + } + return null; +} diff --git a/tests/unit/services/vector-disk-cache.test.ts b/tests/unit/services/vector-disk-cache.test.ts new file mode 100644 index 0000000..b87d965 --- /dev/null +++ b/tests/unit/services/vector-disk-cache.test.ts @@ -0,0 +1,214 @@ +/** + * GIT-98: disk-backed vector cache. + * + * The contract under test: every failure is a MISS (download as before), never + * stale or foreign rows. Real filesystem, isolated GITMEM_DIR per test. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import * as fs from "fs"; +import * as os from "os"; +import * as path from "path"; + +let dir: string; + +async function load() { + vi.resetModules(); + return import("../../../src/services/vector-disk-cache.js"); +} + +const FP = { count: 3, latestUpdatedAt: "2026-09-20T15:25:16.480033+00:00" }; +const ROWS = [ + { id: "a", title: "one", embedding: [0.1, 0.2, 0.3] }, + { id: "b", title: "two", embedding: [0.4, 0.5, 0.6] }, + { id: "c", title: "three", embedding: [0.7, 0.8, 0.9] }, +]; + +describe("vector-disk-cache (GIT-98)", () => { + beforeEach(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), "gitmem-vcache-")); + process.env.GITMEM_DIR = dir; + delete process.env.GITMEM_VECTOR_DISK_CACHE; + }); + + afterEach(() => { + fs.rmSync(dir, { recursive: true, force: true }); + delete process.env.GITMEM_DIR; + delete process.env.GITMEM_VECTOR_DISK_CACHE; + }); + + it("round-trips rows when store and fingerprint match", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + expect(c.writeVectorCache(key, FP, ROWS)).toBe(true); + + const r = c.readVectorCache(key, FP); + expect(r.hit).toBe(true); + if (r.hit) { + expect(r.rows).toEqual(ROWS); + expect(r.bytesOnDisk).toBeGreaterThan(0); + } + }); + + it("misses when the row count changed (insert or deactivation)", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + const r = c.readVectorCache(key, { ...FP, count: 4 }); + expect(r).toEqual({ hit: false, reason: "store changed since cache was written" }); + }); + + it("misses when the newest updated_at changed (edit)", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + const r = c.readVectorCache(key, { ...FP, latestUpdatedAt: "2026-09-21T00:00:00+00:00" }); + expect(r.hit).toBe(false); + }); + + it("never serves another store's rows", async () => { + const c = await load(); + const mine = c.computeStoreKey("https://mine.supabase.co", "gitmem_learnings"); + const theirs = c.computeStoreKey("https://theirs.supabase.co", "gitmem_learnings"); + expect(mine).not.toBe(theirs); + c.writeVectorCache(mine, FP, ROWS); + expect(c.readVectorCache(theirs, FP).hit).toBe(false); + + // Same URL, different table prefix is a different store too. + const prefixed = c.computeStoreKey("https://mine.supabase.co", "orchestra_learnings"); + expect(prefixed).not.toBe(mine); + }); + + it("refuses a file that claims a different storeKey, even at the right path", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + const file = c.getVectorCachePath(key); + const tampered = JSON.parse(fs.readFileSync(file, "utf-8")); + tampered.storeKey = "0000000000000000"; + fs.writeFileSync(file, JSON.stringify(tampered)); + expect(c.readVectorCache(key, FP)).toEqual({ hit: false, reason: "written for a different store" }); + }); + + it("an unknown fingerprint (count -1) never matches and is never persisted", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + const unknown = { count: -1, latestUpdatedAt: null }; + + expect(c.writeVectorCache(key, unknown, ROWS)).toBe(false); + expect(fs.existsSync(c.getVectorCachePath(key))).toBe(false); + + // Even if a file with an identical unknown fingerprint somehow exists, two failures must not "agree". + fs.mkdirSync(path.dirname(c.getVectorCachePath(key)), { recursive: true }); + fs.writeFileSync(c.getVectorCachePath(key), JSON.stringify({ + format: c.VECTOR_CACHE_FORMAT, storeKey: key, fingerprint: unknown, savedAt: "x", rows: ROWS, + })); + expect(c.readVectorCache(key, unknown)).toEqual({ hit: false, reason: "store fingerprint unavailable" }); + }); + + it("treats corrupt, truncated and wrong-format files as misses", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + const file = c.getVectorCachePath(key); + fs.mkdirSync(path.dirname(file), { recursive: true }); + + fs.writeFileSync(file, '{"format":1,"storeKey":"' + key + '","rows":[{"id":"a"'); + expect(c.readVectorCache(key, FP)).toEqual({ hit: false, reason: "corrupt JSON" }); + + fs.writeFileSync(file, "null"); + expect(c.readVectorCache(key, FP).hit).toBe(false); + + fs.writeFileSync(file, JSON.stringify({ format: 999, storeKey: key, fingerprint: FP, rows: ROWS })); + expect(c.readVectorCache(key, FP).hit).toBe(false); + + fs.writeFileSync(file, JSON.stringify({ format: c.VECTOR_CACHE_FORMAT, storeKey: key, fingerprint: FP, rows: "nope" })); + expect(c.readVectorCache(key, FP)).toEqual({ hit: false, reason: "corrupt rows" }); + }); + + it("GITMEM_VECTOR_DISK_CACHE=0 disables both read and write", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + + process.env.GITMEM_VECTOR_DISK_CACHE = "0"; + expect(c.readVectorCache(key, FP).hit).toBe(false); + fs.rmSync(c.getVectorCachePath(key)); + expect(c.writeVectorCache(key, FP, ROWS)).toBe(false); + expect(fs.existsSync(c.getVectorCachePath(key))).toBe(false); + }); + + it("leaves no temp files behind and writes owner-only", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + const cacheDir = path.dirname(c.getVectorCachePath(key)); + expect(fs.readdirSync(cacheDir).filter((f) => f.endsWith(".tmp"))).toEqual([]); + if (process.platform !== "win32") { + expect(fs.statSync(c.getVectorCachePath(key)).mode & 0o777).toBe(0o600); + } + }); + + it("the cache file does not contain the Supabase URL in clear", async () => { + const c = await load(); + const key = c.computeStoreKey("https://secret-project-ref.supabase.co", "gitmem_learnings"); + c.writeVectorCache(key, FP, ROWS); + const file = c.getVectorCachePath(key); + expect(path.basename(file)).not.toContain("secret-project-ref"); + expect(fs.readFileSync(file, "utf-8")).not.toContain("secret-project-ref"); + }); + + describe("single-flight download lock", () => { + it("only one caller becomes leader; release lets the next one in", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + const release = c.tryAcquireDownloadLock(key); + expect(release).toBeTypeOf("function"); + expect(c.tryAcquireDownloadLock(key)).toBeNull(); + release!(); + const again = c.tryAcquireDownloadLock(key); + expect(again).toBeTypeOf("function"); + again!(); + expect(fs.existsSync(`${c.getVectorCachePath(key)}.lock`)).toBe(false); + }); + + it("breaks a lock whose holder is presumed dead", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.tryAcquireDownloadLock(key); // leader "crashes" without releasing + const lock = `${c.getVectorCachePath(key)}.lock`; + const old = new Date(Date.now() - 5 * 60_000); + fs.utimesSync(lock, old, old); + const release = c.tryAcquireDownloadLock(key); + expect(release).toBeTypeOf("function"); + release!(); + }); + + it("a follower receives the leader's rows once the cache lands", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + const release = c.tryAcquireDownloadLock(key)!; + const follower = c.waitForLeaderCache(key, FP, 5_000); + setTimeout(() => { c.writeVectorCache(key, FP, ROWS); release(); }, 300); + const got = await follower; + expect(got?.hit).toBe(true); + expect(got?.rows).toEqual(ROWS); + }); + + it("a follower gives up (null) when the leader releases without a usable cache", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + const release = c.tryAcquireDownloadLock(key)!; + const follower = c.waitForLeaderCache(key, FP, 5_000); + setTimeout(() => release(), 300); // leader failed: no cache written + expect(await follower).toBeNull(); + }); + + it("a follower's wait is bounded when the leader hangs", async () => { + const c = await load(); + const key = c.computeStoreKey("https://x.supabase.co", "gitmem_learnings"); + c.tryAcquireDownloadLock(key); // never released, never writes + const t0 = Date.now(); + expect(await c.waitForLeaderCache(key, FP, 600)).toBeNull(); + expect(Date.now() - t0).toBeLessThan(2_000); + }); + }); +});