Compare commits

..

8 commits

Author SHA1 Message Date
eec90bdd69 chore(release): 10.4.9
Some checks failed
CI / Node 24 (push) Successful in 12m33s
CI / Node 22 (push) Successful in 12m35s
Publish (The Source) / Publish to The Source registry (push) Successful in 12m48s
CI / Bun (latest) (push) Successful in 12m27s
CI / Integration + conformance (Node 22) (push) Failing after 17m6s
2026-09-02 08:20:58 -07:00
2648f56ddf Merge branch 'fix/pending-embed-low-water' into rel/10.4.9-candidate
Some checks failed
CI / Node 22 (push) Successful in 12m24s
CI / Node 24 (push) Successful in 12m14s
CI / Bun (latest) (push) Successful in 12m23s
CI / Integration + conformance (Node 22) (push) Failing after 17m1s
2026-09-01 16:03:37 -07:00
8a2ebacf02 fix(open): pending-embed recovery keeps the crash-recovery contract — foreground, bounded by the mark
Some checks failed
CI / Node 24 (push) Successful in 12m24s
CI / Node 22 (push) Successful in 12m35s
CI / Integration + conformance (Node 22) (push) Failing after 17m18s
CI / Bun (latest) (push) Successful in 12m28s
The delta gate caught the backgrounded fold breaking six pinned
crash-recovery cases: a reopened brain must have its markers re-armed
when open() returns, and a background latch races every consumer of that
contract. The backgrounding is reverted; the low-water mark stays — it
is the part that kills the whole-history scan, and with it the
foreground fold costs the log's tail on any brain that has ever drained.
The unmarked first open after upgrade pays one full scan, once, and the
open narrates it as its own step.
2026-09-01 16:03:33 -07:00
d5147ed608 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
2026-09-01 12:38:00 -07:00
6a89adc468 fix(graph): the verb fast paths honour every requested type, source, and target
Some checks failed
CI / Node 22 (push) Successful in 12m21s
CI / Node 24 (push) Successful in 12m11s
CI / Bun (latest) (push) Successful in 12m24s
CI / Integration + conformance (Node 22) (push) Failing after 16m55s
related() with a verb-type ARRAY returned edges for only the first type —
the storage fast paths collapsed `verbType` (and, in their sibling blocks,
`sourceId` and `targetId`) arrays to their first element, silently
dropping the rest of the ask. Every consumer passing a verb list
under-traversed with no error and no narration: the same quiet-loss class
as the graph-first paging defect, one seam over.

All four fast paths now union over the full requested set, deduped by
edge id, before the metadata filters and pagination run. Pinned in
tests/integration/related-verb-array.test.ts: the second requested type's
edge returns in both array orders, on the anchor side, the target side,
and the type-only path; a one-element array equals the scalar; no
duplicates on overlap; pagination walks the union consistently.
2026-09-01 12:23:03 -07:00
88e79729d3 perf(open): pending-embed recovery is bounded by a low-water mark and runs behind the doors
Some checks failed
CI / Node 22 (push) Successful in 12m22s
CI / Node 24 (push) Successful in 12m33s
CI / Integration + conformance (Node 22) (push) Failing after 17m7s
CI / Bun (latest) (push) Successful in 12m24s
The recovery fold scanned the generation log from generation 1 at every
open, on the open's foreground — O(whole history) on long-lived brains
(measured at two minutes of a large brain's open). Now an advisory mark
records the log's head whenever the pending set drains to empty (and at
clean close when empty); recovery scans from the mark + 1. The mark is
advisory and monotone-safe: stale-low costs a longer scan, never a
marker. The fold itself moves behind the doors as a latched background
task — the embed worker starts when it settles, and awaitPendingEmbeds()
and close() wait on the latch first, so no caller can observe a
half-recovered set. A pending embed's outcome was always eventual;
moving its recovery off the foreground changes when the worker starts,
never whether a marker is honored.

