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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue