--

lmdbq

A persistent, embedded message queue and pub/sub toolkit for Bun and Node.js, backed by mdbxmou (which embeds libmdbx).

Everything is stored in an LMDB environment and survives process restarts. It ships a FIFO Queue, a durable PubSub, a JobQueue with delays/retries/priorities, and a ChunkedQueue.

Features

• FIFO queue — O(1) push / pop / peek / length, named topics, batch push/pop
• Pub/sub — durable append-only topic logs with per-group cursors and live in-process delivery
• Job queue — waiting / active / completed / failed / delayed states, priorities, retries with backoff, progress, and events
• Chunked queue — messages batched into fixed-size chunks with configurable retention
• String or Buffer message types
• Shared, reference-counted environment — safe to open many queues on one path

Use Cases

What this library is a good fit for:

• Durable in-process queues — separate producers and consumers that need to survive a process restart without a broker. Because data lives in an embedded LMDB file, a queue keeps its backlog across crashes and deploys.
• Local task / job orchestration — scheduled or retryable work (emails, image resizing, report generation, webhooks) using the JobQueue's delays, priorities, attempts, and backoff, without standing up Redis or a database server.
• Fan-out event logs — multiple independent consumers need their own read cursor over the same topic (PubSub), or a durable command/event log for aggregate state rebuilding.
• High-volume ordered ingestion — batching writes into fixed-size chunks (ChunkedQueue) to amortize writes for metrics, logs, or telemetry collected on one host.
• Offline-first / single-host services — Bun or Node services that want ACID persistence with zero infrastructure, no network dependency, and no external queue broker.

What it is not for:

• Distributed worker pools across multiple machines — LMDB is a single-process-per-environment (with shared-memory support only within one host), so live delivery and cursor advancement are in-process only.
• Multi-host pub/sub or cross-machine fan-out.
• Message ordering guarantees between separate processes sharing one file (see Concurrency below).

Install

bun add @jream/lmdbq

mdbxmou is a native addon. Building it requires a C++ toolchain (cmake, make, g++) the first time it is installed.

Usage

FIFO Queue

import { Queue, Producer, Consumer, STRING_TYPE } from "@jream/lmdbq";

const queue = new Queue<string>({ path: "./data", topic: "jobs", dataType: STRING_TYPE });

queue.push("one");
queue.pushBatch(["two", "three"]);

queue.length();  // 3
queue.pop();     // "one"
queue.peek();    // "two"
queue.clear();
queue.close();

Producer / consumer split:

const producer = new Producer<string>({ path: "./data", topic: "orders", dataType: STRING_TYPE });
const consumer = new Consumer<string>({ path: "./data", topic: "orders", dataType: STRING_TYPE });

producer.push(["a", "b", "c"]);
consumer.pop();        // "a"
consumer.popBatch(2);  // ["b", "c"]

producer.close();
consumer.close();

Pub/Sub

import { Publisher, Subscriber, STRING_TYPE } from "@jream/lmdbq";

const publisher = new Publisher<string>({ path: "./data", dataType: STRING_TYPE });
const subscriber = new Subscriber<string>({ path: "./data", dataType: STRING_TYPE });

// Pull-based, per-group cursor:
publisher.publish("news", "hello");
subscriber.read("news", { group: "worker-1" }); // [{ id: 1, message: "hello" }]

// Push-based live delivery (same process):
subscriber.subscribe("news", (msg) => console.log(msg.message));
publisher.publish("news", "world"); // logs "world"

publisher.close();
subscriber.close();

Job Queue

import { JobQueue, STRING_TYPE } from "@jream/lmdbq";

const queue = new JobQueue<string>({ path: "./data", dataType: STRING_TYPE });

queue.add("send-email", { jobId: "1", priority: 1, attempts: 3, backoff: 1000 });
queue.add("resize-image", { delay: 60_000 });

const job = queue.getNextJob(); // highest-priority waiting job, now "active"
queue.updateProgress(job.id, 50);
queue.complete(job.id, { ok: true });

// On failure, retries move to "delayed" and back to "waiting" when due.
const failed = queue.fail(job.id, "transient");

Chunked Queue

import { ChunkedQueue, STRING_TYPE } from "@jream/lmdbq";

const queue = new ChunkedQueue<string>({
  path: "./data",
  topic: "logs",
  dataType: STRING_TYPE,
  chunkSize: 64 * 1024, // seal a chunk once it reaches 64KB
  chunksToKeep: 4,      // retain the last 4 consumed chunks
});

queue.push("a");
queue.push("b");
queue.pop();     // "a"
queue.flush();   // persist the write buffer now
queue.close();   // flushes on close

Concurrency & limitations

These behaviors are deliberate and worth knowing before you rely on them:

