diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index 81390eab..87e73e4a 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -5,9 +5,7 @@ import { HNSWNoun, - HNSWVerb, HNSWNounWithMetadata, - HNSWVerbWithMetadata, StatisticsData, NounType } from '../../coreTypes.js' @@ -20,10 +18,6 @@ import { } from '../baseStorage.js' import { getBrainyVersion } from '../../utils/index.js' -// Type aliases for better readability -type HNSWNode = HNSWNoun -type Edge = HNSWVerb - // Node.js modules - dynamically imported to avoid issues in browser environments let fs: any let path: any @@ -75,8 +69,6 @@ export class FileSystemStorage extends BaseStorage { // - Handles 2.5M+ entities with < 10K files per shard // - Eliminates dynamic depth changes that cause path mismatch bugs private readonly SHARDING_DEPTH = 1 as const - private readonly MAX_SHARDS = 256 // Hex range: 00-ff - private cachedShardingDepth: number = this.SHARDING_DEPTH // Always use fixed depth protected rootDir: string private nounsDir!: string private verbsDir!: string @@ -251,40 +243,20 @@ export class FileSystemStorage extends BaseStorage { this.countsFilePath = path.join(this.systemDir, 'counts.json') await this.initializeCounts() - // Detect existing sharding structure and migrate if needed - const detectedDepth = await this.detectExistingShardingDepth() - - if (detectedDepth !== null && detectedDepth !== this.SHARDING_DEPTH) { - // Migration needed: existing structure doesn't match our fixed depth - console.log(`📦 Brainy Storage Migration`) - console.log(` Current structure: depth ${detectedDepth}`) - console.log(` Target structure: depth ${this.SHARDING_DEPTH}`) - console.log(` Entities to migrate: ${this.totalNounCount}`) - - await this.migrateShardingStructure(detectedDepth, this.SHARDING_DEPTH) - - console.log(`✅ Migration complete - now using depth ${this.SHARDING_DEPTH} sharding`) - } else if (detectedDepth === null) { - // The legacy sharding probe inspects `entities/nouns/hnsw` — a 7.x - // directory the 8.0 write path never populates (8.0 stores entities in - // the canonical `entities/nouns///` layout), so it returns - // null for EVERY 8.0 store, new or not. Decide new-vs-existing from the - // canonical layout the DB actually reads/writes so an established brain - // is not mislabeled "New installation" on every boot. - const established = - this.totalNounCount > 0 || (await this.hasCanonicalEntities()) - console.log( - established - ? `📁 Using depth ${this.SHARDING_DEPTH} sharding (${this.totalNounCount} entities)` - : `📁 New installation: using depth ${this.SHARDING_DEPTH} sharding (optimal for 1-2.5M entities)` - ) - } else { - // Already using correct depth - console.log(`📁 Using depth ${this.SHARDING_DEPTH} sharding (${this.totalNounCount} entities)`) - } - - // Always use fixed depth after migration/detection - this.cachedShardingDepth = this.SHARDING_DEPTH + // Boot log: new-vs-established, decided from the canonical layout the + // database actually reads and writes (`entities/nouns///`) + // plus the known noun count. The legacy hnsw sharding-depth probe and + // its depth-migration machinery are gone: the 8.0 write path never + // populated the directory they inspected, so the probe concluded "new + // installation" for every store on every boot and the migration branch + // was unreachable. + const established = + this.totalNounCount > 0 || (await this.hasCanonicalEntities()) + console.log( + established + ? `📁 Using depth ${this.SHARDING_DEPTH} sharding (${this.totalNounCount} entities)` + : `📁 New installation: using depth ${this.SHARDING_DEPTH} sharding (optimal for 1-2.5M entities)` + ) // Initialize GraphAdjacencyIndex and type statistics await super.init() @@ -320,390 +292,15 @@ export class FileSystemStorage extends BaseStorage { } } - /** - * Save a node to storage - * CRITICAL FIX: Added atomic write pattern to prevent file corruption during concurrent imports - */ - protected async saveNode(node: HNSWNode): Promise { - await this.ensureInitialized() - - // Convert connections Map to a serializable format - // CRITICAL: Only save lightweight vector data (no metadata) - // Metadata is saved separately via saveNounMetadata() (2-file system) - const serializableNode = { - id: node.id, - vector: node.vector, - connections: this.mapToObject(node.connections, (set) => - Array.from(set as Set) - ), - level: node.level || 0 - // NO metadata field - saved separately for scalability - } - - const filePath = this.getNodePath(node.id) - const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` - - try { - // ATOMIC WRITE SEQUENCE: - // 1. Write to temp file - await this.ensureDirectoryExists(path.dirname(tempPath)) - await fs.promises.writeFile(tempPath, JSON.stringify(serializableNode, null, 2)) - - // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) - await fs.promises.rename(tempPath, filePath) - } catch (error: any) { - // Clean up temp file on any error - try { - await fs.promises.unlink(tempPath) - } catch (cleanupError) { - // Ignore cleanup errors - } - throw error - } - - // Count tracking happens in baseStorage.saveNounMetadata_internal - // This fixes the race condition where metadata didn't exist yet - } - - /** - * Get a node from storage - */ - protected async getNode(id: string): Promise { - await this.ensureInitialized() - - // Clean, predictable path - no backward compatibility needed - const filePath = this.getNodePath(id) - try { - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedNode = JSON.parse(data) - - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries(parsedNode.connections)) { - connections.set(Number(level), new Set(nodeIds as string[])) - } - - // CRITICAL: Only return lightweight vector data (no metadata) - // Metadata is retrieved separately via getNounMetadata() (2-file system) - return { - id: parsedNode.id, - vector: parsedNode.vector, - connections, - level: parsedNode.level || 0 - // NO metadata field - retrieved separately for scalability - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error reading node ${id}:`, error) - } - return null - } - } - - /** - * Get all nodes from storage - * CRITICAL FIX: Now scans sharded subdirectories (depth=1) - * Previously only scanned flat directory, causing rebuild to find 0 entities - */ - protected async getAllNodes(): Promise { - await this.ensureInitialized() - - const allNodes: HNSWNode[] = [] - try { - // FIX: Use sharded file discovery instead of flat directory read - // This scans all 256 shard subdirectories (00-ff) to find actual files - const files = await this.getAllShardedFiles(this.nounsDir) - - for (const file of files) { - // Extract ID from filename and use sharded path - const id = file.replace('.json', '') - const filePath = this.getNodePath(id) - - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedNode = JSON.parse(data) - - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedNode.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } - - allNodes.push({ - id: parsedNode.id, - vector: parsedNode.vector, - connections, - level: parsedNode.level || 0 - }) - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error reading directory ${this.nounsDir}:`, error) - } - } - return allNodes - } - - /** - * Get nodes by noun type - * CRITICAL FIX: Now scans sharded subdirectories (depth=1) - * @param nounType The noun type to filter by - * @returns Promise that resolves to an array of nodes of the specified noun type - */ - protected async getNodesByNounType(nounType: string): Promise { - await this.ensureInitialized() - - const nouns: HNSWNode[] = [] - try { - // FIX: Use sharded file discovery instead of flat directory read - const files = await this.getAllShardedFiles(this.nounsDir) - - for (const file of files) { - // Extract ID from filename and use sharded path - const nodeId = file.replace('.json', '') - const filePath = this.getNodePath(nodeId) - - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedNode = JSON.parse(data) - - // Filter by noun type using metadata - const metadata = await this.getMetadata(nodeId) - if (metadata && metadata.noun === nounType) { - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedNode.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } - - nouns.push({ - id: parsedNode.id, - vector: parsedNode.vector, - connections, - level: parsedNode.level || 0 - }) - } - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error reading directory ${this.nounsDir}:`, error) - } - } - - return nouns - } - - /** - * Delete a node from storage - */ - protected async deleteNode(id: string): Promise { - await this.ensureInitialized() - - const filePath = this.getNodePath(id) - - // Load metadata to get type for count update (separate storage) - try { - const metadata = await this.getNounMetadata(id) - if (metadata) { - const type = metadata.noun || 'default' - this.decrementEntityCount(type) - } - } catch { - // Metadata might not exist, that's ok - } - - try { - await fs.promises.unlink(filePath) - - // Persist counts periodically - if (this.totalNounCount % 10 === 0) { - await this.persistCounts() - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error deleting node file ${filePath}:`, error) - throw error - } - } - } - - /** - * Save an edge to storage - * CRITICAL FIX: Added atomic write pattern to prevent file corruption during concurrent imports - */ - protected async saveEdge(edge: Edge): Promise { - await this.ensureInitialized() - - // Convert connections Map to a serializable format - // ARCHITECTURAL FIX: Include core relational fields in verb vector file - // These fields are essential for 90% of operations - no metadata lookup needed - const serializableEdge = { - id: edge.id, - vector: edge.vector, - connections: this.mapToObject(edge.connections, (set) => - Array.from(set as Set) - ), - - // CORE RELATIONAL DATA - verb: edge.verb, - sourceId: edge.sourceId, - targetId: edge.targetId, - - // User metadata (if any) - saved separately for scalability - // metadata field is saved separately via saveVerbMetadata() - } - - const filePath = this.getVerbPath(edge.id) - const tempPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).substring(2)}` - - try { - // ATOMIC WRITE SEQUENCE: - // 1. Write to temp file - await this.ensureDirectoryExists(path.dirname(tempPath)) - await fs.promises.writeFile(tempPath, JSON.stringify(serializableEdge, null, 2)) - - // 2. Atomic rename temp → final (crash-safe, prevents truncation during concurrent writes) - await fs.promises.rename(tempPath, filePath) - } catch (error: any) { - // Clean up temp file on any error - try { - await fs.promises.unlink(tempPath) - } catch (cleanupError) { - // Ignore cleanup errors - } - throw error - } - - // Count tracking happens in baseStorage.saveVerbMetadata_internal - // This fixes the race condition where metadata didn't exist yet - } - - /** - * Get an edge from storage - */ - protected async getEdge(id: string): Promise { - await this.ensureInitialized() - - const filePath = this.getVerbPath(id) - try { - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedEdge = JSON.parse(data) - - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries(parsedEdge.connections)) { - connections.set(Number(level), new Set(nodeIds as string[])) - } - - // Return HNSWVerb with core relational fields (NO metadata field) - return { - id: parsedEdge.id, - vector: parsedEdge.vector, - connections, - - // CORE RELATIONAL DATA (read from vector file) - verb: parsedEdge.verb, - sourceId: parsedEdge.sourceId, - targetId: parsedEdge.targetId - - // ✅ NO metadata field - // User metadata retrieved separately via getVerbMetadata() - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error reading edge ${id}:`, error) - } - return null - } - } - - /** - * Get all edges from storage - * CRITICAL FIX: Now scans sharded subdirectories (depth=1) - * Previously only scanned flat directory, causing rebuild to find 0 relationships - */ - protected async getAllEdges(): Promise { - await this.ensureInitialized() - - const allEdges: Edge[] = [] - try { - // FIX: Use sharded file discovery instead of flat directory read - // This scans all 256 shard subdirectories (00-ff) to find actual files - const files = await this.getAllShardedFiles(this.verbsDir) - - for (const file of files) { - // Extract ID from filename and use sharded path - const id = file.replace('.json', '') - const filePath = this.getVerbPath(id) - - const data = await fs.promises.readFile(filePath, 'utf-8') - const parsedEdge = JSON.parse(data) - - // Convert serialized connections back to Map> - const connections = new Map>() - for (const [level, nodeIds] of Object.entries( - parsedEdge.connections - )) { - connections.set(Number(level), new Set(nodeIds as string[])) - } - - // Include core relational fields (NO metadata field) - allEdges.push({ - id: parsedEdge.id, - vector: parsedEdge.vector, - connections, - - // CORE RELATIONAL DATA - verb: parsedEdge.verb, - sourceId: parsedEdge.sourceId, - targetId: parsedEdge.targetId - - // ✅ NO metadata field - // User metadata retrieved separately via getVerbMetadata() - }) - } - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error reading directory ${this.verbsDir}:`, error) - } - } - return allEdges - } - /** - * Delete an edge from storage - */ - protected async deleteEdge(id: string): Promise { - await this.ensureInitialized() - // Delete the HNSWVerb file using sharded path - const filePath = this.getVerbPath(id) - try { - await fs.promises.unlink(filePath) - } catch (error: any) { - if (error.code !== 'ENOENT') { - console.error(`Error deleting edge file ${filePath}:`, error) - throw error - } - } - // CRITICAL: Also delete verb metadata - this is what getVerbs() uses to find verbs - // Without this, getVerbsBySource() will still find "deleted" verbs via their metadata - try { - const metadata = await this.getVerbMetadata(id) - if (metadata) { - const verbType = (metadata.verb || metadata.type || 'default') as string - this.decrementVerbCount(verbType) - await this.deleteVerbMetadata(id) - } - } catch (error) { - // Ignore metadata deletion errors - verb file is already deleted - console.warn(`Failed to delete verb metadata for ${id}:`, error) - } - } + + + + + /** * Primitive operation: Write object to path @@ -2245,39 +1842,33 @@ export class FileSystemStorage extends BaseStorage { */ private async initializeCountsFromDisk(): Promise { try { - // CRITICAL: Detect existing depth before counting - // Can't use getAllShardedFiles() which assumes depth=1 - const existingDepth = await this.detectExistingShardingDepth() - const depthToUse = existingDepth !== null ? existingDepth : this.SHARDING_DEPTH + // Count the CANONICAL 8.0 layout (`entities////…`) — + // the tree saveNoun/getNouns actually read and write. The previous scan + // counted the vestigial `entities/*/hnsw` directories, which the 8.0 + // write path never populates, so a store recovering from a lost or + // corrupted counts.json re-initialized every counter to ZERO on real + // data (wrong stats/boot logs and a mis-sized rebuild-strategy + // decision at open). + const nouns = await this.scanCanonicalEntities('nouns') + this.totalNounCount = nouns.count + const verbs = await this.scanCanonicalEntities('verbs') + this.totalVerbCount = verbs.count - // Count nouns using detected depth - const validNounFiles = await this.getAllFilesAtDepth(this.nounsDir, depthToUse) - this.totalNounCount = validNounFiles.length - - // Count verbs using detected depth - const validVerbFiles = await this.getAllFilesAtDepth(this.verbsDir, depthToUse) - this.totalVerbCount = validVerbFiles.length - - // Sample some files to get type distribution (don't read all) - // Load metadata separately for type information - const sampleSize = Math.min(100, validNounFiles.length) - for (let i = 0; i < sampleSize; i++) { - try { - const file = validNounFiles[i] - const id = file.replace('.json', '') - - // Load metadata from separate storage for type info - const metadata = await this.getNounMetadata(id) - if (metadata) { - const type = metadata.noun || 'default' - this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1) - } - } catch { - // Skip invalid files or missing metadata + // Sample some entities for the type distribution (don't read all). + // Read the metadata files DIRECTLY with fs — this runs inside init(), + // and every guarded accessor (getNounMetadata → ensureInitialized) + // re-enters init() from here, deadlocking the open. (Latent in the old + // code too: its scan of the empty hnsw dirs just never sampled.) + for (const entityDir of nouns.sampleDirs) { + const metadata = await this.readEntityMetadataRaw(entityDir) + if (metadata) { + const type = metadata.noun || 'default' + this.entityCounts.set(type, (this.entityCounts.get(type) || 0) + 1) } } // Extrapolate counts if we sampled + const sampleSize = nouns.sampleDirs.length if (sampleSize < this.totalNounCount && sampleSize > 0) { const multiplier = this.totalNounCount / sampleSize for (const [type, count] of this.entityCounts.entries()) { @@ -2291,6 +1882,62 @@ export class FileSystemStorage extends BaseStorage { } } + /** + * Walk the canonical `entities//<2-hex-shard>//` tree, counting + * one entity per id directory (the layout `getNounVectorPath`/`getNouns` + * use). Returns up to 100 sampled entity directories (absolute paths) — + * nouns feed the type-distribution estimate above. An absent tree (fresh + * store) counts zero. + */ + private async scanCanonicalEntities( + kind: 'nouns' | 'verbs' + ): Promise<{ count: number; sampleDirs: string[] }> { + const base = path.join(this.rootDir, 'entities', kind) + const SAMPLE_MAX = 100 + let count = 0 + const sampleDirs: string[] = [] + try { + const shards = await fs.promises.readdir(base, { withFileTypes: true }) + for (const shard of shards) { + if (!shard.isDirectory() || !/^[0-9a-f]{2}$/i.test(shard.name)) continue + const shardPath = path.join(base, shard.name) + const ids = await fs.promises.readdir(shardPath, { withFileTypes: true }) + for (const entry of ids) { + if (!entry.isDirectory()) continue + count++ + if (sampleDirs.length < SAMPLE_MAX) { + sampleDirs.push(path.join(shardPath, entry.name)) + } + } + } + } catch (error: any) { + if (error?.code !== 'ENOENT') throw error + } + return { count, sampleDirs } + } + + /** + * Read one canonical entity's `metadata.json` (or `.json.gz`) directly with + * fs — NO guarded accessors. Used only by the init-time count recovery, + * where `getNounMetadata`'s `ensureInitialized()` would re-enter `init()`. + * @param entityDir - Absolute `entities///` directory. + * @returns The parsed metadata, or null when absent/unreadable. + */ + private async readEntityMetadataRaw(entityDir: string): Promise { + const base = path.join(entityDir, 'metadata.json') + try { + return JSON.parse(await fs.promises.readFile(base, 'utf-8')) + } catch { + // fall through to the compressed variant + } + try { + const gz = await fs.promises.readFile(`${base}.gz`) + return JSON.parse(zlib.gunzipSync(gz).toString('utf-8')) + } catch { + return null + } + } + /** * Persist counts to filesystem storage */ @@ -2321,307 +1968,9 @@ export class FileSystemStorage extends BaseStorage { // Intelligent Directory Sharding // ============================================= - /** - * Migrate files from one sharding depth to another - * Handles: 0→1 (flat to single-level), 2→1 (deep to single-level) - * Uses atomic file operations and comprehensive error handling - * - * @param fromDepth - Source sharding depth - * @param toDepth - Target sharding depth (must be 1) - */ - private async migrateShardingStructure(fromDepth: number, toDepth: number): Promise { - // Validation - if (fromDepth === toDepth) { - throw new Error(`Migration not needed: already at depth ${toDepth}`) - } - if (toDepth !== 1) { - throw new Error(`Migration only supports target depth 1 (got ${toDepth})`) - } - if (fromDepth !== 0 && fromDepth !== 2) { - throw new Error(`Migration only supports source depth 0 or 2 (got ${fromDepth})`) - } - // Create migration lock to prevent concurrent migrations - const lockFile = path.join(this.systemDir, '.migration-lock') - const lockExists = await this.fileExists(lockFile) - - if (lockExists) { - // Check if lock is stale (> 1 hour old) - try { - const stats = await fs.promises.stat(lockFile) - const lockAge = Date.now() - stats.mtimeMs - const ONE_HOUR = 60 * 60 * 1000 - - if (lockAge < ONE_HOUR) { - throw new Error( - 'Migration already in progress. If this is incorrect, delete .migration-lock file.' - ) - } - - // Lock is stale, remove it - console.log('⚠️ Removing stale migration lock (> 1 hour old)') - await fs.promises.unlink(lockFile) - } catch (error: any) { - if (error.code !== 'ENOENT') { - throw error - } - } - } - - try { - // Create lock file - await fs.promises.writeFile(lockFile, JSON.stringify({ - startedAt: new Date().toISOString(), - fromDepth, - toDepth, - pid: process.pid - })) - - // Discover all files to migrate - console.log('📊 Discovering files to migrate...') - const filesToMigrate = await this.discoverFilesForMigration(fromDepth) - - if (filesToMigrate.length === 0) { - console.log('ℹ️ No files to migrate') - return - } - - console.log(`📦 Migrating ${filesToMigrate.length} files...`) - - // Create all target shard directories upfront - await this.createAllShardDirectories(this.nounsDir) - await this.createAllShardDirectories(this.verbsDir) - - // Migrate files with progress tracking - let migratedCount = 0 - let skippedCount = 0 - const errors: Array<{ file: string; error: string }> = [] - - for (const fileInfo of filesToMigrate) { - try { - await this.migrateFile(fileInfo, fromDepth, toDepth) - migratedCount++ - - // Progress update every 1000 files - if (migratedCount % 1000 === 0) { - const percent = ((migratedCount / filesToMigrate.length) * 100).toFixed(1) - console.log(` 📊 Progress: ${migratedCount}/${filesToMigrate.length} (${percent}%)`) - } - - // Yield to event loop every 100 files to prevent blocking - if (migratedCount % 100 === 0) { - await new Promise(resolve => setImmediate(resolve)) - } - } catch (error: any) { - skippedCount++ - errors.push({ - file: fileInfo.oldPath, - error: error.message - }) - - // Log first few errors - if (errors.length <= 5) { - console.warn(`⚠️ Skipped ${fileInfo.oldPath}: ${error.message}`) - } - } - } - - // Final summary - console.log(`\n✅ Migration Results:`) - console.log(` Migrated: ${migratedCount} files`) - console.log(` Skipped: ${skippedCount} files`) - - if (errors.length > 0) { - console.warn(`\n⚠️ ${errors.length} files could not be migrated`) - if (errors.length > 5) { - console.warn(` (First 5 errors shown above, ${errors.length - 5} more occurred)`) - } - } - - // Cleanup: Remove empty old directories - if (fromDepth === 0) { - // No subdirectories to clean for flat structure - } else if (fromDepth === 2) { - await this.cleanupEmptyDirectories(this.nounsDir, fromDepth) - await this.cleanupEmptyDirectories(this.verbsDir, fromDepth) - } - - // Verification: Count files in new structure - const verifyCount = await this.countFilesInStructure(toDepth) - console.log(`\n🔍 Verification: ${verifyCount} files in new structure`) - - if (verifyCount < migratedCount) { - console.warn(`⚠️ Warning: Verification count (${verifyCount}) < migrated count (${migratedCount})`) - } - } finally { - // Always remove lock file - try { - await fs.promises.unlink(lockFile) - } catch (error) { - // Ignore error if lock file doesn't exist - } - } - } - - /** - * Discover all files that need to be migrated - * Constructs correct oldPath based on source depth - */ - private async discoverFilesForMigration(fromDepth: number): Promise> { - const files: Array<{ oldPath: string; id: string; type: 'noun' | 'verb' }> = [] - - // Discover noun files - const nounFiles = await this.getAllFilesAtDepth(this.nounsDir, fromDepth) - for (const filename of nounFiles) { - const id = filename.replace('.json', '') - - // Construct correct oldPath based on fromDepth - let oldPath: string - switch (fromDepth) { - case 0: - // Flat: nouns/uuid.json - oldPath = path.join(this.nounsDir, `${id}.json`) - break - case 1: - // Single-level: nouns/ab/uuid.json - oldPath = path.join(this.nounsDir, id.substring(0, 2), `${id}.json`) - break - case 2: - // Deep: nouns/ab/cd/uuid.json - oldPath = path.join(this.nounsDir, id.substring(0, 2), id.substring(2, 4), `${id}.json`) - break - default: - throw new Error(`Unsupported fromDepth: ${fromDepth}`) - } - - files.push({ oldPath, id, type: 'noun' }) - } - - // Discover verb files - const verbFiles = await this.getAllFilesAtDepth(this.verbsDir, fromDepth) - for (const filename of verbFiles) { - const id = filename.replace('.json', '') - - // Construct correct oldPath based on fromDepth - let oldPath: string - switch (fromDepth) { - case 0: - // Flat: verbs/uuid.json - oldPath = path.join(this.verbsDir, `${id}.json`) - break - case 1: - // Single-level: verbs/ab/uuid.json - oldPath = path.join(this.verbsDir, id.substring(0, 2), `${id}.json`) - break - case 2: - // Deep: verbs/ab/cd/uuid.json - oldPath = path.join(this.verbsDir, id.substring(0, 2), id.substring(2, 4), `${id}.json`) - break - default: - throw new Error(`Unsupported fromDepth: ${fromDepth}`) - } - - files.push({ oldPath, id, type: 'verb' }) - } - - return files - } - - /** - * Get all files at a specific depth - */ - private async getAllFilesAtDepth(baseDir: string, depth: number): Promise { - const allFiles: string[] = [] - - try { - const dirExists = await this.directoryExists(baseDir) - if (!dirExists) { - return [] - } - - switch (depth) { - case 0: - // Flat: files directly in baseDir - const entries = await fs.promises.readdir(baseDir) - for (const entry of entries) { - if (entry.endsWith('.json')) { - allFiles.push(entry) - } - } - break - - case 1: - // Single-level: baseDir/ab/uuid.json - const shardDirs = await fs.promises.readdir(baseDir) - for (const shard of shardDirs) { - const shardPath = path.join(baseDir, shard) - try { - const stat = await fs.promises.stat(shardPath) - if (stat.isDirectory()) { - const shardFiles = await fs.promises.readdir(shardPath) - for (const file of shardFiles) { - if (file.endsWith('.json')) { - allFiles.push(file) - } - } - } - } catch (error) { - // Skip inaccessible directories - } - } - break - - case 2: - // Deep: baseDir/ab/cd/uuid.json - const level1Dirs = await fs.promises.readdir(baseDir) - for (const level1 of level1Dirs) { - const level1Path = path.join(baseDir, level1) - try { - const level1Stat = await fs.promises.stat(level1Path) - if (level1Stat.isDirectory()) { - const level2Dirs = await fs.promises.readdir(level1Path) - for (const level2 of level2Dirs) { - const level2Path = path.join(level1Path, level2) - try { - const level2Stat = await fs.promises.stat(level2Path) - if (level2Stat.isDirectory()) { - const files = await fs.promises.readdir(level2Path) - for (const file of files) { - if (file.endsWith('.json')) { - allFiles.push(file) - } - } - } - } catch (error) { - // Skip inaccessible directories - } - } - } - } catch (error) { - // Skip inaccessible directories - } - } - break - } - } catch (error) { - // Directory doesn't exist or not accessible - } - - return allFiles - } - - /** - * Create all 256 shard directories (00-ff) - */ - private async createAllShardDirectories(baseDir: string): Promise { - for (let i = 0; i < this.MAX_SHARDS; i++) { - const shard = i.toString(16).padStart(2, '0') - const shardDir = path.join(baseDir, shard) - await this.ensureDirectoryExists(shardDir) - } - } /** * Migrate a single file atomically @@ -2650,116 +1999,20 @@ export class FileSystemStorage extends BaseStorage { await fs.promises.rename(oldPath, newPath) } - /** - * Clean up empty directories after migration - */ - private async cleanupEmptyDirectories(baseDir: string, depth: number): Promise { - try { - if (depth === 2) { - // Clean up level2 and level1 directories - const level1Dirs = await fs.promises.readdir(baseDir) - for (const level1 of level1Dirs) { - const level1Path = path.join(baseDir, level1) - try { - const level1Stat = await fs.promises.stat(level1Path) - if (level1Stat.isDirectory()) { - const level2Dirs = await fs.promises.readdir(level1Path) - for (const level2 of level2Dirs) { - const level2Path = path.join(level1Path, level2) - try { - // Try to remove level2 directory (will fail if not empty) - await fs.promises.rmdir(level2Path) - } catch (error) { - // Directory not empty or other error - ignore - } - } - // Try to remove level1 directory - await fs.promises.rmdir(level1Path) - } - } catch (error) { - // Directory not empty or other error - ignore - } - } - } - } catch (error) { - // Cleanup is best-effort, don't throw - } - } - /** - * Count files in the current structure - */ - private async countFilesInStructure(depth: number): Promise { - let count = 0 - - count += (await this.getAllFilesAtDepth(this.nounsDir, depth)).length - count += (await this.getAllFilesAtDepth(this.verbsDir, depth)).length - - return count - } - - /** - * Detect the actual sharding depth used by existing files - * Examines directory structure to determine current sharding strategy - * Returns null if no files exist yet (new installation) - */ - private async detectExistingShardingDepth(): Promise { - try { - // Check if nouns directory exists and has content - const dirExists = await this.directoryExists(this.nounsDir) - if (!dirExists) { - return null // New installation - } - - const entries = await fs.promises.readdir(this.nounsDir, { withFileTypes: true }) - - // Check if there are any .json files directly in nounsDir (flat structure) - const hasDirectJsonFiles = entries.some((e: any) => e.isFile() && e.name.endsWith('.json')) - if (hasDirectJsonFiles) { - return 0 // Flat structure: nouns/uuid.json - } - - // Check for subdirectories with hex names (sharding directories) - const subdirs = entries.filter((e: any) => e.isDirectory() && /^[0-9a-f]{2}$/i.test(e.name)) - if (subdirs.length === 0) { - return null // No files yet - } - - // Check first subdir to see if it has files or more subdirs - const firstSubdir = subdirs[0].name - const subdirPath = path.join(this.nounsDir, firstSubdir) - const subdirEntries = await fs.promises.readdir(subdirPath, { withFileTypes: true }) - - const hasJsonFiles = subdirEntries.some((e: any) => e.isFile() && e.name.endsWith('.json')) - if (hasJsonFiles) { - return 1 // Single-level sharding: nouns/ab/uuid.json - } - - const hasSubSubdirs = subdirEntries.some((e: any) => e.isDirectory() && /^[0-9a-f]{2}$/i.test(e.name)) - if (hasSubSubdirs) { - return 2 // Deep sharding: nouns/ab/cd/uuid.json - } - - return 1 // Default to single-level if structure is unclear - } catch (error) { - // If we can't read the directory, assume new installation - return null - } - } /** * Whether this store already holds canonical 8.0 entities. * * 8.0 writes nouns to `entities/nouns///vectors.json` (see - * `getNounVectorPath`), but the legacy sharding probe - * ({@link detectExistingShardingDepth}) inspects `entities/nouns/hnsw` — a 7.x - * directory the 8.0 write path never populates. The probe therefore returns - * null for every 8.0 store and cannot tell an established brain from a fresh - * one, which mislabels established stores "New installation" on every boot. - * This checks the canonical shard tree the DB actually reads and writes (the - * same `entities/nouns/` shards `getNounsWithPagination` walks) so boot - * logs are truthful. + * `getNounVectorPath`). This checks that canonical shard tree — the one the + * DB actually reads and writes (the same `entities/nouns/` shards + * `getNounsWithPagination` walks) — so the new-vs-established boot log is + * truthful. (The 7.x hnsw sharding probe this replaced inspected a directory + * the 8.0 write path never populated, so it mislabeled every established + * store "New installation" on every boot; that probe and its depth-migration + * machinery are removed.) * * @returns true if at least one 2-hex shard directory (00–ff) exists under * `entities/nouns/`, i.e. the store has previously persisted entities. @@ -2781,305 +2034,6 @@ export class FileSystemStorage extends BaseStorage { } } - /** - * Get sharding depth - * Always returns 1 (single-level sharding) for optimal balance of - * simplicity, performance, and reliability across all dataset sizes - * - * Single-level sharding (depth=1): - * - 256 shard directories (00-ff) - * - Handles 2.5M+ entities with excellent performance - * - No dynamic depth changes = no path mismatch bugs - * - Industry standard approach (Git uses similar) - */ - private getOptimalShardingDepth(): number { - return this.SHARDING_DEPTH - } - - /** - * Get the path for a node with consistent sharding strategy - * Clean, predictable path generation - */ - private getNodePath(id: string): string { - return this.getShardedPath(this.nounsDir, id) - } - - /** - * Get the path for a verb with consistent sharding strategy - */ - private getVerbPath(id: string): string { - return this.getShardedPath(this.verbsDir, id) - } - - /** - * Universal sharded path generator - * Always uses depth=1 (single-level sharding) for consistency - * - * Format: baseDir/ab/uuid.json - * Where 'ab' = first 2 hex characters of UUID (lowercase) - * - * Validates UUID format and throws descriptive errors - */ - private getShardedPath(baseDir: string, id: string): string { - // Extract first 2 characters for shard directory - const shard = id.substring(0, 2).toLowerCase() - - // Validate shard is valid hex (00-ff) - if (!/^[0-9a-f]{2}$/.test(shard)) { - throw new Error( - `Invalid entity ID format: ${id}. ` + - `Expected UUID starting with 2 hex characters, got '${shard}'. ` + - `IDs must be UUIDs or hex strings.` - ) - } - - // Single-level sharding: baseDir/ab/uuid.json - return path.join(baseDir, shard, `${id}.json`) - } - - /** - * Get all JSON files from the single-level sharded directory structure - * Traverses all shard subdirectories (00-ff) - */ - private async getAllShardedFiles(baseDir: string): Promise { - const allFiles: string[] = [] - - try { - const shardDirs = await fs.promises.readdir(baseDir) - - for (const shardDir of shardDirs) { - const shardPath = path.join(baseDir, shardDir) - - try { - const stat = await fs.promises.stat(shardPath) - - if (stat.isDirectory()) { - const shardFiles = await fs.promises.readdir(shardPath) - for (const file of shardFiles) { - if (file.endsWith('.json')) { - allFiles.push(file) - } - } - } - } catch (shardError) { - // Skip inaccessible shard directories - continue - } - } - - // Sort for consistent ordering - allFiles.sort() - return allFiles - - } catch (error: any) { - if (error.code === 'ENOENT') { - // Directory doesn't exist yet - return [] - } - throw error - } - } - - /** - * Production-scale streaming pagination for very large datasets - * Avoids loading all filenames into memory - */ - private async getVerbsWithPaginationStreaming( - options: { - limit?: number - cursor?: string - filter?: { - verbType?: string | string[] - sourceId?: string | string[] - targetId?: string | string[] - service?: string | string[] - metadata?: Record - } - }, - startIndex: number, - limit: number - ): Promise<{ - items: HNSWVerbWithMetadata[] - totalCount?: number - hasMore: boolean - nextCursor?: string - }> { - const verbs: HNSWVerbWithMetadata[] = [] - let processedCount = 0 - let skippedCount = 0 - let resultCount = 0 - - const depth = this.cachedShardingDepth ?? this.getOptimalShardingDepth() - - try { - // Stream through sharded directories efficiently - // hasMore=false means we reached the end of files, hasMore=true means streaming stopped early - const streamingHasMore = await this.streamShardedFiles( - this.verbsDir, - depth, - async (filename: string, filePath: string) => { - // Skip files until we reach start index - if (skippedCount < startIndex) { - skippedCount++ - return true // continue - } - - // Stop if we have enough results - if (resultCount >= limit) { - return false // stop streaming - more files exist - } - - try { - const id = filename.replace('.json', '') - - // Read verb data and metadata - const data = await fs.promises.readFile(filePath, 'utf-8') - const edge = JSON.parse(data) - const metadata = await this.getVerbMetadata(id) - - // Don't skip verbs without metadata - metadata is optional - // FIX: This was the root cause of the VFS bug (11 versions) - // Verbs can exist without metadata files (e.g., from imports/migrations) - - // Convert connections if needed - let connections = edge.connections - if (connections && typeof connections === 'object' && !(connections instanceof Map)) { - const connectionsMap = new Map>() - for (const [level, nodeIds] of Object.entries(connections)) { - connectionsMap.set(Number(level), new Set(nodeIds as string[])) - } - connections = connectionsMap - } - - // Canonical hydration (single source of truth: - // src/types/reservedFields.ts) — reserved fields top-level, ONLY - // custom fields in `metadata`. The previous hand-rolled - // destructure here missed `subtype` (so streamed verbs lost it) - // and `verb` (so the type key echoed inside custom metadata). - const verbWithMetadata: HNSWVerbWithMetadata = this.hydrateVerbWithMetadata( - { - id: edge.id, - vector: edge.vector, - connections: connections || new Map(), - verb: edge.verb, - sourceId: edge.sourceId, - targetId: edge.targetId - }, - metadata - ) - - // Apply filters - if (options.filter) { - const filter = options.filter - - if (filter.verbType) { - const types = Array.isArray(filter.verbType) ? filter.verbType : [filter.verbType] - if (!types.includes(verbWithMetadata.verb)) return true // continue - } - - if (filter.sourceId) { - const sources = Array.isArray(filter.sourceId) ? filter.sourceId : [filter.sourceId] - if (!sources.includes(verbWithMetadata.sourceId)) return true // continue - } - - if (filter.targetId) { - const targets = Array.isArray(filter.targetId) ? filter.targetId : [filter.targetId] - if (!targets.includes(verbWithMetadata.targetId)) return true // continue - } - } - - verbs.push(verbWithMetadata) - resultCount++ - processedCount++ - return true // continue - - } catch (error) { - console.warn(`Failed to read verb from ${filePath}:`, error) - processedCount++ - return true // continue - } - } - ) - - // CRITICAL FIX: Use streaming result for hasMore, not cached totalVerbCount - // streamingHasMore=false means we exhausted all files - // Also verify we loaded items to prevent infinite loops - const finalHasMore = streamingHasMore && (resultCount > 0 || startIndex === 0) - - return { - items: verbs, - totalCount: this.totalVerbCount || undefined, // Return cached count as hint only - hasMore: finalHasMore, - nextCursor: finalHasMore ? String(startIndex + resultCount) : undefined - } - - } catch (error: any) { - if (error.code === 'ENOENT') { - return { - items: [], - totalCount: 0, - hasMore: false - } - } - throw error - } - } - - /** - * Stream through sharded files without loading all names into memory - * Production-scale implementation for millions of files - */ - /** - * Stream through files in single-level sharded structure - * Calls processor for each file until processor returns false - * Returns true if more files exist (processor stopped early), false if all processed - */ - private async streamShardedFiles( - baseDir: string, - depth: number, - processor: (filename: string, fullPath: string) => Promise - ): Promise { - let hasMore = true - - // Single-level sharding (depth=1): baseDir/ab/uuid.json - try { - const shardDirs = await fs.promises.readdir(baseDir) - const sortedShardDirs = shardDirs.sort() - - for (const shardDir of sortedShardDirs) { - const shardPath = path.join(baseDir, shardDir) - - try { - const stat = await fs.promises.stat(shardPath) - - if (stat.isDirectory()) { - const files = await fs.promises.readdir(shardPath) - const sortedFiles = files.filter((f: string) => f.endsWith('.json')).sort() - - for (const file of sortedFiles) { - const shouldContinue = await processor(file, path.join(shardPath, file)) - - if (!shouldContinue) { - hasMore = false - break - } - } - - if (!hasMore) break - } - } catch (shardError) { - // Skip inaccessible shard directories - continue - } - } - } catch (error: any) { - if (error.code === 'ENOENT') { - hasMore = false - } - } - - return hasMore - } /** * Check if a file exists (handles both sharded and non-sharded) diff --git a/tests/unit/storage/sharding-new-installation-log.test.ts b/tests/unit/storage/sharding-new-installation-log.test.ts index ca674137..c1af8427 100644 --- a/tests/unit/storage/sharding-new-installation-log.test.ts +++ b/tests/unit/storage/sharding-new-installation-log.test.ts @@ -87,6 +87,29 @@ describe('FileSystemStorage boot log: an established brain is not "New installat await teardown(second) }) + it('counts RECOVER from the canonical layout when counts.json is lost (container-restart case)', async () => { + const dir = mkTmp() + + // Establish a store with 3 entities (canonical layout + counts.json). + const first: any = new FileSystemStorage(dir) + await first.init() + for (const c of ['a', 'b', 'c']) await seedOne(first, c.repeat(32)) + await teardown(first) + + // Simulate the lost/corrupted counts.json a container restart can leave. + for (const f of ['counts.json', 'counts.json.gz']) { + rmSync(join(dir, '_system', f), { force: true }) + } + + // Reopen: the recovery scan must count the CANONICAL tree. The removed + // implementation scanned the vestigial `entities/*/hnsw` dirs and + // re-initialized every counter to ZERO on real data. + const second: any = new FileSystemStorage(dir) + await second.init() + expect(await second.getNounCount()).toBe(3) + await teardown(second) + }) + it('hasCanonicalEntities() sees the canonical shard tree — independent of counts.json', async () => { const dir = mkTmp() const s: any = new FileSystemStorage(dir)