perf(storage): simplify cloud adapters to always-on write buffering
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 <noreply@anthropic.com>
This commit is contained in:
parent
6449bb1afe
commit
26510ce7b8
4 changed files with 58 additions and 262 deletions
|
|
@ -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<HNSWNode>
|
||||
|
|
@ -164,12 +161,7 @@ export class AzureBlobStorage extends BaseStorage {
|
|||
this.nounCacheManager = new CacheManager<HNSWNode>(options.cacheConfig)
|
||||
this.verbCacheManager = new CacheManager<Edge>(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<void> {
|
||||
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<void> {
|
||||
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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<HNSWNode>
|
||||
|
|
@ -177,12 +174,7 @@ export class GcsStorage extends BaseStorage {
|
|||
this.nounCacheManager = new CacheManager<HNSWNode>(options.cacheConfig)
|
||||
this.verbCacheManager = new CacheManager<Edge>(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<void> {
|
||||
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<void> {
|
||||
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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<HNSWNode>
|
||||
|
|
@ -171,12 +168,7 @@ export class R2Storage extends BaseStorage {
|
|||
})
|
||||
this.verbCacheManager = new CacheManager<Edge>(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<void> {
|
||||
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<void> {
|
||||
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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue