# Durable queues
`armdb::queue!` declares an engine-owned work queue. Open it through the same
`Db` that owns its data and use its opaque handle only for `push`, `lease`,
`ack`, `nack`, `len`, `dead_len`, `next_delay`, `wait_due_timeout`, and `due`.
There is no public collection, migration, or arbitrary write API.
```rust
armdb::queue!(SearchJobs, payload = SearchJob, name = "search_jobs", var);
let jobs = db.open_queue::<SearchJobs>(
QueueConfig::balanced()
.max_attempts(6)
.backoff_secs(&[60, 300, 900])
.lease_timeout(Duration::from_secs(60))
.build()?,
)?;
```
`QueueConfig::header_config(FixedConfig)` controls the FixedStore header
collection. Its sizing, shard routing, and sync settings are preserved, but
`reversed` is always normalized to `false`: queues must lease the earliest due
row first. For a `var` queue,
`QueueConfig::payload_config(Config)` passes the Bitcask configuration to the
separate `"<name>.payload"` collection unchanged. Use it to select payload
shards, caches, or (with the `encryption` feature) an encryption key:
```rust
let payload = Config::balanced()
.shard_count(4)
.encryption_key([0x42; 32])
.build();
let jobs = db.open_queue::<SearchJobs>(
QueueConfig::balanced()
.header_config(FixedConfig::balanced().grow_step(10_000).build())
.payload_config(payload)
.build()?,
)?;
```
An encrypted payload collection requires `TxConfig::log_encryption_key` for
every queue transition that uses the transaction log: `db.tx().push(...)` and
the queue handle's `ack`/`nack` operations. Without it the transition fails
before it writes queue state. Reopen an encrypted `var` queue with the same
payload configuration and key. A missing or wrong key fails closed.
## Delivery and storage
Delivery is **at least once**. A consumer that dies after its effect but before
`ack` leaves an expiring lease; later delivery is intentional. `LeaseToken` is
also bound to the process-local queue handle that leased it, in addition to its
queue key and lease epoch. A stale token — including one supplied to a different
queue — is rejected instead of acknowledging a later attempt. Tokens do not
survive a restart; lease the row again after reopening. There is no FIFO, priority,
deduplication, per-entity ordering, cursor, or RPC subscription contract.
Handlers must therefore be idempotent: queue payloads should be pointers to
current source state, not snapshots whose delivery order changes their meaning.
The default declaration stores an inline fixed payload (at most 4096 bytes) in
FixedStore. Its payload type must implement `armour_core::GetType`, so the
managed collection fingerprint rejects an incompatible fixed-payload schema on
reopen. It updates slots in place, has no queue compaction, and a fixed
queue's `Queue::stats().compaction_bytes` is consequently always zero. (That
field does not measure the separate payload backend of a `var` queue.) A `var`
declaration stores the 32-byte header in FixedStore and its Rapira payload in a
separate Bitcask collection named `"<name>.payload"`; leasing, retrying and
acknowledging change only the header. `Queue::stats().payload_writes` is a
process-lifetime diagnostic count of body writes: a direct `var` push increments
it once as soon as its payload-store write succeeds (so it includes a payload
orphan if the later header publish fails), while transactional pushes and stream
emissions increment it after their transaction commits successfully. It is never
incremented by a lease.
For a direct `var` `push`, armdb writes the payload before publishing its header.
A crash between those writes leaves an invisible payload orphan, collected on
the next open; it never creates a header whose job body is absent. Transactional
pushes and change-stream emissions instead log both physical rows as one
transactional commit.
## Waiting, clocks, and capacity
Visibility and lease deadlines are persisted wall-clock milliseconds. A backward
clock jump can extend a lease or delay a row; a forward jump can make it due
early. This preserves at-least-once delivery. Every blocking or async wait is
recomputed at most every `QueueConfig::wakeup_recheck` (five seconds by default),
so a backward jump delays a parked waiter by at most one recheck rather than by
the size of the jump. Visibility is checked again under the row lock.
`max_depth` is a soft cap over live rows: concurrent producers can overshoot it.
`Queue::push` refuses eagerly. Inside `db.tx()` the cap is judged once, at
`commit()`, over the transaction's **net** effect per queue —
`live_count − acks + pushes` (explicit pushes and change-stream emissions alike) —
so `ack + push` on a queue that is exactly full commits in either builder order,
while a second push is refused with `QueueFull` and nothing is written. A
dead-letter `nack` grants no credit. A transaction with zero net growth commits
even on a queue already above the cap.
Dead-letter rows are counted separately by `dead_len` and can have optional
retention. A lease scan leaves its row at its stable key, so scanning past leased
work is O(outstanding leases at the front) but all in RAM: a leased row is
recognized and skipped straight off the lock-free iterator, without taking its
shard lock. The `queue` benchmark contains `lease_scan/10k_outstanding` against
an operational target of at most 10 ms per scan. Measured on the development
host (`cargo bench -p armdb --bench queue`, 10,000 outstanding leases):
**1.33 ms** per scan — against **6.70 ms** for the same benchmark when the
lock-free skip is removed and every leased row is claimed under its shard lock
before being rejected. v1 deliberately has no finger cache.
The queue FixedStore preset uses `grow_step = 100_000`. Its in-memory version
vector retains a four-byte word per preallocated slot, so that RAM baseline is
exactly 400,000 bytes (about 0.4 MiB) per shard. Opening the queue also extends
`fixed.data` to 100,000 complete header slots: with the maximum 4096-byte fixed
payload this is roughly 400 MiB of *logical* file length per shard (often sparse
until blocks are written). Choose queue count, payload width, and shard count
with both baselines in mind.
FixedStore versions use a 30-bit modular ring. Replay compares versions only
over a single process/crash-replay horizon, and replication treats a peer more
than 2^29 versions behind as unusably stale; the ambiguous half-ring is never
guessed. A high-rate queue can wrap a slot, but a normal live/replay comparison
does not accumulate that much lag.
Version 1 exposes only the runtime-agnostic `Queue::due()` future for async
waiting. It deliberately provides no executor-specific wrapper or subscription.
`LsmTree` is excluded both as a change-stream source and as a transaction
participant.
## Transactions and streams
Use `db.tx().push`, `ack`, or `nack` when a queue transition must share a crash
boundary with other collections. Change streams are declared through
`SchemaRegistry::change_stream`; their source must use `db.tx()`. The stream
closure is evaluated before the transaction log is appended, its queue writes
are recorded as absolute rows, and replay applies those rows without calling the
closure again. An aborted transaction emits no job.
Transactions are atomic against a process crash by default, but **power-loss
atomicity requires `.durable()`** (or a `TxConfig` with durability enabled by
default). Treat "effect plus ack" and "source row plus stream emission" as
power-loss-safe only under that policy.
A stream commit costs concurrency on its queue. The emitted key is not known
until lowering runs, so planning locks **every shard** of every destination
queue: for the duration of that commit — across the log append, and across the
fsync when it is `.durable()` — `push`, `lease`, `ack` and `nack` on that queue
all block. Roughly one commit in `ID_BLOCK` (1024) additionally flushes the
sequence map inside that window to reserve the next id block. The seq map is
disjoint from every participant, so there is no lock cycle, but a latency
budget should account for it.
## Waiting, and what a wake means
`wait_due_timeout` returning `true` is **advisory**: it says work was due when
the scan ran, not that it is yours. A racing consumer may have taken it, so the
following `lease` legitimately returns zero entries — loop, do not assert. The
same holds for `due()`.
`next_delay()` returns `Ok(None)` when nothing is scheduled and
`Ok(Some(Duration::ZERO))` when work is due right now; an error is reported
rather than folded into `None`, so a caller using it to drive its own timer
cannot mistake a broken clock for an idle queue.
Both forms are bounded by `QueueConfig::wakeup_recheck`, and both end when the
database shuts down: `due()` resolves to an error, `wait_due_timeout` returns
one. Stop writers before `Db::close`; the wake worker itself is joined by
`Db::shutdown`/`Drop`, not by `close`.
## Retention
A variable-payload queue reclaims a dead row's body after the row itself is
swept. That delete can fail — a transient I/O error on the payload collection —
and when it does the id is kept on an in-process retry list (capped at 1024,
with a full orphan pass as the overflow path). Retention runs before each queue
operation and on every TTL sweep.
**A retention failure never fails a queue operation.** It reclaims bytes
belonging to rows that are already gone, so `push`, `lease` and `nack` proceed
and the failure is logged at `warn`; `Db::sweep_expired_now` still returns the
error, which is where an operator should watch for it. What a persistent
failure costs is disk, not availability.
One case does stop live reclamation: after a `DbError::TxFailedAfterLog` the
queue's publication permit stays pinned for the life of the transaction runtime,
so that queue's retry drain and orphan pass do not run again until the process
restarts. The next open recovers with a full orphan pass.
## Boundaries
Queue replication is not a cross-collection cut. Its Fixed header and (for
`var`) Bitcask payload use their normal backend replication paths, but promotion
can observe arbitrary lease deadlines; an overdue lease simply becomes visible
and idempotent processing remains required. A queue is one process's work: armdb
already takes an exclusive database-directory lock, so opening the same `Db`
from two processes is outside this model.
The same boundary applies to a change stream's **source**: replicated writes are
applied straight to the shard and never reach the write hook, so a follower
applying a stream source's rows emits nothing. Events are produced on the node
that took the write, and only there.
`SchemaRegistry::register_queue` classifies the physical queue collections for
the schema graph. Validation and `schema::repair` do not reconstruct queue
history or inspect crash-only payload orphans.