perf: parallel + id-only canonical enumeration (heal-cost dominant term)
The canonical enumeration walk (getNounsWithPagination) hydrated each entity's vector + metadata ONE-AT-A-TIME inside the shard loop — every enumeration paid N x per-op-latency serially, and every index heal enumerates canonical, so this was the dominant multiplier in the measured multi-minute heals (cortex heal-cost decomposition). - Hydration is now 16-way bounded-concurrency (matching the wave cor's native rebuild uses): heal wall-clock becomes ~2xN/16 x per-op instead of N x per-op. Order, cursor resume (skipped nouns still never read), filters, peek/hasMore and totalCount are all preserved — pages are byte-identical to the serial walk. - New getNounIdsWithPagination(): the id-only opt-out for callers that own their IO schedule (an index heal). Unfiltered = ids straight from the shard paths, ZERO per-entity reads; filtered hydrates metadata only (16-way). Same cursor/offset/nextCursor contract, so it is page-compatible with the hydrating walk. Regression: pagination-parallel-hydration.test.ts pins paged==big-page identity, ids==items order, zero-read id-only, and filter parity. 103 storage tests green.
This commit is contained in:
parent
bfa1762107
commit
ec5b93339a
2 changed files with 250 additions and 18 deletions
|
|
@ -262,6 +262,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
private _derivedFamiliesLoaded = false
|
||||
/** Storage-root-relative path of the persisted family registry. */
|
||||
private static readonly DERIVED_FAMILIES_KEY = '_system/derived-artifacts.json'
|
||||
/**
|
||||
* Bounded concurrency for hydrating enumerated nouns during a pagination walk
|
||||
* (readCanonicalObject + getNounMetadata per item). A canonical enumeration —
|
||||
* which every index heal performs — otherwise pays N×per-op-latency serially;
|
||||
* 16-way matches the wave the native rebuild uses so heal wall-clock is
|
||||
* ~2×N/16×per-op instead of N×per-op. Bounded so a huge dataset can't spawn a
|
||||
* read per entity at once.
|
||||
*/
|
||||
private static readonly HYDRATE_CONCURRENCY = 16
|
||||
protected graphIndex?: GraphAdjacencyIndex
|
||||
protected graphIndexPromise?: Promise<GraphAdjacencyIndex>
|
||||
/**
|
||||
|
|
@ -1995,39 +2004,63 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
.map((p) => ({ path: p, id: idFromVectorPath(p) }))
|
||||
.sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
|
||||
|
||||
for (const { path: nounPath, id: nounId } of entries) {
|
||||
if (collected.length >= peekCount) break
|
||||
// Resume: in the cursor's own shard, skip up to AND INCLUDING the cursor id.
|
||||
if (cursor && shard === cursor.shard && nounId <= cursor.id) continue
|
||||
// Resume: in the cursor's own shard, skip up to AND INCLUDING the cursor
|
||||
// id BEFORE hydrating — a cheap id compare (from the path), so skipped
|
||||
// nouns are never read.
|
||||
const toHydrate =
|
||||
cursor && shard === cursor.shard
|
||||
? entries.filter((e) => e.id > cursor.id)
|
||||
: entries
|
||||
|
||||
try {
|
||||
const noun = await this.readCanonicalObject(nounPath)
|
||||
if (!noun) continue
|
||||
const deserialized = this.deserializeNoun(noun)
|
||||
const metadata = await this.getNounMetadata(deserialized.id)
|
||||
if (!metadata) continue
|
||||
// Hydrate in bounded-concurrency batches (16-way) instead of one-at-a-time.
|
||||
// Every canonical enumeration — and every index heal enumerates canonical —
|
||||
// otherwise pays N×per-op-latency SERIALLY (the dominant heal-time term).
|
||||
// Order is preserved (the batch is a slice of the sorted entries and its
|
||||
// results are consumed in order), so offset windows / cursor resume stay
|
||||
// deterministic. peekCount stops the walk; the final batch over-hydrates by
|
||||
// at most BASE_STORAGE_HYDRATE_CONCURRENCY entries (bounded, acceptable).
|
||||
for (
|
||||
let i = 0;
|
||||
i < toHydrate.length && collected.length < peekCount;
|
||||
i += BaseStorage.HYDRATE_CONCURRENCY
|
||||
) {
|
||||
const batch = toHydrate.slice(i, i + BaseStorage.HYDRATE_CONCURRENCY)
|
||||
const hydrated = await Promise.all(
|
||||
batch.map(async ({ path: nounPath }) => {
|
||||
try {
|
||||
const noun = await this.readCanonicalObject(nounPath)
|
||||
if (!noun) return null
|
||||
const deserialized = this.deserializeNoun(noun)
|
||||
const metadata = await this.getNounMetadata(deserialized.id)
|
||||
if (!metadata) return null
|
||||
return { deserialized, metadata }
|
||||
} catch (error) {
|
||||
// Skip nouns that fail to load
|
||||
return null
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
for (const h of hydrated) {
|
||||
if (collected.length >= peekCount) break
|
||||
if (!h) continue
|
||||
const { deserialized, metadata } = h
|
||||
|
||||
// Apply type filter
|
||||
if (filter?.nounType && metadata.noun) {
|
||||
const types = Array.isArray(filter.nounType) ? filter.nounType : [filter.nounType]
|
||||
if (!types.includes(metadata.noun)) {
|
||||
continue
|
||||
}
|
||||
if (!types.includes(metadata.noun)) continue
|
||||
}
|
||||
|
||||
// Apply service filter
|
||||
if (filter?.service) {
|
||||
const services = Array.isArray(filter.service) ? filter.service : [filter.service]
|
||||
if (metadata.service && !services.includes(metadata.service)) {
|
||||
continue
|
||||
}
|
||||
if (metadata.service && !services.includes(metadata.service)) continue
|
||||
}
|
||||
|
||||
// Combine noun + metadata via the canonical hydration helper —
|
||||
// reserved fields top-level, ONLY custom fields in `metadata`.
|
||||
collected.push({ noun: this.hydrateNounWithMetadata(deserialized, metadata), shard })
|
||||
} catch (error) {
|
||||
// Skip nouns that fail to load
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
|
|
@ -2067,6 +2100,111 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Enumerate noun IDS ONLY, without hydrating each entity's vector
|
||||
* and metadata — the opt-out for callers (e.g. an index heal) that own their
|
||||
* own IO schedule and index straight from a stream. Shares the exact
|
||||
* shard-walk, cursor, offset and `nextCursor` contract of
|
||||
* {@link getNounsWithPagination}, so the two are page-compatible.
|
||||
*
|
||||
* - UNFILTERED (the heal case): ids come straight from the shard paths — ZERO
|
||||
* per-entity reads. A full enumeration is O(entries listed), not O(N reads).
|
||||
* - FILTERED: the type/service filter needs metadata, so only the metadata is
|
||||
* hydrated (16-way bounded concurrency), never the full noun.
|
||||
*
|
||||
* @param options - `limit`/`offset`/`cursor`/`filter` — same semantics as
|
||||
* {@link getNounsWithPagination}.
|
||||
* @returns The page of ids plus `totalCount` / `hasMore` / `nextCursor`.
|
||||
*/
|
||||
public async getNounIdsWithPagination(options: {
|
||||
limit: number
|
||||
offset?: number
|
||||
cursor?: string
|
||||
filter?: {
|
||||
nounType?: string | string[]
|
||||
service?: string | string[]
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
}): Promise<{ ids: string[]; totalCount: number; hasMore: boolean; nextCursor?: string }> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const { limit, offset = 0, filter } = options
|
||||
const cursor = this.decodeNounWalkCursor(options.cursor)
|
||||
const collected: Array<{ id: string; shard: number }> = []
|
||||
const peekCount = cursor ? limit + 1 : offset + limit + 1
|
||||
const startShard = cursor ? cursor.shard : 0
|
||||
|
||||
for (let shard = startShard; shard < 256 && collected.length < peekCount; shard++) {
|
||||
const shardHex = shard.toString(16).padStart(2, '0')
|
||||
const shardDir = `entities/nouns/${shardHex}`
|
||||
try {
|
||||
const nounFiles = await this.listCanonicalObjects(shardDir)
|
||||
const entries = nounFiles
|
||||
.filter((p) => p.includes('/vectors.json'))
|
||||
.map((p) => idFromVectorPath(p))
|
||||
.sort((a, b) => (a < b ? -1 : a > b ? 1 : 0))
|
||||
const toWalk =
|
||||
cursor && shard === cursor.shard ? entries.filter((id) => id > cursor.id) : entries
|
||||
|
||||
if (!filter) {
|
||||
// Unfiltered — ids straight from the paths, no reads at all.
|
||||
for (const id of toWalk) {
|
||||
if (collected.length >= peekCount) break
|
||||
collected.push({ id, shard })
|
||||
}
|
||||
} else {
|
||||
// Filtered — hydrate metadata ONLY (16-way) to apply the filter.
|
||||
for (
|
||||
let i = 0;
|
||||
i < toWalk.length && collected.length < peekCount;
|
||||
i += BaseStorage.HYDRATE_CONCURRENCY
|
||||
) {
|
||||
const batch = toWalk.slice(i, i + BaseStorage.HYDRATE_CONCURRENCY)
|
||||
const metas = await Promise.all(
|
||||
batch.map(async (id) => {
|
||||
try {
|
||||
return { id, metadata: await this.getNounMetadata(id) }
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
})
|
||||
)
|
||||
for (const m of metas) {
|
||||
if (collected.length >= peekCount) break
|
||||
if (!m || !m.metadata) continue
|
||||
const metadata = m.metadata
|
||||
if (filter.nounType && metadata.noun) {
|
||||
const types = Array.isArray(filter.nounType) ? filter.nounType : [filter.nounType]
|
||||
if (!types.includes(metadata.noun)) continue
|
||||
}
|
||||
if (filter.service) {
|
||||
const services = Array.isArray(filter.service) ? filter.service : [filter.service]
|
||||
if (metadata.service && !services.includes(metadata.service)) continue
|
||||
}
|
||||
collected.push({ id: m.id, shard })
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// Skip shards with no data
|
||||
}
|
||||
}
|
||||
|
||||
const windowStart = cursor ? 0 : offset
|
||||
const pagePairs = collected.slice(windowStart, windowStart + limit)
|
||||
const ids = pagePairs.map((p) => p.id)
|
||||
const hasMore = collected.length > windowStart + limit
|
||||
const totalCount = filter ? collected.length : Math.max(this.totalNounCount, collected.length)
|
||||
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && pagePairs.length > 0) {
|
||||
const lastPair = pagePairs[pagePairs.length - 1]
|
||||
nextCursor = this.encodeNounWalkCursor(lastPair.shard, lastPair.id)
|
||||
}
|
||||
|
||||
return { ids, totalCount, hasMore, nextCursor }
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Encode a noun-walk resume cursor — the `(shard, nounId)` of the
|
||||
* last returned noun — as an opaque, version-tagged token (`cn1:` prefix lets
|
||||
|
|
|
|||
94
tests/unit/storage/pagination-parallel-hydration.test.ts
Normal file
94
tests/unit/storage/pagination-parallel-hydration.test.ts
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
/**
|
||||
* @module tests/unit/storage/pagination-parallel-hydration
|
||||
* @description The canonical enumeration walk (getNounsWithPagination) hydrated
|
||||
* each item's vector + metadata ONE-AT-A-TIME — N×per-op-latency serially, the
|
||||
* dominant term in an index heal (cortex heal-cost decomposition). It now
|
||||
* hydrates 16-way, and a new getNounIdsWithPagination returns ids WITHOUT
|
||||
* hydration (zero per-entity reads when unfiltered). Both must preserve the exact
|
||||
* pagination contract: same order, cursor continuation, filters, totalCount.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, vi } from 'vitest'
|
||||
import { Brainy, NounType } from '../../../src/index.js'
|
||||
|
||||
describe('paginated enumeration — parallel hydration + id-only (cortex heal-cost)', () => {
|
||||
let brain: any
|
||||
let storage: any
|
||||
const N = 30
|
||||
|
||||
beforeEach(async () => {
|
||||
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||
brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' }, dimensions: 384 })
|
||||
await brain.init()
|
||||
for (let i = 0; i < N; i++) {
|
||||
await brain.add({
|
||||
data: `n${i}`,
|
||||
type: i % 3 === 0 ? NounType.Task : NounType.Concept,
|
||||
metadata: { i }
|
||||
})
|
||||
}
|
||||
await brain.flush()
|
||||
storage = brain.storage
|
||||
})
|
||||
|
||||
/** Page the whole dataset through a small limit via cursor and collect ordered ids. */
|
||||
const pageAll = async (fn: (opts: any) => Promise<any>, key: 'items' | 'ids') => {
|
||||
const out: string[] = []
|
||||
let cursor: string | undefined
|
||||
for (let guard = 0; guard < 1000; guard++) {
|
||||
const page = await fn({ limit: 4, cursor })
|
||||
const batch = key === 'items' ? page.items.map((n: any) => n.id) : page.ids
|
||||
out.push(...batch)
|
||||
if (!page.hasMore) break
|
||||
cursor = page.nextCursor
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
it('parallel hydration yields the SAME ordered pages as one big page', async () => {
|
||||
const big = await storage.getNounsWithPagination({ limit: 1000, offset: 0 })
|
||||
const bigIds = big.items.map((n: any) => n.id)
|
||||
// At least the N we added (a brain also has its VFS root entity).
|
||||
expect(bigIds.length).toBeGreaterThanOrEqual(N)
|
||||
expect(big.totalCount).toBe(bigIds.length)
|
||||
|
||||
const paged = await pageAll((o) => storage.getNounsWithPagination(o), 'items')
|
||||
expect(paged).toEqual(bigIds) // identical order, no dupes, no gaps across pages
|
||||
})
|
||||
|
||||
it('getNounIdsWithPagination returns exactly the same ids, in the same order', async () => {
|
||||
const idsPaged = await pageAll((o) => storage.getNounIdsWithPagination(o), 'ids')
|
||||
const itemsPaged = await pageAll((o) => storage.getNounsWithPagination(o), 'items')
|
||||
expect(idsPaged).toEqual(itemsPaged)
|
||||
expect(new Set(idsPaged).size).toBe(idsPaged.length) // every id exactly once
|
||||
expect(idsPaged.length).toBeGreaterThanOrEqual(N)
|
||||
})
|
||||
|
||||
it('id-only enumeration does ZERO per-entity hydration reads when unfiltered', async () => {
|
||||
const readSpy = vi.spyOn(storage as any, 'readCanonicalObject')
|
||||
await storage.getNounIdsWithPagination({ limit: 1000, offset: 0 })
|
||||
expect(readSpy).not.toHaveBeenCalled()
|
||||
readSpy.mockRestore()
|
||||
|
||||
// The hydrating walk, by contrast, DOES read each entity.
|
||||
const readSpy2 = vi.spyOn(storage as any, 'readCanonicalObject')
|
||||
await storage.getNounsWithPagination({ limit: 1000, offset: 0 })
|
||||
expect(readSpy2.mock.calls.length).toBeGreaterThan(0)
|
||||
readSpy2.mockRestore()
|
||||
})
|
||||
|
||||
it('a type filter matches between the hydrating and id-only walks', async () => {
|
||||
const taskItems = await storage.getNounsWithPagination({
|
||||
limit: 1000,
|
||||
offset: 0,
|
||||
filter: { nounType: NounType.Task }
|
||||
})
|
||||
const taskIds = await storage.getNounIdsWithPagination({
|
||||
limit: 1000,
|
||||
offset: 0,
|
||||
filter: { nounType: NounType.Task }
|
||||
})
|
||||
const expected = Math.ceil(N / 3) // every 3rd is a Task
|
||||
expect(taskItems.items.length).toBe(expected)
|
||||
expect(new Set(taskIds.ids)).toEqual(new Set(taskItems.items.map((n: any) => n.id)))
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue