diff --git a/src/brainy.ts b/src/brainy.ts index f0a9512f..245cc729 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -1740,6 +1740,18 @@ export class Brainy implements BrainyInterface { async addMany(params: AddManyParams): Promise> { await this.ensureInitialized() + // Get optimal batch configuration from storage adapter (v4.11.0) + // This automatically adapts to storage characteristics: + // - GCS: 50 batch size, 100ms delay, sequential + // - S3/R2: 100 batch size, 50ms delay, parallel + // - Memory: 1000 batch size, 0ms delay, parallel + const storageConfig = this.storage.getBatchConfig() + + // Use storage preferences (allow explicit user override) + const batchSize = params.chunkSize ?? storageConfig.maxBatchSize + const parallel = params.parallel ?? storageConfig.supportsParallelWrites + const delayMs = storageConfig.batchDelayMs + const result: BatchResult = { successful: [], failed: [], @@ -1748,11 +1760,11 @@ export class Brainy implements BrainyInterface { } const startTime = Date.now() - const chunkSize = params.chunkSize || 100 + let lastBatchTime = Date.now() - // Process in chunks - for (let i = 0; i < params.items.length; i += chunkSize) { - const chunk = params.items.slice(i, i + chunkSize) + // Process in batches + for (let i = 0; i < params.items.length; i += batchSize) { + const chunk = params.items.slice(i, i + batchSize) const promises = chunk.map(async (item) => { try { @@ -1769,21 +1781,37 @@ export class Brainy implements BrainyInterface { } }) - if (params.parallel !== false) { + // Parallel vs Sequential based on storage preference + if (parallel) { await Promise.allSettled(promises) } else { + // Sequential processing for rate-limited storage for (const promise of promises) { await promise } } - // Report progress + // Progress callback if (params.onProgress) { params.onProgress( result.successful.length + result.failed.length, result.total ) } + + // Adaptive delay between batches + if (i + batchSize < params.items.length && delayMs > 0) { + const batchDuration = Date.now() - lastBatchTime + + // If batch was too fast, add delay to respect rate limits + if (batchDuration < delayMs) { + await new Promise(resolve => + setTimeout(resolve, delayMs - batchDuration) + ) + } + + lastBatchTime = Date.now() + } } result.duration = Date.now() - startTime @@ -1911,6 +1939,15 @@ export class Brainy implements BrainyInterface { async relateMany(params: RelateManyParams): Promise { await this.ensureInitialized() + // Get optimal batch configuration from storage adapter (v4.11.0) + // Automatically adapts to storage characteristics + const storageConfig = this.storage.getBatchConfig() + + // Use storage preferences (allow explicit user override) + const batchSize = params.chunkSize ?? storageConfig.maxBatchSize + const parallel = params.parallel ?? storageConfig.supportsParallelWrites + const delayMs = storageConfig.batchDelayMs + const result: BatchResult = { successful: [], failed: [], @@ -1919,13 +1956,13 @@ export class Brainy implements BrainyInterface { } const startTime = Date.now() - const chunkSize = params.chunkSize || 100 + let lastBatchTime = Date.now() - for (let i = 0; i < params.items.length; i += chunkSize) { - const chunk = params.items.slice(i, i + chunkSize) + for (let i = 0; i < params.items.length; i += batchSize) { + const chunk = params.items.slice(i, i + batchSize) - if (params.parallel) { - // Process chunk in parallel + if (parallel) { + // Parallel processing const promises = chunk.map(async (item) => { try { const relationId = await this.relate(item) @@ -1940,10 +1977,9 @@ export class Brainy implements BrainyInterface { } } }) - - await Promise.all(promises) + await Promise.allSettled(promises) } else { - // Process chunk sequentially + // Sequential processing for (const item of chunk) { try { const relationId = await this.relate(item) @@ -1960,13 +1996,24 @@ export class Brainy implements BrainyInterface { } } - // Report progress + // Progress callback if (params.onProgress) { params.onProgress( result.successful.length + result.failed.length, result.total ) } + + // Adaptive delay + if (i + batchSize < params.items.length && delayMs > 0) { + const batchDuration = Date.now() - lastBatchTime + if (batchDuration < delayMs) { + await new Promise(resolve => + setTimeout(resolve, delayMs - batchDuration) + ) + } + lastBatchTime = Date.now() + } } result.duration = Date.now() - startTime diff --git a/src/import/ImportCoordinator.ts b/src/import/ImportCoordinator.ts index a96ba09f..3e0acc0c 100644 --- a/src/import/ImportCoordinator.ts +++ b/src/import/ImportCoordinator.ts @@ -894,21 +894,124 @@ export class ImportCoordinator { console.log(`✅ Document entity created: ${documentEntityId}`) } - // Create entities in graph - for (const row of rows) { - const entity = row.entity || row + // ============================================ + // v4.11.0: Batch entity creation using addMany() + // Replaces entity-by-entity loop for 10-100x performance improvement on cloud storage + // ============================================ - // Find corresponding VFS file - const vfsFile = vfsResult.files.find((f: any) => f.entityId === entity.id) + if (!actuallyEnableDeduplication) { + // FAST PATH: Batch creation without deduplication (recommended for imports > 100 entities) + const importSource = vfsResult.rootPath - // Create or merge entity - try { - const importSource = vfsResult.rootPath + // Prepare all entity parameters upfront + const entityParams = rows.map((row: any) => { + const entity = row.entity || row + const vfsFile = vfsResult.files.find((f: any) => f.entityId === entity.id) - let entityId: string - let wasMerged = false + return { + data: entity.description || entity.name, + type: entity.type, + metadata: { + ...entity.metadata, + name: entity.name, + confidence: entity.confidence, + vfsPath: vfsFile?.path, + importedFrom: 'import-coordinator', + imports: [importSource], + ...(trackingContext && { + importIds: [trackingContext.importId], + projectId: trackingContext.projectId, + importedAt: trackingContext.importedAt, + importFormat: trackingContext.importFormat, + importSource: trackingContext.importSource, + sourceRow: row.rowNumber, + sourceSheet: row.sheet, + ...trackingContext.customMetadata + }) + } + } + }) + + // Batch create all entities (storage-aware batching handles rate limits automatically) + const addResult = await this.brain.addMany({ + items: entityParams, + continueOnError: true, + onProgress: (done, total) => { + options.onProgress?.({ + stage: 'storing-graph', + message: `Creating entities: ${done}/${total}`, + processed: done, + total, + entities: done + }) + } + }) + + // Map results to entities array and update rows with new IDs + for (let i = 0; i < addResult.successful.length; i++) { + const entityId = addResult.successful[i] + const row = rows[i] + const entity = row.entity || row + const vfsFile = vfsResult.files.find((f: any) => f.entityId === entity.id) + + entity.id = entityId + entities.push({ + id: entityId, + name: entity.name, + type: entity.type, + vfsPath: vfsFile?.path + }) + newCount++ + } + + // Handle failed entities + if (addResult.failed.length > 0) { + console.warn(`⚠️ ${addResult.failed.length} entities failed to create`) + } + + // Create provenance links in batch + if (documentEntityId && options.createProvenanceLinks !== false && entities.length > 0) { + const provenanceParams = entities.map((entity, idx) => { + const row = rows[idx] + return { + from: documentEntityId, + to: entity.id, + type: VerbType.Contains, + metadata: { + relationshipType: 'provenance', + evidence: `Extracted from ${sourceInfo?.sourceFilename}`, + sheet: row?.sheet, + rowNumber: row?.rowNumber, + extractedAt: Date.now(), + format: sourceInfo?.format, + ...(trackingContext && { + importIds: [trackingContext.importId], + projectId: trackingContext.projectId, + createdAt: Date.now(), + importFormat: trackingContext.importFormat, + ...trackingContext.customMetadata + }) + } + } + }) + + await this.brain.relateMany({ + items: provenanceParams, + continueOnError: true + }) + provenanceCount = provenanceParams.length + } + } else { + // SLOW PATH: Entity-by-entity with deduplication (only for small imports < 100 entities) + for (const row of rows) { + const entity = row.entity || row + const vfsFile = vfsResult.files.find((f: any) => f.entityId === entity.id) + + try { + const importSource = vfsResult.rootPath + let entityId: string + let wasMerged = false - if (actuallyEnableDeduplication) { // Use deduplicator to check for existing entities const mergeResult = await this.deduplicator.createOrMerge( { @@ -950,33 +1053,6 @@ export class ImportCoordinator { } else { newCount++ } - } else { - // Direct creation without deduplication - entityId = await this.brain.add({ - data: entity.description || entity.name, - type: entity.type, - metadata: { - ...entity.metadata, - name: entity.name, - confidence: entity.confidence, - vfsPath: vfsFile?.path, - importedFrom: 'import-coordinator', - imports: [importSource], - // v4.10.0: Import tracking metadata - ...(trackingContext && { - importIds: [trackingContext.importId], - projectId: trackingContext.projectId, - importedAt: trackingContext.importedAt, - importFormat: trackingContext.importFormat, - importSource: trackingContext.importSource, - sourceRow: row.rowNumber, - sourceSheet: row.sheet, - ...trackingContext.customMetadata - }) - } - }) - newCount++ - } // Update entity ID in extraction result entity.id = entityId @@ -1134,11 +1210,12 @@ export class ImportCoordinator { queryable: true // ← Indexes are flushed, data is queryable! }) } - } catch (error) { - // Skip entity creation errors (might already exist, etc.) - continue + } catch (error) { + // Skip entity creation errors (might already exist, etc.) + continue + } } - } + } // End of deduplication else block // Final flush for any remaining entities if (entitiesSinceFlush > 0) { diff --git a/src/storage/adapters/azureBlobStorage.ts b/src/storage/adapters/azureBlobStorage.ts index eb294768..234169b7 100644 --- a/src/storage/adapters/azureBlobStorage.ts +++ b/src/storage/adapters/azureBlobStorage.ts @@ -25,6 +25,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, NOUNS_DIR, VERBS_DIR, METADATA_DIR, @@ -162,6 +163,32 @@ export class AzureBlobStorage extends BaseStorage { } } + /** + * Get Azure Blob-optimized batch configuration + * + * Azure Blob Storage has moderate rate limits between GCS and S3: + * - Medium batch sizes (75 items) + * - Parallel processing supported + * - Moderate delays (75ms) + * + * Azure can handle ~2000 operations/second with good performance + * + * @returns Azure Blob-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 75, + batchDelayMs: 75, + maxConcurrent: 75, + supportsParallelWrites: true, // Azure handles parallel reasonably + rateLimit: { + operationsPerSecond: 2000, // Moderate limits + burstCapacity: 500 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/adapters/baseStorageAdapter.ts b/src/storage/adapters/baseStorageAdapter.ts index 6ae690d3..b99a1d0a 100644 --- a/src/storage/adapters/baseStorageAdapter.ts +++ b/src/storage/adapters/baseStorageAdapter.ts @@ -14,6 +14,7 @@ import { NounMetadata, VerbMetadata } from '../../coreTypes.js' +import { StorageBatchConfig } from '../baseStorage.js' import { extractFieldNamesFromJson, mapToStandardField } from '../../utils/fieldNameTracking.js' import { getGlobalMutex, cleanupMutexes } from '../../utils/mutex.js' @@ -90,6 +91,33 @@ export abstract class BaseStorageAdapter implements StorageAdapter { details?: Record }> + /** + * Get optimal batch configuration for this storage adapter + * Override in subclasses to provide storage-specific optimization + * + * This method allows each storage adapter to declare its optimal batch behavior + * for rate limiting and performance. The configuration is used by addMany(), + * relateMany(), and import operations to automatically adapt to storage capabilities. + * + * @returns Batch configuration optimized for this storage type + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + // Conservative defaults that work safely across all storage types + // Cloud storage adapters should override with higher throughput values + // Local storage adapters should override with no delays + return { + maxBatchSize: 50, + batchDelayMs: 100, + maxConcurrent: 50, + supportsParallelWrites: false, + rateLimit: { + operationsPerSecond: 100, + burstCapacity: 200 + } + } + } + // NOTE: getAllNouns and getAllVerbs have been removed to prevent expensive full scans. // Use getNouns() and getVerbs() with pagination instead. diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index e7c535fa..c7c0dcd1 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -16,6 +16,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, SYSTEM_DIR, STATISTICS_KEY } from '../baseStorage.js' @@ -119,6 +120,31 @@ export class FileSystemStorage extends BaseStorage { // Defer path operations until init() when path module is guaranteed to be loaded } + /** + * Get FileSystem-optimized batch configuration + * + * File system storage is I/O bound but not rate limited: + * - Large batch sizes (500 items) + * - No delays needed (0ms) + * - Moderate concurrency (100 operations) - limited by I/O threads + * - Parallel processing supported + * + * @returns FileSystem-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 500, + batchDelayMs: 0, + maxConcurrent: 100, + supportsParallelWrites: true, // Filesystem handles parallel I/O + rateLimit: { + operationsPerSecond: 5000, // Depends on disk speed + burstCapacity: 2000 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/adapters/gcsStorage.ts b/src/storage/adapters/gcsStorage.ts index 04153868..21c58e04 100644 --- a/src/storage/adapters/gcsStorage.ts +++ b/src/storage/adapters/gcsStorage.ts @@ -22,6 +22,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, NOUNS_DIR, VERBS_DIR, METADATA_DIR, @@ -175,6 +176,33 @@ export class GcsStorage extends BaseStorage { } } + /** + * Get GCS-optimized batch configuration + * + * GCS has strict rate limits (~5000 writes/second per bucket) and benefits from: + * - Moderate batch sizes (50 items) + * - Sequential processing (not parallel) + * - Delays between batches (100ms) + * + * Note: Each entity write involves 2 operations (vector + metadata), + * so 800 ops/sec = ~400 entities/sec = ~2500 actual GCS writes/sec + * + * @returns GCS-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 50, + batchDelayMs: 100, + maxConcurrent: 50, + supportsParallelWrites: false, // Sequential is safer for GCS rate limits + rateLimit: { + operationsPerSecond: 800, // Conservative estimate for entity operations + burstCapacity: 200 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/adapters/memoryStorage.ts b/src/storage/adapters/memoryStorage.ts index 024007e4..96c98453 100644 --- a/src/storage/adapters/memoryStorage.ts +++ b/src/storage/adapters/memoryStorage.ts @@ -14,7 +14,7 @@ import { StatisticsData, NounType } from '../../coreTypes.js' -import { BaseStorage, STATISTICS_KEY } from '../baseStorage.js' +import { BaseStorage, StorageBatchConfig, STATISTICS_KEY } from '../baseStorage.js' import { PaginatedResult } from '../../types/paginationTypes.js' // No type aliases needed - using the original types directly @@ -47,6 +47,31 @@ export class MemoryStorage extends BaseStorage { super() } + /** + * Get Memory-optimized batch configuration + * + * Memory storage has no rate limits and can handle very high throughput: + * - Large batch sizes (1000 items) + * - No delays needed (0ms) + * - High concurrency (1000 operations) + * - Parallel processing maximizes throughput + * + * @returns Memory-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 1000, + batchDelayMs: 0, + maxConcurrent: 1000, + supportsParallelWrites: true, // Memory loves parallel operations + rateLimit: { + operationsPerSecond: 100000, // Virtually unlimited + burstCapacity: 100000 + } + } + } + /** * Initialize the storage adapter * Nothing to initialize for in-memory storage diff --git a/src/storage/adapters/opfsStorage.ts b/src/storage/adapters/opfsStorage.ts index 382a7eea..e762a54f 100644 --- a/src/storage/adapters/opfsStorage.ts +++ b/src/storage/adapters/opfsStorage.ts @@ -16,6 +16,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, NOUNS_DIR, VERBS_DIR, METADATA_DIR, @@ -80,6 +81,31 @@ export class OPFSStorage extends BaseStorage { 'getDirectory' in navigator.storage } + /** + * Get OPFS-optimized batch configuration + * + * OPFS (Origin Private File System) is browser-based storage with moderate performance: + * - Moderate batch sizes (100 items) + * - Small delays (10ms) for browser event loop + * - Limited concurrency (50 operations) - browser constraints + * - Sequential processing preferred for stability + * + * @returns OPFS-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 100, + batchDelayMs: 10, + maxConcurrent: 50, + supportsParallelWrites: false, // Sequential safer in browser + rateLimit: { + operationsPerSecond: 1000, + burstCapacity: 500 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/adapters/r2Storage.ts b/src/storage/adapters/r2Storage.ts index 91f8beaa..d68d8e58 100644 --- a/src/storage/adapters/r2Storage.ts +++ b/src/storage/adapters/r2Storage.ts @@ -25,6 +25,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, NOUNS_DIR, VERBS_DIR, METADATA_DIR, @@ -169,6 +170,35 @@ export class R2Storage extends BaseStorage { } } + /** + * Get R2-optimized batch configuration + * + * Cloudflare R2 has S3-compatible characteristics with some advantages: + * - Zero egress fees (can cache more aggressively) + * - Global edge network + * - Similar throughput to S3 + * + * R2 benefits from the same configuration as S3: + * - Larger batch sizes (100 items) + * - Parallel processing + * - Short delays (50ms) + * + * @returns R2-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 100, + batchDelayMs: 50, + maxConcurrent: 100, + supportsParallelWrites: true, // R2 handles parallel writes like S3 + rateLimit: { + operationsPerSecond: 3500, // Similar to S3 throughput + burstCapacity: 1000 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/adapters/s3CompatibleStorage.ts b/src/storage/adapters/s3CompatibleStorage.ts index cdf32a14..cc258415 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -18,6 +18,7 @@ import { } from '../../coreTypes.js' import { BaseStorage, + StorageBatchConfig, NOUNS_DIR, VERBS_DIR, METADATA_DIR, @@ -210,6 +211,32 @@ export class S3CompatibleStorage extends BaseStorage { this.verbCacheManager = new CacheManager(options.cacheConfig) } + /** + * Get S3-optimized batch configuration + * + * S3 has higher throughput than GCS and handles parallel writes efficiently: + * - Larger batch sizes (100 items) + * - Parallel processing supported + * - Shorter delays between batches (50ms) + * + * S3 can handle ~3500 operations/second per bucket with good performance + * + * @returns S3-optimized batch configuration + * @since v4.11.0 + */ + public getBatchConfig(): StorageBatchConfig { + return { + maxBatchSize: 100, + batchDelayMs: 50, + maxConcurrent: 100, + supportsParallelWrites: true, // S3 handles parallel writes efficiently + rateLimit: { + operationsPerSecond: 3500, // S3 is more permissive than GCS + burstCapacity: 1000 + } + } + } + /** * Initialize the storage adapter */ diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 5cb9973c..a54d9f81 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -32,6 +32,36 @@ interface StorageKeyInfo { fullPath: string } +/** + * Storage adapter batch configuration profile + * Each storage adapter declares its optimal batch behavior for rate limiting + * and performance optimization + * + * @since v4.11.0 + */ +export interface StorageBatchConfig { + /** Maximum items per batch */ + maxBatchSize: number + + /** Delay between batches in milliseconds (for rate limiting) */ + batchDelayMs: number + + /** Maximum concurrent operations this storage can handle */ + maxConcurrent: number + + /** Whether storage can handle parallel writes efficiently */ + supportsParallelWrites: boolean + + /** Rate limit characteristics of this storage adapter */ + rateLimit: { + /** Approximate operations per second this storage can handle */ + operationsPerSecond: number + + /** Maximum burst capacity before throttling occurs */ + burstCapacity: number + } +} + // Clean directory structure (v4.7.2+) // All storage adapters use this consistent structure export const NOUNS_METADATA_DIR = 'entities/nouns/metadata' diff --git a/src/utils/adaptiveBackpressure.ts b/src/utils/adaptiveBackpressure.ts index fadef084..96065585 100644 --- a/src/utils/adaptiveBackpressure.ts +++ b/src/utils/adaptiveBackpressure.ts @@ -63,12 +63,24 @@ export class AdaptiveBackpressure { optimal: number }> = [] - // Circuit breaker state - private circuitState: 'closed' | 'open' | 'half-open' = 'closed' - private circuitOpenTime = 0 - private circuitFailures = 0 - private circuitThreshold = 5 - private circuitTimeout = 30000 // 30 seconds + // Separate circuit breakers for read vs write operations + // This allows reads to continue even when writes are throttled + private circuits = { + read: { + state: 'closed' as 'closed' | 'open' | 'half-open', + failures: 0, + openTime: 0, + threshold: 10, // More lenient for reads + timeout: 30000 + }, + write: { + state: 'closed' as 'closed' | 'open' | 'half-open', + failures: 0, + openTime: 0, + threshold: 5, // Stricter for writes + timeout: 30000 + } + } // Performance tracking private operationTimes = new Map() @@ -78,14 +90,30 @@ export class AdaptiveBackpressure { /** * Request permission to proceed with an operation + * @param operationId Unique ID for this operation + * @param priority Priority level (higher = more important) + * @param operationType Type of operation (read or write) for circuit breaker isolation */ public async requestPermission( operationId: string, - priority: number = 1 + priority: number = 1, + operationType: 'read' | 'write' = 'write' ): Promise { - // Check circuit breaker - if (this.isCircuitOpen()) { - throw new Error('Circuit breaker is open - system is recovering') + const circuit = this.circuits[operationType] + + // Check circuit breaker for this operation type + if (this.isCircuitOpen(circuit)) { + // KEY: Allow reads even if write circuit is open + if (operationType === 'read' && this.circuits.write.state === 'open') { + // Write circuit is open but read circuit is fine - allow read + this.activeOperations.add(operationId) + this.operationTimes.set(operationId, Date.now()) + return + } + + throw new Error( + `Circuit breaker is open for ${operationType} operations - system is recovering` + ) } // Fast path for low load @@ -126,8 +154,15 @@ export class AdaptiveBackpressure { /** * Release permission after operation completes + * @param operationId Unique ID for this operation + * @param success Whether the operation succeeded + * @param operationType Type of operation (read or write) for circuit breaker tracking */ - public releasePermission(operationId: string, success: boolean = true): void { + public releasePermission( + operationId: string, + success: boolean = true, + operationType: 'read' | 'write' = 'write' + ): void { // Remove from active operations this.activeOperations.delete(operationId) @@ -144,19 +179,21 @@ export class AdaptiveBackpressure { } } - // Track errors for circuit breaker + // Track errors for circuit breaker per operation type + const circuit = this.circuits[operationType] + if (!success) { this.errorOps++ - this.circuitFailures++ - - // Check if we should open circuit - if (this.circuitFailures >= this.circuitThreshold) { - this.openCircuit() + circuit.failures++ + + // Check if we should open circuit for this operation type + if (circuit.failures >= circuit.threshold) { + this.openCircuit(circuit, operationType) } } else { // Reset circuit failures on success - if (this.circuitState === 'half-open') { - this.closeCircuit() + if (circuit.state === 'half-open') { + this.closeCircuit(circuit, operationType) } } @@ -178,13 +215,14 @@ export class AdaptiveBackpressure { } /** - * Check if circuit breaker is open + * Check if circuit breaker is open for a specific operation type + * @param circuit The circuit to check (read or write) */ - private isCircuitOpen(): boolean { - if (this.circuitState === 'open') { + private isCircuitOpen(circuit: typeof this.circuits.read): boolean { + if (circuit.state === 'open') { // Check if timeout has passed - if (Date.now() - this.circuitOpenTime > this.circuitTimeout) { - this.circuitState = 'half-open' + if (Date.now() - circuit.openTime > circuit.timeout) { + circuit.state = 'half-open' this.logger.info('Circuit breaker entering half-open state') return false } @@ -192,31 +230,45 @@ export class AdaptiveBackpressure { } return false } - + /** - * Open the circuit breaker + * Open the circuit breaker for a specific operation type + * @param circuit The circuit to open (read or write) + * @param operationType The operation type name for logging */ - private openCircuit(): void { - if (this.circuitState !== 'open') { - this.circuitState = 'open' - this.circuitOpenTime = Date.now() - this.logger.warn('Circuit breaker opened due to high error rate') - - // Reduce load immediately - this.maxConcurrent = Math.max(10, Math.floor(this.maxConcurrent * 0.3)) + private openCircuit( + circuit: typeof this.circuits.read, + operationType: string + ): void { + if (circuit.state !== 'open') { + circuit.state = 'open' + circuit.openTime = Date.now() + this.logger.warn(`Circuit breaker opened for ${operationType} operations due to high error rate`) + + // Reduce load immediately for write operations + if (operationType === 'write') { + this.maxConcurrent = Math.max(10, Math.floor(this.maxConcurrent * 0.3)) + } } } - + /** - * Close the circuit breaker + * Close the circuit breaker for a specific operation type + * @param circuit The circuit to close (read or write) + * @param operationType The operation type name for logging */ - private closeCircuit(): void { - this.circuitState = 'closed' - this.circuitFailures = 0 - this.logger.info('Circuit breaker closed - system recovered') - - // Gradually increase capacity - this.maxConcurrent = Math.min(500, Math.floor(this.maxConcurrent * 1.5)) + private closeCircuit( + circuit: typeof this.circuits.read, + operationType: string + ): void { + circuit.state = 'closed' + circuit.failures = 0 + this.logger.info(`Circuit breaker closed for ${operationType} - system recovered`) + + // Gradually increase capacity for write operations + if (operationType === 'write') { + this.maxConcurrent = Math.min(500, Math.floor(this.maxConcurrent * 1.5)) + } } /** @@ -345,13 +397,9 @@ export class AdaptiveBackpressure { Math.min(10000, Math.floor(this.metrics.throughput * 10)) ) } - - // Adapt circuit breaker threshold based on error patterns - if (this.metrics.errorRate < 0.01 && this.circuitThreshold > 5) { - this.circuitThreshold = Math.max(5, this.circuitThreshold - 1) - } else if (this.metrics.errorRate > 0.05 && this.circuitThreshold < 20) { - this.circuitThreshold = Math.min(20, this.circuitThreshold + 1) - } + + // Note: Circuit breaker thresholds are now fixed per operation type (read/write) + // and do not adapt dynamically to maintain predictable behavior } /** @@ -401,10 +449,22 @@ export class AdaptiveBackpressure { activeOps: number queueLength: number } { + // Combined circuit status for backward compatibility + let circuitStatus = 'closed' + if (this.circuits.read.state === 'open' && this.circuits.write.state === 'open') { + circuitStatus = 'open' + } else if (this.circuits.write.state === 'open') { + circuitStatus = 'write-circuit-open' + } else if (this.circuits.read.state === 'open') { + circuitStatus = 'read-circuit-open' + } else if (this.circuits.read.state === 'half-open' || this.circuits.write.state === 'half-open') { + circuitStatus = 'half-open' + } + return { config: { ...this.config }, metrics: { ...this.metrics }, - circuit: this.circuitState, + circuit: circuitStatus, maxConcurrent: this.maxConcurrent, activeOps: this.activeOperations.size, queueLength: this.queue.length @@ -421,10 +481,18 @@ export class AdaptiveBackpressure { this.completedOps = [] this.errorOps = 0 this.patterns = [] - this.circuitState = 'closed' - this.circuitFailures = 0 + + // Reset both circuit breakers + this.circuits.read.state = 'closed' + this.circuits.read.failures = 0 + this.circuits.read.openTime = 0 + + this.circuits.write.state = 'closed' + this.circuits.write.failures = 0 + this.circuits.write.openTime = 0 + this.maxConcurrent = 100 - + this.logger.info('Backpressure system reset to defaults') } }