diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 8a491e3e..5fdf129f 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -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 /** @@ -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 + } + }): 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 diff --git a/tests/unit/storage/pagination-parallel-hydration.test.ts b/tests/unit/storage/pagination-parallel-hydration.test.ts new file mode 100644 index 00000000..ada324bb --- /dev/null +++ b/tests/unit/storage/pagination-parallel-hydration.test.ts @@ -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, 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))) + }) +})