@k2b/sync

NATS-native distributed synchronization primitives for TypeScript and Bun.

Sync v6 is a small, explicit layer over NATS Core, JetStream, KV, and Object Store. It does not hide the distributed semantics: durable work is at-least-once, handlers must be idempotent, ordering exists only where you declare it, and resource configuration drift fails loudly instead of being patched silently.

Nine primitives: queue, topic, job, pump, scheduler, mutex, ephemeral, objectStore, and the local retry helper.

Requirements

• NATS Server 2.14+ with JetStream enabled (three-node cluster recommended for production)
• Bun (or Node 22+) with the official NATS.js client for the connection (Sync pins its own @nats-io/* dependencies to exactly 3.4.0)
• An already connected, caller-owned NatsConnection — Sync never reads environment variables, loads credentials, or creates infrastructure

Installation

bun add @k2b/sync @nats-io/transport-node

Getting started

import { connect } from "@nats-io/transport-node";
import { createSync } from "@k2b/sync";

const connection = await connect({ servers: ["nats://nats-0:4222", "nats://nats-1:4222", "nats://nats-2:4222"] });

const sync = createSync({
  connection,
  namespace: "cloud-prod",   // deployment isolation (dev / staging / prod)
  application: "notebooks",  // ownership + diagnostics metadata
});

const runs = sync.job<{ runId: string }>({ id: "workflow-runs" });

await sync.ready(); // verifies the server, provisions resources, fails on drift

createSync() and the primitive factories perform no I/O. ready() verifies the connection and server version, creates missing resources, and compares every existing resource against its declaration — an incompatible difference throws ResourceDriftError with a field diff and mutates nothing.

Shutdown order:

await sync.drain({ timeoutMs: 30_000 }); // stop pulls, finish handlers, nak leftovers
await connection.drain();                // the connection is yours

The two concurrency knobs

These have exactly one meaning everywhere:

• process({ concurrency: 64 }) — at most 64 handlers run simultaneously in this worker handle in this process. Never a cluster limit.
• delivery.maxInFlight: 512 — the durable NATS consumer's global unacknowledged-delivery ceiling (MaxAckPending), shared by all pods.

Four pods × concurrency: 64 with maxInFlight: 512 ⇒ at most min(256, 512) running handlers. A dead pod occupies its unacknowledged slots until ackWaitMs expires, then the work is redelivered elsewhere.

Sync pulls messages only for currently free local slots — there is no hidden prefetch buffer, so unclaimed work stays on the server for other pods.

Queue

Durable work with at-least-once delivery, competing consumers, delay, retries, and a DLQ.

const emails = sync.queue<{ to: string }>({
  id: "emails",
  delivery: { ackWaitMs: 30_000, maxAttempts: 5, maxInFlight: 1_000, backoffMs: [1_000, 5_000, 30_000] },
  retention: { maxAgeMs: 7 * 24 * 3_600_000, maxBytes: 1024 ** 3 },
});

await emails.send({ data: { to: "[email protected]" }, idempotencyKey: "welcome:42" });
await emails.send({ data: { to: "[email protected]" }, delayMs: 60_000 }); // broker-side delay

const worker = await emails.process({ concurrency: 8 }, async (message) => {
  await deliver(message.data);        // resolves → acked; throws → retried, then DLQ
  await message.heartbeat();          // long handlers reset ackWait
});

const dead = await emails.deadLetters.list();
await emails.deadLetters.requeue({ messageId: dead[0].messageId, idempotencyKey: "retry-1" });

// Admin inspection: bounded oldest-first pages, including deleted-cursor recovery.
const page = await emails.deadLetters.page({ limit: 20 });
const next = page.nextCursor
  ? await emails.deadLetters.page({ limit: 20, cursor: page.nextCursor })
  : null;
const first = page.entries[0];
const detail = first
  ? await emails.deadLetters.get({ messageId: first.messageId, streamSequence: first.streamSequence })
  : null;

• idempotencyKey deduplicates within dedupeWindowMs (default 2 min), scoped per tenant.
• delayMs/at use one-shot NATS message schedules — no consumer slot is occupied while waiting.
• reader() gives manual ack() / retry() / deadLetter() settlement per message.
• sendBatch([...]) enqueues up to 1000 messages atomically (all-or-nothing; no dedupe ids — NATS batches exclude them). pause()/resume() stop and restart global delivery without touching publishers. send({ ttlMs }) makes work expire if not settled in time.
• ordering: { mode: "partitioned", partitions: 64 } hashes orderingKey to a stable partition with strictly serial per-partition delivery — including across handler failures: partitioned retries happen in place (the delivery is held with heartbeats through the backoff) so younger messages can never overtake a retrying one. The partition count becomes the global in-flight ceiling; this is for per-aggregate processing, not general fan-out.
• Without retention, a queue or job keeps work for 7 days and up to 256 messages at its payload limit (33 MiB at the 128 KiB default, at most 1 GiB). JetStream reserves maxBytes on every replica as soon as a stream exists, so size it for the backlog you expect. deadLetterRetention: { maxAgeMs?, maxBytes? } sizes the dead-letter stream separately; each field defaults to retention. Jobs take only { maxBytes? } there, because terminalRetentionMs sets their dead-letter age.
• At maxBytes or maxMessages, the default whenFull: "discard_oldest" accepts every send and silently drops the oldest pending work. whenFull: "reject" keeps accepted work and fails new sends with StoreFullError, so producers can back off or shed load. NATS cannot schedule messages on a stream that rejects new work, so these queues and jobs refuse delayMs, at, and job resubmit({ delayMs }) with SyncUsageError; retry backoff still works. maxAgeMs expires pending work in both modes. An existing stream cannot switch modes in place: drain it, delete it, and declare it again.
• Under whenFull: "reject", job continuations need room too, because resubmit() publishes the successor while its parent still occupies the stream. At the limit, a plain continuation fails like a handler error: the handler runs again, and the job is dead-lettered after maxAttempts. A coalesced continuation keeps its successor and waits for room without running the handler again, so a stream that holds only such parents stalls. Keep producers well below the limit of jobs that continue themselves.

Upgrading from 6.x:

• Queues and jobs declared without retention used 1 GiB for both the work and the dead-letter stream. Those existing streams now differ from the smaller default, and ready() throws ResourceDriftError on max_bytes. Keep the old limits by declaring retention: { maxAgeMs: 7 * 24 * 3_600_000, maxBytes: 1024 ** 3 }, or lower max_bytes of both streams to the new default (for example with nats stream edit <stream> --max-bytes <bytes>) once they hold less than that.
• A job's terminalRetentionMs must be a positive integer. 6.x accepted 0 as "no age limit" for dead letters; that now throws RangeError. Declare a long explicit age instead, such as 365 * 24 * 3_600_000. The existing dead-letter stream then fails ready() with ResourceDriftError on max_age until it has the same age (for example with nats stream edit <stream> --max-age <age>).

Queue, job, topic-consumer, and scheduler pause/resume calls wait for the broker to apply the requested state. Already delivered work can finish. A concurrent control request can prevent confirmation and cause the call to fail.

Job

The normal shape for background tasks: a queue plus a required idempotent key, retry policy, and bounded fan-out.

const runs = sync.job<{ runId: string }>({ id: "workflow-runs" });

await runs.submit({ key: `run:${runId}`, input: { runId } }); // duplicate keys dedupe (windowed)
await runs.submit({ key: "reindex", input, coalesce: true }); // ≤1 queued-or-running; key frees on completion

await runs.submitMany(
  runIds.map((runId) => ({ key: `run:${runId}`, input: { runId } })),
  { publishConcurrency: 128, maxPendingBytes: 8 * 1024 * 1024 }, // bounded promises AND bytes
);

await runs.process(
  {
    concurrency: 64,
    onError: async ({ context, error }) => {
      await persistFailure(context.jobId, error);
      return context.failureCount < 2 ? { action: "retry", delayMs: 5_000 } : { action: "dead_letter", reason: "gave up" };
    },
  },
  async (context) => {
    await runWorkflow(context.input.runId, { signal: context.signal });
  },
);

Jobs do not store results or expose join() — durable domain status belongs in your database. submitMany is not atomic: on failure a BatchSubmitError reports the accepted and duplicate counts and prior accepted items stay accepted.

coalesce: true keeps the first accepted input for each tenant and key. Both submit and submitMany use the same claim: duplicates join that job and do not replace its input. If publication fails with an unknown outcome, retrying the submission repairs the original pending job. Handlers still need to be idempotent and call heartbeat() during long work.

context.resubmit({ input, delayMs }) requests a continuation after the handler succeeds. Coalesced continuations retain the key across the handoff and preserve the original ordering key and metadata. Dead-letter requeue also respects an active coalesced key.

An accepted coalesced message whose claim record is missing (lost or expired) adopts a fresh claim and runs instead of being acknowledged silently; the run is reported as a redelivery event with detail.orphanClaimAdopted.

Upgrading existing 6.2.0 deployments: the coalescing repair (6.2.1+) requires stopping all 6.2.0 producers and workers before starting the corrected code. Do not run old and corrected writers together: old workers can overwrite or delete a newer claim. Existing queued jobs can adopt their legacy claims. A legacy pending claim without a queued message contains no recoverable input; a submission reports SyncUsageError instead of inventing a receipt. Reconcile that specific job with application-owned state before clearing its old claim and submitting it again. No streams or claims are automatically reset.

Topic

A retained event log with four deliberately different reads:

const events = sync.topic<NotebookEvent>({
  id: "notebook-events",
  retention: { maxAgeMs: 24 * 3_600_000, maxBytes: 256 * 1024 * 1024 },
});

const receipt = await events.publish({ data: event, tenantId: workspaceId });

// Optimistic per-tenant event sourcing: append only if nothing was written since.
await events.publish({ data: event, tenantId: workspaceId, expectedAfter: receipt.cursor }); // ConflictError on lost races
await events.publishBatch({ tenantId: workspaceId, events: [...], expectedAfter: receipt.cursor }); // atomic multi-event append

// 1. live(): core NATS broadcast — best-effort, no replay, every listener sees it.
for await (const event of events.live({ tenantId: workspaceId })) notifySockets(event);

// 2. replay(): from a cursor to the head captured at start, then ends.
for await (const event of events.replay({ tenantId: workspaceId, after: cursor })) apply(event);

// 3. follow(): like replay but stays open for new events.
for await (const event of events.follow({ tenantId: workspaceId, after: cursor })) apply(event);

// 4. process(): named durable consumer — pods with the same name compete,
//    different names own independent cursors.
await events.process({ consumer: "search-indexer", concurrency: 4 }, async (event) => index(event));

Cursors are opaque and resource-bound (CursorMismatchError elsewhere). If a cursor points below the retained window — in replay(), follow(), or a fresh process({ start: { after } }) consumer — Sync throws RetentionGapError instead of silently skipping; its resumeAfter cursor resumes from the first retained event without losing it. live() events carry no cursor and are suitable for invalidate-then-read, not as durable acceptance evidence.

tenantId on replay/follow is filtered on the server: a tenant-scoped read receives only that tenant's events, however many other tenants share the topic. That makes one topic with tenantId per entity the recommended shape for per-entity logs (documents, records, mailboxes). JetStream reserves every stream's maxBytes (times replicas) against the account, so a topic per entity multiplies the reservation by the number of entities; a shared topic keeps it constant.

Gap detection on a shared topic: a limits stream only loses its front (age, bytes, message count). A tenant read reports RetentionGapError when the stream no longer retains the position right after your cursor, because Sync cannot tell whether the removed events belonged to your tenant. Other tenants' events between yours are never a gap, and an idle follow() keeps its position current while others write. A cursor that is only persisted when its own tenant changes therefore goes stale once the window passes it. Keep persisted per-tenant cursors fresh: read head() before latestCursor({ tenantId }); once you have applied the tenant's events up to that latest cursor, you may persist the head() cursor instead (the tenant has no events in between). Refresh idle tenants the same way before the retained window reaches their cursor.

sync.listTopics({ idPrefix }) lists topics of the namespace that exist on the broker (not only the ones declared in this process), and topic.destroy() deletes a topic's event and dead-letter streams without provisioning anything. Together they retire old per-entity topics after their events are captured elsewhere.

deadLetterRetention: { maxAgeMs?, maxBytes? } sizes the consumer dead-letter stream independently; both default to retention. Its maxBytes must hold one dead letter (payload limit plus 4 KiB).

Retain updates until snapshots cover them

For a document log that must keep every accepted update, opt into rejection and omit the topic's age limit:

const documents = sync.topic<DocumentUpdate>({
  id: "document-updates",
  whenFull: "reject",
  retention: { maxBytes: 1024 ** 3 }, // no maxAgeMs: no age expiry
  deadLetterRetention: { maxAgeMs: 7 * 24 * 3_600_000, maxBytes: 1024 ** 2 },
});

At maxBytes or maxMessages, JetStream rejects new events without evicting accepted ones. publish() and publishBatch() throw StoreFullError when capacity refusal is confirmed. Back off and retry after snapshots free room; keep unaccepted input available to the writer. Transport timeouts have an unknown outcome and remain transport errors; they are not safe rejections.

Persist snapshots in your application's durable store, then compute the minimum covered stream sequence across all documents in this topic and call await documents.prune({ through: documents.cursorAt(minimumSequence) }). The watermark is inclusive. A missing snapshot contributes sequence zero; never omit a stalled document. Advance idle documents' persisted cursors only after proving that they have no uncovered events up to the new cursor. Serialize watermark computation with snapshot creation, restoration and document lifecycle changes so coverage cannot regress underneath pruning.

prune() returns { purged }, removes only the stream prefix at or below through, and preserves concurrent appends above it. It rejects foreign cursors and watermarks beyond the current head. Retrying after a crash is safe, including when the snapshot committed but pruning did not. Readers behind the pruned prefix must load a covering snapshot before replaying. There is no per-tenant purge: holes in the middle of a shared stream would break the retained-window gap detection. One stalled snapshot can fill the stream and block new writes; monitor the backlog and repair that snapshot.

Sync owns retention and bounded deletion. The application owns snapshot durability and the safe watermark; Sync cannot verify an external SQL commit. Acknowledging a topic consumer alone neither creates a snapshot nor prunes events. Keep snapshot work independent of new writes so a full log can recover.

The existing whenFull: "discard_oldest" default and explicitly configured ages are unchanged. An omitted topic maxAgeMs also leaves its DLQ without age expiry unless deadLetterRetention.maxAgeMs is set. Changing an existing stream's discard mode or age fails with ResourceDriftError; Sync never changes retention in place. Migrate only after persisting coverage of its accepted events. Other primitives keep their existing age requirements.

Topic DLQ entries can be inspected and deleted through deadLetters; opt-in recovery targets only the original consumer.

Pump

Checkpointed draining of a finite source (imports, backfills, reindexing). The KV run record is the truth; per-item checkpoints mean a crash repeats only ambiguous items.

const reindex = sync.pump<{ mailbox: string }, string, { key: string }>({
  id: "mail-reindex",
  batchSize: 100,
  dispatchConcurrency: 16,
  pull: async ({ input, cursor, limit }) => fetchPage(input.mailbox, cursor, limit),
  dispatch: async ({ item }) => indexItem(item.key), // must be idempotent by item.key
});

await reindex.start({ key: "mailbox:42", input: { mailbox: "42" } });
await reindex.process({ concurrency: 4 });
const state = await reindex.get({ key: "mailbox:42" }); // queued | running | waiting | completed | failed | canceled

Wake-ups are repairable: process() reconciles lost wake-ups from KV state on start, and reconcile() is callable explicitly.

Scheduler

NATS 2.14 message schedules are the clock: the broker produces durable ticks even while every application process is offline.

const cron = sync.scheduler({
  id: "maintenance",
  // Tick retention is per schedule (age + count). Global byte limits are
  // deliberately not configurable: with discard-old they would eventually
  // evict the broker-side schedule definitions and silently stop the clock.
  retention: { maxAgeMs: 7 * 24 * 3_600_000, maxTicksPerSchedule: 10_000 },
});

await cron.create({
  id: "cleanup",
  cron: "0 3 * * *",            // five-field cron, minute resolution
  timezone: "Europe/Berlin",
  misfire: "latest",             // or "all": execute every retained slot
  process: async (context) => {
    await cleanup({ signal: context.signal });
  },
});

await cron.process({ concurrency: 4 }); // runs of one schedule never overlap
await cron.runNow({ id: "cleanup", requestId: "manual-1" }); // durably accepted

misfire: "latest" coalesces ticks that accumulated during downtime and executes only the newest retained slot — the newest accepted slot is never lost. runNow returning means the run is durably accepted, not that it started or finished; repeating a requestId deduplicates within the 120 s duplicate window (like every other idempotency key in v6). delete() cancels the broker schedule and drops its retained ticks; a later create() starts fresh.

Mutex

KV compare-and-set leases with monotonic fencing.

const locks = sync.mutex({ id: "provider-refresh", ttlMs: 10_000 });

const result = await locks.withLock({ resource: "tenant:42" }, async (lock) => {
  // lock.fence is a monotonic bigint — persist and compare it if stale
  // writes to external systems after lease expiry must be excluded.
  return refresh(lock.fence);
});

acquire(), extend() and release() decide ownership from authoritative stream reads, bypassing follower-served KV direct reads. extend() retries a CAS conflict at most twice after the first attempt, rereading the current record each time. It only renews the same owner while the stored lease is unexpired and preserves the original acquisition fence. false means the lease could not be renewed; even exhausted revision contention does not prove that another owner acquired it.

An unsuccessful extend() emits lock_lost (type LockLostEvent) with LockLostDetail:

• lock: resource key;
• reason: missing (no live record, including delete/purge/TTL markers), owner_changed, expired, or revision_conflict (three rejected CAS writes while the final read still shows the same unexpired owner);
• acquisitionRevision: acquisition fence as a decimal string;
• attemptedRevision: last rejected CAS revision, or null if no CAS ran;
• observedRevision: final authoritative revision (0 for no message).

Loss diagnostics contain no owner tokens or stored values. A stopped runtime returns false without renewing or emitting a loss event. These read, retry, and diagnostic corrections ship in 7.0.1; no resource migration is needed.

A lease alone cannot stop an expired owner from writing to PostgreSQL afterwards. Consumers needing strict exclusion compare the fence or make effects idempotent.

Ephemeral

Presence, service registry, and transient state on NATS KV with per-key TTL.

const registry = sync.ephemeral<{ url: string }>({ id: "services", ttlMs: 15_000 });

await registry.upsert({ key: "api/pod-1", value: { url } });
await registry.touch({ key: "api/pod-1" });          // heartbeat: refresh TTL

const snap = await registry.snapshot({ prefix: "api/" });
for await (const event of registry.watch({ after: snap.revision })) {
  // upsert | delete | expire | resync_required
}

A watch without after starts by replaying the current entries as upserts, then streams changes. If the watch revision fell out of history, one explicit resync_required event is emitted and the watch ends — take a fresh snapshot; Sync never silently skips ahead. TTLs round up to whole seconds (NATS minimum 1s); entries read back via snapshot/watch report updatedAt and omit expiresAt (a reader cannot know custom per-key TTLs).

Object store

Explicit large-artifact storage. Sync never auto-offloads oversized payloads — you upload explicitly and pass the returned ObjectRef (a plain JSON value) through queues and jobs.

const artifacts = sync.objectStore({
  id: "workflow-artifacts",
  retention: { maxAgeMs: 7 * 24 * 3_600_000, maxBytes: 100 * 1024 ** 3 },
  maxObjectBytes: 512 * 1024 ** 2,
});

const ref = await artifacts.put({ key: `runs/${runId}/input`, body: readableStream });
await runs.submit({ key: runId, input: { runId, artifact: ref } });

const stored = await artifacts.get(ref); // null if deleted or replaced since
// get() bounds the wait between body chunks (idleTimeoutMs, default 30 s):
// an object purged mid-read errors the stream instead of hanging it.

Streaming both ways, digest-verified, byte-limited mid-stream (ObjectTooLargeError). References do not pin objects: choose bucket retention larger than your maximum queue residence plus retry window, and delete() explicitly when an artifact is no longer shared. Permanent end-user files belong in your application's object storage, not here.

Retry

A local, transport-free helper — also importable from the browser-safe subpath @k2b/sync/retry.

import { retry, expBackoff, isRetryableTransportError } from "@k2b/sync/retry";

const result = await retry({
  run: async () => fetchThing(),
  after: async ({ ctx }) => {
    if (ctx.error && isRetryableTransportError(ctx.error) && ctx.attempt < 5) {
      ctx.reschedule({ delayMs: ctx.expBackoff() });
    }
  },
});

Diagnostics

sync.health();            // { state, connection, pendingResources, driftedResources, activeWorkers, activeHandlers, droppedEvents }
await sync.resources();   // sanitized per-resource summaries (messages, bytes, consumers, DLQ depth, ...)
for await (const event of sync.events()) { ... } // bounded structured events; slow readers drop events, never block work

Observers are contained: a throwing or slow observer can never alter transport settlement.

sync.controls() returns typed controls for queue, job, and scheduler resources and topics whose process() has started. Topic controls retain inspect/delete access after the worker stops; direct recovery requires an active handler registered with recoverDeadLetters: true. It preserves the original event identity and only invokes that consumer, never republishing the topic. Historical DLQ entries without original cursor metadata remain inspect/delete-only. See the topic recovery contract for timeout and at-least-once behavior.

The inventory includes resources already declared in this process. Reading the inventory performs no I/O and does not provision resources or start workers. Queue and job controls expose their existing deadLetters store; scheduler controls expose list(), runNow(), and awaitRun().

Identify controls by { namespace, kind, id }; a queue and job may share an ID. Repeated declarations produce one inventory entry. Scheduler inspection reports handlerAvailable across all local handles of that scheduler, without merging their handlers or workers. Controls remain discoverable after workers stop. The inventory is process-local, not broker-wide, and supplies no authorization: applications must protect any administrative route that exposes it.

Resource model

User-provided ids never become raw NATS names. Every resource is identified by { namespace, kind, id }, hashed into stable stream/KV/bucket names (S6_Q_…, KV_S6_E_…) and lower-case subject tokens (sync.v6.<ns>.queue.<hash>.t.<tenant>.work). The full identity, owner, and API version are stamped into resource metadata.

• owner defaults to application; every application opening a shared resource must declare the same configuration and owner.
• Drift (any semantic difference between declaration and live resource) throws ResourceDriftError and never mutates the resource.
• Two conflicting declarations of one resource in the same process fail before any I/O.

Semantics you must build on

• Durable delivery (queue, job, durable topic consumers, pump, scheduler) is at-least-once. Handlers and sinks must be idempotent.
• A successful publish/submit means the stream quorum accepted the message — not that a handler ran.
• A late ack of a delivery that was already redelivered and settled elsewhere is accepted idempotently by NATS; it is not detectable as "stale". StaleDeliveryError is thrown when an ack cannot be confirmed at all (e.g. the consumer was deleted).
• Ordering exists only in partitioned queues (per key) and within a single topic reader.
• Payload limits are enforced locally on the complete encoded envelope (default 128 KiB; ephemeral values 4 KiB) before publish.

Development

docker compose -f compose.nats.yml up -d --wait  # persistent 3-node NATS 2.14 cluster (ports 14222-14224)
cd packages/sync
bun run test         # parallel suite against the real cluster
bun run test:serial  # fault suite (node restarts) — must run alone
bun run typecheck

Migrating from v5

v6 is a hard cut: Redis is gone, @k2b/sync/browser is gone, ratelimit is gone, and no v5 state is migrated. See MIGRATION.md.

License

MIT