Safe Async
Timeout-bounded promises, AbortSignal-aware coordination, Promise.race-shaped helpers, and settled-state queries for the framework's async surfaces (external-db queries, cluster coordination, queue operations, audit chain writes).
Hazards this module addresses: races between interleaved awaits, unbounded retries masking real failures, hangs from unresponsive backends, and partial results from operator-supplied drivers.
Surface: - Async coordination: withTimeout, withSignal, withTimeoutSignal, sleep, repeating, flushLoop, safeAwait, parallel, asyncRetry - Async state objects: Mutex, Semaphore, Once, CircuitBreaker - Sync helpers used by async pipelines: safeInvoke (callback wrapper with optional onError), makeDropCallback (factory for log-stream-style onDrop callbacks), makeScheduledFlush (idempotent setTimeout coalesce-and-flush helper)
Design posture: - AbortSignal everywhere. Every time-bounded primitive accepts an AbortSignal and aborts cleanly when it fires. - Error.cause preserved. Wrapper errors set .cause to the original failure so debugging traces back to the root. - No leaked Promises. Mutex / Semaphore release on path-out in finally blocks — even on cancellation. - Bounded by default. Semaphore / parallel have explicit limits and reject over-the-limit acquisitions rather than growing unboundedly. - Fail loud. Errors propagate; primitives never silently swallow. safeAwait is the opt-in {error, value} tuple form for callers that want to log-and-continue.
Best-practice notes for callers: - Pair withTimeout with external-db / network calls where operator-supplied drivers might hang. Puts a ceiling on each individual attempt. - Wrap chain-writes with Mutex.runExclusive. Audit chain hashing reads the previous tip and writes a successor; without serialization, concurrent record() calls can hash against the same prev-tip and fork the chain. - Use Once for boot-time lazy init (counter primer, schema check). Multiple concurrent first-callers correctly wait on the same in-flight init Promise. - Use safeAwait for fire-and-forget paths (audit hooks in middleware) — preserves "log + continue" without unhandled- rejection warnings. - Prefer Promise.allSettled over Promise.all when partial failure is acceptable (multiple log sinks; one down shouldn't block the others).
b.safeAsync.withTimeout(promise, ms, opts?) #
{
signal: AbortSignal, // aborts the wrapper with async/aborted
name: string, // diagnostic label baked into error messages
}Race a Promise against a wall-clock deadline. On timeout the wrapper rejects with SafeAsyncError (.code = "async/timeout"); the underlying Promise keeps running in the background since the framework cannot cancel an arbitrary async operation. Pair with AbortSignal-aware I/O when the caller also wants the work itself to stop. opts.signal aborts the wrapper with .code = "async/aborted"; opts.name is included in the timeout message for diagnostics.
var b = require("blamejs");
// Bound an HTTP call to 5s.
var fetchUser = Promise.resolve({ id: 42, name: "alice" });
var user = await b.safeAsync.withTimeout(fetchUser, 5000, { name: "fetchUser" });
user.id;
// → 42
// Timeout surfaces as SafeAsyncError(async/timeout).
var hang = new Promise(function () {});
try { await b.safeAsync.withTimeout(hang, 10, { name: "stuck" }); }
catch (e) { e.code; }
// → "async/timeout"b.safeAsync.withSignal(promise, signal) #
Race a Promise against an AbortSignal. When the signal aborts the wrapper rejects with SafeAsyncError (.code = "async/aborted", .cause = signal.reason). The underlying Promise continues running in the background — only the wrapper's resolution is short-circuited. Useful for plumbing one signal through a chain of awaits where some intermediates aren't signal-aware.
var b = require("blamejs");
// Propagate an AbortSignal through a non-signal-aware Promise.
var ctrl = new AbortController();
var slow = new Promise(function (resolve) { setTimeout(resolve, 50, "done"); });
var wrapped = b.safeAsync.withSignal(slow, ctrl.signal);
ctrl.abort();
try { await wrapped; }
catch (e) { e.code; }
// → "async/aborted"b.safeAsync.sleep(ms, opts?) #
{
signal: AbortSignal, // aborts mid-sleep with async/aborted
unref: boolean, // default false; true to not keep the process alive
}Promise that resolves after ms milliseconds. opts.signal aborts the sleep cleanly — the wrapper rejects with SafeAsyncError (.code = "async/aborted"). opts.unref flips the timer to non-process-holding (default false, so await sleep(ms) reads naturally as "I'm waiting, this IS my work"). ms <= 0 resolves immediately; non-finite ms rejects.
var b = require("blamejs");
// Backoff between retries.
var t0 = Date.now();
await b.safeAsync.sleep(20);
(Date.now() - t0) >= 18;
// → true
// Abort mid-sleep — propagates as SafeAsyncError(async/aborted).
var ctrl = new AbortController();
setTimeout(function () { ctrl.abort(); }, 5);
try { await b.safeAsync.sleep(1000, { signal: ctrl.signal }); }
catch (e) { e.code; }
// → "async/aborted"b.safeAsync.withTimeoutSignal(signal, ms) #
Compose an existing AbortSignal with a fresh wall-clock timeout. Returns an AbortSignal that fires when EITHER the input signal aborts OR ms milliseconds elapse — exactly the shape I/O primitives like fetch({ signal }) already accept. Edge cases: neither argument supplied returns null (a naturally falsy "no signal needed" value most signal-accepting APIs treat as no-op); only signal returns it unchanged; only ms returns AbortSignal.timeout(ms).
var b = require("blamejs");
// Add a 5s deadline on top of the user's existing AbortSignal.
var userCtrl = new AbortController();
var sig = b.safeAsync.withTimeoutSignal(userCtrl.signal, 5000);
sig instanceof AbortSignal;
// → true
// No user signal + no timeout returns null (no-abort sentinel).
b.safeAsync.withTimeoutSignal(null, 0);
// → nullb.safeAsync.safeAwait(promise) #
Go-style [error, value] tuple wrapper. Never throws — a rejected Promise becomes [error, null], a resolved Promise becomes [null, value]. Replaces try/catch scaffolding around fire-and-forget paths (audit hooks in middleware, optional lookups) where the caller wants to log-and-continue without unhandled-rejection warnings. For settled-state inspection of many concurrent Promises the standard Promise.allSettled pairs naturally with this idiom.
var b = require("blamejs");
// Resolved Promise → [null, value].
var ok = await b.safeAsync.safeAwait(Promise.resolve(42));
ok[0];
// → null
ok[1];
// → 42
// Rejected Promise → [error, null].
var bad = await b.safeAsync.safeAwait(Promise.reject(new Error("nope")));
bad[0].message;
// → "nope"
// Pair with Promise.allSettled for bulk settled-state inspection.
var results = await Promise.all([
b.safeAsync.safeAwait(Promise.resolve("a")),
b.safeAsync.safeAwait(Promise.reject(new Error("b-failed"))),
b.safeAsync.safeAwait(Promise.resolve("c")),
]);
results.filter(function (r) { return r[0] === null; }).length;
// → 2b.safeAsync.safeInvoke(callback, payload, onError) #
Drop-silent operator-callback invoker. Calls callback(payload) if callback is a function, routes any throw to onError(e) if supplied, and silently swallows nested throws from onError itself. Used by every drop-callback / completion-callback / failure-callback site in the framework so a buggy operator callback can never crash the request that triggered the audit hook. Hot-path observability sink — drop-silent by design.
var b = require("blamejs");
// Happy path: callback runs with the payload.
var seen = null;
b.safeAsync.safeInvoke(function (p) { seen = p; }, { reason: "buffer-full", batch: [1, 2] });
seen.reason;
// → "buffer-full"
// Throw routed to onError; original caller never sees it.
var caught = null;
b.safeAsync.safeInvoke(
function () { throw new Error("boom"); },
{ batch: [] },
function (e) { caught = e.message; }
);
caught;
// → "boom"b.safeAsync.makeDropCallback(onDrop, onError) #
Factory for the canonical log-stream-sink onDrop wrapper. Returns a closure (reason, batch, err) => void that calls onDrop with the framework-canonical payload shape { reason, batch, error }, routing any throw from the operator callback to onError. Every sink (cloudwatch / otlp-grpc / otlp-http / syslog / webhook) previously rolled its own three-line _emitDrop wrapper — this factory removes that duplication.
var b = require("blamejs");
var dropped = [];
var emit = b.safeAsync.makeDropCallback(
function (info) { dropped.push(info); },
function (e) { console.warn("onDrop threw: " + e.message); }
);
emit("buffer-full", [{ id: 1 }], new Error("queue overflow"));
dropped[0].reason;
// → "buffer-full"
dropped[0].error.message;
// → "queue overflow"b.safeAsync.makeScheduledFlush(delayMs, flushFn) #
Idempotent setTimeout coalesce-and-flush scheduler used by every log-stream sink to batch buffered writes. Returns { schedule, cancel, isPending } — calling schedule() repeatedly within delayMs collapses to a single deferred flushFn() call. The timer is unref'd so a pending flush never keeps the process alive; async rejections from flushFn are swallowed (best-effort sink — operators see drops via the sink's own onDrop). Throws TypeError on bad arguments at construction time.
var b = require("blamejs");
// Coalesce many schedule() calls into one flush after delayMs.
var flushed = 0;
var sched = b.safeAsync.makeScheduledFlush(20, function () { flushed += 1; });
sched.schedule();
sched.schedule();
sched.schedule();
sched.isPending();
// → true
await b.safeAsync.sleep(40);
flushed;
// → 1b.safeAsync.makeBufferedEnqueue(buffer, opts) #
{
batchSize: number, // flush when buffer reaches this depth
bufferLimit: number, // drop oldest once buffer exceeds this depth
flush: Function, // () => Promise — non-awaited batch drain
schedule: Function, // () => void — coalescing deferred flush
onOverflow: Function, // (dropped) => void — drop accounting (optional)
}Backpressure enqueue for batching egress sinks. Returns an enqueue(entry) function that pushes entry onto the operator-owned buffer array with drop-oldest overflow protection, then either kicks a flush when the batch is full or defers to a coalescing scheduler. Resolves { accepted: true, queued } with the post-enqueue depth.
This is the shared hot-path decision every batching log-stream sink (CloudWatch, OTLP/HTTP, webhook) makes per record: bound the buffer, surface dropped records to drop accounting, and trigger delivery on a full batch without awaiting it. Bounding is mandatory — an unbounded buffer behind a slow or dead collector is an out-of-memory vector.
opts.flush returns the in-flight drain promise (its rejection is swallowed here — the sink reports failures through its own onDrop). opts.schedule is the coalescing deferral (typically a makeScheduledFlush handle's schedule). opts.onOverflow(dropped) is invoked with the evicted record so the caller can increment its drop counter and emit a drop event. Validates wiring at construction (TypeError) so a sink author's typo surfaces at setup, not under load.
var b = require("blamejs");
var buffer = [];
var dropped = 0;
var sched = b.safeAsync.makeScheduledFlush(20, drain);
var enqueue = b.safeAsync.makeBufferedEnqueue(buffer, {
batchSize: 100,
bufferLimit: 1000,
flush: drain,
schedule: sched.schedule,
onOverflow: function () { dropped += 1; },
});
function drain() { buffer.length = 0; return Promise.resolve(); }
await enqueue({ message: "hi" });
// → { accepted: true, queued: 1 }b.safeAsync.makeDrainingClose(opts) #
{
scheduler: Object, // { cancel() } — the coalescing flush handle
getInflight: Function, // () => Promise|null — current in-flight drain
flush: Function, // () => Promise — final drain
markClosed: Function, // () => void — flip the closed flag last
}Graceful shutdown for a batching egress sink. Returns an async close() that cancels the coalescing scheduler, awaits any in-flight drain, runs one final flush, then marks the sink closed — in that order, because the order is load-bearing.
The flush loop typically guards on !closed to stop pulling from the buffer; flipping closed first would strand the records an operator queued in the moment before shutdown. Draining before the flip is the difference between a clean shutdown and silently dropped tail records (lost logs, lost audit). This primitive encodes that invariant once so each sink can't reintroduce the reorder.
opts.getInflight is read at close time (not construction) so it observes whatever drain is running then; its rejection is swallowed — the sink surfaces flush failures through its own onDrop. opts.flush runs the final drain; opts.markClosed flips the sink's closed flag.
var b = require("blamejs");
var closed = false, inflight = null, buffer = [];
var sched = b.safeAsync.makeScheduledFlush(20, drain);
function drain() { inflight = Promise.resolve(); return inflight; }
var close = b.safeAsync.makeDrainingClose({
scheduler: sched,
getInflight: function () { return inflight; },
flush: drain,
markClosed: function () { closed = true; },
});
await close();
closed;
// → trueb.safeAsync.makeBatchDrain(opts) #
{
buffer: Array, // operator-owned record buffer
batchSize: number, // default splice width
scheduler: Object, // { schedule() } — reschedule when records remain
isClosed: Function, // () => boolean — polled each iteration
sendBatch: Function, // (batch) => Promise — throw ⇒ permanent reject
onRetryExhausted: Function, // (batch, err) => void — permanent-reject accounting
takeBatch: Function, // (buffer) => batch — optional; default splice(0, batchSize)
beforeDrain: Function, // () => Promise — optional pre-loop step
onBeforeDrainFail: Function,// (records, err) => void — optional; beforeDrain threw
}The drain loop behind a batching egress sink. Owns the single-flight latch and returns { flush, getInflight, isInFlight }. Calling flush() while a drain is in progress returns that same in-flight promise (one drain at a time); otherwise it pulls batches off the operator-owned buffer and ships each via opts.sendBatch until the buffer empties or the sink closes, rescheduling itself if records remain.
opts.sendBatch(batch) is the per-sink transport (serialize + send, typically wrapped in retry); a throw means the batch is permanently rejected — the loop reports it via opts.onRetryExhausted(batch, err) and stops (the buffer keeps the rest for the next cycle). opts.isClosed is polled each iteration so a shutdown stops the loop promptly.
Two optional hooks cover sinks that need more than a plain splice: opts.takeBatch(buffer) returns the next batch (default buffer.splice(0, batchSize)) for sinks with a byte-size cap; and opts.beforeDrain() runs once before the loop (e.g. ensure a remote log stream exists) — if it throws, the whole buffer is drained to opts.onBeforeDrainFail(records, err) as a permanent drop, since every batch would hit the same failure.
var b = require("blamejs");
var buffer = [{ n: 1 }, { n: 2 }];
var sent = [];
var sched = b.safeAsync.makeScheduledFlush(20, function () {});
var drain = b.safeAsync.makeBatchDrain({
buffer: buffer,
batchSize: 10,
scheduler: sched,
isClosed: function () { return false; },
sendBatch: function (batch) { sent.push(batch); return Promise.resolve(); },
onRetryExhausted: function () {},
});
await drain.flush();
sent.length;
// → 1b.safeAsync.makeBatchingSink(opts) #
{
batchSize: number, // flush at this depth
bufferLimit: number, // drop oldest past this depth
maxBatchAgeMs: number, // coalescing flush delay
sendBatch: Function, // (batch) => Promise — transport
onDrop: Function, // ({reason,batch,error}) => void (optional)
prepareRecord: Function, // (record) => {entry}|{rejected,...} (optional)
takeBatch: Function, // () => batch (optional; byte-cap sinks)
beforeDrain: Function, // () => Promise (optional pre-drain step)
beforeDrainDropKind: string, // drop kind when beforeDrain fails
}The complete batching egress-sink core — buffer, drop accounting, single-flight drain, and graceful close, wired together. Returns { emit, close, flush, stats }. A sink built on this provides only its transport (sendBatch) and config; the bounded buffer, overflow and retry-exhaustion drop counting, batch-full flushing, and drain-before-close shutdown all come from here.
It composes the three lower-level primitives — makeBufferedEnqueue (backpressure), makeBatchDrain (single-flight drain), makeDrainingClose (shutdown) — so every sink shares one implementation of the parts that are easy to get subtly wrong (an unbounded buffer is an OOM vector; flipping closed before the final drain strands tail records).
opts.sendBatch(batch) is the transport; a throw means permanent rejection (counted as a drop, reported via onDrop "retry-exhausted"). opts.prepareRecord(record) optionally transforms or rejects a record before buffering — return { entry } to buffer entry, or { rejected: true, reason, dropKind?, drop?, error? } to refuse it (e.g. an oversize event past a provider's hard cap). opts.takeBatch and opts.beforeDrain are forwarded to the drain (byte-cap batching; a pre-drain handshake whose failure drops the buffer under opts.beforeDrainDropKind).
var b = require("blamejs");
var sent = [];
var sink = b.safeAsync.makeBatchingSink({
batchSize: 2,
bufferLimit: 100,
maxBatchAgeMs: 50,
sendBatch: function (batch) { sent.push(batch); return Promise.resolve(); },
});
await sink.emit({ message: "a" });
await sink.emit({ message: "b" }); // batch full → flush
await sink.close();
sent.length;
// → 1b.safeAsync.parallel(items, fn, opts?) #
{
concurrency: number, // 1..256; default 8
signal: AbortSignal, // refuses to dispatch further items; in-flight run to settle
}Bounded-concurrency mapAsync. Runs fn(item, index) over items with at most opts.concurrency in-flight at a time and resolves with results in INPUT order (not completion order). Worker-loop scheduling: a fixed pool of workers each pull the next index from a shared cursor as soon as their previous task settles — avoids the Promise.all-batched-chunks pitfall where a long-pole straggler leaves workers idle. The first rejection is propagated; still-in-flight calls finish in the background (operator-supplied promises may not be signal-aware). opts.concurrency validates at config time (1..256, default 8) and throws on out-of-range so typos surface immediately.
var b = require("blamejs");
var urls = ["a", "b", "c", "d"];
var fetchOne = function (u) { return Promise.resolve("loaded:" + u); };
var results = await b.safeAsync.parallel(urls, fetchOne, { concurrency: 2 });
results;
// → ["loaded:a", "loaded:b", "loaded:c", "loaded:d"]
// First rejection wins; remaining workers drain.
try {
await b.safeAsync.parallel([1, 2, 3], function (n) {
if (n === 2) return Promise.reject(new Error("bad-2"));
return Promise.resolve(n);
}, { concurrency: 1 });
} catch (e) {
e.message;
// → "bad-2"
}b.safeAsync.repeating(fn, intervalMs, opts?) #
{
unref: boolean, // default true
onError: function(error), // captures sync throws + Promise rejections
name: string, // diagnostic label
}Bounded-cadence interval timer with consistent unref + cancel semantics. Replaces the scattered setInterval ceremony where each caller hand-rolled t.unref() and a corresponding clearInterval in shutdown. fn may be sync or async; if async, the next tick fires intervalMs after the prior fn() STARTED (fixed-rate, matching setInterval). Promise rejections are captured by opts.onError if provided, otherwise silently dropped — a repeating timer is fire-and-forget by definition and an unhandled rejection here would crash the process. opts.unref defaults true; set false for cluster heartbeat-style timers that must hold the loop open. Returns { stop }.
var b = require("blamejs");
var ticks = 0;
var sweep = b.safeAsync.repeating(function () { ticks += 1; }, 10, {
unref: true,
name: "tick-counter",
});
await b.safeAsync.sleep(35);
sweep.stop();
ticks >= 2;
// → trueb.safeAsync.flushLoop(fn, intervalMs, opts?) #
{
onError: function(error), // captures sync throws + Promise rejections
name: string, // diagnostic label
}After-completion background flusher. Schedules fn(), awaits its settle (resolve OR reject), then schedules the next call intervalMs later. Differs from repeating (fixed-rate, no overlap protection) — flushLoop is the right shape for background flushers that must never overlap two flushes and shouldn't accumulate backlog when one flush is slow. Always unref'd; opts.onError catches rejections, otherwise they're silently dropped. Returns { stop }.
var b = require("blamejs");
var flushes = 0;
var loop = b.safeAsync.flushLoop(function () {
flushes += 1;
return Promise.resolve();
}, 10, { name: "telemetry-flush" });
await b.safeAsync.sleep(35);
loop.stop();
flushes >= 1;
// → trueb.safeAsync.keyedSerializer() #
Serializes async work per key: run(key, fn) queues fn behind any in-flight or queued work for the same key and runs it once they settle, so a read-modify-write or a check-then-create on a shared store cannot interleave with another call for the same key in the same process. Different keys run concurrently. The per-key chain is dropped once it drains, so the map does not grow without bound.
In-process only: it serializes calls within ONE process. A registry shared across processes still needs its backend's own atomic create / unique constraint to refuse a cross-process duplicate.
var reg = b.safeAsync.keyedSerializer();
// concurrent register("acme") calls apply one-at-a-time, so the second
// sees the first's row and is refused as a duplicate:
await reg.run("acme", function () { return register("acme", row); });Last updated 2026-08-08T16:39:49.652Z by seeder.