Module memory 

In-memory backend for tests and local development.

Contract:

• MemoryBackend::new(); Clone-able handle over shared Arc state.
• publish with delay uses tokio::time::sleep in a spawned task (so it is compatible with tokio::time::pause() / advance() in tests).
• consume honours prefetch (at most prefetch un-acked deliveries per consumer).
• retry re-publishes next after delay, then acks the original.
• defer holds next for delay and then inserts it by priority, then acks the original. The hold is a tokio::time::sleep too, so it is virtual-time friendly.
• Every enqueue (plain, delayed or deferred) goes through one priority insertion: the envelope lands behind every pending envelope whose priority is greater than or equal to its own and ahead of the rest, so equal priorities stay FIFO. A hold ends inside that same critical section, so no envelope is ever counted by both deferred and pending.
• dead_letter moves the envelope into an inspectable dead_letters(queue) list with reason.
• Test helpers: pending(queue) -> usize, deferred(queue) -> usize, dead_letters(queue) -> Vec<(Envelope, String)>, acked(queue) -> Vec<Envelope>.
• close ends all consumer streams. Afterwards declare, publish, defer, retry and dead_letter all fail with Error::ShutDown; a plain ack still records. A delayed publish or a deferral whose timer fires after the close is dropped, exactly like a broker that went away mid-wait.

Implementation notes:

• Each consumer owns a tokio::sync::Semaphore with prefetch permits. A permit is moved into every Delivery handed out and released on ack / retry / dead-letter, which is what caps the number of outstanding deliveries. The semaphore is dropped from the queue when the consumer task ends.
• Dropping a stream requeues every message that was produced for it but not handed out yet, so a message can never vanish just because a consumer went away.
• A delivery that is dropped without being settled releases its permit but the message is not requeued (unlike a real broker). Tests should settle deliveries.

Structs

MemoryBackend

An in-memory Backend. Cloning gives another handle onto the same state.