• Same-process coordination is safe and recommended. Queue, PubSub, and JobQueue are internally consistent when used from a single process (multiple instances sharing one path are fine; the shared environment is reference-counted).
• One LMDB environment is open per directory, not per process. The shared Store keyed by resolve(path) means two different processes opening the same directory each open the file. LMDB writes use MDBX_NOSTICKYTHREADS-style transactions, but there is no fencing or cross-process locking, so concurrent readers/writers from separate processes are not guaranteed consistent. Prefer one process per path.
• No browser/worker support. This is a native addon (mdbxmou → libmdbx) and requires a C++ toolchain to build on first install.
• close() may not persist every write immediately. Queue, PubSub, and JobQueue write in committed transactions, but ChunkedQueue buffers messages in memory until a chunk fills, or flush() / close() is called. Call flush() before expecting data to be durable-persisted from another instance.
• IDs / cursor counts are per-path, monotonic, and not reused across clears, so a clear() resets cursors but never rewrites history. Plan around this if you replay from cursors after a clear.

Configuration

Raw environment access

Every store instance is reference-counted per directory. If you need to reach past the queue abstractions against the same LMDB environment, use the exported Store:

import { Store } from "@jream/lmdbq";

// Acquire returns a shared handle for a path; release when done.
const store = Store.acquire("./data");
store.client; // the underlying mdbxmou MDBX_Env
store.getOrCreateMap({ name: "my-index", keyMode: 1, valueFlag: 1 });
store.release();

Use Store.acquire/store.release() rather than opening MDBX_Env yourself: it keeps the reference count correct so the environment closes only when the last consumer goes away.

interface QueueOptions {
  path: string;            // LMDB environment directory
  topic: string;           // queue/topic name
  dataType?: DataType;     // "string" | "buffer"  (default "buffer")
  maxDbi?: number;         // max named databases  (default 32)
}

interface ChunkedQueueOptions extends QueueOptions {
  chunkSize?: number;      // bytes per chunk  (default 64 * 1024)
  chunksToKeep?: number;   // consumed chunks to retain  (default 4)
}

interface JobQueueOptions {
  path: string;
  name?: string;           // queue name  (default "default")
  dataType?: DataType;
  maxDbi?: number;
}

API

Queue<T>

| Method | Returns | Description | | --------------------- | ---------------- | ------------------------------------------------------ | | push(message) | number | Append a message; returns the new length | | pushBatch(messages) | number | Append several messages atomically | | pop() | T \| undefined | Remove and return the head message | | popBatch(count) | T[] | Remove and return up to count head messages | | peek() | T \| undefined | Return the head message without removing it | | length() | number | Number of queued messages | | isEmpty() | boolean | Whether the queue is empty | | clear() | void | Remove all messages and reset cursors | | close() | void | Release the shared environment |

Producer<T> / Consumer<T>

| Producer | Consumer | Description | | --------------------- | --------------------- | ------------------------------------ | | push(message) | pop() | Single message | | push(messages) | popBatch(count) | Batch | | length() | peek() | Inspect | | close() | length() / close()| Lifecycle |

PubSub<T> / Publisher<T> / Subscriber<T>

| Method | Returns | Description | | --------------------------------- | ------------------ | -------------------------------------------------- | | publish(topic, message) | number | Append to a topic; returns the sequence id | | read(topic, { group?, count? }) | PubSubMessage[] | Read unread messages, advancing the group cursor | | subscribe(topic, handler, opts) | () => void | Replay backlog then deliver live; returns unsubscribe | | length(topic) | number | Total messages published to a topic | | close() | void | Release the shared environment |

JobQueue<T>

| Method | Returns | Description | | ------------------------------- | -------------------------------- | ------------------------------------------ | | add(data, opts?) | Job<T> | Add a job; dedupes by opts.jobId | | getJob(jobId) | Job<T> \| undefined | Fetch a job by id | | getJobs(states?) | Job<T>[] | List jobs, optionally filtered by state | | getNextJob() | Job<T> \| undefined | Promote due jobs; pick highest priority | | complete(jobId, returnValue?) | Job<T> \| undefined | Mark completed | | fail(jobId, reason?) | Job<T> \| undefined | Fail; retries with backoff when attempts remain | | updateProgress(jobId, n) | void | Set progress (0-100) | | counts() | Record<JobState, number> | Jobs per state | | remove(jobId) / clear() | boolean / void | Delete jobs | | on(event, handler) | void | Subscribe to lifecycle events |

Job events: waiting, active, completed, failed, retry, progress.

ChunkedQueue<T>

| Method | Returns | Description | | -------------- | ---------------- | -------------------------------------------- | | push(message)| number | Append; seals a chunk when full | | pop() | T \| undefined | Remove and return the head message | | length() | number | Number of queued messages | | chunkCount() | number | Number of sealed chunks on disk | | flush() | void | Persist the write buffer now | | clear() | void | Remove all messages and reset cursors | | close() | void | Flush pending writes and release the environment |

Constants

| Constant | Value | | ------------- | ---------- | | STRING_TYPE | "string" | | BUFFER_TYPE | "buffer" |

Testing

bun test

License

MIT License - see LICENSE.md