brainy/src/storage/adapters/memoryStorage.ts
David Snelling b20666e020 refactor(8.0): final cleanup — drop HnswProvider alias + config.hnsw + 'hnsw' surface (scaffold step 6)
Step 6 of the brainy 8.0 rename scaffolding. Removes every legacy 'hnsw'-
flavoured public name introduced as a compat shim in steps 1-5. The
algorithm-neutral surface is now the only surface.

REMOVED — PHASE A (type aliases)

src/plugin.ts
- Removed `export type HnswProvider = VectorIndexProvider` alias.
- Removed `export type DiskAnnProvider = VectorIndexProvider` alias.

src/index.ts
- Removed `export const HNSWIndex = JsHnswVectorIndex` alias.

src/types/brainy.types.ts
- Removed `BrainyStats.indexHealth.hnsw` field (the new `vector` field
  carries the same boolean).

src/brainy.ts
- `stats()` no longer emits the legacy `hnsw` field on `indexHealth`.

REMOVED — PHASE B (storage adapter rename, 8 adapters)

Repo-wide rename: `saveHNSWData` → `saveVectorIndexData` and
`getHNSWData` → `getVectorIndexData` across:

- src/storage/adapters/baseStorageAdapter.ts (abstract declarations)
- src/storage/adapters/fileSystemStorage.ts
- src/storage/adapters/gcsStorage.ts
- src/storage/adapters/r2Storage.ts
- src/storage/adapters/s3CompatibleStorage.ts
- src/storage/adapters/azureBlobStorage.ts
- src/storage/adapters/opfsStorage.ts
- src/storage/adapters/memoryStorage.ts
- src/storage/adapters/historicalStorageAdapter.ts
- src/brainy.ts (call sites)
- src/hnsw/hnswIndex.ts + src/hnsw/typeAwareHNSWIndex.ts (call sites)

The default-delegation wrappers added in scaffold step 4 are removed
(would have been duplicate declarations after the rename).

REMOVED — PHASE C (cache category 'hnsw')

src/utils/unifiedCache.ts
- Cache-category union narrowed from
    'hnsw' | 'vectors' | 'metadata' | 'embedding' | 'other'
  to
    'vectors' | 'metadata' | 'embedding' | 'other'
- typeAccessCounts, typeSizes, typeCounts, accessRatios, sizeRatios,
  and the fairness-check iterator all drop the 'hnsw' key.
- Per planning § 2.5: pre-8.0 'hnsw' cache entries are an in-memory
  category. Cache is rebuildable; entries naturally don't exist after
  a restart, so no migration path is needed.

src/hnsw/hnswIndex.ts
- 3 cache.set() call sites migrated from category 'hnsw' to 'vectors'.
- Cache key prefix `hnsw:vector:` → `vector:`.
- typeCounts/typeSizes/typeAccessCounts accessor renames
  `.hnsw` → `.vectors`.

tests/unit/utils/unifiedCache-eviction.test.ts
- Test fixtures updated: cache.set(..., 'hnsw', ...) → cache.set(..., 'vectors', ...).
- Stats assertion `stats.typeSizes.hnsw` → `stats.typeSizes.vectors`.

REMOVED — PHASE D (config.hnsw)

src/types/brainy.types.ts
- Removed `BrainyConfig.hnsw` field entirely. The 8.0 surface is
  `BrainyConfig.vector.{recall, quantization, vectorStorage, advanced}`.

src/brainy.ts
- normalizeConfig() no longer emits a `hnsw` field on Required<BrainyConfig>.
- setupIndex() rewired:
  - Reads from `this.config.vector` (not `this.config.hnsw`).
  - Calls `resolveJsHnswConfig(this.config.vector)` to translate the
    `recall` preset into M / efConstruction / efSearch knobs (with
    `advanced.hnsw` overrides winning when supplied).
- Imports `resolveJsHnswConfig` from './utils/recallPreset.js'.