Pinned in tests/integration/pending-embed-low-water.test.ts: the drain
writes the mark and the next open scans from mark + 1; a pending embed
enqueued after the mark survives an unclean stop; open arms the fold as
a background latch the barrier waits on; a clean close writes the mark
even without a drain.
2026-09-01 12:17:55 -07:00
077cbc0b6f fix(find): connected finds are graph-first — neighbours, then the filter over those ids, then the page
Some checks failed
CI / Node 22 (push) Successful in 12m18s
CI / Node 24 (push) Successful in 12m13s
CI / Integration + conformance (Node 22) (push) Failing after 16m56s
CI / Bun (latest) (push) Successful in 12m24s
With `connected` present, find() materialized the whole-store filtered id
list, paged it, hydrated the page, and only then intersected with the
neighbour set. Every such call paid O(store) for the filter and the
hydration of rows that were never neighbours, and a neighbour outside the
first page of the filtered STORE was silently dropped — the answer depended
on the store's order and the page size.

The neighbour set is now the candidate universe: resolved first from the
adjacency, the metadata filter evaluated over those ids only through the
provider's own evaluation (a new optional `filterIdsWithin` door on
MetadataIndexProvider; the reference index implements it from its own
getIdsForFilter so the two can never disagree; a provider without it is
served by the whole-store answer intersected here), `orderBy` sorts the
whole neighbour set before the page is cut, and the vector leg walks the
neighbours as its candidate set. The text leg of a hybrid find keeps its
post-intersection — it has no candidate door.

Pinned in tests/integration/find-connected-order.test.ts: paging reaches
every matching neighbour and never a non-neighbour; a `missing` negation is
evaluated over the neighbours; the index is asked about the neighbour ids
only and hydration is one page; orderBy sorts the whole set; the vector leg
stays inside the neighbours; an edgeless anchor answers [] before the
filter is asked.
2026-09-01 11:29:44 -07:00
5e3b343a0e fix(storage): counts persistence is single-flight, coalesced, and never races its own temp file
Some checks failed
CI / Node 24 (push) Successful in 12m24s
CI / Node 22 (push) Successful in 12m35s
CI / Integration + conformance (Node 22) (push) Failing after 17m25s
CI / Bun (latest) (push) Successful in 12m19s
persistCounts() was write-through on every count change with no
serialization, and the atomic writer named its temp file with millisecond
granularity. Two persists inside one millisecond shared the temp path: both
wrote it, the first rename consumed it, the second rename found nothing —
ENOENT, roughly 1,500 times a day on a busy production brain, with a full
ledger write per change behind it. No data was lost (the surviving rename
carried a complete ledger and the next change re-persisted), but the race
was real and the write rate absurd.

flushCounts() now runs exactly one persist at a time; requests arriving
during it collapse into one trailing pass that carries the burst's final
state — N changes cost at most two writes. writeFileAtomic() adds a
per-process sequence to the temp name so no two writes can share a path.
Pinned: a 25-change burst → ≤2 ledger writes, zero errors, ledger equal to
memory; parallel real writes land complete; three same-instant atomic
writes own three distinct temp paths.
2026-09-01 09:32:23 -07:00
13 changed files with 842 additions and 78 deletions

View file

@ -2,6 +2,17 @@
All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines.
### [10.4.9](https://source.soulcraft.com/soulcraftlabs/open-brainy/compare/v10.4.6...v10.4.9) (2026-09-02)
- Merge branch 'fix/pending-embed-low-water' into rel/10.4.9-candidate (2648f56d)
- fix(open): pending-embed recovery keeps the crash-recovery contract — foreground, bounded by the mark (8a2ebacf)
- Merge branches 'fix/connected-find-order', 'fix/pending-embed-low-water' and 'fix/related-verb-array' into rel/10.4.9-candidate (d5147ed6)
- fix(graph): the verb fast paths honour every requested type, source, and target (6a89adc4)
- perf(open): pending-embed recovery is bounded by a low-water mark and runs behind the doors (88e79729)
- fix(find): connected finds are graph-first — neighbours, then the filter over those ids, then the page (077cbc0b)
- fix(storage): counts persistence is single-flight, coalesced, and never races its own temp file (5e3b343a)
### [10.4.6](https://source.soulcraft.com/soulcraftlabs/open-brainy/compare/v10.4.5...v10.4.6) (2026-08-31)
- fix(transact): metadata-index ops take their JSON-safe view at the crossing, not at construction (73500e7d)

