Thread-Safe Buffered Batch Writer for Streaming Records into a Database

Problem Statement

In high-throughput event processing pipelines, inserting records into a database one by one creates excessive network overhead and connection contention. Instead, records are typically collected in memory and written out in batches.

Implement a thread-safe BatchWriter class that receives records from multiple concurrent producer threads, buffers them, and writes them to an external database client.

A batch must be flushed to the database when either of the following conditions is met:

1. The number of buffered records reaches batch_size.
2. More than flush_interval_ms milliseconds have elapsed since the oldest record in the current batch was added.

Crucially, external database I/O is slow. Your design must not hold the main buffer lock while invoking db_client.insert_batch(...) so that producer threads are not blocked during network calls.

Class Interface

class DatabaseClient:
    def insert_batch(self, records: list[dict]) -> None:
        """Writes a batch of records to the database. May take significant time."""
        pass

class BatchWriter:
    def __init__(self, db_client: DatabaseClient, batch_size: int, flush_interval_ms: int, max_capacity: int):
        """Initializes the writer with size and time thresholds, and backpressure limit."""
        pass

    def write(self, record: dict) -> None:
        """Adds a record to the buffer. If the buffer is full (max_capacity), blocks until space is available."""
        pass

    def flush(self) -> None:
        """Forces an immediate flush of all currently buffered records."""
        pass

    def close(self) -> None:
        """Flushes all remaining buffered records and terminates background timer threads cleanly."""
        pass

Requirements and Constraints

• batch_size >= 1, flush_interval_ms >= 10, max_capacity >= batch_size.
• write() must be thread-safe and handle backpressure: if total pending records reach max_capacity, producer threads must block until records are drained.
• Do not perform db_client.insert_batch inside a critical section locking the incoming queue; snapshot or swap the batch under the lock, release the lock, and then write.
• Calling close() should prevent any further writes, flush all remaining in-flight records, and wait for any active batch inserts to finish.

Example Scenario

# Configuration: batch_size=3, flush_interval_ms=500
writer = BatchWriter(db_client, batch_size=3, flush_interval_ms=500, max_capacity=10)

# Thread A writes record 1 at t=0ms
# Thread B writes record 2 at t=10ms
# At t=500ms, timer expires -> db_client.insert_batch([rec1, rec2]) is executed

# Thread A writes records 3, 4, 5 in rapid succession at t=600ms
# Threshold reached -> db_client.insert_batch([rec3, rec4, rec5]) is executed immediately

writer.close() # Ensures no records are dropped

Edge Cases to Consider

• close() called while producers are actively calling write().
• Empty flushes triggered by timers when no records have arrived.
• Ensuring the background timer resets accurately after an eager size-based flush.