fix(storage): add universal batch metadata reading to ALL storage adapters

CRITICAL FIX: getMetadataBatch was only implemented in S3CompatibleStorage,
causing other adapters to fall back to individual calls = socket exhaustion\!

Universal Implementation:
- Add getMetadataBatch() to MemoryStorage (in-memory batch processing)
- Add getMetadataBatch() to FileSystemStorage (10 concurrent file reads)
- Add getMetadataBatch() to OPFSStorage (10 concurrent OPFS operations)
- Enhanced S3CompatibleStorage with adaptive delays and timeout handling

Enhanced Debugging:
- Log storage adapter type and batch availability
- Clear fallback warnings if batch processing unavailable
- Progress reporting with success rates
- Better timeout and error handling

This ensures socket exhaustion prevention works regardless of storage adapter.
Services using Memory/FileSystem/OPFS storage will now use batch processing
instead of 1400+ individual getMetadata() calls during initialization.

Expected result: Services initialize successfully across ALL storage types
This commit is contained in:
David Snelling 2025-08-07 12:14:23 -07:00
parent befa1c2c8d
commit 3456ba332a
5 changed files with 137 additions and 8 deletions

View file

@ -492,6 +492,45 @@ export class FileSystemStorage extends BaseStorage {
}
}
/**
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
* FileSystem implementation uses controlled concurrency to prevent too many file reads
*/
public async getMetadataBatch(ids: string[]): Promise<Map<string, any>> {
await this.ensureInitialized()
const results = new Map<string, any>()
const batchSize = 10 // Process 10 files at a time
// Process in batches to avoid overwhelming the filesystem
for (let i = 0; i < ids.length; i += batchSize) {
const batch = ids.slice(i, i + batchSize)
const batchPromises = batch.map(async (id) => {
try {
const metadata = await this.getMetadata(id)
return { id, metadata }
} catch (error) {
console.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
await new Promise(resolve => setImmediate(resolve))
}
return results
}
/**
* Save noun metadata to storage
*/

View file

@ -481,6 +481,25 @@ export class MemoryStorage extends BaseStorage {
return JSON.parse(JSON.stringify(metadata))
}
/**
* 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) {
const metadata = this.metadata.get(id)
if (metadata) {
// Deep clone to prevent mutation
results.set(id, JSON.parse(JSON.stringify(metadata)))
}
}
return results
}
/**
* Save noun metadata to storage
*/

View file

@ -624,6 +624,45 @@ export class OPFSStorage extends BaseStorage {
}
}
/**
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
* OPFS implementation uses controlled concurrency for file operations
*/
public async getMetadataBatch(ids: string[]): Promise<Map<string, any>> {
await this.ensureInitialized()
const results = new Map<string, any>()
const batchSize = 10 // Process 10 files at a time
// Process in batches to avoid overwhelming OPFS
for (let i = 0; i < ids.length; i += batchSize) {
const batch = ids.slice(i, i + batchSize)
const batchPromises = batch.map(async (id) => {
try {
const metadata = await this.getMetadata(id)
return { id, metadata }
} catch (error) {
console.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
await new Promise(resolve => setImmediate(resolve))
}
return results
}
/**
* Save verb metadata to storage
*/

View file

@ -1929,29 +1929,55 @@ export class S3CompatibleStorage extends BaseStorage {
for (let i = 0; i < ids.length; i += batchSize) {
const batch = ids.slice(i, i + batchSize)
// Process batch with concurrency control
// Process batch with concurrency control and enhanced retry logic
const batchPromises = batch.map(async (id) => {
try {
const metadata = await this.getMetadata(id)
// Add timeout wrapper for individual metadata reads
const metadata = await Promise.race([
this.getMetadata(id),
new Promise<null>((_, reject) =>
setTimeout(() => reject(new Error('Metadata read timeout')), 5000) // 5 second timeout
)
])
return { id, metadata }
} catch (error) {
// Don't fail entire batch if one metadata read fails
this.logger.debug(`Failed to read metadata for ${id}:`, error)
// Enhanced error handling for minor socket issues
const errorMessage = error instanceof Error ? error.message : String(error)
if (errorMessage.includes('timeout') || errorMessage.includes('ECONNRESET')) {
this.logger.debug(`⏰ Metadata timeout for ${id} (normal during initial indexing):`, errorMessage)
} else {
this.logger.debug(`Failed to read metadata for ${id}:`, error)
}
return { id, metadata: null }
}
})
const batchResults = await Promise.all(batchPromises)
// Add results to map
// Track error rates to adjust delays
let errorCount = 0
for (const { id, metadata } of batchResults) {
if (metadata !== null) {
results.set(id, metadata)
} else {
errorCount++
}
}
// Yield to prevent socket exhaustion between batches
await new Promise(resolve => setImmediate(resolve))
// Adaptive delay based on error rates (helps with S3 rate limiting)
const errorRate = errorCount / batch.length
if (errorRate > 0.5) {
// High error rate - add extra delay
await new Promise(resolve => setTimeout(resolve, 2000)) // 2 second delay
prodLog.debug(`🐌 High error rate (${(errorRate * 100).toFixed(1)}%) - adding extra delay`)
} else if (errorRate > 0.2) {
// Moderate error rate - add modest delay
await new Promise(resolve => setTimeout(resolve, 500)) // 500ms delay
prodLog.debug(`⚡ Moderate error rate (${(errorRate * 100).toFixed(1)}%) - adding modest delay`)
} else {
// Low error rate - normal yield
await new Promise(resolve => setImmediate(resolve))
}
}
return results