4
package-lock.json generated
View file

@ -1,12 +1,12 @@
{
"name": "@soulcraftlabs/brainy",
"version": "10.4.6",
"version": "10.4.9",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@soulcraftlabs/brainy",
"version": "10.4.6",
"version": "10.4.9",
"license": "MIT",
"dependencies": {
"@msgpack/msgpack": "^3.1.2",

View file

@ -1,6 +1,6 @@
{
"name": "@soulcraftlabs/brainy",
"version": "10.4.6",
"version": "10.4.9",
"brainyContract": 1,
"description": "Universal Knowledge Protocol™ - World's first Triple Intelligence database unifying vector, graph, and document search in one API. Stage 3 CANONICAL: 42 nouns × 127 verbs covering 96-97% of all human knowledge.",
"main": "dist/index.js",

View file

@ -1820,6 +1820,11 @@ 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) {
// Foreground, as the crash-recovery contract pins it: a reopened brain
// has its markers re-armed when open() returns. The low-water mark
// bounds this to the log's tail on any brain that has ever drained —
// milliseconds — so the foreground cost is the unmarked first open
// only, once per upgraded brain.
try {
await step(
'bridge-pending-embed-sidecars',
@ -1828,7 +1833,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
)
await step(
'recover-pending-embeds',
'folding the generation log\'s deferred-embed markers back into the pending set',
'folding the generation log\'s deferred-embed markers (from the low-water mark) into the pending set',
() => this.recoverPendingEmbedsFromLog()
)
if (this._pendingEmbedIds.size > 0) {
@ -2408,6 +2413,17 @@ 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'
/**
* @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 +2451,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 +2495,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 stays on the open's foreground
* the crash-recovery contract pins that a reopened brain has its markers
* re-armed when open() returns and the mark is what makes that cheap.
* 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 +2515,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 ?? []) {
@ -7470,7 +7536,37 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// JS path — there the materialized `candidateIds` restricts the walk instead.
let preResolvedAllowedIds: OpaqueIdSet | undefined
if (params.where || params.type || params.subtype || params.service || params.excludeVFS) {
// Graph-first law (10.4.8, BRAINY-PROD-LATENCY-TRIAD rounds 44/45): with
// `connected` present the NEIGHBOUR SET is the candidate universe. It is
// resolved first from the adjacency (O(neighbours)), the metadata filter
// is evaluated over those ids only, and paging happens LAST. The earlier
// order materialized the whole-store filtered id list, paged it, hydrated
// the page, and only then intersected with the neighbours — O(store) per
// call, and a neighbour outside the first page was silently dropped.
let graphFirstIds: string[] | null = null
if (hasGraphCriteria) {
graphFirstIds = await this.resolveConnectedIds(params)
if (hiddenIds.size > 0) {
graphFirstIds = graphFirstIds.filter((id) => !hiddenIds.has(id))
}
if (
graphFirstIds.length > 0 &&
(params.where || params.type || params.subtype || params.service || params.excludeVFS)
) {
preResolvedFilter = this.buildMetadataFilter(params)
graphFirstIds = await this.filterIdsWithinBelted(preResolvedFilter, graphFirstIds)
}
if (graphFirstIds.length === 0) {
return []
}
if (!hasVectorSearchCriteria) {
return await this.pageConnectedIds(params, graphFirstIds)
}
// The vector leg walks ONLY the neighbours (its candidate walk). The
// filter is already applied above, so no opaque universe is produced —
// it would describe the whole store, not the neighbour set.
preResolvedMetadataIds = graphFirstIds
} else if (params.where || params.type || params.subtype || params.service || params.excludeVFS) {
preResolvedFilter = this.buildMetadataFilter(params)
preResolvedMetadataIds = await this.filterIdsBelted(preResolvedFilter)
@ -7659,9 +7755,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
// Graph search component with O(1) traversal
if (params.connected) {
results = await this.executeGraphSearch(params, results)
// The text leg of a hybrid find has no candidate door, so its hits are
// held to the neighbour set here; the vector leg walked only the neighbours.
if (graphFirstIds !== null && results.length > 0) {
const neighbourSet = new Set(graphFirstIds)
results = results.filter((r) => neighbourSet.has(r.id))
}
// Apply fusion scoring if requested
@ -12776,6 +12874,29 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
/**
* The id-scoped twin of {@link filterIdsBelted}: evaluate `filter` over `ids`
* only, through the provider's own evaluation so the answer can never drift
* from `getIdsForFilter`'s. A provider without the door is served by its
* whole-store answer intersected here (the reference index implements the
* door itself). Same belt: field refusals cross as `BrainyFieldRefusal`.
*/
private async filterIdsWithinBelted(filter: unknown, ids: readonly string[]): Promise<string[]> {
this.ensureIndexesLoaded(['metadata'])
const mip = this.metadataIndex as unknown as MetadataIndexProvider
try {
if (typeof mip.filterIdsWithin === 'function') {
return await mip.filterIdsWithin(filter, ids)
}
const matched = new Set(await this.metadataIndex.getIdsForFilter(filter))
return ids.filter((id) => matched.has(id))
} catch (err) {
const normalized = asBrainyFieldRefusal(err)
if (normalized) throw normalized
throw err
}
}
async getIndexStatus(): Promise<{
initialized: boolean
/** `true` once open()'s index-build-if-needed step has run. Named for API
@ -15759,16 +15880,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
/**
* Execute graph search component.
* Resolve `params.connected` to the neighbour id set the graph-first
* find's candidate universe (deterministic traversal order, anchors excluded).
*
* Honors the full `GraphConstraints` contract: multi-hop `depth` (breadth-first via
* `neighbors()`), `via`/`type` verb-type filtering, and `direction`. Previously this read
* only `from`/`to`/`direction` and did a single 1-hop `getNeighbors()`, so `depth` and `via`
* were silently ignored `find({ connected: { from, depth: 3 } })` returned only the
* immediate neighbour at every depth.
* `neighbors()`), `via`/`type` verb-type filtering, and `direction`. An empty set
* is re-verified against the adjacency before it is believed a not-serving
* adjacency throws rather than answering `[]` as truth.
*/
private async executeGraphSearch(params: FindParams<T>, existingResults: Result<T>[]): Promise<Result<T>[]> {
if (!params.connected) return existingResults
private async resolveConnectedIds(params: FindParams<T>): Promise<string[]> {
if (!params.connected) return []
const { from, to, depth, direction = 'both' } = params.connected
const via = params.connected.via ?? params.connected.type
@ -15822,8 +15943,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
if (anchorInt === undefined) return new Set() // unmapped → no relations
const verbTypeIndex = TypeUtils.getVerbIndex(via as VerbType)
// No limit: match the JS BFS exactly — overall result limiting happens
// downstream against existingResults.
// No limit: match the JS BFS exactly — the page is cut downstream,
// after the metadata filter, by pageConnectedIds / the candidate walk.
const reachedInts = await provider.findConnectedSubtype(
anchorInt, verbTypeIndex, subtypeArr[0], effectiveDepth, null
)
@ -15908,22 +16029,44 @@ export class Brainy<T = any> implements BrainyInterface<T> {
await this.verifyGraphAdjacencyLive()
}
// Filter existing results to only connected entities
if (existingResults.length > 0) {
return existingResults.filter(r => connectedIds.has(r.id))
}
return [...connectedIds]
}
// Batch-load connected entities for fast cloud-storage performance
/**
* Page and hydrate an already-filtered neighbour set the pure graph (and
* graph + metadata) find's tail. `orderBy` sorts the WHOLE set by field value
* before the page is cut (never the page after), null values last on `asc`
* and first on `desc`; without `orderBy` the traversal order stands.
*/
private async pageConnectedIds(params: FindParams<T>, ids: string[]): Promise<Result<T>[]> {
const limit = params.limit || 10
const offset = params.offset || 0
let ordered = ids
if (params.orderBy) {
const field = params.orderBy
const asc = (params.order || 'asc') === 'asc'
const valued = await Promise.all(
ids.map(async (id) => ({ id, value: await this.metadataIndex.getFieldValueForEntity(id, field) }))
)
valued.sort((a, b) => {
if (a.value == null && b.value == null) return 0
if (a.value == null) return asc ? 1 : -1
if (b.value == null) return asc ? -1 : 1
if (a.value === b.value) return 0
const comparison = a.value < b.value ? -1 : 1
return asc ? comparison : -comparison
})
ordered = valued.map((v) => v.id)
}
const pageIds = ordered.slice(offset, offset + limit)
const entitiesMap = await this.batchGet(pageIds)
const results: Result<T>[] = []
const ids = [...connectedIds]
const entitiesMap = await this.batchGet(ids)
for (const id of ids) {
for (const id of pageIds) {
const entity = entitiesMap.get(id)
if (entity) {
results.push(this.createResult(id, 1.0, entity))
}
}
return results
}
@ -19443,6 +19586,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* terminal releases have run.
*/
async close(): Promise<void> {
if (this._pendingEmbedIds.size === 0) await this.writeEmbedLowWater()
let closeFailure: unknown = null
try {
await this.closeDurableSteps()

View file

@ -411,6 +411,19 @@ export interface MetadataIndexProvider {
* @returns The matching id universe as an opaque set.
*/
getIdSetForFilter?(filter: any): Promise<OpaqueIdSet>
/**
* @description OPTIONAL: evaluate `filter` over `ids` ONLY and return the
* survivors in the caller's order the door a graph-first
* `find({ connected, where })` walks. The neighbour set is the universe there,
* so the filter must cost O(|ids|) membership checks, never a whole-store
* materialization. A native index answers from its roaring filter result
* (membership by entity int); the reference index answers from its own
* `getIdsForFilter`, so the two doors can never disagree. Absent Brainy
* intersects `getIdsForFilter`'s answer with `ids` itself (correct, O(store)).
* @param filter - The same filter shape accepted by `getIdsForFilter`.
* @param ids - The candidate ids (canonical). The answer is a subsequence.
*/
filterIdsWithin?(filter: any, ids: readonly string[]): Promise<string[]>
getIdsForTextQuery(query: string): Promise<Array<{ id: string; matchCount: number }>>
getSortedIdsForFilter(filter: any, orderBy: string, order?: 'asc' | 'desc', topK?: number): Promise<string[]>
getFilterValues(field: string): Promise<string[]>

View file

@ -1089,6 +1089,10 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
// Counts changed since the last persist? Drives the write-through flush.
protected pendingCountPersist = false
/** The one persist running right now, if any (single-flight law — see flushCounts). */
private countPersistInFlight: Promise<void> | null = null
/** The one trailing persist a burst has queued behind the in-flight one. */
private countPersistTrailing: Promise<void> | null = null
/**
* Get total noun count - O(1) operation
@ -1341,15 +1345,46 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
return
}
try {
// Persist to storage (implemented by subclass)
await this.persistCounts()
this.pendingCountPersist = false
} catch (error) {
console.error('CRITICAL: Failed to flush counts to storage:', error)
// Keep pending flag set so we retry on next operation
throw error
// SINGLE-FLIGHT, COALESCED. Counts are write-through on every change, so
// a burst of writes used to launch one persist per change, all in flight
// together. Two of them inside the same millisecond shared the atomic
// writer's temp path (`.tmp-<pid>-<ms>`): both wrote it, the first rename
// consumed it, the second rename found nothing — ENOENT, ~1,500 times a
// day on a busy production brain, with a full ledger write per change
// behind it. Now exactly one persist runs at a time; requests that arrive
// while it runs collapse into ONE trailing persist that carries the final
// state. A burst of N changes costs at most two writes and never races
// itself.
if (this.countPersistInFlight) {
// The in-flight write may have already serialised a stale snapshot —
// ask for one more pass after it, and let every caller in this burst
// await that same pass.
if (!this.countPersistTrailing) {
this.countPersistTrailing = this.countPersistInFlight
.catch(() => undefined)
.then(() => {
this.countPersistTrailing = null
return this.flushCounts()
})
}
return this.countPersistTrailing
}
this.countPersistInFlight = (async () => {
try {
// Persist to storage (implemented by subclass)
this.pendingCountPersist = false
await this.persistCounts()
} catch (error) {
// Keep the flag set so the next operation retries.
this.pendingCountPersist = true
console.error('CRITICAL: Failed to flush counts to storage:', error)
throw error
} finally {
this.countPersistInFlight = null
}
})()
return this.countPersistInFlight
}
/**

View file

@ -2400,8 +2400,15 @@ export class FileSystemStorage extends BaseStorage {
* Atomic write via temp-file-then-rename so concurrent readers never see a
* half-written lock JSON. Reused by writer-lock writes + heartbeat.
*/
/** Monotonic per-process sequence so two atomic writes never share a temp path. */
private static atomicWriteSeq = 0
private async writeFileAtomic(filePath: string, contents: string): Promise<void> {
const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}`
// pid + timestamp alone collided: two writers of the same target inside
// one millisecond shared this path, and the loser's rename found the
// winner had already moved it (ENOENT). The sequence makes every call's
// temp path its own.
const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}-${++FileSystemStorage.atomicWriteSeq}`
await fs.promises.writeFile(tmp, contents)
await fs.promises.rename(tmp, filePath)
}

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

@ -2575,6 +2575,19 @@ export class MetadataIndexManager implements MetadataIndexProvider {
/** Once-per-field flag for the fallback-degradation announcement. */
private static announcedFallbackSorts = new Set<string>()
/**
* Evaluate `filter` over `ids` only the graph-first find's door (the
* neighbour set filtered by id, never the store filtered and then
* intersected). This index answers from its own `getIdsForFilter`, so the
* two doors cannot disagree; the cost is that of the filter over this
* in-memory index, and the answer keeps the caller's order.
*/
async filterIdsWithin(filter: any, ids: readonly string[]): Promise<string[]> {
if (ids.length === 0) return []
const matched = new Set(await this.getIdsForFilter(filter))
return ids.filter((id) => matched.has(id))
}
async getSortedIdsForFilter(
filter: any,
orderBy: string,

View file

@ -0,0 +1,111 @@
/**
* @module tests/integration/counts-persist-single-flight
* @description Regression for a production race in FileSystemStorage's
* counts ledger: `persistCounts()` was write-through on every count change
* with no serialization, and the atomic writer named its temp file with
* millisecond granularity (`.tmp-<pid>-<ms>`). Two persists inside one
* millisecond shared the temp path both wrote it, the first rename
* consumed it, the second rename found nothing: ENOENT, ~1,500 times a day
* on a busy production brain, with a full ledger write per change behind it.
*
* Under pin: persists are single-flight and coalesced one in flight, at
* most one trailing pass carrying the burst's final state and every atomic
* write owns a unique temp path. A burst of N count changes costs at most
* two ledger writes, never errors, and leaves a ledger equal to memory.
*/
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
import * as fs from 'node:fs'
import * as os from 'node:os'
import * as path from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
describe('counts persistence is single-flight, coalesced, and never races its own temp file', () => {
let dir: string
let brain: any
beforeEach(async () => {
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-counts-race-'))
brain = new Brainy({
requireSubtype: false,
storage: { type: 'filesystem', path: dir },
dimensions: 384,
silent: true
})
await brain.init()
})
afterEach(async () => {
vi.restoreAllMocks()
await brain.close()
fs.rmSync(dir, { recursive: true, force: true })
})
it('a burst of concurrent count changes → at most two ledger writes, zero errors, ledger == memory', async () => {
const storage = brain.storage
const countsPath: string = storage.countsFilePath
expect(countsPath, 'the filesystem adapter persists a counts ledger').toBeTruthy()
// Let init's own persists settle so the burst is measured alone.
await storage.flushCounts?.()
const renameSpy = vi.spyOn(fs.promises, 'rename')
const errorSpy = vi.spyOn(console, 'error')
// Twenty-five concurrent count changes — the shape of a write burst; each
// used to launch its own persist.
const BURST = 25
await Promise.all(
Array.from({ length: BURST }, () => storage.scheduleCountPersist())
)
const ledgerRenames = renameSpy.mock.calls.filter(([, to]) => String(to) === countsPath)
expect(ledgerRenames.length, 'single-flight + one trailing pass').toBeLessThanOrEqual(2)
expect(ledgerRenames.length, 'the burst was persisted at all').toBeGreaterThanOrEqual(1)
const persistErrors = errorSpy.mock.calls.filter((args) => String(args[0]).includes('persisting counts'))
expect(persistErrors).toEqual([])
const ledger = JSON.parse(fs.readFileSync(countsPath, 'utf-8'))
expect(ledger.totalNounCount).toBe(storage.totalNounCount)
expect(ledger.totalVerbCount).toBe(storage.totalVerbCount)
})
it('real writes in parallel: the ledger lands complete and no persist error is logged', async () => {
const storage = brain.storage
const countsPath: string = storage.countsFilePath
const errorSpy = vi.spyOn(console, 'error')
await Promise.all(
Array.from({ length: 12 }, (_, i) =>
brain.add({ data: `burst row ${i}`, type: NounType.Thing })
)
)
await storage.flushCounts?.()
const persistErrors = errorSpy.mock.calls.filter((args) => String(args[0]).includes('persisting counts'))
expect(persistErrors).toEqual([])
const ledger = JSON.parse(fs.readFileSync(countsPath, 'utf-8'))
expect(ledger.totalNounCount).toBe(storage.totalNounCount)
expect(await brain.getNounCount()).toBe(ledger.totalNounCount)
})
it('every atomic write owns its own temp path — two writes in one millisecond never collide', async () => {
const storage = brain.storage
const tmpNames: string[] = []
vi.spyOn(fs.promises, 'writeFile').mockImplementation(async (p: any) => {
tmpNames.push(String(p))
})
vi.spyOn(fs.promises, 'rename').mockImplementation(async () => undefined)
const target = path.join(dir, 'probe.json')
await Promise.all([
storage.writeFileAtomic(target, '{"a":1}'),
storage.writeFileAtomic(target, '{"a":2}'),
storage.writeFileAtomic(target, '{"a":3}')
])
const probeTmps = tmpNames.filter((n) => n.startsWith(`${target}.tmp-`))
expect(probeTmps.length).toBe(3)
expect(new Set(probeTmps).size, 'no two writes shared a temp path').toBe(3)
})
})

View file

@ -0,0 +1,165 @@
/**
* @module tests/integration/find-connected-order
* @description The graph-first law for `find({ connected })` (10.4.8).
*
* With `connected` present the neighbour set is the candidate universe: it is
* resolved from the adjacency first, the metadata filter is evaluated over
* those ids only, and the page is cut last. The earlier order materialized the
* whole-store filtered id list, paged it, hydrated the page, and only then
* intersected with the neighbours so a neighbour outside the first page of
* the filtered STORE was silently dropped, and every call paid O(store).
*
* These pins hold both halves. The answer: every matching neighbour is
* reachable by paging, a non-neighbour never appears, a negation (`missing`)
* is evaluated over the neighbours, `orderBy` sorts the whole neighbour set
* before the page is cut, and the vector leg walks the neighbours only. The
* cost shape: the metadata index is asked about the neighbour ids only, and
* hydration is one page never the store.
*/
import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest'
import { Brainy } from '../../src/brainy'
import { NounType, VerbType } from '../../src/types/graphTypes'
import { v5 } from '../../src/universal/uuid'
import { generateTestVector } from '../helpers/test-factory'
/** Matching rows that are NOT neighbours — added FIRST, so the whole-store filtered list leads with them. */
const NOISE = 120
/** Matching rows that ARE neighbours of the anchor. */
const NEIGHBOURS = 30
/** Neighbours carrying `retracted: true` — excluded by the `missing` negation. */
const RETRACTED = 4
describe('find({ connected }) is graph-first: neighbours → filter → page', () => {
let brain: Brainy<any>
const anchor = 'anchor'
const sharedVector = generateTestVector()
const neighbourIds = new Set(Array.from({ length: NEIGHBOURS }, (_, i) => v5(`nb-${i}`)))
beforeAll(async () => {
brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' } })
await brain.init()
await brain.add({
id: anchor,
data: 'the anchor',
type: NounType.Person,
metadata: { kind: 'anchor' },
vector: generateTestVector()
})
for (let i = 0; i < NOISE; i++) {
await brain.add({
id: `noise-${i}`,
data: `noise ${i}`,
type: NounType.Person,
metadata: { kind: 'note', rank: 1000 + i },
vector: sharedVector
})
}
for (let i = 0; i < NEIGHBOURS; i++) {
await brain.add({
id: `nb-${i}`,
data: `neighbour ${i}`,
type: NounType.Person,
metadata: { kind: 'note', rank: i + 1, ...(i < RETRACTED ? { retracted: true } : {}) },
vector: sharedVector
})
await brain.relate({ from: anchor, to: `nb-${i}`, type: VerbType.Knows })
}
})
afterAll(async () => {
brain = null as any
})
it('returns the matching neighbours page by page — none dropped, never a non-neighbour', async () => {
const seen = new Set<string>()
for (let offset = 0; offset <= NEIGHBOURS; offset += 10) {
const page = await brain.find({
connected: { from: anchor, direction: 'out' },
where: { kind: 'note' },
limit: 10,
offset
})
expect(page).toHaveLength(offset < NEIGHBOURS ? 10 : 0)
for (const r of page) {
expect(neighbourIds.has(r.entity.id)).toBe(true)
expect(seen.has(r.entity.id)).toBe(false)
seen.add(r.entity.id)
}
}
expect(seen.size).toBe(NEIGHBOURS)
})
it('evaluates a negation (`missing`) over the neighbour set, not the store', async () => {
const results = await brain.find({
connected: { from: anchor, direction: 'out' },
where: { kind: 'note', retracted: { missing: true } },
limit: 100
})
expect(results).toHaveLength(NEIGHBOURS - RETRACTED)
for (const r of results) {
expect(neighbourIds.has(r.entity.id)).toBe(true)
expect(r.entity.metadata.retracted).toBeUndefined()
}
})
it('asks the metadata index about the neighbour ids only, and hydrates one page', async () => {
const index = (brain as any).metadataIndex
const within = vi.spyOn(index, 'filterIdsWithin')
const hydrate = vi.spyOn(brain as any, 'batchGet')
try {
const results = await brain.find({
connected: { from: anchor, direction: 'out' },
where: { kind: 'note' },
limit: 10
})
expect(results).toHaveLength(10)
expect(within).toHaveBeenCalledTimes(1)
const askedIds = within.mock.calls[0][1] as string[]
expect(askedIds).toHaveLength(NEIGHBOURS)
for (const id of askedIds) expect(neighbourIds.has(id)).toBe(true)
expect(hydrate).toHaveBeenCalledTimes(1)
expect(hydrate.mock.calls[0][0]).toHaveLength(10)
} finally {
within.mockRestore()
hydrate.mockRestore()
}
})
it('orders the WHOLE neighbour set before cutting the page', async () => {
const results = await brain.find({
connected: { from: anchor, direction: 'out' },
where: { kind: 'note' },
orderBy: 'rank',
order: 'desc',
limit: 5
})
expect(results.map((r) => r.entity.metadata.rank)).toEqual([30, 29, 28, 27, 26])
})
it('walks the vector leg over the neighbours only', async () => {
const results = await brain.find({
vector: sharedVector,
connected: { from: anchor, direction: 'out' },
where: { kind: 'note' },
limit: 5
})
expect(results).toHaveLength(5)
for (const r of results) expect(neighbourIds.has(r.entity.id)).toBe(true)
})
it('an anchor without neighbours answers [] before the filter is asked', async () => {
const index = (brain as any).metadataIndex
const within = vi.spyOn(index, 'filterIdsWithin')
try {
const results = await brain.find({
connected: { from: 'noise-0', direction: 'out' },
where: { kind: 'note' },
limit: 10
})
expect(results).toEqual([])
expect(within).not.toHaveBeenCalled()
} finally {
within.mockRestore()
}
})
})

View file

@ -0,0 +1,141 @@
/**
* @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` on the open's foreground the crash-recovery
* contract keeps markers re-armed when open() returns. 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', () => {
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)
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('a reopened brain has its pending set settled when open() returns', 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 crash-recovery contract: markers are re-armed by open itself —
// no latch, no background race. (Here the drain landed, so zero.)
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)
})
})