soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

fix: eliminate cloud storage write amplification and rate limiting

Browse source 
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:
David Snelling 2026-01-31 09:09:36 -08:00
parent 23e1c56ae0
commit 92d9420a5c
10 changed files with 674 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/brainy.ts
View file

@ -6530,6 +6530,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
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

54
src/storage/adapters/azureBlobStorage.ts
View file

@ -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<Edge>(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}`)
}
/**

54
src/storage/adapters/gcsStorage.ts
View file

@ -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<Edge>(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}`)
}
/**

62
src/storage/adapters/r2Storage.ts
View file

@ -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<Edge>(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<void> {
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}`)
}
/**

90
src/storage/adapters/s3CompatibleStorage.ts
View file

@ -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<HNSWNode>(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,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}`)
}
/**

14
src/storage/baseStorage.ts
View file

@ -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

152
src/utils/metadataIndex.ts
View file

@ -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<string, ChunkData>() // "field:chunkId" -> ChunkData
private dirtySparseIndices = new Map<string, SparseIndex>() // 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<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
* 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<void> {
// 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

132
src/utils/metadataWriteBuffer.ts Normal file
View 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
}
}