Merge branches 'fix/connected-find-order', 'fix/pending-embed-low-water' and 'fix/related-verb-array' into rel/10.4.9-candidate
Some checks failed
CI / Node 22 (push) Failing after 7m58s
CI / Node 24 (push) Failing after 7m38s
CI / Bun (latest) (push) Successful in 12m29s
CI / Integration + conformance (Node 22) (push) Failing after 17m6s

This commit is contained in:
David Snelling 2026-09-01 12:38:00 -07:00
commit d5147ed608
4 changed files with 412 additions and 66 deletions

View file

@ -1820,31 +1820,31 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// 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<T = any> implements BrainyInterface<T> {
*/
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<void> | 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<T = any> implements BrainyInterface<T> {
*/
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<void> {
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<T = any> implements BrainyInterface<T> {
* 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<T = any> implements BrainyInterface<T> {
private async recoverPendingEmbedsFromLog(): Promise<void> {
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<T = any> implements BrainyInterface<T> {
* before I proceed" callers use this; nothing else ever needs to wait.
*/
public async awaitPendingEmbeds(): Promise<void> {
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<T = any> implements BrainyInterface<T> {
* terminal releases have run.
*/
async close(): Promise<void> {
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()

View file

@ -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<string>()
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<string>()
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<string>()
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<string>()
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)

View file

@ -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<Brainy<any>> => {
const brain = new Brainy<any>({
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()
})
})

View file

@ -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<any>
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)
})
})