@kido/batch-bufferCopy Anchor Link

Architecture Paradigm: Dual Count/Timer Write-Behind Batching + Non-Blocking In-Memory & Redis List Storage Backends + Multi-Policy Backpressure Gates + Binary-Split Fallback Batch Inserter.

💥 The Real-World BottleneckCopy Anchor Link

High-throughput telemetry, audit logging, analytics events, and bulk CSV data imports trigger severe relational database bottlenecks:

1. Row-by-Row INSERT Latency Multiplier:

   Copy Text

   -- The Naive Disaster: 1,000 incoming logs per second executed individually
   INSERT INTO activity_logs (user_id, action) VALUES ($1, $2);

   Executing 1,000 separate SQL queries incurs 1,000 network round-trips and 1,000 disk WAL flushes, capping write throughput at 30-60 writes/s and exhausting connection pools.
2. The "All-or-Nothing" Bulk Insert Failure: When inserting a batch of 500 rows, if a single row violates a unique constraint or schema validation, the entire transaction aborts. In naive implementations, 499 valid rows are discarded because of 1 corrupt record.
3. Out-of-Memory Crashes During Large CSV Imports: Reading a 200MB or 1-million-line CSV file into memory using fs.readFileSync blows up the V8 heap (JavaScript heap out of memory), crashing the entire backend process.

🏛️ 3-Tier Production Evolution LadderCopy Anchor Link

📦 Architecture & Core API ExportsCopy Anchor Link

@kido/batch-buffer is partitioned into 4 high-performance slices:

1. Hybrid Batch Engine (src/hybrid-batcher.ts)Copy Anchor Link

• HybridBatcher<T>: Core batch coordinator supporting dual triggers:
  • maxBatchSize: Number of items before an immediate flush triggers (e.g., 50 items).
  • flushIntervalMs: Maximum time items remain buffered before flushing (e.g., 1000ms).
• Concurrency mutex prevents overlapping flush operations during slow database I/O spikes.
• Multi-policy backpressure gates: DROP_OLDEST, DROP_NEWEST, or BLOCK when buffer reaches capacity.

2. Storage Backends (src/backends/)Copy Anchor Link

• MemoryBackend: High-speed zero-dependency FIFO buffer running entirely in process memory.
• RedisBackend: Distributed Redis/Valkey list driver utilizing atomic Lua scripts to drain batches across multiple application pods without race conditions.

3. Read-Your-Own-Writes Audit Adapter (src/adapters/buffered-audit-adapter.ts)Copy Anchor Link

• BufferedAuditStorageAdapter: Seamlessly buffers audit logs while providing Read-Your-Own-Writes visibility: queries inspect both the committed database and the active in-memory buffer.

4. Streaming CSV Processor (src/stream-csv/)Copy Anchor Link

• processCsvStream(stream, options): Bounded-memory streaming CSV parser reading millions of rows chunk-by-chunk using Node streams.
• insertBatchWithFallback(items, inserter): Binary-split recursive fallback engine that rescues valid records when a bulk insert transaction encounters a constraint error.

💻 Production Implementation & Code SnippetCopy Anchor Link

Activity Log Buffering & Streaming CSV IngestionCopy Anchor Link

import { 
  HybridBatcher, 
  MemoryBackend,
  processCsvStream,
  insertBatchWithFallback 
} from '@kido/batch-buffer';
import fs from 'node:fs';

// 1. Initialize Hybrid Write-Behind Buffer (50 items or 1 second)
export const logBatcher = new HybridBatcher({
  backend: new MemoryBackend({ maxCapacity: 10000 }),
  maxBatchSize: 50,
  flushIntervalMs: 1000,
  async onFlush(batch) {
    // Single bulk insert replaces 50 individual SQL operations
    await db.activityLog.createMany({
      data: batch,
      skipDuplicates: true,
    });
    console.log(`Flushed batch of ${batch.length} activity logs to DB.`);
  },
});

// Fast non-blocking enqueue (< 0.1ms)
export function logUserAction(userId: string, action: string) {
  logBatcher.push({
    userId,
    action,
    timestamp: new Date(),
  });
}

// 2. High-Performance Bounded-Memory CSV Import with Binary-Split Fallback
export async function importLargeProductCatalog(filePath: string) {
  const readStream = fs.createReadStream(filePath);

  let totalImported = 0;

  await processCsvStream(readStream, {
    batchSize: 500,
    async onBatch(rows) {
      // Binary-split fallback guarantees valid rows are saved even if 1 row fails
      const result = await insertBatchWithFallback(rows, async (chunk) => {
        await db.product.createMany({ data: chunk });
      });

      totalImported += result.successfulCount;
    },
  });

  return { success: true, totalImported };
}

⚡ Benchmark & Verified k6 PerformanceCopy Anchor Link

Benchmarked against 100,000 log writes on PostgreSQL 18 with NVMe storage:

🚀 Surgical Extraction & CLI InstallationCopy Anchor Link

# Install directly via Kido CLI
npx @kido/cli add @kido/batch-buffer