diff --git a/package-lock.json b/package-lock.json index f7c3a436..be9fce16 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@soulcraft/brainy", - "version": "0.54.0", + "version": "0.54.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@soulcraft/brainy", - "version": "0.54.0", + "version": "0.54.1", "license": "MIT", "dependencies": { "@aws-sdk/client-s3": "^3.540.0", diff --git a/package.json b/package.json index 6317baca..cd445c1f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@soulcraft/brainy", - "version": "0.54.0", + "version": "0.54.1", "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 067dac2b..5213217d 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -25,6 +25,8 @@ import { CacheManager } from '../cacheManager.js' import { createModuleLogger } from '../../utils/logger.js' import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js' import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js' +import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js' +import { getCoalescer, RequestCoalescer } from '../../utils/requestCoalescer.js' // Type aliases for better readability type HNSWNode = HNSWNoun @@ -114,6 +116,18 @@ export class S3CompatibleStorage extends BaseStorage { // Adaptive backpressure for automatic flow control private backpressure = getGlobalBackpressure() + + // Write buffers for bulk operations + private nounWriteBuffer: WriteBuffer | null = null + private verbWriteBuffer: WriteBuffer | null = null + + // Request coalescer for deduplication + private requestCoalescer: RequestCoalescer | null = null + + // High-volume mode detection + private highVolumeMode = false + private lastVolumeCheck = 0 + private volumeCheckInterval = 5000 // Operation executors for timeout and retry handling private operationExecutors: StorageOperationExecutors @@ -314,6 +328,12 @@ export class S3CompatibleStorage extends BaseStorage { this.nounCacheManager.setStorageAdapters(nounStorageAdapter, nounStorageAdapter) this.verbCacheManager.setStorageAdapters(verbStorageAdapter, verbStorageAdapter) + // Initialize write buffers for high-volume scenarios + this.initializeBuffers() + + // Initialize request coalescer + this.initializeCoalescer() + this.isInitialized = true this.logger.info(`Initialized ${this.serviceType} storage with bucket ${this.bucketName}`) } catch (error) { @@ -324,6 +344,270 @@ export class S3CompatibleStorage extends BaseStorage { } } + /** + * Initialize write buffers for high-volume scenarios + */ + private initializeBuffers(): void { + const storageId = `${this.serviceType}-${this.bucketName}` + + // Create noun write buffer + this.nounWriteBuffer = getWriteBuffer( + `${storageId}-nouns`, + 'noun', + async (items) => { + // Bulk write nouns to S3 + await this.bulkWriteNouns(items) + } + ) + + // Create verb write buffer + this.verbWriteBuffer = getWriteBuffer( + `${storageId}-verbs`, + 'verb', + async (items) => { + // Bulk write verbs to S3 + await this.bulkWriteVerbs(items) + } + ) + } + + /** + * Initialize request coalescer + */ + private initializeCoalescer(): void { + const storageId = `${this.serviceType}-${this.bucketName}` + + this.requestCoalescer = getCoalescer( + storageId, + async (batch) => { + // Process coalesced operations + await this.processCoalescedBatch(batch) + } + ) + } + + /** + * 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 pending operations + const backpressureStatus = this.backpressure.getStatus() + const socketMetrics = this.socketManager.getMetrics() + + const shouldEnableHighVolume = + backpressureStatus.queueLength > 100 || + socketMetrics.pendingRequests > 50 || + this.pendingOperations > 20 + + if (shouldEnableHighVolume && !this.highVolumeMode) { + this.highVolumeMode = true + this.logger.info('Enabling high-volume mode for optimized batching') + + // Adjust buffer parameters for high volume + if (this.nounWriteBuffer) { + this.nounWriteBuffer.adjustForLoad(backpressureStatus.queueLength) + } + if (this.verbWriteBuffer) { + this.verbWriteBuffer.adjustForLoad(backpressureStatus.queueLength) + } + if (this.requestCoalescer) { + this.requestCoalescer.adjustParameters(backpressureStatus.queueLength) + } + } else if (!shouldEnableHighVolume && this.highVolumeMode) { + this.highVolumeMode = false + this.logger.info('Disabling high-volume mode') + } + } + + /** + * Bulk write nouns to S3 + */ + private async bulkWriteNouns(items: Map): Promise { + const { PutObjectCommand } = await import('@aws-sdk/client-s3') + + // Process in parallel with limited concurrency + const promises: Promise[] = [] + const batchSize = 10 // Process 10 at a time + const entries = Array.from(items.entries()) + + for (let i = 0; i < entries.length; i += batchSize) { + const batch = entries.slice(i, i + batchSize) + + const batchPromise = Promise.all( + batch.map(async ([id, node]) => { + const serializableNode = { + ...node, + connections: this.mapToObject(node.connections, (set) => + Array.from(set as Set) + ) + } + + const key = `${this.nounPrefix}${id}.json` + const body = JSON.stringify(serializableNode, null, 2) + + await this.s3Client!.send( + new PutObjectCommand({ + Bucket: this.bucketName, + Key: key, + Body: body, + ContentType: 'application/json' + }) + ) + }) + ).then(() => {}) // Convert Promise to Promise + + promises.push(batchPromise) + } + + await Promise.all(promises) + } + + /** + * Bulk write verbs to S3 + */ + private async bulkWriteVerbs(items: Map): Promise { + const { PutObjectCommand } = await import('@aws-sdk/client-s3') + + // Process in parallel with limited concurrency + const promises: Promise[] = [] + const batchSize = 10 + const entries = Array.from(items.entries()) + + for (let i = 0; i < entries.length; i += batchSize) { + const batch = entries.slice(i, i + batchSize) + + const batchPromise = Promise.all( + batch.map(async ([id, edge]) => { + const serializableEdge = { + ...edge, + connections: this.mapToObject(edge.connections, (set) => + Array.from(set as Set) + ) + } + + const key = `${this.verbPrefix}${id}.json` + const body = JSON.stringify(serializableEdge, null, 2) + + await this.s3Client!.send( + new PutObjectCommand({ + Bucket: this.bucketName, + Key: key, + Body: body, + ContentType: 'application/json' + }) + ) + }) + ).then(() => {}) // Convert Promise to Promise + + promises.push(batchPromise) + } + + await Promise.all(promises) + } + + /** + * Process coalesced batch of operations + */ + private async processCoalescedBatch(batch: any[]): Promise { + // Group operations by type + const writes: any[] = [] + const reads: any[] = [] + const deletes: any[] = [] + + for (const op of batch) { + if (op.type === 'write') { + writes.push(op) + } else if (op.type === 'read') { + reads.push(op) + } else if (op.type === 'delete') { + deletes.push(op) + } + } + + // Process in order: deletes, writes, reads + if (deletes.length > 0) { + await this.processBulkDeletes(deletes) + } + if (writes.length > 0) { + await this.processBulkWrites(writes) + } + if (reads.length > 0) { + await this.processBulkReads(reads) + } + } + + /** + * Process bulk deletes + */ + private async processBulkDeletes(deletes: any[]): Promise { + const { DeleteObjectCommand } = await import('@aws-sdk/client-s3') + + await Promise.all( + deletes.map(async (op) => { + await this.s3Client!.send( + new DeleteObjectCommand({ + Bucket: this.bucketName, + Key: op.key + }) + ) + }) + ) + } + + /** + * Process bulk writes + */ + private async processBulkWrites(writes: any[]): Promise { + const { PutObjectCommand } = await import('@aws-sdk/client-s3') + + await Promise.all( + writes.map(async (op) => { + await this.s3Client!.send( + new PutObjectCommand({ + Bucket: this.bucketName, + Key: op.key, + Body: JSON.stringify(op.data), + ContentType: 'application/json' + }) + ) + }) + ) + } + + /** + * Process bulk reads + */ + private async processBulkReads(reads: any[]): Promise { + const { GetObjectCommand } = await import('@aws-sdk/client-s3') + + await Promise.all( + reads.map(async (op) => { + try { + const response = await this.s3Client!.send( + new GetObjectCommand({ + Bucket: this.bucketName, + Key: op.key + }) + ) + + if (response.Body) { + const data = await response.Body.transformToString() + op.data = JSON.parse(data) + } + } catch (error) { + op.data = null + } + }) + ) + } + /** * Dynamically adjust batch size based on memory pressure and error rates */ @@ -413,6 +697,15 @@ export class S3CompatibleStorage extends BaseStorage { */ protected async saveNode(node: HNSWNode): Promise { await this.ensureInitialized() + + // Check if we should use high-volume mode + this.checkVolumeMode() + + // Use write buffer in high-volume mode + if (this.highVolumeMode && this.nounWriteBuffer) { + await this.nounWriteBuffer.add(node.id, node) + return + } // Apply backpressure before starting operation const requestId = await this.applyBackpressure() @@ -832,6 +1125,15 @@ export class S3CompatibleStorage extends BaseStorage { */ protected async saveEdge(edge: Edge): Promise { await this.ensureInitialized() + + // Check if we should use high-volume mode + this.checkVolumeMode() + + // Use write buffer in high-volume mode + if (this.highVolumeMode && this.verbWriteBuffer) { + await this.verbWriteBuffer.add(edge.id, edge) + return + } // Apply backpressure before starting operation const requestId = await this.applyBackpressure() diff --git a/src/utils/requestCoalescer.ts b/src/utils/requestCoalescer.ts new file mode 100644 index 00000000..290f3b76 --- /dev/null +++ b/src/utils/requestCoalescer.ts @@ -0,0 +1,398 @@ +/** + * Request Coalescer + * Batches and deduplicates operations to reduce S3 API calls + * Automatically flushes based on size, time, or pressure + */ + +import { createModuleLogger } from './logger.js' + +interface CoalescedOperation { + type: 'write' | 'read' | 'delete' + key: string + data?: any + resolve: (value: any) => void + reject: (error: any) => void + timestamp: number +} + +interface BatchStats { + totalOperations: number + coalescedOperations: number + deduplicated: number + batchesProcessed: number + averageBatchSize: number +} + +/** + * Coalesces multiple operations into efficient batches + */ +export class RequestCoalescer { + private logger = createModuleLogger('RequestCoalescer') + + // Operation queues by type + private writeQueue = new Map() + private readQueue = new Map() + private deleteQueue = new Map() + + // Batch configuration + private maxBatchSize = 100 + private maxBatchAge = 100 // ms - flush quickly under load + private minBatchSize = 10 // Don't flush until we have enough + + // Flush timers + private flushTimer: NodeJS.Timeout | null = null + private lastFlush = Date.now() + + // Statistics + private stats: BatchStats = { + totalOperations: 0, + coalescedOperations: 0, + deduplicated: 0, + batchesProcessed: 0, + averageBatchSize: 0 + } + + // Processor function + private processor: (batch: CoalescedOperation[]) => Promise + + constructor( + processor: (batch: CoalescedOperation[]) => Promise, + options?: { + maxBatchSize?: number + maxBatchAge?: number + minBatchSize?: number + } + ) { + this.processor = processor + + if (options) { + this.maxBatchSize = options.maxBatchSize || this.maxBatchSize + this.maxBatchAge = options.maxBatchAge || this.maxBatchAge + this.minBatchSize = options.minBatchSize || this.minBatchSize + } + } + + /** + * Add a write operation to be coalesced + */ + public async write(key: string, data: any): Promise { + return new Promise((resolve, reject) => { + // Check if we already have a pending write for this key + const existing = this.writeQueue.get(key) + + if (existing && existing.length > 0) { + // Replace the data but resolve all promises + const last = existing[existing.length - 1] + last.data = data // Use latest data + + // Add this promise to be resolved + existing.push({ + type: 'write', + key, + data, + resolve, + reject, + timestamp: Date.now() + }) + + this.stats.deduplicated++ + } else { + // New write operation + this.writeQueue.set(key, [{ + type: 'write', + key, + data, + resolve, + reject, + timestamp: Date.now() + }]) + } + + this.stats.totalOperations++ + this.checkFlush() + }) + } + + /** + * Add a read operation to be coalesced + */ + public async read(key: string): Promise { + return new Promise((resolve, reject) => { + // Check if we already have a pending read for this key + const existing = this.readQueue.get(key) + + if (existing && existing.length > 0) { + // Coalesce with existing read + existing.push({ + type: 'read', + key, + resolve, + reject, + timestamp: Date.now() + }) + + this.stats.deduplicated++ + } else { + // New read operation + this.readQueue.set(key, [{ + type: 'read', + key, + resolve, + reject, + timestamp: Date.now() + }]) + } + + this.stats.totalOperations++ + this.checkFlush() + }) + } + + /** + * Add a delete operation to be coalesced + */ + public async delete(key: string): Promise { + return new Promise((resolve, reject) => { + // Cancel any pending writes for this key + if (this.writeQueue.has(key)) { + const writes = this.writeQueue.get(key)! + writes.forEach(op => op.reject(new Error('Cancelled by delete'))) + this.writeQueue.delete(key) + this.stats.deduplicated += writes.length + } + + // Cancel any pending reads for this key + if (this.readQueue.has(key)) { + const reads = this.readQueue.get(key)! + reads.forEach(op => op.resolve(null)) // Return null for deleted items + this.readQueue.delete(key) + this.stats.deduplicated += reads.length + } + + // Check if we already have a pending delete + const existing = this.deleteQueue.get(key) + + if (existing && existing.length > 0) { + // Coalesce with existing delete + existing.push({ + type: 'delete', + key, + resolve, + reject, + timestamp: Date.now() + }) + + this.stats.deduplicated++ + } else { + // New delete operation + this.deleteQueue.set(key, [{ + type: 'delete', + key, + resolve, + reject, + timestamp: Date.now() + }]) + } + + this.stats.totalOperations++ + this.checkFlush() + }) + } + + /** + * Check if we should flush the queues + */ + private checkFlush(): void { + const totalSize = this.writeQueue.size + this.readQueue.size + this.deleteQueue.size + const now = Date.now() + const age = now - this.lastFlush + + // Immediate flush conditions + if (totalSize >= this.maxBatchSize) { + this.flush('size_limit') + return + } + + // Age-based flush + if (age >= this.maxBatchAge && totalSize >= this.minBatchSize) { + this.flush('age_limit') + return + } + + // Schedule a flush if not already scheduled + if (!this.flushTimer && totalSize > 0) { + const delay = Math.max(10, this.maxBatchAge - age) + this.flushTimer = setTimeout(() => { + this.flush('timer') + }, delay) + } + } + + /** + * Flush all queued operations + */ + public async flush(reason: string = 'manual'): Promise { + // Clear timer + if (this.flushTimer) { + clearTimeout(this.flushTimer) + this.flushTimer = null + } + + // Collect all operations into a single batch + const batch: CoalescedOperation[] = [] + + // Process deletes first (highest priority) + this.deleteQueue.forEach((ops) => { + // Only take the first operation per key (others are duplicates) + if (ops.length > 0) { + batch.push(ops[0]) + this.stats.coalescedOperations += ops.length + } + }) + + // Then writes + this.writeQueue.forEach((ops) => { + if (ops.length > 0) { + // Use the last write (most recent data) + const lastWrite = ops[ops.length - 1] + batch.push(lastWrite) + this.stats.coalescedOperations += ops.length + } + }) + + // Then reads + this.readQueue.forEach((ops) => { + if (ops.length > 0) { + batch.push(ops[0]) + this.stats.coalescedOperations += ops.length + } + }) + + // Clear queues + const allOps = [ + ...Array.from(this.deleteQueue.values()).flat(), + ...Array.from(this.writeQueue.values()).flat(), + ...Array.from(this.readQueue.values()).flat() + ] + + this.deleteQueue.clear() + this.writeQueue.clear() + this.readQueue.clear() + + if (batch.length === 0) { + return + } + + // Update stats + this.stats.batchesProcessed++ + this.stats.averageBatchSize = + (this.stats.averageBatchSize * (this.stats.batchesProcessed - 1) + batch.length) / + this.stats.batchesProcessed + + this.logger.debug(`Flushing batch of ${batch.length} operations (${allOps.length} total) - reason: ${reason}`) + + // Process the batch + try { + await this.processor(batch) + + // Resolve all promises + allOps.forEach(op => { + if (op.type === 'read') { + // Find the result for this read + const result = batch.find(b => b.key === op.key && b.type === 'read') + op.resolve(result?.data || null) + } else { + op.resolve(undefined) + } + }) + } catch (error) { + // Reject all promises + allOps.forEach(op => op.reject(error)) + + this.logger.error('Batch processing failed:', error) + } + + this.lastFlush = Date.now() + } + + /** + * Get current statistics + */ + public getStats(): BatchStats { + return { ...this.stats } + } + + /** + * Get current queue sizes + */ + public getQueueSizes(): { + writes: number + reads: number + deletes: number + total: number + } { + return { + writes: this.writeQueue.size, + reads: this.readQueue.size, + deletes: this.deleteQueue.size, + total: this.writeQueue.size + this.readQueue.size + this.deleteQueue.size + } + } + + /** + * Adjust batch parameters based on load + */ + public adjustParameters(pending: number): void { + if (pending > 10000) { + // Extreme load - batch aggressively + this.maxBatchSize = 500 + this.maxBatchAge = 50 + this.minBatchSize = 50 + } else if (pending > 1000) { + // High load - larger batches + this.maxBatchSize = 200 + this.maxBatchAge = 100 + this.minBatchSize = 20 + } else if (pending > 100) { + // Moderate load + this.maxBatchSize = 100 + this.maxBatchAge = 200 + this.minBatchSize = 10 + } else { + // Low load - optimize for latency + this.maxBatchSize = 50 + this.maxBatchAge = 500 + this.minBatchSize = 5 + } + } + + /** + * Force immediate flush of all operations + */ + public async forceFlush(): Promise { + await this.flush('force') + } +} + +// Global coalescer instances by storage type +const coalescers = new Map() + +/** + * Get or create a coalescer for a storage instance + */ +export function getCoalescer( + storageId: string, + processor: (batch: any[]) => Promise +): RequestCoalescer { + if (!coalescers.has(storageId)) { + coalescers.set(storageId, new RequestCoalescer(processor)) + } + return coalescers.get(storageId)! +} + +/** + * Clear all coalescers + */ +export function clearCoalescers(): void { + coalescers.clear() +} \ No newline at end of file diff --git a/src/utils/writeBuffer.ts b/src/utils/writeBuffer.ts new file mode 100644 index 00000000..b99734d6 --- /dev/null +++ b/src/utils/writeBuffer.ts @@ -0,0 +1,406 @@ +/** + * Write Buffer + * Accumulates writes and flushes them in bulk to reduce S3 operations + * Implements intelligent deduplication and compression + */ + +import { HNSWNoun, HNSWVerb } from '../coreTypes.js' +import { createModuleLogger } from './logger.js' +import { getGlobalBackpressure } from './adaptiveBackpressure.js' + +interface BufferedWrite { + id: string + data: T + timestamp: number + type: 'noun' | 'verb' | 'metadata' + retryCount: number +} + +interface FlushResult { + successful: number + failed: number + duration: number +} + +/** + * High-performance write buffer for bulk operations + */ +export class WriteBuffer { + private logger = createModuleLogger('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) + private maxRetries = 3 // Maximum retry attempts + + // State + private flushTimer: NodeJS.Timeout | null = null + private isFlushing = false + private lastFlush = Date.now() + private pendingFlush: Promise | null = null + + // Statistics + private totalWrites = 0 + private totalFlushes = 0 + private failedWrites = 0 + private duplicatesRemoved = 0 + + // Write function + private writeFunction: (items: Map) => Promise + private type: 'noun' | 'verb' | 'metadata' + + // Backpressure integration + private backpressure = getGlobalBackpressure() + + constructor( + type: 'noun' | 'verb' | 'metadata', + writeFunction: (items: Map) => Promise, + options?: { + maxBufferSize?: number + flushInterval?: number + minFlushSize?: number + } + ) { + this.type = type + this.writeFunction = writeFunction + + if (options) { + this.maxBufferSize = options.maxBufferSize || this.maxBufferSize + this.flushInterval = options.flushInterval || this.flushInterval + this.minFlushSize = options.minFlushSize || this.minFlushSize + } + + // Start periodic flush + this.startPeriodicFlush() + } + + /** + * Add item to buffer + */ + public async add(id: string, data: T): Promise { + // Check if we're already at capacity + if (this.buffer.size >= this.maxBufferSize) { + // Wait for current flush to complete + if (this.pendingFlush) { + await this.pendingFlush + } + + // Force flush if still at capacity + if (this.buffer.size >= this.maxBufferSize) { + await this.flush('capacity') + } + } + + // Check for duplicate and update if newer + const existing = this.buffer.get(id) + if (existing) { + // Update with newer data + existing.data = data + existing.timestamp = Date.now() + this.duplicatesRemoved++ + } else { + // Add new item + this.buffer.set(id, { + id, + data, + timestamp: Date.now(), + type: this.type, + retryCount: 0 + }) + } + + this.totalWrites++ + + // Check if we should flush + this.checkFlush() + } + + /** + * Check if we should flush + */ + private checkFlush(): void { + const bufferSize = this.buffer.size + const timeSinceFlush = Date.now() - this.lastFlush + + // Immediate flush conditions + if (bufferSize >= this.maxBufferSize) { + this.flush('size') + return + } + + // Time-based flush with minimum size + if (timeSinceFlush >= this.flushInterval && bufferSize >= this.minFlushSize) { + this.flush('time') + return + } + + // Adaptive flush based on system load + const backpressureStatus = this.backpressure.getStatus() + if (backpressureStatus.queueLength > 1000 && bufferSize > 10) { + // System under pressure - flush smaller batches more frequently + this.flush('pressure') + } + } + + /** + * Flush buffer to storage + */ + public async flush(reason: string = 'manual'): Promise { + // Prevent concurrent flushes + if (this.isFlushing) { + if (this.pendingFlush) { + return this.pendingFlush + } + return { successful: 0, failed: 0, duration: 0 } + } + + // Nothing to flush + if (this.buffer.size === 0) { + return { successful: 0, failed: 0, duration: 0 } + } + + this.isFlushing = true + const startTime = Date.now() + + // Create flush promise + this.pendingFlush = this.doFlush(reason, startTime) + + try { + const result = await this.pendingFlush + return result + } finally { + this.isFlushing = false + this.pendingFlush = null + } + } + + /** + * Perform the actual flush + */ + private async doFlush(reason: string, startTime: number): Promise { + const itemsToFlush = new Map() + const flushingItems = new Map>() + + // Take items from buffer + let count = 0 + for (const [id, item] of this.buffer.entries()) { + itemsToFlush.set(id, item.data) + flushingItems.set(id, item) + count++ + + // Limit batch size for better performance + if (count >= 500) { + break + } + } + + // Remove from buffer + for (const id of itemsToFlush.keys()) { + this.buffer.delete(id) + } + + this.logger.debug(`Flushing ${itemsToFlush.size} ${this.type} items - reason: ${reason}`) + + try { + // Request permission from backpressure system + const opId = `flush-${Date.now()}` + await this.backpressure.requestPermission(opId, 2) // Higher priority + + try { + // Perform bulk write + await this.writeFunction(itemsToFlush) + + // Success + this.backpressure.releasePermission(opId, true) + this.totalFlushes++ + this.lastFlush = Date.now() + + const duration = Date.now() - startTime + this.logger.info(`Flushed ${itemsToFlush.size} items in ${duration}ms`) + + return { + successful: itemsToFlush.size, + failed: 0, + duration + } + } catch (error) { + // Release with error + this.backpressure.releasePermission(opId, false) + throw error + } + } catch (error) { + this.logger.error(`Flush failed: ${error}`) + + // Put items back with retry count + for (const [id, item] of flushingItems.entries()) { + item.retryCount++ + + if (item.retryCount < this.maxRetries) { + // Put back for retry + this.buffer.set(id, item) + } else { + // Max retries exceeded + this.failedWrites++ + this.logger.error(`Max retries exceeded for ${this.type} ${id}`) + } + } + + const duration = Date.now() - startTime + + return { + successful: 0, + failed: itemsToFlush.size, + duration + } + } + } + + /** + * Start periodic flush timer + */ + private startPeriodicFlush(): void { + if (this.flushTimer) { + return + } + + this.flushTimer = setInterval(() => { + if (this.buffer.size > 0) { + const timeSinceFlush = Date.now() - this.lastFlush + + // Flush if we have items and enough time has passed + if (timeSinceFlush >= this.flushInterval) { + this.flush('periodic').catch(error => { + this.logger.error('Periodic flush failed:', error) + }) + } + } + }, Math.min(100, this.flushInterval / 2)) + } + + /** + * Stop periodic flush timer + */ + public stop(): void { + if (this.flushTimer) { + clearInterval(this.flushTimer) + this.flushTimer = null + } + } + + /** + * Force flush all pending writes + */ + public async forceFlush(): Promise { + // Flush everything regardless of size + const oldMinSize = this.minFlushSize + this.minFlushSize = 0 + + try { + const result = await this.flush('force') + + // Flush any remaining items + while (this.buffer.size > 0) { + const additionalResult = await this.flush('force-remaining') + result.successful += additionalResult.successful + result.failed += additionalResult.failed + result.duration += additionalResult.duration + } + + return result + } finally { + this.minFlushSize = oldMinSize + } + } + + /** + * Get buffer statistics + */ + public getStats(): { + bufferSize: number + totalWrites: number + totalFlushes: number + failedWrites: number + duplicatesRemoved: number + avgFlushSize: number + } { + return { + bufferSize: this.buffer.size, + totalWrites: this.totalWrites, + totalFlushes: this.totalFlushes, + failedWrites: this.failedWrites, + duplicatesRemoved: this.duplicatesRemoved, + avgFlushSize: this.totalFlushes > 0 ? this.totalWrites / this.totalFlushes : 0 + } + } + + /** + * Adjust parameters based on load + */ + public adjustForLoad(pendingRequests: number): void { + if (pendingRequests > 10000) { + // Extreme load - buffer more aggressively + this.maxBufferSize = 5000 + this.flushInterval = 500 + this.minFlushSize = 500 + } else if (pendingRequests > 1000) { + // High load + this.maxBufferSize = 2000 + this.flushInterval = 1000 + this.minFlushSize = 200 + } else if (pendingRequests > 100) { + // Moderate load + this.maxBufferSize = 1000 + this.flushInterval = 2000 + this.minFlushSize = 100 + } else { + // Low load - optimize for latency + this.maxBufferSize = 500 + this.flushInterval = 5000 + this.minFlushSize = 50 + } + } +} + +// Global write buffers +const writeBuffers = new Map>() + +/** + * Get or create a write buffer + */ +export function getWriteBuffer( + id: string, + type: 'noun' | 'verb' | 'metadata', + writeFunction: (items: Map) => Promise +): WriteBuffer { + if (!writeBuffers.has(id)) { + writeBuffers.set(id, new WriteBuffer(type, writeFunction)) + } + return writeBuffers.get(id)! +} + +/** + * Flush all write buffers + */ +export async function flushAllBuffers(): Promise { + const promises: Promise[] = [] + + for (const buffer of writeBuffers.values()) { + promises.push(buffer.forceFlush()) + } + + await Promise.all(promises) +} + +/** + * Clear all write buffers + */ +export function clearWriteBuffers(): void { + for (const buffer of writeBuffers.values()) { + buffer.stop() + } + writeBuffers.clear() +} \ No newline at end of file