Handlers
A buffer in front of somewhere records are sent: a log collector, an analytics endpoint, an audit sink. Records go in one at a time and go out in batches, when a batch is full or when the oldest record in it has waited long enough.
The rest of it exists because the far end is not always there. A failed batch is retried on a backoff with jitter, so a collector coming back does not meet every node at once. A far end that keeps failing opens a circuit breaker, so the buffer stops spending the caller's time on something that is down and tries again once a probe succeeds. A batch out of attempts goes to the dead-letter handler rather than back onto the queue.
The buffer is bounded, so a collector that is down does not become memory growth: past maxBufferSize records are dropped, which is what keeps the process serving requests.
Shutdown flushes what is buffered, within shutdownTimeoutMs.
b.handlers.create(opts) #
{
name: string, // used in logs and stats; default "anonymous"
flush: function, // async (batch) → void; required
maxBatch: number, // records per batch; default 100
maxAgeMs: number, // oldest-record wait; default 1000
maxBufferSize: number, // buffered records before dropping; default 10000
shutdownTimeoutMs: number, // flush budget at shutdown; default 30000
retry: object, // { maxAttempts, baseDelayMs, maxDelayMs, jitterFactor }
breaker: object, // { failureThreshold, cooldownMs, successThreshold }
deadLetter: function, // (batch, err) for a batch out of attempts
onError: function, // (err) for a failure the handler swallowed
}Open a buffered handler and answer the handle to work it. The handle carries emit(record), drain() for the batches already buffered, shutdown() and shutdownSync(), getStats(), size() and name.
A batch goes out when it reaches maxBatch records or when the oldest record in it has waited maxAgeMs, whichever comes first. flush is given the batch and is expected to be async; whether it settles is what decides the retry.
retry is { maxAttempts, baseDelayMs, maxDelayMs, jitterFactor }, and breaker is { failureThreshold, cooldownMs, successThreshold }. With the breaker open, emit still buffers and nothing is sent until the cooldown lets a probe through.
deadLetter receives a batch that has used its attempts. Without one those records are dropped, so a deployment that has to account for every record supplies it.
var b = require("@blamejs/core");
var sink = b.handlers.create({
name: "otlp-logs",
maxBatch: 200,
maxAgeMs: b.constants.TIME.seconds(2),
flush: async function (batch) { await post(batch); },
deadLetter: function (batch) { spillToDisk(batch); },
});
sink.emit({ level: "info", message: "ready" });
await sink.shutdown();