From 26510ce7b863db51e0339e8840e4f924410bd088 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Tue, 2 Dec 2025 13:43:04 -0800 Subject: [PATCH] perf(storage): simplify cloud adapters to always-on write buffering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes dynamic high-volume mode switching complexity from all cloud storage adapters (GCS, S3, R2, Azure). Write buffering is now always enabled for consistent, predictable performance. Changes: - Remove highVolumeMode, lastVolumeCheck, volumeCheckInterval properties - Remove checkVolumeMode() methods (~100 lines in S3 alone) - Remove BRAINY_FORCE_HIGH_VOLUME env var checks - Always use write buffer when available - Preserve v6.2.6 cache-before-buffer fix for consistency Benefits: - Consistent behavior regardless of load - Predictable performance characteristics - -204 lines of code complexity - Cloud storage always benefits from batching 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/storage/adapters/azureBlobStorage.ts | 70 +++--------- src/storage/adapters/gcsStorage.ts | 72 +++---------- src/storage/adapters/r2Storage.ts | 64 +++-------- src/storage/adapters/s3CompatibleStorage.ts | 114 ++------------------ 4 files changed, 58 insertions(+), 262 deletions(-) diff --git a/src/storage/adapters/azureBlobStorage.ts b/src/storage/adapters/azureBlobStorage.ts index 9b5c80dc..1dda69d2 100644 --- a/src/storage/adapters/azureBlobStorage.ts +++ b/src/storage/adapters/azureBlobStorage.ts @@ -104,11 +104,8 @@ export class AzureBlobStorage extends BaseStorage { // Request coalescer for deduplication private requestCoalescer: RequestCoalescer | null = null - // High-volume mode detection - private highVolumeMode = false - private lastVolumeCheck = 0 - private volumeCheckInterval = 1000 // Check every second - private forceHighVolumeMode = false // Environment variable override + // v6.2.7: Write buffering always enabled for consistent performance + // Removes dynamic mode switching complexity - cloud storage always benefits from batching // Multi-level cache manager for efficient data access private nounCacheManager: CacheManager @@ -164,12 +161,7 @@ export class AzureBlobStorage extends BaseStorage { this.nounCacheManager = new CacheManager(options.cacheConfig) this.verbCacheManager = new CacheManager(options.cacheConfig) - // Check for high-volume mode override - if (typeof process !== 'undefined' && process.env?.BRAINY_FORCE_HIGH_VOLUME === 'true') { - this.forceHighVolumeMode = true - this.highVolumeMode = true - prodLog.info('🚀 High-volume mode FORCED via BRAINY_FORCE_HIGH_VOLUME environment variable') - } + // v6.2.7: Write buffering always enabled - no env var check needed } /** @@ -460,32 +452,7 @@ export class AzureBlobStorage extends BaseStorage { } } - /** - * Check if high-volume mode should be enabled - */ - private checkVolumeMode(): void { - if (this.forceHighVolumeMode) { - return // Already forced on - } - - const now = Date.now() - if (now - this.lastVolumeCheck < this.volumeCheckInterval) { - return - } - - this.lastVolumeCheck = now - - // Enable high-volume mode if we have many pending operations - const shouldEnable = this.pendingOperations > 20 - - if (shouldEnable && !this.highVolumeMode) { - this.highVolumeMode = true - prodLog.info('🚀 High-volume mode ENABLED (pending operations:', this.pendingOperations, ')') - } else if (!shouldEnable && this.highVolumeMode && !this.forceHighVolumeMode) { - this.highVolumeMode = false - prodLog.info('🐌 High-volume mode DISABLED (pending operations:', this.pendingOperations, ')') - } - } + // v6.2.7: Removed checkVolumeMode() - write buffering always enabled for cloud storage /** * Flush noun buffer to Azure @@ -521,30 +488,25 @@ export class AzureBlobStorage extends BaseStorage { /** * Save a node to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() - // ALWAYS check if we should use high-volume mode (critical for detection) - this.checkVolumeMode() + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.nounWriteBuffer) { + this.logger.trace(`📝 BUFFERING: Adding noun ${node.id} to write buffer`) - // 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)`) - - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency - // Without this, add() returns but relate() can't find the entity (cloud storage production bug) + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency if (node.vector && Array.isArray(node.vector) && node.vector.length > 0) { this.nounCacheManager.set(node.id, node) } 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)`) } - // Direct write in normal mode + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveNodeDirect(node) } @@ -1030,25 +992,23 @@ export class AzureBlobStorage extends BaseStorage { /** * Save an edge to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() - // Check volume mode - this.checkVolumeMode() - - // Use write buffer in high-volume mode - if (this.highVolumeMode && this.verbWriteBuffer) { + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.verbWriteBuffer) { this.logger.trace(`📝 BUFFERING: Adding verb ${edge.id} to write buffer`) - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency this.verbCacheManager.set(edge.id, edge) await this.verbWriteBuffer.add(edge.id, edge) return } - // Direct write in normal mode + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveEdgeDirect(edge) } diff --git a/src/storage/adapters/gcsStorage.ts b/src/storage/adapters/gcsStorage.ts index d699e008..ac254485 100644 --- a/src/storage/adapters/gcsStorage.ts +++ b/src/storage/adapters/gcsStorage.ts @@ -109,11 +109,8 @@ export class GcsStorage extends BaseStorage { // Request coalescer for deduplication private requestCoalescer: RequestCoalescer | null = null - // High-volume mode detection - MUCH more aggressive - private highVolumeMode = false - private lastVolumeCheck = 0 - private volumeCheckInterval = 1000 // Check every second, not 5 - private forceHighVolumeMode = false // Environment variable override + // v6.2.7: Write buffering always enabled for consistent performance + // Removes dynamic mode switching complexity - cloud storage always benefits from batching // Multi-level cache manager for efficient data access private nounCacheManager: CacheManager @@ -177,12 +174,7 @@ export class GcsStorage extends BaseStorage { this.nounCacheManager = new CacheManager(options.cacheConfig) this.verbCacheManager = new CacheManager(options.cacheConfig) - // Check for high-volume mode override - if (typeof process !== 'undefined' && process.env?.BRAINY_FORCE_HIGH_VOLUME === 'true') { - this.forceHighVolumeMode = true - this.highVolumeMode = true - prodLog.info('🚀 High-volume mode FORCED via BRAINY_FORCE_HIGH_VOLUME environment variable') - } + // v6.2.7: Write buffering always enabled - no env var check needed } /** @@ -371,32 +363,7 @@ export class GcsStorage extends BaseStorage { } } - /** - * Check if high-volume mode should be enabled - */ - private checkVolumeMode(): void { - if (this.forceHighVolumeMode) { - return // Already forced on - } - - const now = Date.now() - if (now - this.lastVolumeCheck < this.volumeCheckInterval) { - return - } - - this.lastVolumeCheck = now - - // Enable high-volume mode if we have many pending operations - const shouldEnable = this.pendingOperations > 20 - - if (shouldEnable && !this.highVolumeMode) { - this.highVolumeMode = true - prodLog.info('🚀 High-volume mode ENABLED (pending operations:', this.pendingOperations, ')') - } else if (!shouldEnable && this.highVolumeMode && !this.forceHighVolumeMode) { - this.highVolumeMode = false - prodLog.info('🐌 High-volume mode DISABLED (pending operations:', this.pendingOperations, ')') - } - } + // v6.2.7: Removed checkVolumeMode() - write buffering always enabled for cloud storage /** * Flush noun buffer to GCS @@ -432,31 +399,25 @@ export class GcsStorage extends BaseStorage { /** * Save a node to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() - // ALWAYS check if we should use high-volume mode (critical for detection) - this.checkVolumeMode() + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.nounWriteBuffer) { + this.logger.trace(`📝 BUFFERING: Adding noun ${node.id} to write buffer`) - // 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)`) - - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency - // Without this, add() returns but relate() can't find the entity (GCS production bug) - // The buffer flushes asynchronously, but cache ensures immediate reads succeed + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency if (node.vector && Array.isArray(node.vector) && node.vector.length > 0) { this.nounCacheManager.set(node.id, node) } 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)`) } - // Direct write in normal mode + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveNodeDirect(node) } @@ -831,26 +792,23 @@ export class GcsStorage extends BaseStorage { /** * Save an edge to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() - // Check volume mode - this.checkVolumeMode() - - // Use write buffer in high-volume mode - if (this.highVolumeMode && this.verbWriteBuffer) { + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.verbWriteBuffer) { this.logger.trace(`📝 BUFFERING: Adding verb ${edge.id} to write buffer`) - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency - // Without this, relate() might not find the verb immediately after creation + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency this.verbCacheManager.set(edge.id, edge) await this.verbWriteBuffer.add(edge.id, edge) return } - // Direct write in normal mode + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveEdgeDirect(edge) } diff --git a/src/storage/adapters/r2Storage.ts b/src/storage/adapters/r2Storage.ts index 7a025773..49b6f8c0 100644 --- a/src/storage/adapters/r2Storage.ts +++ b/src/storage/adapters/r2Storage.ts @@ -112,11 +112,8 @@ export class R2Storage extends BaseStorage { // Request coalescer for deduplication private requestCoalescer: RequestCoalescer | null = null - // High-volume mode detection (R2-specific thresholds) - private highVolumeMode = false - private lastVolumeCheck = 0 - private volumeCheckInterval = 800 // Check more frequently on R2 - private forceHighVolumeMode = false + // v6.2.7: Write buffering always enabled for consistent performance + // Removes dynamic mode switching complexity - cloud storage always benefits from batching // Multi-level cache manager for efficient data access private nounCacheManager: CacheManager @@ -171,12 +168,7 @@ export class R2Storage extends BaseStorage { }) this.verbCacheManager = new CacheManager(options.cacheConfig) - // Check for high-volume mode override - if (typeof process !== 'undefined' && process.env?.BRAINY_FORCE_HIGH_VOLUME === 'true') { - this.forceHighVolumeMode = true - this.highVolumeMode = true - prodLog.info('🚀 R2: High-volume mode FORCED via environment variable') - } + // v6.2.7: Write buffering always enabled - no env var check needed } /** @@ -417,32 +409,7 @@ export class R2Storage extends BaseStorage { } } - /** - * Check if high-volume mode should be enabled - */ - private checkVolumeMode(): void { - if (this.forceHighVolumeMode) { - return - } - - const now = Date.now() - if (now - this.lastVolumeCheck < this.volumeCheckInterval) { - return - } - - this.lastVolumeCheck = now - - // R2 threshold: enable at 15 pending operations (lower than S3/GCS) - const shouldEnable = this.pendingOperations > 15 - - if (shouldEnable && !this.highVolumeMode) { - this.highVolumeMode = true - prodLog.info('🚀 R2: High-volume mode ENABLED (pending:', this.pendingOperations, ')') - } else if (!shouldEnable && this.highVolumeMode && !this.forceHighVolumeMode) { - this.highVolumeMode = false - prodLog.info('🐌 R2: High-volume mode DISABLED (pending:', this.pendingOperations, ')') - } - } + // v6.2.7: Removed checkVolumeMode() - write buffering always enabled for cloud storage /** * Flush noun buffer to R2 @@ -477,18 +444,16 @@ export class R2Storage extends BaseStorage { /** * Save a node to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() - this.checkVolumeMode() - - // Use write buffer in high-volume mode - if (this.highVolumeMode && this.nounWriteBuffer) { + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.nounWriteBuffer) { this.logger.trace(`📝 BUFFERING: Adding noun ${node.id} to write buffer`) - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency - // Without this, add() returns but relate() can't find the entity (cloud storage production bug) + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency if (node.vector && Array.isArray(node.vector) && node.vector.length > 0) { this.nounCacheManager.set(node.id, node) } @@ -497,7 +462,7 @@ export class R2Storage extends BaseStorage { return } - // Direct write in normal mode + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveNodeDirect(node) } @@ -761,18 +726,23 @@ export class R2Storage extends BaseStorage { // Verb storage methods (similar to noun methods - implementing key methods for space) + /** + * Save an edge to storage + * v6.2.7: Always uses write buffer for consistent performance + */ protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() - this.checkVolumeMode() - if (this.highVolumeMode && this.verbWriteBuffer) { - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.verbWriteBuffer) { + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency this.verbCacheManager.set(edge.id, edge) await this.verbWriteBuffer.add(edge.id, edge) return } + // Fallback to direct write if buffer not initialized (shouldn't happen after init) await this.saveEdgeDirect(edge) } diff --git a/src/storage/adapters/s3CompatibleStorage.ts b/src/storage/adapters/s3CompatibleStorage.ts index 19e0011d..9e4c4418 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -151,11 +151,8 @@ export class S3CompatibleStorage extends BaseStorage { // Request coalescer for deduplication private requestCoalescer: RequestCoalescer | null = null - // High-volume mode detection - MUCH more aggressive - private highVolumeMode = false - private lastVolumeCheck = 0 - private volumeCheckInterval = 1000 // Check every second, not 5 - private forceHighVolumeMode = false // Environment variable override + // v6.2.7: Write buffering always enabled for consistent performance + // Removes dynamic mode switching complexity - cloud storage always benefits from batching // Operation executors for timeout and retry handling private operationExecutors: StorageOperationExecutors @@ -710,92 +707,8 @@ export class S3CompatibleStorage extends BaseStorage { ) } - /** - * Check if we should enable high-volume mode - */ - private checkVolumeMode(): void { - const now = Date.now() - if (now - this.lastVolumeCheck < this.volumeCheckInterval) { - return - } - - this.lastVolumeCheck = now - - // Check environment variable override - const envThreshold = process.env.BRAINY_BUFFER_THRESHOLD - const threshold = envThreshold ? parseInt(envThreshold) : 0 // Default to 0 for immediate activation! - - // 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() - - // Reasonable high-volume detection - only activate under real load - const isTestEnvironment = process.env.NODE_ENV === 'test' - const explicitlyDisabled = process.env.BRAINY_FORCE_BUFFERING === 'false' - - // Use reasonable thresholds instead of emergency aggressive ones - const reasonableThreshold = Math.max(threshold, 10) // At least 10 pending operations - const highSocketUtilization = 0.8 // 80% socket utilization - const highRequestRate = 50 // 50 requests per second - const significantErrors = 5 // 5 consecutive errors - - const shouldEnableHighVolume = - !isTestEnvironment && // Disable in test environment - !explicitlyDisabled && // Allow explicit disabling - (this.forceHighVolumeMode || // Environment override - backpressureStatus.queueLength >= reasonableThreshold || // High queue backlog - socketMetrics.pendingRequests >= reasonableThreshold || // Many pending requests - this.pendingOperations >= reasonableThreshold || // Many pending ops - socketMetrics.socketUtilization >= highSocketUtilization || // High socket pressure - (socketMetrics.requestsPerSecond >= highRequestRate) || // High request rate - (this.consecutiveErrors >= significantErrors)) // Significant error pattern - - if (shouldEnableHighVolume && !this.highVolumeMode) { - this.highVolumeMode = true - 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(queueLength) - const stats = this.nounWriteBuffer.getStats() - this.logger.warn(` Noun Buffer: ${stats.bufferSize} items, ${stats.totalWrites} total writes`) - } - if (this.verbWriteBuffer) { - 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(queueLength) - const sizes = this.requestCoalescer.getQueueSizes() - this.logger.warn(` Coalescer: ${sizes.total} queued operations`) - } - - } else if (!shouldEnableHighVolume && this.highVolumeMode && !this.forceHighVolumeMode) { - this.highVolumeMode = false - 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)}%`) - } - } - + // v6.2.7: Removed checkVolumeMode() - write buffering always enabled for cloud storage + /** * Bulk write nouns to S3 */ @@ -1059,30 +972,25 @@ export class S3CompatibleStorage extends BaseStorage { /** * Save a node to storage + * v6.2.7: Always uses write buffer for consistent performance */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() - - // 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)`) - // v6.2.6: CRITICAL FIX - Populate cache BEFORE buffering for read-after-write consistency - // Without this, add() returns but relate() can't find the entity (cloud storage production bug) - // The buffer flushes asynchronously, but cache ensures immediate reads succeed + // v6.2.7: Always use write buffer - cloud storage benefits from batching + if (this.nounWriteBuffer) { + this.logger.trace(`📝 BUFFERING: Adding noun ${node.id} to write buffer`) + + // v6.2.6: Populate cache BEFORE buffering for read-after-write consistency if (node.vector && Array.isArray(node.vector) && node.vector.length > 0) { this.nounCacheManager.set(node.id, node) } 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)`) } + // Fallback to direct write if buffer not initialized (shouldn't happen after init) // Apply backpressure before starting operation const requestId = await this.applyBackpressure()