# buffered-sink

## Purpose

`BufferedSink` is an in-memory queuing wrapper around any `Sink` implementation. It decouples the hot path of log record production from the potentially slower I/O performed by the underlying sink, allowing burst writes to be absorbed without blocking callers. The buffer is bounded by a configurable `maxQueueSize`, and when that limit is reached the sink applies one of two overflow policies: silently drop the excess record (`"drop"`) or drain the head record inline before pushing the new one (`"block"`). The `"block"` policy never parks writers in an external waiter list; it always makes forward progress by awaiting `inner.write` of the oldest record before accepting the new one. This eliminates the deadlock hazard of the original "wait until someone calls `flush()` externally" design.

This class is designed for high-throughput server processes where log writes must not stall request handling, and for test harnesses where controlling when records reach the underlying sink is essential for deterministic assertions.

## Features

- Wraps any `Sink` and delegates `write`, `flush`, and `close` calls
- Bounded in-memory queue with configurable `maxQueueSize` (default 1000)
- Two overflow policies: `"drop"` silently discards excess records; `"block"` suspends the caller via a `Promise` until queue capacity frees up
- `dropped` property tracks cumulative records discarded under the `"drop"` policy
- `flush(signal?)` drains the queue sequentially, forwarding each record to the inner sink; throws `LoggerError(LOGGER.FLUSH)` if the provided `AbortSignal` fires mid-drain
- `close()` flushes all pending records then delegates to `inner.close()` if implemented
- Re-entrant flush guard — concurrent `flush()` calls short-circuit rather than double-draining
- `"block"` policy drains inline on overflow — no external `flush()` is required for `write()` to make forward progress, eliminating the deadlock hazard of waiter-list designs
- Constructor rejects non-positive `maxQueueSize` with `LoggerError(LOGGER.CONFIG)`

## Usage: Basic Example

```ts
import { BufferedSink } from "@suaveplan/logger/sink";
import { JsonSink } from "@suaveplan/logger/sink";

const json = new JsonSink({ writer: (line) => process.stdout.write(line + "\n") });

const buffer = new BufferedSink({
  sink: json,
  maxQueueSize: 500,
  overflowPolicy: "drop",
});

// Write a record — enqueued immediately, not yet flushed to stdout
await buffer.write(record);

// Drain all pending records to the underlying JsonSink
await buffer.flush();
```

## Usage: Advanced Example

```ts
import { BufferedSink } from "@suaveplan/logger/sink";
import { JsonSink } from "@suaveplan/logger/sink";

const sink = new BufferedSink({
  sink: new JsonSink({ writer: appendToFile }),
  maxQueueSize: 2000,
  overflowPolicy: "block", // back-pressure: callers await capacity
});

// Graceful shutdown with a 5-second deadline
process.on("SIGTERM", async () => {
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), 5_000);

  try {
    await sink.flush(ac.signal);
  } catch {
    // AbortSignal fired — some records may remain in the queue
    console.error(`Dropped ${sink.dropped} records during shutdown`);
  } finally {
    clearTimeout(timer);
    await sink.close();
  }
});
```

## API Reference

### `class BufferedSink implements Sink`

**Constructor**: `new BufferedSink(options: BufferedSinkOptions)`

| Option | Type | Default | Description |
|---|---|---|---|
| `sink` | `Sink` | required | The underlying sink that receives flushed records |
| `maxQueueSize` | `number` | `1000` | Maximum queued records before overflow policy triggers |
| `overflowPolicy` | `"drop" \| "block"` | `"drop"` | Action when queue is full |

**Properties**

- `dropped: number` — Cumulative count of records discarded under the `"drop"` policy. Does not reset between flushes.

**Methods**

- `write(record: LogRecord): Promise<void>` — Enqueue a record. Under `"drop"` policy, records beyond `maxQueueSize` are silently counted and discarded. Under `"block"` policy, when the queue is at capacity `write()` first awaits `inner.write` of the head record, then pushes; this ensures forward progress without requiring an external `flush()` and rules out the deadlock that a parked-writer design would expose if no one ever flushed.
- `flush(signal?: AbortSignal): Promise<void>` — Drain the entire queue by forwarding records to the inner sink in FIFO order. Throws `LoggerError(LOGGER.FLUSH)` if the signal is aborted. Throws `LoggerError(LOGGER.SINK)` if the inner sink's `write` rejects. Re-entrant calls return immediately.
- `close(): Promise<void>` — Flush then close the inner sink (if it implements `close()`).

## Implementation Notes

`BufferedSink` uses a simple array queue (`shift()` for FIFO) rather than a ring buffer. For typical logging workloads (hundreds of records per second) the GC pressure is negligible. The `"block"` policy makes forward progress inline (drains the head before pushing) so concurrent overflowing writers serialise on `inner.write` rather than queuing in a waiter list — back-pressure is expressed naturally through awaited promises. The design favours simplicity over micro-optimisation since the bottleneck is always the inner sink's I/O, not the queue operations.

There is no background flush timer — `BufferedSink` is purely demand-driven. If you need periodic auto-flush, start your own `setInterval` that calls `flush()` and clears it on shutdown.