NOT IN THIS COMMIT (deliberately)

- Persisted file path migration `_system/hnsw-*.json` →
  `_system/vector-index-*.json`. Requires dual-read logic on boot
  across 8 storage adapters. Per integration doc lines 531-539:
  "Reader accepts either spelling on load; writer emits the new spelling
  only; brains self-migrate on the next persist after upgrade." This is
  a separate body of work and lands in a follow-up commit before 8.0 GA.
- `config.hnswPersistMode` top-level field is unchanged. It's not in
  the rename inventory; a future commit can fold it into
  `config.vector.persistMode` if desired.
- strictConfig enforcement wiring. The field is accepted by
  normalizeConfig() with default 'warn'; the actual warning emission at
  knob-mismatch sites lands when the config-resolution layer is touched
  for the persisted-path migration.

VERIFICATION

- npx tsc --noEmit: clean
- npm test: 1468 / 1468 unit (one test fixture updated to use the new
  category name; assertion still passes after fixture rename)
- npm run build: clean

The 8.0 PR's user-facing surface is now algorithm-neutral end-to-end:
VectorIndexProvider, config.vector.recall, JsHnswVectorIndex,
saveVectorIndexData, 'vectors' cache category, indexHealth.vector,
strictConfig — all standalone. No legacy 'hnsw' name remains in any
public type or method signature.
2026-06-09 13:28:53 -07:00

512 lines
16 KiB
TypeScript

