diff --git a/docs/operations/cloud-run-filestore-guide.md b/docs/operations/cloud-run-filestore-guide.md new file mode 100644 index 00000000..76907278 --- /dev/null +++ b/docs/operations/cloud-run-filestore-guide.md @@ -0,0 +1,138 @@ +# Cloud Run + Filestore (NFS) Deployment Guide + +> Eliminate cloud storage rate limiting by using brainy's FileSystem adapter with a Filestore NFS mount on Cloud Run. + +## Why Filestore? + +Brainy's metadata index generates 26-40 file writes per `brain.add()` call. On GCS, each write is a cloud API call subject to rate limiting (~1 write/sec per object), causing HTTP 429 errors and 61-second latency for a single add operation. + +With Filestore: +- **Local disk speed**: Each write is ~1ms (vs 50-300ms for cloud API) +- **No rate limits**: NFS has no per-object write throttling +- **Zero code changes**: Use brainy's existing `FileSystemStorage` adapter +- **Same Cloud Run deployment**: Just add a volume mount + +## Prerequisites + +- Google Cloud project with Filestore API enabled +- Cloud Run service (gen2 execution environment required for NFS) +- VPC connector or Direct VPC egress configured + +## Step 1: Create a Filestore Instance + +```bash +gcloud filestore instances create brainy-store \ + --zone=us-central1-b \ + --tier=BASIC_SSD \ + --file-share=name=brainy_data,capacity=1TB \ + --network=name=default +``` + +**Tier recommendations:** +- **BASIC_SSD** (~$370/month for 1 TiB): Good balance of performance and cost +- **BASIC_HDD** (~$204/month for 1 TiB): Lower cost, sufficient for moderate workloads +- **ENTERPRISE** or **ZONAL**: Higher IOPS for heavy workloads + +Note the IP address from the output — you'll need it for the Cloud Run mount. + +## Step 2: Configure Cloud Run NFS Volume Mount + +```yaml +# service.yaml +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + name: my-brainy-service +spec: + template: + metadata: + annotations: + run.googleapis.com/execution-environment: gen2 + run.googleapis.com/vpc-access-connector: projects/PROJECT/locations/REGION/connectors/CONNECTOR + spec: + containers: + - image: gcr.io/PROJECT/my-brainy-app + volumeMounts: + - name: brainy-nfs + mountPath: /mnt/brainy + volumes: + - name: brainy-nfs + nfs: + server: FILESTORE_IP # e.g., 10.0.0.2 + path: /brainy_data +``` + +Deploy: + +```bash +gcloud run services replace service.yaml --region=us-central1 +``` + +## Step 3: Configure Brainy to Use FileSystem Adapter + +```typescript +import { Brainy } from '@soulcraft/brainy' +import { FileSystemStorage } from '@soulcraft/brainy/storage' + +const storage = new FileSystemStorage({ + basePath: '/mnt/brainy/brain-data' +}) + +const brain = new Brainy({ storage }) +await brain.init() +``` + +That's it. No GCS adapter, no rate limits, no retry logic needed. + +## Caveats + +### Single-Writer Recommendation + +NFS does not provide strong file locking guarantees on Cloud Run. For best results: + +- Use a **single Cloud Run instance** (max-instances=1) for write operations +- Multiple read-only instances can safely read from the same Filestore mount +- For multi-writer scenarios, use the GCS adapter with the write buffer (rate limit protection built in) + +### Filestore Permissions + +Cloud Run gen2 instances run as a specific service account. Ensure the Filestore instance allows access from your VPC network. No additional IAM permissions are needed — Filestore uses NFS network-level access. + +### Cost Comparison + +| Solution | Monthly Cost (1 TiB) | Write Latency | Rate Limits | +|----------|---------------------|---------------|-------------| +| GCS Standard | ~$20 storage + API ops | 50-300ms/write | 1 write/sec/object | +| GCS HNS | ~$20 storage + API ops | 50-300ms/write | 8,000 writes/sec | +| Filestore SSD | ~$370 fixed | ~1ms/write | None | +| Filestore HDD | ~$204 fixed | ~5ms/write | None | + +Filestore costs more for storage but eliminates all rate limiting issues and provides significantly lower write latency. + +### When to Choose Filestore vs GCS + +**Choose Filestore when:** +- Write-heavy workloads (frequent `brain.add()`, chat applications) +- Low-latency requirements +- Single-writer architecture is acceptable + +**Choose GCS (with write buffer) when:** +- Multi-instance deployments need shared storage +- Cost optimization for large datasets (lifecycle policies, Autoclass) +- Read-heavy workloads with infrequent writes + +## Verification + +After deploying, verify the mount works: + +```bash +# In Cloud Run container +ls -la /mnt/brainy/ +# Should show the Filestore share contents + +# Test write performance +dd if=/dev/zero of=/mnt/brainy/test bs=1M count=100 +# Expected: ~100MB/s+ for SSD tier +``` + +Then run your brainy application and confirm `brain.add()` completes without rate limit errors. diff --git a/docs/operations/cost-optimization-gcs.md b/docs/operations/cost-optimization-gcs.md index 59ab3ecf..a5ecbb34 100644 --- a/docs/operations/cost-optimization-gcs.md +++ b/docs/operations/cost-optimization-gcs.md @@ -1,6 +1,8 @@ # Google Cloud Storage Cost Optimization Guide for Brainy > **Cost Impact**: Reduce storage costs from $138k/year to $8.3k/year at 500TB scale (**94% savings**) +> **Disclaimer**: Cost savings percentages are PROJECTED based on published cloud provider pricing at 500TB scale. Actual savings depend on access patterns, data lifecycle, and workload characteristics. These are not measured benchmarks. + ## Overview Brainy provides enterprise-grade cost optimization features for Google Cloud Storage, including lifecycle policies and Autoclass for automatic tier management. @@ -414,6 +416,49 @@ export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json" - After: $740,000/year - **Savings: $2,120,000/year (74%)** +## Strategy 5: HNS Buckets for Write-Heavy Workloads + +### When to Use HNS (Hierarchical Namespace) + +If you experience HTTP 429 rate limit errors during `brain.add()` operations (especially with many metadata fields), GCS Hierarchical Namespace buckets provide significantly higher write throughput with zero code changes. + +### Why It Helps + +Standard GCS buckets enforce ~1 write/sec per object path. Brainy's metadata index writes multiple chunk and sparse index files per `brain.add()` call, which can exceed these limits during burst writes. + +HNS buckets provide: +- **8x higher initial QPS** (8,000 writes/sec vs 1,000 for standard buckets) +- **Better handling of hierarchical key patterns** (which brainy uses for sharded storage) +- **No code changes required** — create a new HNS-enabled bucket and point brainy at it + +### Setup + +```bash +# Create HNS-enabled bucket +gcloud storage buckets create gs://my-brainy-hns \ + --location=us-central1 \ + --uniform-bucket-level-access \ + --enable-hierarchical-namespace +``` + +Then configure brainy to use the new bucket: + +```typescript +const storage = new GcsStorage({ + bucketName: 'my-brainy-hns' +}) +``` + +### Trade-offs + +- HNS buckets do not support object versioning (irrelevant for brainy — uses its own COW versioning) +- HNS is available in most GCS regions +- Pricing is the same as standard buckets + +### Alternative: Cloud Run with Filestore/NFS + +For the highest write throughput with zero rate limiting, see the [Cloud Run Filestore Guide](cloud-run-filestore-guide.md) — mount a Filestore NFS volume and use brainy's FileSystem adapter instead of GCS. + --- **Last Updated**: 2025-10-17 diff --git a/src/brainy.ts b/src/brainy.ts index 39a66cff..2e589cf8 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -6530,6 +6530,14 @@ export class Brainy implements BrainyInterface { this.originalConsole = undefined } + // Drain the metadata write buffer if the storage adapter has one + if (this.storage && 'metadataWriteBuffer' in this.storage) { + const buffer = (this.storage as any).metadataWriteBuffer + if (buffer && typeof buffer.destroy === 'function') { + await buffer.destroy() + } + } + // Storage doesn't have close in current interface // We'll just mark as not initialized this.initialized = false diff --git a/src/storage/adapters/azureBlobStorage.ts b/src/storage/adapters/azureBlobStorage.ts index ae6fdd7b..c0cb4a54 100644 --- a/src/storage/adapters/azureBlobStorage.ts +++ b/src/storage/adapters/azureBlobStorage.ts @@ -37,6 +37,7 @@ import { import { BrainyError } from '../../errors/brainyError.js' import { CacheManager } from '../cacheManager.js' import { createModuleLogger, prodLog } from '../../utils/logger.js' +import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js' import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js' import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js' import { getCoalescer, RequestCoalescer } from '../../utils/requestCoalescer.js' @@ -201,6 +202,12 @@ export class AzureBlobStorage extends BaseStorage { this.verbCacheManager = new CacheManager(options.cacheConfig) // Write buffering always enabled - no env var check needed + + // Initialize metadata write buffer for cloud rate limit protection + this.metadataWriteBuffer = new MetadataWriteBuffer( + (path, data) => this.writeObjectToPath(path, data), + { maxBufferSize: 200, flushIntervalMs: 200, concurrencyLimit: 10 } + ) } /** @@ -897,20 +904,45 @@ export class AzureBlobStorage extends BaseStorage { // Lazy container validation for progressive init await this.ensureValidatedForWrite() - try { - this.logger.trace(`Writing object to path: ${path}`) + const MAX_RETRIES = 5 + let lastError: any - const blockBlobClient = this.containerClient!.getBlockBlobClient(path) - const content = JSON.stringify(data, null, 2) - await blockBlobClient.upload(content, content.length, { - blobHTTPHeaders: { blobContentType: 'application/json' } - }) + for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { + try { + this.logger.trace(`Writing object to path: ${path}`) - this.logger.trace(`Object written successfully to ${path}`) - } catch (error) { - this.logger.error(`Failed to write object to ${path}:`, error) - throw new Error(`Failed to write object to ${path}: ${error}`) + const blockBlobClient = this.containerClient!.getBlockBlobClient(path) + const content = JSON.stringify(data, null, 2) + await blockBlobClient.upload(content, content.length, { + blobHTTPHeaders: { blobContentType: 'application/json' } + }) + + this.logger.trace(`Object written successfully to ${path}`) + if (attempt > 0) { + this.clearThrottlingState() + } + return + } catch (error: any) { + lastError = error + + if (this.isThrottlingError(error) && attempt < MAX_RETRIES) { + const baseDelay = Math.min(100 * Math.pow(2, attempt), 5000) + const jitter = baseDelay * 0.2 * (Math.random() - 0.5) + const delay = Math.round(baseDelay + jitter) + this.logger.warn( + `Throttled writing ${path} (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms` + ) + await this.handleThrottling(error) + await new Promise(resolve => setTimeout(resolve, delay)) + continue + } + + break + } } + + this.logger.error(`Failed to write object to ${path}:`, lastError) + throw new Error(`Failed to write object to ${path}: ${lastError}`) } /** diff --git a/src/storage/adapters/gcsStorage.ts b/src/storage/adapters/gcsStorage.ts index b65c8ed7..99a7b3e3 100644 --- a/src/storage/adapters/gcsStorage.ts +++ b/src/storage/adapters/gcsStorage.ts @@ -34,6 +34,7 @@ import { import { BrainyError } from '../../errors/brainyError.js' import { CacheManager } from '../cacheManager.js' import { createModuleLogger, prodLog } from '../../utils/logger.js' +import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js' import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js' import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js' import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js' @@ -240,6 +241,12 @@ export class GcsStorage extends BaseStorage { this.verbCacheManager = new CacheManager(options.cacheConfig) // Write buffering always enabled - no env var check needed + + // Initialize metadata write buffer for cloud rate limit protection + this.metadataWriteBuffer = new MetadataWriteBuffer( + (path, data) => this.writeObjectToPath(path, data), + { maxBufferSize: 200, flushIntervalMs: 200, concurrencyLimit: 10 } + ) } /** @@ -840,20 +847,45 @@ export class GcsStorage extends BaseStorage { // Lazy bucket validation for progressive init await this.ensureValidatedForWrite() - try { - this.logger.trace(`Writing object to path: ${path}`) + const MAX_RETRIES = 5 + let lastError: any - const file = this.bucket!.file(path) - await file.save(JSON.stringify(data, null, 2), { - contentType: 'application/json', - resumable: false - }) + for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { + try { + this.logger.trace(`Writing object to path: ${path}`) - this.logger.trace(`Object written successfully to ${path}`) - } catch (error) { - this.logger.error(`Failed to write object to ${path}:`, error) - throw new Error(`Failed to write object to ${path}: ${error}`) + const file = this.bucket!.file(path) + await file.save(JSON.stringify(data, null, 2), { + contentType: 'application/json', + resumable: false + }) + + this.logger.trace(`Object written successfully to ${path}`) + if (attempt > 0) { + this.clearThrottlingState() + } + return + } catch (error: any) { + lastError = error + + if (this.isThrottlingError(error) && attempt < MAX_RETRIES) { + const baseDelay = Math.min(100 * Math.pow(2, attempt), 5000) + const jitter = baseDelay * 0.2 * (Math.random() - 0.5) + const delay = Math.round(baseDelay + jitter) + this.logger.warn( + `Throttled writing ${path} (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms` + ) + await this.handleThrottling(error) + await new Promise(resolve => setTimeout(resolve, delay)) + continue + } + + break + } } + + this.logger.error(`Failed to write object to ${path}:`, lastError) + throw new Error(`Failed to write object to ${path}: ${lastError}`) } /** diff --git a/src/storage/adapters/r2Storage.ts b/src/storage/adapters/r2Storage.ts index 73ac4500..4777c559 100644 --- a/src/storage/adapters/r2Storage.ts +++ b/src/storage/adapters/r2Storage.ts @@ -37,6 +37,7 @@ import { import { BrainyError } from '../../errors/brainyError.js' import { CacheManager } from '../cacheManager.js' import { createModuleLogger, prodLog } from '../../utils/logger.js' +import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js' import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js' import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js' import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js' @@ -169,6 +170,12 @@ export class R2Storage extends BaseStorage { this.verbCacheManager = new CacheManager(options.cacheConfig) // Write buffering always enabled - no env var check needed + + // Initialize metadata write buffer for cloud rate limit protection + this.metadataWriteBuffer = new MetadataWriteBuffer( + (path, data) => this.writeObjectToPath(path, data), + { maxBufferSize: 200, flushIntervalMs: 200, concurrencyLimit: 10 } + ) } /** @@ -611,24 +618,49 @@ export class R2Storage extends BaseStorage { protected async writeObjectToPath(path: string, data: any): Promise { await this.ensureInitialized() - try { - this.logger.trace(`Writing object to path: ${path}`) + const MAX_RETRIES = 5 + let lastError: any - const { PutObjectCommand } = await import('@aws-sdk/client-s3') - await this.s3Client!.send( - new PutObjectCommand({ - Bucket: this.bucketName, - Key: path, - Body: JSON.stringify(data, null, 2), - ContentType: 'application/json' - }) - ) + for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { + try { + this.logger.trace(`Writing object to path: ${path}`) - this.logger.trace(`Object written successfully to ${path}`) - } catch (error) { - this.logger.error(`Failed to write object to ${path}:`, error) - throw new Error(`Failed to write object to ${path}: ${error}`) + const { PutObjectCommand } = await import('@aws-sdk/client-s3') + await this.s3Client!.send( + new PutObjectCommand({ + Bucket: this.bucketName, + Key: path, + Body: JSON.stringify(data, null, 2), + ContentType: 'application/json' + }) + ) + + this.logger.trace(`Object written successfully to ${path}`) + if (attempt > 0) { + this.clearThrottlingState() + } + return + } catch (error: any) { + lastError = error + + if (this.isThrottlingError(error) && attempt < MAX_RETRIES) { + const baseDelay = Math.min(100 * Math.pow(2, attempt), 5000) + const jitter = baseDelay * 0.2 * (Math.random() - 0.5) + const delay = Math.round(baseDelay + jitter) + this.logger.warn( + `Throttled writing ${path} (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms` + ) + await this.handleThrottling(error) + await new Promise(resolve => setTimeout(resolve, delay)) + continue + } + + break + } } + + this.logger.error(`Failed to write object to ${path}:`, lastError) + throw new Error(`Failed to write object to ${path}: ${lastError}`) } /** diff --git a/src/storage/adapters/s3CompatibleStorage.ts b/src/storage/adapters/s3CompatibleStorage.ts index 92c531b7..21049f53 100644 --- a/src/storage/adapters/s3CompatibleStorage.ts +++ b/src/storage/adapters/s3CompatibleStorage.ts @@ -35,6 +35,7 @@ import { import { BrainyError } from '../../errors/brainyError.js' import { CacheManager } from '../cacheManager.js' import { createModuleLogger, prodLog } from '../../utils/logger.js' +import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js' import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js' import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js' import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js' @@ -256,6 +257,12 @@ export class S3CompatibleStorage extends BaseStorage { // Initialize cache managers this.nounCacheManager = new CacheManager(options.cacheConfig) this.verbCacheManager = new CacheManager(options.cacheConfig) + + // Initialize metadata write buffer for cloud rate limit protection + this.metadataWriteBuffer = new MetadataWriteBuffer( + (path, data) => this.writeObjectToPath(path, data), + { maxBufferSize: 200, flushIntervalMs: 200, concurrencyLimit: 10 } + ) } /** @@ -1801,43 +1808,68 @@ export class S3CompatibleStorage extends BaseStorage { // Lazy bucket validation for progressive init await this.ensureValidatedForWrite() - // Apply backpressure before starting operation - const requestId = await this.applyBackpressure() + const MAX_RETRIES = 5 + let lastError: any - try { - const { PutObjectCommand } = await import('@aws-sdk/client-s3') - const body = JSON.stringify(data, null, 2) + for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { + // Apply backpressure before each attempt + const requestId = await this.applyBackpressure() - this.logger.trace(`Writing object to path: ${path}`) + try { + const { PutObjectCommand } = await import('@aws-sdk/client-s3') + const body = JSON.stringify(data, null, 2) - await this.s3Client!.send( - new PutObjectCommand({ - Bucket: this.bucketName, - Key: path, - Body: body, - ContentType: 'application/json' + this.logger.trace(`Writing object to path: ${path}`) + + await this.s3Client!.send( + new PutObjectCommand({ + Bucket: this.bucketName, + Key: path, + Body: body, + ContentType: 'application/json' + }) + ) + + this.logger.debug(`Object written successfully to ${path}`) + + // Log the change for efficient synchronization + await this.appendToChangeLog({ + timestamp: Date.now(), + operation: 'add', + entityType: 'metadata', + entityId: path, + data: data }) - ) - this.logger.debug(`Object written successfully to ${path}`) + // Release backpressure on success + this.releaseBackpressure(true, requestId) + if (attempt > 0) { + this.clearThrottlingState() + } + return + } catch (error: any) { + // Release backpressure on error + this.releaseBackpressure(false, requestId) + lastError = error - // Log the change for efficient synchronization - await this.appendToChangeLog({ - timestamp: Date.now(), - operation: 'add', - entityType: 'metadata', - entityId: path, - data: data - }) + if (this.isThrottlingError(error) && attempt < MAX_RETRIES) { + const baseDelay = Math.min(100 * Math.pow(2, attempt), 5000) + const jitter = baseDelay * 0.2 * (Math.random() - 0.5) + const delay = Math.round(baseDelay + jitter) + this.logger.warn( + `Throttled writing ${path} (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms` + ) + await this.handleThrottling(error) + await new Promise(resolve => setTimeout(resolve, delay)) + continue + } - // Release backpressure on success - this.releaseBackpressure(true, requestId) - } catch (error) { - // Release backpressure on error - this.releaseBackpressure(false, requestId) - this.logger.error(`Failed to write object to ${path}:`, error) - throw new Error(`Failed to write object to ${path}: ${error}`) + break + } } + + this.logger.error(`Failed to write object to ${path}:`, lastError) + throw new Error(`Failed to write object to ${path}: ${lastError}`) } /** diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index f622cdff..bf8c9569 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -30,6 +30,7 @@ import { BlobStorage, type COWStorageAdapter } from './cow/BlobStorage.js' import { CommitLog } from './cow/CommitLog.js' import { unwrapBinaryData, wrapBinaryData } from './cow/binaryDataCodec.js' import { prodLog } from '../utils/logger.js' +import { MetadataWriteBuffer } from '../utils/metadataWriteBuffer.js' /** * Storage key analysis result @@ -217,6 +218,11 @@ export abstract class BaseStorage extends BaseStorageAdapter { // Track if type counts have been rebuilt (prevent repeated rebuilds) private typeCountsRebuilt = false + // Write buffer for cloud storage adapters — deduplicates rapid writes to the same path + // FileSystem adapter does NOT use this (local writes are already fast) + // Initialized by cloud adapters in their init() method + protected metadataWriteBuffer: MetadataWriteBuffer | null = null + /** * Analyze a storage key to determine its routing and path * @param id - The key to analyze (UUID or system key) @@ -585,8 +591,12 @@ export abstract class BaseStorage extends BaseStorageAdapter { // This ensures readWithInheritance() returns data immediately, fixing "Source entity not found" bug this.writeCache.set(branchPath, data) - // Write to storage (async) - await this.writeObjectToPath(branchPath, data) + // Use write buffer if available (cloud adapters), otherwise write directly (filesystem) + if (this.metadataWriteBuffer) { + await this.metadataWriteBuffer.write(branchPath, data) + } else { + await this.writeObjectToPath(branchPath, data) + } // Cache is NOT cleared here anymore - persists until flush() // This provides a safety net for immediate queries after batch writes diff --git a/src/utils/metadataIndex.ts b/src/utils/metadataIndex.ts index 2f8fb18d..6869f130 100644 --- a/src/utils/metadataIndex.ts +++ b/src/utils/metadataIndex.ts @@ -129,6 +129,12 @@ export class MetadataIndexManager { private chunkManager: ChunkManager private chunkingStrategy: AdaptiveChunkingStrategy + // Deferred write tracking: accumulate chunk/sparse index writes during add/remove + // operations and flush them in a single concurrent batch at the end. + // This eliminates per-field sequential writes that cause cloud storage rate limiting. + private dirtyChunks = new Map() // "field:chunkId" -> ChunkData + private dirtySparseIndices = new Map() // field -> SparseIndex + // Roaring Bitmap Support // EntityIdMapper for UUID ↔ integer conversion private idMapper: EntityIdMapper @@ -651,6 +657,118 @@ export class MetadataIndexManager { this.unifiedCache.set(unifiedKey, sparseIndex, 'metadata', size, 200) } + /** + * Flush all deferred chunk and sparse index writes accumulated during add/remove operations. + * Writes are deduplicated (same chunk/field written once even if updated multiple times) + * and executed concurrently via Promise.all for maximum throughput. + */ + private async flushDirtyMetadata(): Promise { + if (this.dirtyChunks.size === 0 && this.dirtySparseIndices.size === 0) { + return + } + + const promises: Promise[] = [] + + // Save all dirty chunks (deduplicated — same chunk written once even if updated multiple times) + for (const [_key, chunk] of this.dirtyChunks) { + promises.push(this.chunkManager.saveChunk(chunk)) + } + + // Save all dirty sparse indices (deduplicated — same field's index written once) + for (const [field, sparseIndex] of this.dirtySparseIndices) { + promises.push(this.saveSparseIndex(field, sparseIndex)) + } + + // Execute all writes concurrently + await Promise.all(promises) + + this.dirtyChunks.clear() + this.dirtySparseIndices.clear() + } + + /** + * Split a chunk without saving immediately — returns the new chunks for deferred save. + * Used by addToChunkedIndex() to keep splits within the deferred write batch. + */ + private async splitChunkDeferred( + chunk: ChunkData, + sparseIndex: SparseIndex + ): Promise<{ chunk1: ChunkData; chunk2: ChunkData }> { + const values = Array.from(chunk.entries.keys()).sort() + const midpoint = Math.floor(values.length / 2) + + // Create two new chunks with roaring bitmaps + const entries1 = new Map() + const entries2 = new Map() + + for (let i = 0; i < values.length; i++) { + const value = values[i] + const bitmap = chunk.entries.get(value)! + + if (i < midpoint) { + entries1.set(value, new RoaringBitmap32(bitmap.toArray())) + } else { + entries2.set(value, new RoaringBitmap32(bitmap.toArray())) + } + } + + // Create chunk objects without saving (just allocate IDs and set up data) + const chunkId1 = this.chunkManager['getNextChunkId'](chunk.field) + const chunk1: ChunkData = { + chunkId: chunkId1, + field: chunk.field, + entries: entries1, + lastUpdated: Date.now() + } + + const chunkId2 = this.chunkManager['getNextChunkId'](chunk.field) + const chunk2: ChunkData = { + chunkId: chunkId2, + field: chunk.field, + entries: entries2, + lastUpdated: Date.now() + } + + // Update chunk cache (for read-after-write consistency within this operation) + this.chunkManager['chunkCache'].set(`${chunk.field}:${chunkId1}`, chunk1) + this.chunkManager['chunkCache'].set(`${chunk.field}:${chunkId2}`, chunk2) + + // Update sparse index + sparseIndex.removeChunk(chunk.chunkId) + + const descriptor1: ChunkDescriptor = { + chunkId: chunk1.chunkId, + field: chunk1.field, + valueCount: entries1.size, + idCount: Array.from(entries1.values()).reduce((sum, bitmap) => sum + bitmap.size, 0), + zoneMap: this.chunkManager.calculateZoneMap(chunk1), + lastUpdated: Date.now(), + splitThreshold: 80, + mergeThreshold: 20 + } + + const descriptor2: ChunkDescriptor = { + chunkId: chunk2.chunkId, + field: chunk2.field, + valueCount: entries2.size, + idCount: Array.from(entries2.values()).reduce((sum, bitmap) => sum + bitmap.size, 0), + zoneMap: this.chunkManager.calculateZoneMap(chunk2), + lastUpdated: Date.now(), + splitThreshold: 80, + mergeThreshold: 20 + } + + sparseIndex.registerChunk(descriptor1, this.chunkManager.createBloomFilter(chunk1)) + sparseIndex.registerChunk(descriptor2, this.chunkManager.createBloomFilter(chunk2)) + + // Delete old chunk from storage (this still writes immediately as it's a deletion) + await this.chunkManager.deleteChunk(chunk.field, chunk.chunkId) + + prodLog.debug(`Split chunk ${chunk.field}:${chunk.chunkId} into ${chunk1.chunkId} and ${chunk2.chunkId} (deferred save)`) + + return { chunk1, chunk2 } + } + /** * Get IDs for a value using chunked sparse index with roaring bitmaps * Now fully lazy-loaded via UnifiedCache (no local sparseIndices Map) @@ -934,7 +1052,8 @@ export class MetadataIndexManager { // Add to chunk await this.chunkManager.addToChunk(targetChunk, normalizedValue, id) - await this.chunkManager.saveChunk(targetChunk) + // Defer chunk save — mark dirty instead of writing immediately + this.dirtyChunks.set(`${targetChunk.field}:${targetChunk.chunkId}`, targetChunk) // Update chunk descriptor in sparse index const updatedZoneMap = this.chunkManager.calculateZoneMap(targetChunk) @@ -955,11 +1074,15 @@ export class MetadataIndexManager { // Check if chunk needs splitting if (targetChunk.entries.size > 80) { - await this.chunkManager.splitChunk(targetChunk, sparseIndex) + const { chunk1, chunk2 } = await this.splitChunkDeferred(targetChunk, sparseIndex) + // Mark split result chunks as dirty instead of the original + this.dirtyChunks.delete(`${targetChunk.field}:${targetChunk.chunkId}`) + this.dirtyChunks.set(`${chunk1.field}:${chunk1.chunkId}`, chunk1) + this.dirtyChunks.set(`${chunk2.field}:${chunk2.chunkId}`, chunk2) } - // Save sparse index - await this.saveSparseIndex(field, sparseIndex) + // Defer sparse index save — mark dirty instead of writing immediately + this.dirtySparseIndices.set(field, sparseIndex) } /** @@ -980,7 +1103,8 @@ export class MetadataIndexManager { const chunk = await this.chunkManager.loadChunk(field, chunkId) if (chunk && chunk.entries.has(normalizedValue)) { await this.chunkManager.removeFromChunk(chunk, normalizedValue, id) - await this.chunkManager.saveChunk(chunk) + // Defer chunk save — mark dirty instead of writing immediately + this.dirtyChunks.set(`${chunk.field}:${chunk.chunkId}`, chunk) // Update sparse index const updatedZoneMap = this.chunkManager.calculateZoneMap(chunk) @@ -991,7 +1115,8 @@ export class MetadataIndexManager { lastUpdated: Date.now() }) - await this.saveSparseIndex(field, sparseIndex) + // Defer sparse index save — mark dirty instead of writing immediately + this.dirtySparseIndices.set(field, sparseIndex) break } } @@ -1431,7 +1556,11 @@ export class MetadataIndexManager { await this.yieldToEventLoop() } } - + + // Flush all dirty chunks and sparse indices accumulated during this add operation + // This batches writes that were previously sequential per-field into a single concurrent flush + await this.flushDirtyMetadata() + // Adaptive auto-flush based on usage patterns if (!skipFlush) { const timeSinceLastFlush = Date.now() - this.lastFlushTime @@ -1518,6 +1647,9 @@ export class MetadataIndexManager { // Invalidate cache this.metadataCache.invalidatePattern(`field_values_${field}`) } + + // Flush all dirty chunks and sparse indices accumulated during remove + await this.flushDirtyMetadata() } else { // Remove from all indexes (slower, requires scanning all field indexes) // This should be rare - prefer providing metadata when removing @@ -1545,6 +1677,9 @@ export class MetadataIndexManager { } } } + + // Flush all dirty chunks and sparse indices accumulated during scan-remove + await this.flushDirtyMetadata() } } @@ -2233,6 +2368,9 @@ export class MetadataIndexManager { * NOTE: Sparse indices are flushed immediately in add/remove operations */ async flush(): Promise { + // Flush any deferred chunk/sparse writes first + await this.flushDirtyMetadata() + // Check if we have anything to flush if (this.dirtyFields.size === 0) { return // Nothing to flush diff --git a/src/utils/metadataWriteBuffer.ts b/src/utils/metadataWriteBuffer.ts new file mode 100644 index 00000000..a0c0df50 --- /dev/null +++ b/src/utils/metadataWriteBuffer.ts @@ -0,0 +1,132 @@ +/** + * Metadata Write Buffer — Deduplicates rapid writes to the same cloud storage path. + * + * When multiple brain.add() calls happen in rapid succession (e.g., chat: store message + * + create conversation + auto-title), the SAME sparse index and chunk files get written + * repeatedly. This buffer deduplicates writes to the same cloud storage path across + * multiple operations using a time-windowed buffer. + * + * Latest data wins — if the same path is written 5 times in 200ms, only the final + * version is actually sent to cloud storage. + * + * NOT used by FileSystem adapter — local writes are already fast (~1ms), and buffering + * would add unnecessary latency. + */ + +import { prodLog } from './logger.js' + +export class MetadataWriteBuffer { + private pendingWrites = new Map() + private flushTimer: ReturnType | null = null + private isFlushing = false + private pendingFlushPromise: Promise | null = null + private writeFunction: (path: string, data: any) => Promise + + private maxBufferSize: number + private flushIntervalMs: number + private concurrencyLimit: number + + constructor( + writeFunction: (path: string, data: any) => Promise, + options?: { + maxBufferSize?: number + flushIntervalMs?: number + concurrencyLimit?: number + } + ) { + this.writeFunction = writeFunction + this.maxBufferSize = options?.maxBufferSize ?? 200 + this.flushIntervalMs = options?.flushIntervalMs ?? 200 + this.concurrencyLimit = options?.concurrencyLimit ?? 10 + this.startPeriodicFlush() + } + + /** + * Buffer a write to the given path. Latest data wins — if the same path + * is written multiple times before flush, only the last version is sent. + */ + async write(path: string, data: any): Promise { + this.pendingWrites.set(path, data) + + if (this.pendingWrites.size >= this.maxBufferSize) { + await this.flush() + } + } + + /** + * Flush all pending writes to cloud storage. + * Respects concurrency limits to avoid overwhelming the cloud API. + */ + async flush(): Promise { + if (this.isFlushing) { + if (this.pendingFlushPromise) await this.pendingFlushPromise + return + } + if (this.pendingWrites.size === 0) return + + this.isFlushing = true + const writes = new Map(this.pendingWrites) + this.pendingWrites.clear() + + this.pendingFlushPromise = this.doFlush(writes) + try { + await this.pendingFlushPromise + } finally { + this.isFlushing = false + this.pendingFlushPromise = null + } + } + + private async doFlush(writes: Map): Promise { + const entries = Array.from(writes.entries()) + + for (let i = 0; i < entries.length; i += this.concurrencyLimit) { + const batch = entries.slice(i, i + this.concurrencyLimit) + const results = await Promise.allSettled( + batch.map(([path, data]) => this.writeFunction(path, data)) + ) + + // Log failures but don't throw — individual write errors are handled by retry logic + for (let j = 0; j < results.length; j++) { + if (results[j].status === 'rejected') { + const [path] = batch[j] + prodLog.warn(`MetadataWriteBuffer: failed to write ${path}:`, (results[j] as PromiseRejectedResult).reason) + } + } + } + } + + private startPeriodicFlush(): void { + this.flushTimer = setInterval(() => { + if (this.pendingWrites.size > 0) { + this.flush().catch((err) => { + prodLog.warn('MetadataWriteBuffer: periodic flush error:', err) + }) + } + }, this.flushIntervalMs) + + // Prevent timer from keeping the process alive + if (this.flushTimer && typeof this.flushTimer === 'object' && 'unref' in this.flushTimer) { + this.flushTimer.unref() + } + } + + /** + * Drain all pending writes and stop the periodic flush timer. + * Must be called during close/destroy to ensure all data is written. + */ + async destroy(): Promise { + if (this.flushTimer) { + clearInterval(this.flushTimer) + this.flushTimer = null + } + await this.flush() + } + + /** + * Get the number of pending writes in the buffer. + */ + get size(): number { + return this.pendingWrites.size + } +}