diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index 3dea8018..5f2232c1 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -225,11 +225,16 @@ export class FileSystemStorage extends BaseStorage { const isNew = !(await this.fileExists(this.getNodePath(node.id))) // 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 = { - ...node, + 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) @@ -269,12 +274,14 @@ export class FileSystemStorage extends BaseStorage { 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, - metadata: parsedNode.metadata + level: parsedNode.level || 0 + // NO metadata field - retrieved separately for scalability } } catch (error: any) { if (error.code !== 'ENOENT') { @@ -414,11 +421,15 @@ export class FileSystemStorage extends BaseStorage { const isNew = !(await this.fileExists(this.getVerbPath(edge.id))) // Convert connections Map to a serializable format + // CRITICAL: Only save lightweight vector data (no metadata) + // Metadata is saved separately via saveVerbMetadata() (2-file system) const serializableEdge = { - ...edge, + id: edge.id, + vector: edge.vector, connections: this.mapToObject(edge.connections, (set) => Array.from(set as Set) ) + // NO metadata field - saved separately for scalability } const filePath = this.getVerbPath(edge.id) @@ -678,7 +689,9 @@ export class FileSystemStorage extends BaseStorage { const batchPromises = batch.map(async (id) => { try { - const metadata = await this.getMetadata(id) + // 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) return { id, metadata } } catch (error) { console.debug(`Failed to read metadata for ${id}:`, error) diff --git a/src/storage/adapters/gcsStorage.ts b/src/storage/adapters/gcsStorage.ts index 042c6820..8d21312f 100644 --- a/src/storage/adapters/gcsStorage.ts +++ b/src/storage/adapters/gcsStorage.ts @@ -430,14 +430,19 @@ export class GcsStorage extends BaseStorage { this.logger.trace(`Saving node ${node.id}`) // 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 = { - ...node, + id: node.id, + vector: node.vector, connections: Object.fromEntries( Array.from(node.connections.entries()).map(([level, nounIds]) => [ level, Array.from(nounIds) ]) - ) + ), + level: node.level || 0 + // NO metadata field - saved separately for scalability } // Get the GCS key with UUID-based sharding @@ -517,12 +522,14 @@ export class GcsStorage extends BaseStorage { connections.set(Number(level), new Set(nounIds as string[])) } + // CRITICAL: Only return lightweight vector data (no metadata) + // Metadata is retrieved separately via getNounMetadata() (2-file system) const node: HNSWNode = { id: data.id, vector: data.vector, connections, - level: data.level || 0, - metadata: data.metadata // CRITICAL: Include metadata for entity reconstruction + level: data.level || 0 + // NO metadata field - retrieved separately for scalability } // Update cache @@ -741,14 +748,18 @@ export class GcsStorage extends BaseStorage { this.logger.trace(`Saving edge ${edge.id}`) // Convert connections Map to serializable format + // CRITICAL: Only save lightweight vector data (no metadata) + // Metadata is saved separately via saveVerbMetadata() (2-file system) const serializableEdge = { - ...edge, + id: edge.id, + vector: edge.vector, connections: Object.fromEntries( Array.from(edge.connections.entries()).map(([level, verbIds]) => [ level, Array.from(verbIds) ]) ) + // NO metadata field - saved separately for scalability } // Get the GCS key with UUID-based sharding @@ -1320,6 +1331,53 @@ export class GcsStorage extends BaseStorage { }) } + /** + * Batch fetch metadata for multiple noun IDs (efficient for large queries) + * Uses smaller batches to prevent GCS socket exhaustion + * @param ids Array of noun IDs to fetch metadata for + * @returns Map of ID to metadata + */ + public async getMetadataBatch(ids: string[]): Promise> { + await this.ensureInitialized() + + const results = new Map() + const batchSize = 10 // Smaller batches for metadata to prevent socket exhaustion + + // Process in smaller batches + for (let i = 0; i < ids.length; i += batchSize) { + const batch = ids.slice(i, i + batchSize) + + const batchPromises = batch.map(async (id) => { + try { + // 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) + return { id, metadata } + } catch (error: any) { + // Handle GCS-specific errors + if (this.isThrottlingError(error)) { + await this.handleThrottling(error) + } + this.logger.debug(`Failed to read metadata for ${id}:`, error) + return { id, metadata: null } + } + }) + + const batchResults = await Promise.all(batchPromises) + + for (const { id, metadata } of batchResults) { + if (metadata !== null) { + results.set(id, metadata) + } + } + + // Small yield between batches to prevent overwhelming GCS + await new Promise(resolve => setImmediate(resolve)) + } + + return results + } + /** * Clear all data from storage */ diff --git a/src/storage/adapters/memoryStorage.ts b/src/storage/adapters/memoryStorage.ts index a1e59381..74a4af6f 100644 --- a/src/storage/adapters/memoryStorage.ts +++ b/src/storage/adapters/memoryStorage.ts @@ -52,12 +52,14 @@ export class MemoryStorage extends BaseStorage { const isNew = !this.nouns.has(noun.id) // Create a deep copy to avoid reference issues + // CRITICAL: Only save lightweight vector data (no metadata) + // Metadata is saved separately via saveNounMetadata() (2-file system) const nounCopy: HNSWNoun = { id: noun.id, vector: [...noun.vector], connections: new Map(), - level: noun.level || 0, - metadata: noun.metadata + level: noun.level || 0 + // NO metadata field - saved separately for scalability } // Copy connections @@ -88,12 +90,14 @@ export class MemoryStorage extends BaseStorage { } // Return a deep copy to avoid reference issues + // CRITICAL: Only return lightweight vector data (no metadata) + // Metadata is retrieved separately via getNounMetadata() (2-file system) const nounCopy: HNSWNoun = { id: noun.id, vector: [...noun.vector], connections: new Map(), - level: noun.level || 0, - metadata: noun.metadata + level: noun.level || 0 + // NO metadata field - retrieved separately for scalability } // Copy connections @@ -592,7 +596,9 @@ export class MemoryStorage extends BaseStorage { // Memory storage can handle all IDs at once since it's in-memory for (const id of ids) { - const metadata = await this.getMetadata(id) + // 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) } diff --git a/src/storage/adapters/opfsStorage.ts b/src/storage/adapters/opfsStorage.ts index cabbeb74..149993a5 100644 --- a/src/storage/adapters/opfsStorage.ts +++ b/src/storage/adapters/opfsStorage.ts @@ -201,12 +201,16 @@ export class OPFSStorage extends BaseStorage { await this.ensureInitialized() try { - // 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 serializableNoun = { - ...noun, + id: noun.id, + vector: noun.vector, connections: this.mapToObject(noun.connections, (set) => Array.from(set as Set) - ) + ), + level: noun.level || 0 + // NO metadata field - saved separately for scalability } // Use UUID-based sharding for nouns @@ -387,12 +391,15 @@ export class OPFSStorage extends BaseStorage { await this.ensureInitialized() try { - // Convert connections Map to a serializable format + // CRITICAL: Only save lightweight vector data (no metadata) + // Metadata is saved separately via saveVerbMetadata() (2-file system) const serializableEdge = { - ...edge, + id: edge.id, + vector: edge.vector, connections: this.mapToObject(edge.connections, (set) => Array.from(set as Set) ) + // NO metadata field - saved separately for scalability } // Use UUID-based sharding for verbs diff --git a/src/storage/adapters/s3CompatibleStorage.ts b/src/storage/adapters/s3CompatibleStorage.ts index 0b9e2e6f..0c8b3cae 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -959,11 +959,16 @@ export class S3CompatibleStorage extends BaseStorage { this.logger.trace(`Saving node ${node.id}`) // 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 = { - ...node, + 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 } // Import the PutObjectCommand only when needed @@ -1022,6 +1027,15 @@ export class S3CompatibleStorage extends BaseStorage { verifyError ) } + + // Increment noun count - always increment total, and increment by type if metadata exists + this.totalNounCount++ + const metadata = await this.getNounMetadata(node.id) + if (metadata && metadata.type) { + const currentCount = this.entityCounts.get(metadata.type) || 0 + this.entityCounts.set(metadata.type, currentCount + 1) + } + // Release backpressure on success this.releaseBackpressure(true, requestId) } catch (error) { @@ -1438,11 +1452,15 @@ export class S3CompatibleStorage extends BaseStorage { try { // Convert connections Map to a serializable format + // CRITICAL: Only save lightweight vector data (no metadata) + // Metadata is saved separately via saveVerbMetadata() (2-file system) const serializableEdge = { - ...edge, + id: edge.id, + vector: edge.vector, connections: this.mapToObject(edge.connections, (set) => Array.from(set as Set) ) + // NO metadata field - saved separately for scalability } // Import the PutObjectCommand only when needed @@ -1468,7 +1486,15 @@ export class S3CompatibleStorage extends BaseStorage { vector: edge.vector } }) - + + // Increment verb count - always increment total, and increment by type if metadata exists + this.totalVerbCount++ + const metadata = await this.getVerbMetadata(edge.id) + if (metadata && metadata.type) { + const currentCount = this.verbCounts.get(metadata.type) || 0 + this.verbCounts.set(metadata.type, currentCount + 1) + } + // Release backpressure on success this.releaseBackpressure(true, requestId) } catch (error) { @@ -2114,9 +2140,11 @@ export class S3CompatibleStorage extends BaseStorage { const batchPromises = batch.map(async (id) => { try { // Add timeout wrapper for individual metadata reads + // CRITICAL: Use getNounMetadata() instead of deprecated getMetadata() + // This ensures we fetch from the correct noun metadata store (2-file system) const metadata = await Promise.race([ - this.getMetadata(id), - new Promise((_, reject) => + this.getNounMetadata(id), + new Promise((_, reject) => setTimeout(() => reject(new Error('Metadata read timeout')), 5000) // 5 second timeout ) ]) diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 4e09e20c..a3a14017 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -169,12 +169,37 @@ export abstract class BaseStorage extends BaseStorageAdapter { */ public async saveNoun(noun: HNSWNoun): Promise { await this.ensureInitialized() + // Validate noun type before saving - storage boundary protection - const metadata = await this.getNounMetadata(noun.id) - if (metadata?.noun) { - validateNounType(metadata.noun) + if (noun.metadata?.noun) { + validateNounType(noun.metadata.noun) + } + + // Save both the HNSWNoun vector data and metadata separately (2-file system) + try { + // Save the lightweight HNSWNoun vector file first + await this.saveNoun_internal(noun) + + // Then save the metadata to separate file (if present) + if (noun.metadata) { + await this.saveNounMetadata(noun.id, noun.metadata) + } + } catch (error) { + console.error(`[ERROR] Failed to save noun ${noun.id}:`, error) + + // Attempt cleanup - remove noun file if metadata failed + try { + const nounExists = await this.getNoun_internal(noun.id) + if (nounExists) { + console.log(`[CLEANUP] Attempting to remove orphaned noun file ${noun.id}`) + await this.deleteNoun_internal(noun.id) + } + } catch (cleanupError) { + console.error(`[ERROR] Failed to cleanup orphaned noun ${noun.id}:`, cleanupError) + } + + throw new Error(`Failed to save noun ${noun.id}: ${error instanceof Error ? error.message : String(error)}`) } - return this.saveNoun_internal(noun) } /** @@ -200,7 +225,17 @@ export abstract class BaseStorage extends BaseStorageAdapter { */ public async deleteNoun(id: string): Promise { await this.ensureInitialized() - return this.deleteNoun_internal(id) + + // Delete both the vector file and metadata file (2-file system) + await this.deleteNoun_internal(id) + + // Delete metadata file (if it exists) + try { + await this.deleteNounMetadata(id) + } catch (error) { + // Ignore if metadata file doesn't exist + console.debug(`No metadata file to delete for noun ${id}`) + } } /** @@ -762,8 +797,17 @@ export abstract class BaseStorage extends BaseStorageAdapter { */ public async deleteVerb(id: string): Promise { await this.ensureInitialized() - return this.deleteVerb_internal(id) + // Delete both the vector file and metadata file (2-file system) + await this.deleteVerb_internal(id) + + // Delete metadata file (if it exists) + try { + await this.deleteVerbMetadata(id) + } catch (error) { + // Ignore if metadata file doesn't exist + console.debug(`No metadata file to delete for verb ${id}`) + } } /** * Get graph index (lazy initialization) @@ -887,6 +931,16 @@ export abstract class BaseStorage extends BaseStorageAdapter { return this.readObjectFromPath(keyInfo.fullPath) } + /** + * Delete noun metadata from storage + * Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded) + */ + public async deleteNounMetadata(id: string): Promise { + await this.ensureInitialized() + const keyInfo = this.analyzeKey(id, 'noun-metadata') + return this.deleteObjectFromPath(keyInfo.fullPath) + } + /** * Save verb metadata to storage * Routes to correct sharded location based on UUID