/**
* Memory Storage Adapter
* In-memory storage adapter for environments where persistent storage is not available or needed
*/
import {
GraphVerb,
HNSWNoun,
HNSWVerb,
NounMetadata,
VerbMetadata,
HNSWNounWithMetadata,
HNSWVerbWithMetadata,
StatisticsData,
NounType
} from '../../coreTypes.js'
import { BaseStorage, StorageBatchConfig, STATISTICS_KEY } from '../baseStorage.js'
import { PaginatedResult } from '../../types/paginationTypes.js'
// No type aliases needed - using the original types directly
/**
* In-memory storage adapter
* Uses Maps to store data in memory
*/
export class MemoryStorage extends BaseStorage {
// Removed redundant Maps (nouns, verbs) - objectStore handles all storage via type-first paths
private statistics: StatisticsData | null = null
// Unified object store for primitive operations (replaces metadata, nounMetadata, verbMetadata)
private objectStore: Map<string, any> = new Map()
// Raw binary-blob store, keyed by blob key. Holds opaque byte payloads
// (column-store segments, batch vectors) verbatim — no JSON envelope. In-memory
// storage has no local file, so getBinaryBlobPath() returns null.
private blobStore: Map<string, Buffer> = new Map()
// Backward compatibility aliases
private get metadata(): Map<string, any> {
return this.objectStore
}
private get nounMetadata(): Map<string, any> {
return this.objectStore
}
private get verbMetadata(): Map<string, any> {
return this.objectStore
}
constructor() {
super()
}
/**
* Get Memory-optimized batch configuration
*
* Memory storage has no rate limits and can handle very high throughput:
* - Large batch sizes (1000 items)
* - No delays needed (0ms)
* - High concurrency (1000 operations)
* - Parallel processing maximizes throughput
*
* @returns Memory-optimized batch configuration
*/
public getBatchConfig(): StorageBatchConfig {
return {
maxBatchSize: 1000,
batchDelayMs: 0,
maxConcurrent: 1000,
supportsParallelWrites: true, // Memory loves parallel operations
rateLimit: {
operationsPerSecond: 100000, // Virtually unlimited
burstCapacity: 100000
}
}
}
/**
* Initialize the storage adapter
* Calls super.init() to initialize GraphAdjacencyIndex and type statistics
*/
public async init(): Promise<void> {
await super.init()
}
// Removed saveNoun_internal and getNoun_internal - using BaseStorage's type-first implementation
/**
* Get nouns with pagination and filtering
* Returns HNSWNounWithMetadata[] (includes metadata field)
* @param options Pagination and filtering options
* @returns Promise that resolves to a paginated result of nouns with metadata
*/
// Removed public method overrides (getNouns, getNounsWithPagination, getVerbs) - using BaseStorage's type-first implementation
// Removed getNounsByNounType_internal and deleteNoun_internal - using BaseStorage's type-first implementation
// Removed saveVerb_internal and getVerb_internal - using BaseStorage's type-first implementation
// Removed verb *_internal method overrides - using BaseStorage's type-first implementation
/**
* Primitive operation: Write object to path
* All metadata operations use this internally via base class routing
*/
protected async writeObjectToPath(path: string, data: any): Promise<void> {
// Store in unified object store using path as key
this.objectStore.set(path, JSON.parse(JSON.stringify(data)))
}
/**
* Primitive operation: Read object from path
* All metadata operations use this internally via base class routing
*/
protected async readObjectFromPath(path: string): Promise<any | null> {
const data = this.objectStore.get(path)
if (!data) {
return null
}
return JSON.parse(JSON.stringify(data))
}
/**
* Primitive operation: Delete object from path
* All metadata operations use this internally via base class routing
*/
protected async deleteObjectFromPath(path: string): Promise<void> {
this.objectStore.delete(path)
}
/**
* Primitive operation: List objects under path prefix
* All metadata operations use this internally via base class routing
*/
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
const paths: string[] = []
for (const key of this.objectStore.keys()) {
if (key.startsWith(prefix)) {
paths.push(key)
}
}
return paths.sort()
}
// ===========================================================================
// Raw binary-blob primitive
// ===========================================================================
/**
* Persist a raw binary blob in memory under `key`. A defensive copy of the
* bytes is stored so later mutations to the caller's buffer don't corrupt the
* stored blob. Overwrites any existing blob at the same key.
*
* @param key - The blob key.
* @param data - The exact bytes to store.
*/
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
this.blobStore.set(key, Buffer.from(data))
}
/**
* Load a copy of the bytes stored under `key`, or `null` if absent. A copy is
* returned so callers cannot mutate the stored blob in place.
*
* @param key - The blob key.
* @returns The blob bytes, or `null` if absent.
*/
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
const data = this.blobStore.get(key)
return data ? Buffer.from(data) : null
}
/**
* Delete the blob stored under `key`. Missing blobs are ignored.
*
* @param key - The blob key.
*/
public async deleteBinaryBlob(key: string): Promise<void> {
this.blobStore.delete(key)
}
/**
* In-memory storage has no local filesystem path to mmap, so this always
* returns `null`. Callers must use {@link loadBinaryBlob} instead.
*
* @param _key - The blob key (unused).
* @returns Always `null`.
*/
public getBinaryBlobPath(_key: string): string | null {
return null
}
/**
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
* Memory storage implementation is simple since all data is already in memory
*/
public async getMetadataBatch(ids: string[]): Promise<Map<string, any>> {
const results = new Map<string, any>()
// Memory storage can handle all IDs at once since it's in-memory
for (const id of ids) {
// CRITICAL: Use getNounMetadata() instead of deprecated getMetadata()
// This ensures we fetch from the correct noun metadata store (2-file system)
const metadata = await this.getNounMetadata(id)
if (metadata) {
results.set(id, metadata)
}
}
return results
}
/**
* Clear all data from storage
* Clears objectStore (type-first paths)
* Also clears writeCache to prevent stale data after clear
*/
public async clear(): Promise<void> {
this.objectStore.clear()
this.blobStore.clear()
this.statistics = null
this.totalNounCount = 0
this.totalVerbCount = 0
this.entityCounts.clear()
this.verbCounts.clear()
// Clear the statistics cache
this.statisticsCache = null
this.statisticsModified = false
// Clear write-through cache (inherited from BaseStorage)
// Without this, readWithInheritance() would return stale cached data
// after clear(), causing "ghost" entities to appear
this.clearWriteCache()
}
/**
* Get information about storage usage and capacity
* Uses BaseStorage counts
*/
public async getStorageStatus(): Promise<{
type: string
used: number
quota: number | null
details?: Record<string, any>
}> {
return {
type: 'memory',
used: 0, // In-memory storage doesn't have a meaningful size
quota: null, // In-memory storage doesn't have a quota
details: {
nodeCount: this.totalNounCount,
edgeCount: this.totalVerbCount,
objectStoreSize: this.objectStore.size
}
}
}
/**
* Check if COW has been explicitly disabled via clear()
* No-op for MemoryStorage (doesn't persist)
* @returns Always false (marker doesn't persist in memory)
* @protected
*/
/**
* Removed checkClearMarker() and createClearMarker() methods
* COW is now always enabled - marker files are no longer used
*/
/**
* Save statistics data to storage
* @param statistics The statistics data to save
*/
protected async saveStatisticsData(statistics: StatisticsData): Promise<void> {
// For memory storage, we just need to store the statistics in memory
// Create a deep copy to avoid reference issues
this.statistics = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated,
// Include serviceActivity if present
...(statistics.serviceActivity && {
serviceActivity: Object.fromEntries(
Object.entries(statistics.serviceActivity).map(([k, v]) => [k, {...v}])
)
}),
// Include services if present
...(statistics.services && {
services: statistics.services.map(s => ({...s}))
}),
// Include distributedConfig if present
...(statistics.distributedConfig && {
distributedConfig: JSON.parse(JSON.stringify(statistics.distributedConfig))
})
}
// Since this is in-memory, there's no need for time-based partitioning
// or legacy file handling
}
/**
* Get statistics data from storage
* @returns Promise that resolves to the statistics data or null if not found
*/
protected async getStatisticsData(): Promise<StatisticsData | null> {
if (!this.statistics) {
// CRITICAL FIX: Statistics don't exist yet (first init)
// Return minimal stats with counts instead of null
// This prevents HNSW from seeing entityCount=0 during index rebuild
return {
nounCount: {},
verbCount: {},
metadataCount: {},
hnswIndexSize: 0,
totalNodes: this.totalNounCount,
totalEdges: this.totalVerbCount,
totalMetadata: 0,
lastUpdated: new Date().toISOString()
}
}
// Return a deep copy to avoid reference issues
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
// CRITICAL FIX: Populate totalNodes and totalEdges from in-memory counts
// HNSW rebuild depends on these fields to determine entity count
totalNodes: this.totalNounCount,
totalEdges: this.totalVerbCount,
lastUpdated: this.statistics.lastUpdated,
// Include serviceActivity if present
...(this.statistics.serviceActivity && {
serviceActivity: Object.fromEntries(
Object.entries(this.statistics.serviceActivity).map(([k, v]) => [k, {...v}])
)
}),
// Include services if present
...(this.statistics.services && {
services: this.statistics.services.map(s => ({...s}))
}),
// Include distributedConfig if present
...(this.statistics.distributedConfig && {
distributedConfig: JSON.parse(JSON.stringify(this.statistics.distributedConfig))
})
}
// Since this is in-memory, there's no need for fallback mechanisms
// to check multiple storage locations
}
/**
* Initialize counts from in-memory storage - O(1) operation
*/
protected async initializeCounts(): Promise<void> {
// Scan objectStore paths (ID-first structure) to count entities
this.entityCounts.clear()
this.verbCounts.clear()
let totalNouns = 0
let totalVerbs = 0
// Scan all paths in objectStore
for (const path of this.objectStore.keys()) {
// Count nouns (entities/nouns/{shard}/{id}/vectors.json)
const nounMatch = path.match(/^entities\/nouns\/[0-9a-f]{2}\/[^/]+\/vectors\.json$/)
if (nounMatch) {
// Type is in metadata, not path - just count total
totalNouns++
}
// Count verbs (entities/verbs/{shard}/{id}/vectors.json)
const verbMatch = path.match(/^entities\/verbs\/[0-9a-f]{2}\/[^/]+\/vectors\.json$/)
if (verbMatch) {
// Type is in metadata, not path - just count total
totalVerbs++
}
}
this.totalNounCount = totalNouns
this.totalVerbCount = totalVerbs
}
/**
* Persist counts to storage - no-op for memory storage
*/
protected async persistCounts(): Promise<void> {
// No persistence needed for in-memory storage
// Counts are always accurate from the live data structures
}
// =============================================
// HNSW Index Persistence
// =============================================
/**
* Get vector for a noun
* Uses BaseStorage's type-first implementation
*/
public async getNounVector(id: string): Promise<number[] | null> {
const noun = await this.getNoun(id)
return noun ? [...noun.vector] : null
}
// CRITICAL FIX: Mutex locks for HNSW concurrency control
// Even in-memory operations need serialization to prevent async race conditions
private hnswLocks = new Map<string, Promise<void>>()
/**
* Save HNSW graph data for a noun
*
* CRITICAL FIX: Mutex locking to prevent race conditions during concurrent HNSW updates
* Even in-memory operations can race due to async/await interleaving
* Prevents data corruption when multiple entities connect to same neighbor simultaneously
*/
public async saveVectorIndexData(nounId: string, hnswData: {
level: number
connections: Record<string, string[]>
}): Promise<void> {
const path = `hnsw/${nounId}.json`
// MUTEX LOCK: Wait for any pending operations on this entity
while (this.hnswLocks.has(path)) {
await this.hnswLocks.get(path)
}
// Acquire lock by creating a promise that we'll resolve when done
let releaseLock!: () => void
const lockPromise = new Promise<void>(resolve => { releaseLock = resolve })
this.hnswLocks.set(path, lockPromise)
try {
// Read existing data (if exists)
let existingNode: any = {}
const existing = this.objectStore.get(path)
if (existing) {
existingNode = existing
}
// Preserve id and vector, update only HNSW graph metadata
const updatedNode = {
...existingNode, // Preserve all existing fields
level: hnswData.level,
connections: hnswData.connections
}
// Write atomically (in-memory, but now serialized by mutex)
this.objectStore.set(path, JSON.parse(JSON.stringify(updatedNode)))
} finally {
// Release lock
this.hnswLocks.delete(path)
releaseLock()
}
}
/**
* Get HNSW graph data for a noun
*/
public async getVectorIndexData(nounId: string): Promise<{
level: number
connections: Record<string, string[]>
} | null> {
const path = `hnsw/${nounId}.json`
const data = await this.readObjectFromPath(path)
return data || null
}
/**
* Save HNSW system data (entry point, max level)
*
* CRITICAL FIX: Mutex locking to prevent race conditions
*/
public async saveHNSWSystem(systemData: {
entryPointId: string | null
maxLevel: number
}): Promise<void> {
const path = 'system/hnsw-system.json'
// MUTEX LOCK: Wait for any pending operations
while (this.hnswLocks.has(path)) {
await this.hnswLocks.get(path)
}
// Acquire lock
let releaseLock!: () => void
const lockPromise = new Promise<void>(resolve => { releaseLock = resolve })
this.hnswLocks.set(path, lockPromise)
try {
// Write atomically (serialized by mutex)
this.objectStore.set(path, JSON.parse(JSON.stringify(systemData)))
} finally {
// Release lock
this.hnswLocks.delete(path)
releaseLock()
}
}
/**
* Get HNSW system data
*/
public async getHNSWSystem(): Promise<{
entryPointId: string | null
maxLevel: number
} | null> {
const path = 'system/hnsw-system.json'
const data = await this.readObjectFromPath(path)
return data || null
}
}