/** * @module tests/integration/wait-for-indexed * @description THE READ BARRIER — `brain.waitForIndexed(path?, opts?)`. A * consumer that writes and then semantically recalls gets ONE honest barrier * instead of guessing. The contract pinned here: * * 1. SEMANTIC LEG: a deferred add followed by `waitForIndexed('semantic')` * resolves only after the vector landed — the row is vector-searchable * the moment the barrier returns. * 2. TYPED TIMEOUT: `timeoutMs` expiry REJECTS with * WaitForIndexedTimeoutError carrying the leg + the pending count and * naming the gauge — never a silent partial wait. * 3. NO-ARG: every projection at the head; today that means the deferred * embed backlog is drained. * 4. SYNCHRONOUS LEGS: metadata/graph/aggregation resolve immediately by * design today (they update inside the write path) — even while the * semantic backlog is wedged. * 5. GAUGES: getIndexStatus().projections carries the per-leg numbers, and * the top-level pendingEmbeds compat field agrees with the semantic one. * 6. GENERATION REFINEMENT: an empty backlog satisfies any generation * immediately; a non-empty one falls back to the full drain. */ import { describe, it, expect, afterEach, vi } from 'vitest' import { Brainy, WaitForIndexedTimeoutError } from '../../src/index.js' import { NounType } from '../../src/types/graphTypes.js' const brains: Brainy[] = [] async function memBrain(): Promise { const b = new Brainy({ storage: { type: 'memory' }, requireSubtype: false }) await b.init() brains.push(b) return b } /** * Abandon a poisoned in-flight embed run (its embed promise never resolves — * production is covered by the worker's 60s hang guard; the test takes the * white-box shortcut for speed), then drain so teardown never wedges. */ async function unwedge(brain: Brainy): Promise { ;(brain as unknown as { _embedWorkerFlight: Promise | null })._embedWorkerFlight = null await brain.awaitPendingEmbeds() } afterEach(async () => { vi.restoreAllMocks() for (const b of brains.splice(0)) await b.close().catch(() => {}) }) describe('waitForIndexed — the read barrier', () => { it("SEMANTIC LEG: deferred add → waitForIndexed('semantic') resolves and the row is vector-searchable after", async () => { const brain = await memBrain() const embedSpy = vi.spyOn(brain, 'embed') const id = await brain.add({ data: 'the quarterly revenue report for the northern region', type: NounType.Document, deferEmbedding: true, metadata: { kind: 'report' } }) expect(embedSpy, 'no embed on the ack path').not.toHaveBeenCalled() expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1) await brain.waitForIndexed('semantic') // The barrier's meaning: backlog drained, vector real, row searchable. expect(brain.pendingEmbedCount(), 'barrier means drained').toBe(0) const after = await brain.get(id, { includeVectors: true }) expect((after!.vector as number[]).length, 'real vector after the barrier').toBeGreaterThan(0) const hits = await brain.find({ query: 'the quarterly revenue report for the northern region', searchMode: 'semantic', limit: 5 }) expect(hits.map((r) => r.id), 'vector-searchable after the barrier').toContain(id) }) it('TYPED TIMEOUT: a hung embedder + timeoutMs rejects with the typed error naming the pending count and the gauge', async () => { const brain = await memBrain() const hang = vi .spyOn(brain, 'embed') .mockImplementation(() => new Promise(() => {})) await brain.add({ data: 'never lands while the embedder hangs', type: NounType.Document, deferEmbedding: true, metadata: {} }) expect(brain.pendingEmbedCount()).toBe(1) let caught: unknown try { await brain.waitForIndexed('semantic', { timeoutMs: 200 }) } catch (e) { caught = e } expect(caught, 'expiry REJECTS — never a silent partial wait').toBeInstanceOf( WaitForIndexedTimeoutError ) const err = caught as WaitForIndexedTimeoutError expect(err.path).toBe('semantic') expect(err.timeoutMs).toBe(200) expect(err.pendingEmbeds).toBeGreaterThanOrEqual(1) // The message names what was still pending and the gauge to check. expect(err.message).toContain(`${err.pendingEmbeds} deferred embed`) expect(err.message).toContain('getIndexStatus().projections.semantic.pendingEmbeds') hang.mockRestore() await unwedge(brain) expect(brain.pendingEmbedCount()).toBe(0) }) it('NO-ARG: waitForIndexed() waits on the pending-embed drain (every projection at the head)', async () => { const brain = await memBrain() await brain.add({ data: 'a deferred capture that the bare barrier must cover', type: NounType.Document, deferEmbedding: true, metadata: {} }) expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1) await brain.waitForIndexed() expect( brain.pendingEmbedCount(), 'the bare barrier drained the only asynchronous projection' ).toBe(0) }) it('SYNCHRONOUS LEGS: metadata/graph/aggregation resolve immediately — even while the semantic backlog is wedged', async () => { const brain = await memBrain() // Quiet brain first: all three legs resolve on a brain with no backlog. await brain.add({ data: 'quiet row', type: NounType.Document, metadata: { q: 1 } }) await brain.awaitPendingEmbeds() await brain.waitForIndexed('metadata') await brain.waitForIndexed('graph') await brain.waitForIndexed('aggregation') // The stronger pin: these projections update inside the write path today, // so their leg resolves immediately BY DESIGN — independent of a wedged // semantic backlog. (If any of them incorrectly delegated to the embed // drain, this test would hang.) const hang = vi .spyOn(brain, 'embed') .mockImplementation(() => new Promise(() => {})) await brain.add({ data: 'wedged deferred row', type: NounType.Document, deferEmbedding: true, metadata: {} }) expect(brain.pendingEmbedCount()).toBe(1) await brain.waitForIndexed('metadata') await brain.waitForIndexed('graph') await brain.waitForIndexed('aggregation') hang.mockRestore() await unwedge(brain) }) it('GAUGES: getIndexStatus().projections carries the per-leg shape, and the compat field agrees', async () => { const brain = await memBrain() await brain.add({ data: 'gauge row', type: NounType.Document, metadata: { g: 1 } }) await brain.awaitPendingEmbeds() const status = await brain.getIndexStatus() expect(status.projections).toEqual({ semantic: { pendingEmbeds: 0 }, metadata: { synchronous: true }, graph: { synchronous: true }, aggregation: { pendingBackfills: 0, pendingCatchUps: 0 } }) // Compat: the existing top-level gauge stays and agrees. expect(status.pendingEmbeds).toBe(0) // The semantic gauge is honest while a backlog exists. const hang = vi .spyOn(brain, 'embed') .mockImplementation(() => new Promise(() => {})) await brain.add({ data: 'backlogged row', type: NounType.Document, deferEmbedding: true, metadata: {} }) const busy = await brain.getIndexStatus() expect(busy.projections.semantic.pendingEmbeds).toBeGreaterThanOrEqual(1) expect(busy.pendingEmbeds).toBe(busy.projections.semantic.pendingEmbeds) hang.mockRestore() await unwedge(brain) }) it('GENERATION REFINEMENT: an empty backlog satisfies any generation immediately; a non-empty one falls back to the full drain', async () => { const brain = await memBrain() await brain.add({ data: 'generation row', type: NounType.Document, metadata: {} }) await brain.awaitPendingEmbeds() // Empty backlog: the semantic watermark is at the head — >= any committed G. await brain.waitForIndexed('semantic', { generation: 1 }) // Non-empty backlog: the conservative full drain (a superset of the // requested wait, never a partial one). await brain.add({ data: 'second generation row', type: NounType.Document, deferEmbedding: true, metadata: {} }) await brain.waitForIndexed('semantic', { generation: 1 }) expect(brain.pendingEmbedCount(), 'the fallback is the full drain').toBe(0) }) })