diff --git a/package-lock.json b/package-lock.json index be9fce16..8c0d74a3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@soulcraft/brainy", - "version": "0.54.1", + "version": "0.54.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@soulcraft/brainy", - "version": "0.54.1", + "version": "0.54.2", "license": "MIT", "dependencies": { "@aws-sdk/client-s3": "^3.540.0", diff --git a/package.json b/package.json index cd445c1f..f99215dd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@soulcraft/brainy", - "version": "0.54.1", + "version": "0.54.2", "description": "A vector graph database using HNSW indexing with Origin Private File System storage", "main": "dist/index.js", "module": "dist/index.js", diff --git a/src/storage/adapters/s3CompatibleStorage.ts b/src/storage/adapters/s3CompatibleStorage.ts index 5213217d..fb1ab43f 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -124,10 +124,11 @@ export class S3CompatibleStorage extends BaseStorage { // Request coalescer for deduplication private requestCoalescer: RequestCoalescer | null = null - // High-volume mode detection + // High-volume mode detection - MUCH more aggressive private highVolumeMode = false private lastVolumeCheck = 0 - private volumeCheckInterval = 5000 + private volumeCheckInterval = 1000 // Check every second, not 5 + private forceHighVolumeMode = false // Environment variable override // Operation executors for timeout and retry handling private operationExecutors: StorageOperationExecutors @@ -397,32 +398,67 @@ export class S3CompatibleStorage extends BaseStorage { this.lastVolumeCheck = now - // Check pending operations + // Check environment variable override + const envThreshold = process.env.BRAINY_BUFFER_THRESHOLD + const threshold = envThreshold ? parseInt(envThreshold) : 1 // Default to 1! + + // Force enable from environment + if (process.env.BRAINY_FORCE_BUFFERING === 'true') { + this.forceHighVolumeMode = true + } + + // Get metrics const backpressureStatus = this.backpressure.getStatus() const socketMetrics = this.socketManager.getMetrics() + // MUCH more aggressive detection - trigger on almost any load const shouldEnableHighVolume = - backpressureStatus.queueLength > 100 || - socketMetrics.pendingRequests > 50 || - this.pendingOperations > 20 + this.forceHighVolumeMode || // Environment override + backpressureStatus.queueLength > threshold || // Configurable threshold + socketMetrics.pendingRequests > threshold || // Socket pressure + this.pendingOperations > threshold || // Any pending ops + socketMetrics.socketUtilization > 0.1 || // Even 10% socket usage + (socketMetrics.requestsPerSecond > 10) || // High request rate + (this.consecutiveErrors > 0) // Any errors at all if (shouldEnableHighVolume && !this.highVolumeMode) { this.highVolumeMode = true - this.logger.info('Enabling high-volume mode for optimized batching') + this.logger.warn(`๐Ÿšจ HIGH-VOLUME MODE ACTIVATED ๐Ÿšจ`) + this.logger.warn(` Queue Length: ${backpressureStatus.queueLength}`) + this.logger.warn(` Pending Requests: ${socketMetrics.pendingRequests}`) + this.logger.warn(` Pending Operations: ${this.pendingOperations}`) + this.logger.warn(` Socket Utilization: ${(socketMetrics.socketUtilization * 100).toFixed(1)}%`) + this.logger.warn(` Requests/sec: ${socketMetrics.requestsPerSecond}`) + this.logger.warn(` Consecutive Errors: ${this.consecutiveErrors}`) + this.logger.warn(` Threshold: ${threshold}`) // Adjust buffer parameters for high volume + const queueLength = Math.max(backpressureStatus.queueLength, socketMetrics.pendingRequests, 100) + if (this.nounWriteBuffer) { - this.nounWriteBuffer.adjustForLoad(backpressureStatus.queueLength) + this.nounWriteBuffer.adjustForLoad(queueLength) + const stats = this.nounWriteBuffer.getStats() + this.logger.warn(` Noun Buffer: ${stats.bufferSize} items, ${stats.totalWrites} total writes`) } if (this.verbWriteBuffer) { - this.verbWriteBuffer.adjustForLoad(backpressureStatus.queueLength) + this.verbWriteBuffer.adjustForLoad(queueLength) + const stats = this.verbWriteBuffer.getStats() + this.logger.warn(` Verb Buffer: ${stats.bufferSize} items, ${stats.totalWrites} total writes`) } if (this.requestCoalescer) { - this.requestCoalescer.adjustParameters(backpressureStatus.queueLength) + this.requestCoalescer.adjustParameters(queueLength) + const sizes = this.requestCoalescer.getQueueSizes() + this.logger.warn(` Coalescer: ${sizes.total} queued operations`) } - } else if (!shouldEnableHighVolume && this.highVolumeMode) { + + } else if (!shouldEnableHighVolume && this.highVolumeMode && !this.forceHighVolumeMode) { this.highVolumeMode = false - this.logger.info('Disabling high-volume mode') + this.logger.info('โœ… High-volume mode deactivated - load normalized') + } + + // Log current status every 10 checks when in high-volume mode + if (this.highVolumeMode && (now % 10000) < this.volumeCheckInterval) { + this.logger.info(`๐Ÿ“Š High-volume mode status: Queue=${backpressureStatus.queueLength}, Pending=${socketMetrics.pendingRequests}, Sockets=${(socketMetrics.socketUtilization * 100).toFixed(1)}%`) } } @@ -698,13 +734,16 @@ export class S3CompatibleStorage extends BaseStorage { protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() - // Check if we should use high-volume mode + // ALWAYS check if we should use high-volume mode (critical for detection) this.checkVolumeMode() // Use write buffer in high-volume mode if (this.highVolumeMode && this.nounWriteBuffer) { + this.logger.trace(`๐Ÿ“ BUFFERING: Adding noun ${node.id} to write buffer (high-volume mode active)`) await this.nounWriteBuffer.add(node.id, node) return + } else if (!this.highVolumeMode) { + this.logger.trace(`๐Ÿ“ DIRECT WRITE: Saving noun ${node.id} directly (high-volume mode inactive)`) } // Apply backpressure before starting operation @@ -1126,13 +1165,16 @@ export class S3CompatibleStorage extends BaseStorage { protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() - // Check if we should use high-volume mode + // ALWAYS check if we should use high-volume mode (critical for detection) this.checkVolumeMode() // Use write buffer in high-volume mode if (this.highVolumeMode && this.verbWriteBuffer) { + this.logger.trace(`๐Ÿ“ BUFFERING: Adding verb ${edge.id} to write buffer (high-volume mode active)`) await this.verbWriteBuffer.add(edge.id, edge) return + } else if (!this.highVolumeMode) { + this.logger.trace(`๐Ÿ“ DIRECT WRITE: Saving verb ${edge.id} directly (high-volume mode inactive)`) } // Apply backpressure before starting operation diff --git a/src/utils/writeBuffer.ts b/src/utils/writeBuffer.ts index b99734d6..b62f98d8 100644 --- a/src/utils/writeBuffer.ts +++ b/src/utils/writeBuffer.ts @@ -31,10 +31,10 @@ export class WriteBuffer { // Buffer storage private buffer = new Map>() - // Configuration - private maxBufferSize = 1000 // Maximum items before forced flush - private flushInterval = 1000 // Flush every second - private minFlushSize = 100 // Minimum items to flush (unless timeout) + // Configuration - More aggressive for high volume + private maxBufferSize = 2000 // Allow larger buffers + private flushInterval = 500 // Flush more frequently (0.5 seconds) + private minFlushSize = 50 // Lower minimum to flush sooner private maxRetries = 3 // Maximum retry attempts // State @@ -115,6 +115,11 @@ export class WriteBuffer { this.totalWrites++ + // Log buffer growth periodically + if (this.totalWrites % 100 === 0) { + this.logger.info(`๐Ÿ“ˆ BUFFER GROWTH: ${this.buffer.size} ${this.type} items buffered (${this.totalWrites} total writes, ${this.duplicatesRemoved} deduplicated)`) + } + // Check if we should flush this.checkFlush() } @@ -203,7 +208,7 @@ export class WriteBuffer { this.buffer.delete(id) } - this.logger.debug(`Flushing ${itemsToFlush.size} ${this.type} items - reason: ${reason}`) + this.logger.warn(`๐Ÿ”„ BUFFERING: Flushing ${itemsToFlush.size} ${this.type} items (buffer had ${this.buffer.size + itemsToFlush.size}) - reason: ${reason}`) try { // Request permission from backpressure system @@ -220,7 +225,7 @@ export class WriteBuffer { this.lastFlush = Date.now() const duration = Date.now() - startTime - this.logger.info(`Flushed ${itemsToFlush.size} items in ${duration}ms`) + this.logger.warn(`๐Ÿš€ BATCH FLUSH: ${itemsToFlush.size} ${this.type} items โ†’ 1 bulk S3 operation (${duration}ms, reason: ${reason})`) return { successful: itemsToFlush.size, diff --git a/tests/emergency-high-volume-test.js b/tests/emergency-high-volume-test.js new file mode 100644 index 00000000..d2835df6 --- /dev/null +++ b/tests/emergency-high-volume-test.js @@ -0,0 +1,81 @@ +/** + * Emergency test to verify high-volume buffering activates correctly + * Simulates the exact bluesky-package scenario + */ + +const { BrainyData } = await import('../dist/index.js') + +console.log('๐Ÿงช Testing high-volume buffering activation...') + +// Create Brainy with memory storage for testing +const brainy = new BrainyData({ + dimensions: 384, + storage: { + type: 'memory' + } +}) + +console.log('๐Ÿ“Š Starting rapid fire operations (simulating bluesky firehose)...') + +// Simulate the exact pattern that's failing +let operationCount = 0 +const startTime = Date.now() + +// Fire off 1000 operations without awaiting (like bluesky firehose) +const promises = [] +for (let i = 0; i < 1000; i++) { + const promise = brainy.add({ + id: `firehose-${i}`, + data: { + type: 'bluesky_post', + content: `Message ${i} from firehose`, + timestamp: Date.now() + }, + metadata: { + source: 'bluesky', + batch: Math.floor(i / 100) + } + }).catch(error => { + console.error(`Failed to add item ${i}:`, error.message) + return null + }) + + promises.push(promise) + operationCount++ + + // Log every 100 operations + if (i > 0 && i % 100 === 0) { + console.log(`๐Ÿ“ˆ Fired ${i} operations (not awaited)`) + // Small delay to allow buffer detection + await new Promise(resolve => setTimeout(resolve, 10)) + } +} + +console.log(`๐Ÿš€ All ${operationCount} operations fired! Now waiting for completion...`) + +// Wait for all operations to complete +const results = await Promise.all(promises) +const successful = results.filter(r => r !== null).length +const failed = results.filter(r => r === null).length + +const duration = Date.now() - startTime +console.log(`\nโœ… Test completed in ${duration}ms`) +console.log(` Successful: ${successful}/${operationCount}`) +console.log(` Failed: ${failed}/${operationCount}`) + +if (successful < operationCount * 0.9) { + console.error('โŒ Test failed - too many operations failed') + process.exit(1) +} else { + console.log('โœ… Test passed - high-volume operations completed successfully') +} + +// Check if we have data +const searchResults = await brainy.search('bluesky', { k: 5 }) +console.log(`๐Ÿ” Search results: ${searchResults.length} items found`) + +console.log('\n๐ŸŽฏ Key things to check in logs:') +console.log(' 1. "๐Ÿšจ HIGH-VOLUME MODE ACTIVATED ๐Ÿšจ" message') +console.log(' 2. "๐Ÿ“ BUFFERING: Adding noun/verb to write buffer" messages') +console.log(' 3. "๐Ÿš€ BATCH FLUSH:" messages showing bulk operations') +console.log(' 4. "๐Ÿ“ˆ BUFFER GROWTH:" messages showing buffer accumulation') \ No newline at end of file