diff --git a/src/brainy.ts b/src/brainy.ts index 17fa4ad9..696a5f87 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -1820,31 +1820,31 @@ export class Brainy implements BrainyInterface { // a deferred write's ack and its background embed DELAYED a vector; // this is where it lands. if (!this.isReadOnly) { - try { - await step( - 'bridge-pending-embed-sidecars', - 'migrating any pre-log deferred-embed marker files into the generation log', - () => this.bridgeLegacyPendingEmbedSidecars() - ) - await step( - 'recover-pending-embeds', - 'folding the generation log\'s deferred-embed markers back into the pending set', - () => this.recoverPendingEmbedsFromLog() - ) - if (this._pendingEmbedIds.size > 0) { - prodLog.info( - `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` + - `session — resuming in the background` + // BEHIND THE DOORS (the open pays nothing here): the bridge + the + // recovery fold run as one latched background task; the embed worker + // starts when it settles. A pending embed's outcome was always + // eventual — moving its recovery off the open's foreground changes + // when the worker starts, never whether a marker is honored. + // awaitPendingEmbeds() and close() wait on the latch first. + this._pendingEmbedRecovery = (async () => { + try { + await this.bridgeLegacyPendingEmbedSidecars() + await this.recoverPendingEmbedsFromLog() + if (this._pendingEmbedIds.size > 0) { + prodLog.info( + `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` + + `session — resuming in the background` + ) + const t = setTimeout(() => this.kickEmbedWorker(), 0) + ;(t as { unref?: () => void }).unref?.() + } + } catch (err) { + prodLog.warn( + `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` + + `the log's markers remain durable; recovery retries next open` ) - const t = setTimeout(() => this.kickEmbedWorker(), 0) - ;(t as { unref?: () => void }).unref?.() } - } catch (err) { - prodLog.warn( - `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` + - `the log's markers remain durable; recovery retries next open` - ) - } + })() } // PHASE 4 of 5 — "VFS bootstrap": shutdown-hook registration, blob @@ -2408,6 +2408,19 @@ export class Brainy implements BrainyInterface { */ private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/' + /** + * Storage-root-relative path of the ADVISORY pending-embed low-water mark: + * `{ generation, writtenAt }`, written whenever the pending set drains to + * empty (and at clean close when empty). Every marker in facts at or below + * `generation` is consumed, so recovery scans from `generation + 1`. The + * mark is advisory and monotone-safe: stale-low costs a longer scan, never + * a lost marker; it is never required for correctness. + */ + private static readonly PENDING_EMBED_LOWWATER_PATH = '_system/pending_embeds_lowwater.json' + + /** Resolves when the background pending-embed recovery fold has settled (open arms it). */ + private _pendingEmbedRecovery: Promise | null = null + /** * @description Mark a deferred embed pending (MT5): the id joins the * in-memory fast-path set and the returned `embed.pending` record is @@ -2435,6 +2448,40 @@ export class Brainy implements BrainyInterface { */ private clearPendingEmbed(id: string): void { this._pendingEmbedIds.delete(id) + if (this._pendingEmbedIds.size === 0) this.maybeWriteEmbedLowWater() + } + + /** + * @description Advance the advisory low-water mark: called at drain-to-empty + * (and at clean close when empty), it records the fact log's CURRENT head — + * with the set empty, every marker at or below the head has been consumed, + * so the next open's recovery fold scans only what comes after. Fire-and- + * forget at the drain (close() awaits the core); loud on failure: a missed + * write costs the next open a longer scan, never a marker. No-op without a + * fact log (no durable markers exist there) and on read-only opens. + */ + private maybeWriteEmbedLowWater(): void { + void this.writeEmbedLowWater() + } + + /** The awaitable core of {@link maybeWriteEmbedLowWater} — close() awaits it. */ + private async writeEmbedLowWater(): Promise { + if (this.isReadOnly) return + const log = this.generationStore ? this.generationStore.getFactLog() : null + if (!log) return + const generation = log.headGeneration() + if (!(generation > 0)) return + try { + await this.storage.writeRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH, { + generation, + writtenAt: Date.now() + }) + } catch (err) { + prodLog.warn( + `[Brainy] pending-embed low-water write failed at generation ${generation}: ` + + `${(err as Error).message} — the next open scans from the previous mark` + ) + } } /** @@ -2445,9 +2492,14 @@ export class Brainy implements BrainyInterface { * survives the fold is exactly the set of acknowledged deferred writes * whose vectors have not landed. * - * BOUND (honest): no durable low-water mark exists for the earliest - * unconsumed pending, so the fold scans the log's committed facts from - * generation 1 — a sequential read of the log at open, O(log bytes). + * BOUND: the scan starts at the advisory low-water mark + * ({@link Brainy.PENDING_EMBED_LOWWATER_PATH}) — the log head at which the + * pending set last drained to empty — so a settled brain reads only the + * facts since then, not its whole history. Without a mark (first open + * after upgrade) it scans from generation 1, once; a stale-low mark costs + * a longer scan, never a marker. The fold runs BEHIND the doors (open + * arms it as a background task and the embed worker starts when it + * settles); {@link awaitPendingEmbeds} and close() wait for it first. * It is SKIPPED WHOLESALE when the log has never had a v2 tail * ({@link FactLog.hasV2History} — v1 facts cannot carry marker records), * so pre-cutover brains pay nothing; on a mixed log the scan still reads @@ -2460,7 +2512,18 @@ export class Brainy implements BrainyInterface { private async recoverPendingEmbedsFromLog(): Promise { const log = this.generationStore.getFactLog() if (!log || !log.hasV2History()) return - const scan = log.scanFacts({ fromGeneration: 1 }) + let fromGeneration = 1 + try { + const mark = (await this.storage.readRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH)) as { + generation?: number + } | null + if (mark && typeof mark.generation === 'number' && mark.generation > 0) { + fromGeneration = mark.generation + 1 + } + } catch { + // No mark (or unreadable): scan from 1 — correctness over cost. + } + const scan = log.scanFacts({ fromGeneration }) for await (const batch of scan.batches()) { for (const fact of batch.facts) { for (const record of fact.records ?? []) { @@ -2647,6 +2710,7 @@ export class Brainy implements BrainyInterface { * before I proceed" callers use this; nothing else ever needs to wait. */ public async awaitPendingEmbeds(): Promise { + if (this._pendingEmbedRecovery) await this._pendingEmbedRecovery while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) { this.kickEmbedWorker() await (this._embedWorkerFlight ?? Promise.resolve()) @@ -19520,6 +19584,19 @@ export class Brainy implements BrainyInterface { * terminal releases have run. */ async close(): Promise { + if (this._pendingEmbedRecovery) { + // Settle the background marker fold before the durable steps — its scan + // is bounded by the low-water mark (a full scan happens at most once, + // on the first open after upgrade). + const settleStart = Date.now() + await this._pendingEmbedRecovery + const settleMs = Date.now() - settleStart + if (settleMs >= 1000) { + prodLog.info(`[Brainy] close: pending-embed recovery settled in ${settleMs}ms`) + } + this._pendingEmbedRecovery = null + } + if (this._pendingEmbedIds.size === 0) await this.writeEmbedLowWater() let closeFailure: unknown = null try { await this.closeDurableSteps() diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index d8bcb780..a1cc2e35 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -2942,19 +2942,33 @@ export abstract class BaseStorage extends BaseStorageAdapter { !options.filter.service && !options.filter.metadata ) { - const sourceId = Array.isArray(options.filter.sourceId) - ? options.filter.sourceId[0] - : options.filter.sourceId + const sourceIds = Array.isArray(options.filter.sourceId) + ? options.filter.sourceId + : [options.filter.sourceId] - const verbType = Array.isArray(options.filter.verbType) - ? options.filter.verbType[0] - : options.filter.verbType + // EVERY requested verb type is honoured — an array used to collapse to + // its first element here, silently dropping the rest of the ask. + const verbTypes = new Set( + Array.isArray(options.filter.verbType) + ? options.filter.verbType + : [options.filter.verbType] + ) - // Get verbs by source, then filter by type (O(1) graph lookup + O(n) type filter), - // then apply the subtype / visibility metadata filters on the candidate set. - const verbsBySource = await this.getVerbsBySource_internal(sourceId) + // Get verbs by source (union over every requested source), filter by the + // requested type SET (O(1) graph lookup + O(n) type filter), then apply + // the subtype / visibility metadata filters on the candidate set. + const bySource: HNSWVerbWithMetadata[] = [] + const seenVerbIds = new Set() + for (const oneSource of sourceIds) { + for (const v of await this.getVerbsBySource_internal(oneSource)) { + if (!seenVerbIds.has(v.id)) { + seenVerbIds.add(v.id) + bySource.push(v) + } + } + } const filteredVerbs = this.applyVerbMetadataFilters( - verbsBySource.filter(v => v.verb === verbType), + bySource.filter(v => verbTypes.has(v.verb)), options.filter ) @@ -2985,16 +2999,22 @@ export abstract class BaseStorage extends BaseStorageAdapter { !options.filter.service && !options.filter.metadata ) { - const sourceId = Array.isArray(options.filter.sourceId) - ? options.filter.sourceId[0] - : options.filter.sourceId - - // Get verbs by source directly (hydrated with metadata), then apply the - // subtype / visibility metadata filters on the O(degree) candidate set. - const verbsBySource = this.applyVerbMetadataFilters( - await this.getVerbsBySource_internal(sourceId), - options.filter - ) + // EVERY requested source is honoured — an array used to collapse to + // its first element here, silently dropping the rest of the ask. + const onlySourceIds = Array.isArray(options.filter.sourceId) + ? options.filter.sourceId + : [options.filter.sourceId] + const sourceUnion: HNSWVerbWithMetadata[] = [] + const seenSourceVerbIds = new Set() + for (const oneSource of onlySourceIds) { + for (const v of await this.getVerbsBySource_internal(oneSource)) { + if (!seenSourceVerbIds.has(v.id)) { + seenSourceVerbIds.add(v.id) + sourceUnion.push(v) + } + } + } + const verbsBySource = this.applyVerbMetadataFilters(sourceUnion, options.filter) // Apply pagination const paginatedVerbs = verbsBySource.slice(offset, offset + limit) @@ -3023,16 +3043,22 @@ export abstract class BaseStorage extends BaseStorageAdapter { !options.filter.service && !options.filter.metadata ) { - const targetId = Array.isArray(options.filter.targetId) - ? options.filter.targetId[0] - : options.filter.targetId - - // Get verbs by target directly (hydrated with metadata), then apply the - // subtype / visibility metadata filters on the O(degree) candidate set. - const verbsByTarget = this.applyVerbMetadataFilters( - await this.getVerbsByTarget_internal(targetId), - options.filter - ) + // EVERY requested target is honoured — an array used to collapse to + // its first element here, silently dropping the rest of the ask. + const onlyTargetIds = Array.isArray(options.filter.targetId) + ? options.filter.targetId + : [options.filter.targetId] + const targetUnion: HNSWVerbWithMetadata[] = [] + const seenTargetVerbIds = new Set() + for (const oneTarget of onlyTargetIds) { + for (const v of await this.getVerbsByTarget_internal(oneTarget)) { + if (!seenTargetVerbIds.has(v.id)) { + seenTargetVerbIds.add(v.id) + targetUnion.push(v) + } + } + } + const verbsByTarget = this.applyVerbMetadataFilters(targetUnion, options.filter) // Apply pagination const paginatedVerbs = verbsByTarget.slice(offset, offset + limit) @@ -3061,16 +3087,25 @@ export abstract class BaseStorage extends BaseStorageAdapter { !options.filter.service && !options.filter.metadata ) { - const verbType = Array.isArray(options.filter.verbType) - ? options.filter.verbType[0] - : options.filter.verbType + // EVERY requested verb type is honoured — an array used to collapse to + // its first element here, silently dropping the rest of the ask. + const verbTypes = Array.isArray(options.filter.verbType) + ? options.filter.verbType + : [options.filter.verbType] - // Get verbs by type directly (hydrated with metadata), then apply the - // subtype / visibility metadata filters on the candidate set. - const verbsByType = this.applyVerbMetadataFilters( - await this.getVerbsByType_internal(verbType), - options.filter - ) + // Get verbs by each requested type (hydrated with metadata), deduped by + // id, then apply the subtype / visibility metadata filters on the set. + const byType: HNSWVerbWithMetadata[] = [] + const seenTypeVerbIds = new Set() + for (const oneType of verbTypes) { + for (const v of await this.getVerbsByType_internal(oneType)) { + if (!seenTypeVerbIds.has(v.id)) { + seenTypeVerbIds.add(v.id) + byType.push(v) + } + } + } + const verbsByType = this.applyVerbMetadataFilters(byType, options.filter) // Apply pagination const paginatedVerbs = verbsByType.slice(offset, offset + limit) diff --git a/tests/integration/pending-embed-low-water.test.ts b/tests/integration/pending-embed-low-water.test.ts new file mode 100644 index 00000000..ff01b349 --- /dev/null +++ b/tests/integration/pending-embed-low-water.test.ts @@ -0,0 +1,145 @@ +/** + * @module tests/integration/pending-embed-low-water + * @description The pending-embed recovery fold is bounded and background (10.4.9). + * + * The fold used to scan the generation log from generation 1 at EVERY open, + * on the open's foreground — O(whole history) per open on long-lived brains. + * Now: an advisory low-water mark (`_system/pending_embeds_lowwater.json`) + * records the committed generation whenever the pending set drains to empty, + * recovery scans from `mark + 1`, and the fold runs behind the doors as a + * latched background task the worker, `awaitPendingEmbeds()` and `close()` + * wait on. The mark is advisory: stale-low costs a longer scan, never a + * marker — a pending embed enqueued before a crash is still recovered. + */ +import { describe, it, expect, afterEach, vi } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../src/brainy' +import { NounType } from '../../src/types/graphTypes' + +const LOWWATER_PATH = '_system/pending_embeds_lowwater.json' + +describe('pending-embed recovery: bounded by the low-water mark, behind the doors', () => { + const roots: string[] = [] + const dir = (): string => { + const d = mkdtempSync(join(tmpdir(), 'brainy-lowwater-')) + roots.push(d) + return d + } + const open = async (root: string): Promise> => { + const brain = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: root } + }) + await brain.init() + return brain + } + + afterEach(() => { + for (const d of roots.splice(0)) rmSync(d, { recursive: true, force: true }) + }) + + it('drain-to-empty writes the mark, and the next open scans from mark + 1', async () => { + const root = dir() + const brain = await open(root) + // Hold the worker so the pending state is observable, then release it. + const realKick = (brain as any).kickEmbedWorker.bind(brain) + ;(brain as any).kickEmbedWorker = () => {} + await brain.add({ + id: 'row-1', + data: 'the first deferred row', + type: NounType.Thing, + deferEmbedding: true + }) + expect(brain.pendingEmbedCount()).toBeGreaterThan(0) + ;(brain as any).kickEmbedWorker = realKick + await brain.awaitPendingEmbeds() + // The drain wrote the advisory mark (fire-and-forget: settle the microtask). + await new Promise((r) => setTimeout(r, 50)) + const mark = (await (brain as any).storage.readRawObject(LOWWATER_PATH)) as { + generation: number + } | null + expect(mark).not.toBeNull() + expect(mark!.generation).toBeGreaterThan(0) + await brain.close() + + const brain2 = await open(root) + const log = (brain2 as any).generationStore.getFactLog() + const scanSpy = vi.spyOn(log, 'scanFacts') + try { + await (brain2 as any).recoverPendingEmbedsFromLog() + expect(scanSpy).toHaveBeenCalledTimes(1) + const opts = scanSpy.mock.calls[0][0] as { fromGeneration?: number } + expect(opts.fromGeneration).toBeGreaterThanOrEqual(mark!.generation + 1) + } finally { + scanSpy.mockRestore() + await brain2.close() + } + }) + + it('a pending embed enqueued after the mark survives an unclean stop', async () => { + const root = dir() + const brain = await open(root) + await brain.add({ id: 'settled', data: 'lands before the mark', type: NounType.Thing }) + await brain.awaitPendingEmbeds() + await new Promise((r) => setTimeout(r, 50)) + + // A deferred write whose embed never lands: block the worker, then drop + // the instance without close() — the unclean-stop shape. + ;(brain as any).kickEmbedWorker = () => {} + await brain.add({ + id: 'orphan', + data: 'enqueued then abandoned', + type: NounType.Thing, + deferEmbedding: true + }) + expect(brain.pendingEmbedCount()).toBeGreaterThan(0) + // No close(): simulate the crash by releasing only the writer lock so the + // next open can proceed. + await (brain as any).storage.releaseWriterLock() + + const brain2 = await open(root) + await (brain2 as any)._pendingEmbedRecovery + expect(brain2.pendingEmbedCount()).toBeGreaterThan(0) + await brain2.awaitPendingEmbeds() + expect(brain2.pendingEmbedCount()).toBe(0) + await brain2.close() + // Reap the crashed instance: its fence is gone, so close() fails loudly — + // swallow that here; the point is clearing its watchers and registry entry. + await brain.close().catch(() => undefined) + }) + + it('open arms the fold as a background latch; awaitPendingEmbeds waits on it', async () => { + const root = dir() + const brain = await open(root) + await brain.add({ id: 'a-row', data: 'some data', type: NounType.Thing }) + await brain.awaitPendingEmbeds() + await brain.close() + + const brain2 = await open(root) + // The latch exists the moment init() returns (writable filesystem brain)… + expect((brain2 as any)._pendingEmbedRecovery).not.toBeNull() + // …and the barrier settles it before answering. + await brain2.awaitPendingEmbeds() + expect(brain2.pendingEmbedCount()).toBe(0) + await brain2.close() + }) + + it('a clean close with an empty set writes the mark even if no drain happened', async () => { + const root = dir() + const brain = await open(root) + await brain.add({ id: 'r1', data: 'row one', type: NounType.Thing }) + await brain.awaitPendingEmbeds() + await brain.close() + // Read the mark back through the storage door (the adapter owns the + // on-disk encoding), on a fresh instance. + const brain2 = await open(root) + const mark = (await (brain2 as any).storage.readRawObject(LOWWATER_PATH)) as { + generation: number + } | null + expect(mark).not.toBeNull() + expect(mark!.generation).toBeGreaterThan(0) + await brain2.close() + }) +}) diff --git a/tests/integration/related-verb-array.test.ts b/tests/integration/related-verb-array.test.ts new file mode 100644 index 00000000..36a49850 --- /dev/null +++ b/tests/integration/related-verb-array.test.ts @@ -0,0 +1,89 @@ +/** + * @module tests/integration/related-verb-array + * @description related() honours EVERY verb type in an array (10.4.9). + * + * The storage fast paths for `sourceId + verbType` and `verbType` collapsed a + * verb-type ARRAY to its first element — `related({ from, type: [a, b] })` + * silently returned only `a` edges, whichever order the array came in. The + * same quiet-loss class as the graph-first paging defect, one seam over. + * These pins seed a store where the SECOND requested type's edge must come + * back, on every path the collapse lived in. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest' +import { Brainy } from '../../src/brainy' +import { NounType, VerbType } from '../../src/types/graphTypes' +import { v5 } from '../../src/universal/uuid' + +describe('related() with a verb-type array returns every requested type', () => { + let brain: Brainy + + beforeAll(async () => { + brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' } }) + await brain.init() + for (const id of ['a', 'b', 'c', 'd']) { + await brain.add({ id, data: `node ${id}`, type: NounType.Person }) + } + await brain.relate({ from: 'a', to: 'b', type: VerbType.Supports }) + await brain.relate({ from: 'a', to: 'c', type: VerbType.RelatedTo }) + await brain.relate({ from: 'a', to: 'd', type: VerbType.Knows }) + await brain.relate({ from: 'b', to: 'c', type: VerbType.RelatedTo }) + }) + + afterAll(async () => { + brain = null as any + }) + + it('from + type array: the second type\'s edge comes back, both orders', async () => { + for (const types of [ + [VerbType.Supports, VerbType.RelatedTo], + [VerbType.RelatedTo, VerbType.Supports] + ]) { + const edges = await brain.related({ from: 'a', type: types }) + const targets = new Set(edges.map((e) => e.to)) + expect(targets.has(v5('b')), `types [${types}] missing Supports edge`).toBe(true) + expect(targets.has(v5('c')), `types [${types}] missing RelatedTo edge`).toBe(true) + expect(targets.has(v5('d'))).toBe(false) + expect(edges).toHaveLength(2) + } + }) + + it('a single-element array behaves exactly like the scalar', async () => { + const scalar = await brain.related({ from: 'a', type: VerbType.Supports }) + const array = await brain.related({ from: 'a', type: [VerbType.Supports] }) + expect(array.map((e) => e.id).sort()).toEqual(scalar.map((e) => e.id).sort()) + expect(array).toHaveLength(1) + }) + + it('no duplicate edges when types overlap the same edge set', async () => { + const edges = await brain.related({ + from: 'a', + type: [VerbType.Supports, VerbType.RelatedTo, VerbType.Knows] + }) + const ids = edges.map((e) => e.id) + expect(new Set(ids).size).toBe(ids.length) + expect(edges).toHaveLength(3) + }) + + it('type-only asks (no anchor) honour the whole array too', async () => { + const edges = await brain.related({ type: [VerbType.Supports, VerbType.Knows] }) + const verbs = new Set(edges.map((e) => e.type)) + expect(verbs.has(VerbType.Supports)).toBe(true) + expect(verbs.has(VerbType.Knows)).toBe(true) + expect(edges).toHaveLength(2) + }) + + it('to + type array: the target side honours every type too', async () => { + const edges = await brain.related({ to: 'c', type: [VerbType.RelatedTo, VerbType.Supports] }) + const froms = new Set(edges.map((e) => e.from)) + expect(froms.has(v5('a'))).toBe(true) + expect(froms.has(v5('b'))).toBe(true) + expect(edges).toHaveLength(2) + }) + + it('pagination stays consistent across the union', async () => { + const page1 = await brain.related({ from: 'a', type: [VerbType.Supports, VerbType.RelatedTo, VerbType.Knows], limit: 2 }) + const page2 = await brain.related({ from: 'a', type: [VerbType.Supports, VerbType.RelatedTo, VerbType.Knows], limit: 2, offset: 2 }) + const all = [...page1, ...page2].map((e) => e.id) + expect(new Set(all).size).toBe(3) + }) +})