fix: eliminate cloud storage write amplification and rate limiting
brain.add() was generating 26-40 immediate cloud writes per call, causing HTTP 429 rate limit errors and high latency on GCS/S3/R2/Azure. Three-layer fix: (1) deferred metadata writes with dirty-marking, (2) MetadataWriteBuffer for write coalescing, (3) retry/backoff on all cloud storage adapters.
This commit is contained in:
parent
23e1c56ae0
commit
92d9420a5c
10 changed files with 674 additions and 75 deletions
138
docs/operations/cloud-run-filestore-guide.md
Normal file
138
docs/operations/cloud-run-filestore-guide.md
Normal file
|
|
@ -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.
|
||||||
|
|
@ -1,6 +1,8 @@
|
||||||
# Google Cloud Storage Cost Optimization Guide for Brainy
|
# 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**)
|
> **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
|
## Overview
|
||||||
|
|
||||||
Brainy provides enterprise-grade cost optimization features for Google Cloud Storage, including lifecycle policies and Autoclass for automatic tier management.
|
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
|
- After: $740,000/year
|
||||||
- **Savings: $2,120,000/year (74%)**
|
- **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
|
**Last Updated**: 2025-10-17
|
||||||
|
|
|
||||||
|
|
@ -6530,6 +6530,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
this.originalConsole = undefined
|
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
|
// Storage doesn't have close in current interface
|
||||||
// We'll just mark as not initialized
|
// We'll just mark as not initialized
|
||||||
this.initialized = false
|
this.initialized = false
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,7 @@ import {
|
||||||
import { BrainyError } from '../../errors/brainyError.js'
|
import { BrainyError } from '../../errors/brainyError.js'
|
||||||
import { CacheManager } from '../cacheManager.js'
|
import { CacheManager } from '../cacheManager.js'
|
||||||
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
||||||
|
import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js'
|
||||||
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
||||||
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
||||||
import { getCoalescer, RequestCoalescer } from '../../utils/requestCoalescer.js'
|
import { getCoalescer, RequestCoalescer } from '../../utils/requestCoalescer.js'
|
||||||
|
|
@ -201,6 +202,12 @@ export class AzureBlobStorage extends BaseStorage {
|
||||||
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
||||||
|
|
||||||
// Write buffering always enabled - no env var check needed
|
// 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,6 +904,10 @@ export class AzureBlobStorage extends BaseStorage {
|
||||||
// Lazy container validation for progressive init
|
// Lazy container validation for progressive init
|
||||||
await this.ensureValidatedForWrite()
|
await this.ensureValidatedForWrite()
|
||||||
|
|
||||||
|
const MAX_RETRIES = 5
|
||||||
|
let lastError: any
|
||||||
|
|
||||||
|
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
|
||||||
try {
|
try {
|
||||||
this.logger.trace(`Writing object to path: ${path}`)
|
this.logger.trace(`Writing object to path: ${path}`)
|
||||||
|
|
||||||
|
|
@ -907,10 +918,31 @@ export class AzureBlobStorage extends BaseStorage {
|
||||||
})
|
})
|
||||||
|
|
||||||
this.logger.trace(`Object written successfully to ${path}`)
|
this.logger.trace(`Object written successfully to ${path}`)
|
||||||
} catch (error) {
|
if (attempt > 0) {
|
||||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
this.clearThrottlingState()
|
||||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
|
||||||
}
|
}
|
||||||
|
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}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -34,6 +34,7 @@ import {
|
||||||
import { BrainyError } from '../../errors/brainyError.js'
|
import { BrainyError } from '../../errors/brainyError.js'
|
||||||
import { CacheManager } from '../cacheManager.js'
|
import { CacheManager } from '../cacheManager.js'
|
||||||
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
||||||
|
import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js'
|
||||||
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
||||||
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
||||||
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
||||||
|
|
@ -240,6 +241,12 @@ export class GcsStorage extends BaseStorage {
|
||||||
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
||||||
|
|
||||||
// Write buffering always enabled - no env var check needed
|
// 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,6 +847,10 @@ export class GcsStorage extends BaseStorage {
|
||||||
// Lazy bucket validation for progressive init
|
// Lazy bucket validation for progressive init
|
||||||
await this.ensureValidatedForWrite()
|
await this.ensureValidatedForWrite()
|
||||||
|
|
||||||
|
const MAX_RETRIES = 5
|
||||||
|
let lastError: any
|
||||||
|
|
||||||
|
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
|
||||||
try {
|
try {
|
||||||
this.logger.trace(`Writing object to path: ${path}`)
|
this.logger.trace(`Writing object to path: ${path}`)
|
||||||
|
|
||||||
|
|
@ -850,10 +861,31 @@ export class GcsStorage extends BaseStorage {
|
||||||
})
|
})
|
||||||
|
|
||||||
this.logger.trace(`Object written successfully to ${path}`)
|
this.logger.trace(`Object written successfully to ${path}`)
|
||||||
} catch (error) {
|
if (attempt > 0) {
|
||||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
this.clearThrottlingState()
|
||||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
|
||||||
}
|
}
|
||||||
|
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}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -37,6 +37,7 @@ import {
|
||||||
import { BrainyError } from '../../errors/brainyError.js'
|
import { BrainyError } from '../../errors/brainyError.js'
|
||||||
import { CacheManager } from '../cacheManager.js'
|
import { CacheManager } from '../cacheManager.js'
|
||||||
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
||||||
|
import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js'
|
||||||
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
||||||
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
||||||
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
||||||
|
|
@ -169,6 +170,12 @@ export class R2Storage extends BaseStorage {
|
||||||
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
||||||
|
|
||||||
// Write buffering always enabled - no env var check needed
|
// 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,6 +618,10 @@ export class R2Storage extends BaseStorage {
|
||||||
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
||||||
await this.ensureInitialized()
|
await this.ensureInitialized()
|
||||||
|
|
||||||
|
const MAX_RETRIES = 5
|
||||||
|
let lastError: any
|
||||||
|
|
||||||
|
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
|
||||||
try {
|
try {
|
||||||
this.logger.trace(`Writing object to path: ${path}`)
|
this.logger.trace(`Writing object to path: ${path}`)
|
||||||
|
|
||||||
|
|
@ -625,10 +636,31 @@ export class R2Storage extends BaseStorage {
|
||||||
)
|
)
|
||||||
|
|
||||||
this.logger.trace(`Object written successfully to ${path}`)
|
this.logger.trace(`Object written successfully to ${path}`)
|
||||||
} catch (error) {
|
if (attempt > 0) {
|
||||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
this.clearThrottlingState()
|
||||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
|
||||||
}
|
}
|
||||||
|
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}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,7 @@ import {
|
||||||
import { BrainyError } from '../../errors/brainyError.js'
|
import { BrainyError } from '../../errors/brainyError.js'
|
||||||
import { CacheManager } from '../cacheManager.js'
|
import { CacheManager } from '../cacheManager.js'
|
||||||
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
import { createModuleLogger, prodLog } from '../../utils/logger.js'
|
||||||
|
import { MetadataWriteBuffer } from '../../utils/metadataWriteBuffer.js'
|
||||||
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
import { getGlobalSocketManager } from '../../utils/adaptiveSocketManager.js'
|
||||||
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
import { getGlobalBackpressure } from '../../utils/adaptiveBackpressure.js'
|
||||||
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
import { getWriteBuffer, WriteBuffer } from '../../utils/writeBuffer.js'
|
||||||
|
|
@ -256,6 +257,12 @@ export class S3CompatibleStorage extends BaseStorage {
|
||||||
// Initialize cache managers
|
// Initialize cache managers
|
||||||
this.nounCacheManager = new CacheManager<HNSWNode>(options.cacheConfig)
|
this.nounCacheManager = new CacheManager<HNSWNode>(options.cacheConfig)
|
||||||
this.verbCacheManager = new CacheManager<Edge>(options.cacheConfig)
|
this.verbCacheManager = new CacheManager<Edge>(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,7 +1808,11 @@ export class S3CompatibleStorage extends BaseStorage {
|
||||||
// Lazy bucket validation for progressive init
|
// Lazy bucket validation for progressive init
|
||||||
await this.ensureValidatedForWrite()
|
await this.ensureValidatedForWrite()
|
||||||
|
|
||||||
// Apply backpressure before starting operation
|
const MAX_RETRIES = 5
|
||||||
|
let lastError: any
|
||||||
|
|
||||||
|
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
|
||||||
|
// Apply backpressure before each attempt
|
||||||
const requestId = await this.applyBackpressure()
|
const requestId = await this.applyBackpressure()
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
|
@ -1832,12 +1843,33 @@ export class S3CompatibleStorage extends BaseStorage {
|
||||||
|
|
||||||
// Release backpressure on success
|
// Release backpressure on success
|
||||||
this.releaseBackpressure(true, requestId)
|
this.releaseBackpressure(true, requestId)
|
||||||
} catch (error) {
|
if (attempt > 0) {
|
||||||
|
this.clearThrottlingState()
|
||||||
|
}
|
||||||
|
return
|
||||||
|
} catch (error: any) {
|
||||||
// Release backpressure on error
|
// Release backpressure on error
|
||||||
this.releaseBackpressure(false, requestId)
|
this.releaseBackpressure(false, requestId)
|
||||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
lastError = error
|
||||||
throw new Error(`Failed to write object to ${path}: ${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}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,7 @@ import { BlobStorage, type COWStorageAdapter } from './cow/BlobStorage.js'
|
||||||
import { CommitLog } from './cow/CommitLog.js'
|
import { CommitLog } from './cow/CommitLog.js'
|
||||||
import { unwrapBinaryData, wrapBinaryData } from './cow/binaryDataCodec.js'
|
import { unwrapBinaryData, wrapBinaryData } from './cow/binaryDataCodec.js'
|
||||||
import { prodLog } from '../utils/logger.js'
|
import { prodLog } from '../utils/logger.js'
|
||||||
|
import { MetadataWriteBuffer } from '../utils/metadataWriteBuffer.js'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Storage key analysis result
|
* Storage key analysis result
|
||||||
|
|
@ -217,6 +218,11 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
||||||
// Track if type counts have been rebuilt (prevent repeated rebuilds)
|
// Track if type counts have been rebuilt (prevent repeated rebuilds)
|
||||||
private typeCountsRebuilt = false
|
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
|
* Analyze a storage key to determine its routing and path
|
||||||
* @param id - The key to analyze (UUID or system key)
|
* @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 ensures readWithInheritance() returns data immediately, fixing "Source entity not found" bug
|
||||||
this.writeCache.set(branchPath, data)
|
this.writeCache.set(branchPath, data)
|
||||||
|
|
||||||
// Write to storage (async)
|
// 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)
|
await this.writeObjectToPath(branchPath, data)
|
||||||
|
}
|
||||||
|
|
||||||
// Cache is NOT cleared here anymore - persists until flush()
|
// Cache is NOT cleared here anymore - persists until flush()
|
||||||
// This provides a safety net for immediate queries after batch writes
|
// This provides a safety net for immediate queries after batch writes
|
||||||
|
|
|
||||||
|
|
@ -129,6 +129,12 @@ export class MetadataIndexManager {
|
||||||
private chunkManager: ChunkManager
|
private chunkManager: ChunkManager
|
||||||
private chunkingStrategy: AdaptiveChunkingStrategy
|
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<string, ChunkData>() // "field:chunkId" -> ChunkData
|
||||||
|
private dirtySparseIndices = new Map<string, SparseIndex>() // field -> SparseIndex
|
||||||
|
|
||||||
// Roaring Bitmap Support
|
// Roaring Bitmap Support
|
||||||
// EntityIdMapper for UUID ↔ integer conversion
|
// EntityIdMapper for UUID ↔ integer conversion
|
||||||
private idMapper: EntityIdMapper
|
private idMapper: EntityIdMapper
|
||||||
|
|
@ -651,6 +657,118 @@ export class MetadataIndexManager {
|
||||||
this.unifiedCache.set(unifiedKey, sparseIndex, 'metadata', size, 200)
|
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<void> {
|
||||||
|
if (this.dirtyChunks.size === 0 && this.dirtySparseIndices.size === 0) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
const promises: Promise<void>[] = []
|
||||||
|
|
||||||
|
// 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<string, RoaringBitmap32>()
|
||||||
|
const entries2 = new Map<string, RoaringBitmap32>()
|
||||||
|
|
||||||
|
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
|
* Get IDs for a value using chunked sparse index with roaring bitmaps
|
||||||
* Now fully lazy-loaded via UnifiedCache (no local sparseIndices Map)
|
* Now fully lazy-loaded via UnifiedCache (no local sparseIndices Map)
|
||||||
|
|
@ -934,7 +1052,8 @@ export class MetadataIndexManager {
|
||||||
|
|
||||||
// Add to chunk
|
// Add to chunk
|
||||||
await this.chunkManager.addToChunk(targetChunk, normalizedValue, id)
|
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
|
// Update chunk descriptor in sparse index
|
||||||
const updatedZoneMap = this.chunkManager.calculateZoneMap(targetChunk)
|
const updatedZoneMap = this.chunkManager.calculateZoneMap(targetChunk)
|
||||||
|
|
@ -955,11 +1074,15 @@ export class MetadataIndexManager {
|
||||||
|
|
||||||
// Check if chunk needs splitting
|
// Check if chunk needs splitting
|
||||||
if (targetChunk.entries.size > 80) {
|
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
|
// Defer sparse index save — mark dirty instead of writing immediately
|
||||||
await this.saveSparseIndex(field, sparseIndex)
|
this.dirtySparseIndices.set(field, sparseIndex)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -980,7 +1103,8 @@ export class MetadataIndexManager {
|
||||||
const chunk = await this.chunkManager.loadChunk(field, chunkId)
|
const chunk = await this.chunkManager.loadChunk(field, chunkId)
|
||||||
if (chunk && chunk.entries.has(normalizedValue)) {
|
if (chunk && chunk.entries.has(normalizedValue)) {
|
||||||
await this.chunkManager.removeFromChunk(chunk, normalizedValue, id)
|
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
|
// Update sparse index
|
||||||
const updatedZoneMap = this.chunkManager.calculateZoneMap(chunk)
|
const updatedZoneMap = this.chunkManager.calculateZoneMap(chunk)
|
||||||
|
|
@ -991,7 +1115,8 @@ export class MetadataIndexManager {
|
||||||
lastUpdated: Date.now()
|
lastUpdated: Date.now()
|
||||||
})
|
})
|
||||||
|
|
||||||
await this.saveSparseIndex(field, sparseIndex)
|
// Defer sparse index save — mark dirty instead of writing immediately
|
||||||
|
this.dirtySparseIndices.set(field, sparseIndex)
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -1432,6 +1557,10 @@ export class MetadataIndexManager {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 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
|
// Adaptive auto-flush based on usage patterns
|
||||||
if (!skipFlush) {
|
if (!skipFlush) {
|
||||||
const timeSinceLastFlush = Date.now() - this.lastFlushTime
|
const timeSinceLastFlush = Date.now() - this.lastFlushTime
|
||||||
|
|
@ -1518,6 +1647,9 @@ export class MetadataIndexManager {
|
||||||
// Invalidate cache
|
// Invalidate cache
|
||||||
this.metadataCache.invalidatePattern(`field_values_${field}`)
|
this.metadataCache.invalidatePattern(`field_values_${field}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Flush all dirty chunks and sparse indices accumulated during remove
|
||||||
|
await this.flushDirtyMetadata()
|
||||||
} else {
|
} else {
|
||||||
// Remove from all indexes (slower, requires scanning all field indexes)
|
// Remove from all indexes (slower, requires scanning all field indexes)
|
||||||
// This should be rare - prefer providing metadata when removing
|
// 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
|
* NOTE: Sparse indices are flushed immediately in add/remove operations
|
||||||
*/
|
*/
|
||||||
async flush(): Promise<void> {
|
async flush(): Promise<void> {
|
||||||
|
// Flush any deferred chunk/sparse writes first
|
||||||
|
await this.flushDirtyMetadata()
|
||||||
|
|
||||||
// Check if we have anything to flush
|
// Check if we have anything to flush
|
||||||
if (this.dirtyFields.size === 0) {
|
if (this.dirtyFields.size === 0) {
|
||||||
return // Nothing to flush
|
return // Nothing to flush
|
||||||
|
|
|
||||||
132
src/utils/metadataWriteBuffer.ts
Normal file
132
src/utils/metadataWriteBuffer.ts
Normal file
|
|
@ -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<string, any>()
|
||||||
|
private flushTimer: ReturnType<typeof setInterval> | null = null
|
||||||
|
private isFlushing = false
|
||||||
|
private pendingFlushPromise: Promise<void> | null = null
|
||||||
|
private writeFunction: (path: string, data: any) => Promise<void>
|
||||||
|
|
||||||
|
private maxBufferSize: number
|
||||||
|
private flushIntervalMs: number
|
||||||
|
private concurrencyLimit: number
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
writeFunction: (path: string, data: any) => Promise<void>,
|
||||||
|
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<void> {
|
||||||
|
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<void> {
|
||||||
|
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<string, any>): Promise<void> {
|
||||||
|
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<void> {
|
||||||
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue