@gnldev/events weekly downloads statistics

@gnldev/events

A durable event/notification bus on top of the journal. emit writes to an append-only log; each consumer processes events with its own ack marker: exactly-once marking + at-least-once delivery. The marker is written after the handler succeeds — if the handler throws or the process dies, the event isn't lost, a later poll retries it (on the backoff schedule below, until maxAttempts). The cost: between handler success and marking, a crash (or a concurrent poll race) can cause redelivery → write the handler as idempotent, or use something durable inside it (runDurable/claim). Fan-out: N consumers → each gets every event at least once, via its own marker stream.

Install: pnpm add @gnldev/events — or use it from a repo clone: pnpm install && pnpm -r build.

npm i @gnldev/events   # peer: @gnldev/durable

import { emit, createConsumer } from '@gnldev/events';
import { SqliteStorage } from '@gnldev/durable/sqlite';

const storage = new SqliteStorage('runs.db');   // emit/createConsumer take the WORK store

// Publish (idempotent: same id → a single event).
await emit(storage.work, 'refunds', { orderId: 'o1', amount: 50 }, { id: 'refund-o1' });

// Consume (each consumer name gets its own marker stream; the handler must be idempotent).
const consumer = createConsumer(storage.work, 'refunds', (payload, meta) => notify(payload), { name: 'emailer' });
await consumer.poll();   // or consumer.start()

API

• emit(work, topic, payload, { id?, maxDepth?, resourceId?, orgId? }) → eventId — work is storage.work. With an owner, the handler's meta carries resourceId/orgId, so the run it starts is that user's. An explicit id is then a name within that owner (the returned id is the stored one), maxDepth is counted per organization, and an organization-scoped work store is refused (no consumer polls it). A system event (no owner) is stored and delivered exactly as before. The owner never travels in the payload. It is recorded in the id the engine writes (ownedName), and meta is read from that id. A payload that carries its own "owner" field — a forwarded webhook, say — is delivered as bytes, never read as someone's. A system id or a topic cannot start with the owned-name marker (~o~), so no caller can spell one either.
• createConsumer(work, topic, handler, { name, pollMs?, backoff?, maxPollMs?, maxAttempts?, retryDelayMs?, leaseMs?, maxHoldMs? }) → { poll, start, stop } — name is required to separate fan-out acks; maxAttempts/retryDelayMs govern the dead-letter below.
• eventEraser(work) erases one person's events on every topic, with every consumer's ack, attempt and dead-letter marker for them, without knowing a topic or consumer name. Pass it to eraseSubject from @gnldev/durable with the queue's and the scheduler's erasers (example). An owned event's markers are keyed by its id first (evtack:<id>:<topic>:<consumer>); a system event's keys are unchanged. The work store needs deleteIdPrefix and deletePrefix, or the eraser throws.
• listDeadEvents(work, topic, consumer) → DeadEvent[] — everything quarantined for that consumer, with status (quarantined | released | delivered), attempts, error, releases and the original payload. A management call: it reads the whole topic log, so don't put it in a poll loop.
• retryDeadEvent(work, topic, consumer, eventId) → boolean — hands a quarantined event back for delivery to that one consumer (fresh attempt budget, no backoff wait). Calling it again on an event that is still released is safe and re-asserts the release. false has three causes, and the third is not "nothing to do":
  1. the event was never quarantined;
  2. it has already been delivered (the ack marker is the single source of truth for that);
  3. the release was abandoned — the dead-letter record kept being rewritten underneath the read-modify-write faster than it could complete (5 lost compare-and-swaps in a row: two operators, two tabs, a retry script racing a human). Nothing was overwritten, it is logged on console.warn, and the remedy is to call it again. Silently clobbering the other writer is the alternative, and the compare-and-swap exists to refuse it: without it, two releases landing together both read the same record and both wrote releases: n + 1 from it — both returned true and the record showed releases: 1 (measured). Distinguish it from (1) and (2) by the record itself: it is still there and still quarantined.

How it works

An append-only log per topic; consumer ack markers are separated by name → the same event flows to multiple consumers. Since the marker is written after handler success, the delivery guarantee is at-least-once: on a handler error the marker isn't written, that consumer's page cursor doesn't advance past the event, and a later poll retries it; marking itself is exactly-once via CAS (ackOnce).

One failing event does not hold back the ones behind it: the pass keeps going through the rest of the log, only the persisted bookmark stays parked. That parking is what makes retries safe, and it is also why retries are finite — see the dead-letter below.…