lmdb-queue 0.1.1

An embedded, crash-resilient, persistent queue engine for Python.

LMDB-QUEUE

Persistent, crash-safe job queue for Python. No broker. No daemon. Just disk.

Jobs survive process restarts. Workers run as threads or separate processes on the same machine. Powered by LMDB, a memory-mapped database that writes atomically and recovers instantly after a crash.

---

Install

pip install lmdb-queue

uv add lmdb-queue

---

Usage

Basic

from equeue import Queue

with Queue("./myqueue") as q:
    q.put({"task": "send_email", "to": "[e-mail]"})

    record = q.get(timeout=5.0)
    try:
        send(record.payload)
        record.ack()       # job done
    except Exception:
        record.nack()      # re-queue, or mark FAILED after too many retries

Context manager (auto ack / nack)

with Queue("./myqueue") as q:
    with q.processing(timeout=5.0) as record:
        process(record.payload)
    # ack() on success, nack() on any exception

Async

from equeue import AsyncQueue

async with AsyncQueue("./myqueue") as q:
    await q.put({"task": "resize_image"})

    async with q.processing() as record:
        await handle(record.payload)

---

The Record

get() returns a Record. Completion always goes through the record, never through a bare job ID. This prevents one worker from accidentally finishing a job held by another.

Field	Meaning
payload	Job data passed to put()
job_id	Stable integer ID, useful for logs
retries	How many times this job was nacked
enqueued_at	Unix timestamp from put()

---

Configuration

q = Queue(
    "./myqueue",
    lease_time=30.0,        # seconds before an idle job can be re-queued
    max_retries=3,          # nacks before FAILED (0 = one attempt only)
    map_size=2**30,         # LMDB virtual size in bytes (default 1 GiB)
    sync=False,             # True = safer, slower (fsync every write)
    do_recover=True,        # background thread re-queues expired jobs
    recover_interval=15.0,
    do_vacuum=True,         # background thread removes old DONE jobs
    vacuum_interval=300.0,
)

---

Statistics

q.stats()
# {
#     "pending":   4,   # waiting for a worker
#     "running":   1,   # claimed by a worker
#     "done":    120,   # finished successfully
#     "failed":    2,   # exceeded retry limit
#     "total":   127,   # ever enqueued; never decreases
# }

---

Job lifecycle

PENDING  -->  RUNNING  -->  DONE  -->  (vacuumed)
                  |
                  +--> PENDING   nack() with retries left, or lease expired
                  |
                  +--> FAILED    nack() with no retries left (kept on disk)

---

Exceptions

Exception	When
QueueEmpty	get(timeout=...) found no job in time
QueueClosed	Operation on a closed queue
QueueCorrupted	Wrong claim token, double ack, or broken disk state

---

Notes

• At-least-once delivery. A job can run more than once if a worker crashes or its lease expires. Make handlers idempotent.
• Payload types. Values must be msgpack-serializable: dicts, lists, strings, numbers, bytes, None.
• map_size is not disk usage. It is virtual address space. Disk grows as data is written.
• Not a distributed broker. EQueue is for one machine. It does not replace Redis or Kafka.

---

Docs

• User guide
• RFC: architecture and behaviour contracts

uv run pytest              # all tests
uv run pytest -m contract  # RFC contract tests only