armdb 0.11.0

use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
use crate::Key;
use crate::queue::fixed::QueueFixedRow;
use crate::txn::participant::TxParticipant;
use crate::txn::{Action, Op, Tx, check_op_limits};
use crate::{DbError, DbResult, LeaseToken, Queue, QueueKey, QueueSpec};
use zerocopy::FromBytes;

#[doc(hidden)]
pub trait QueueTx<N: QueueSpec> {
    fn push<'d>(
        tx: Tx<'d>,
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Tx<'d>>;
    fn ack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>>;
    fn nack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>>;
}

impl<'d> Tx<'d> {
    /// Enqueue a queue row as part of this transaction.
    pub fn push<N>(
        self,
        queue: &Arc<Queue<N>>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Self>
    where
        N: QueueSpec,
        N::Mode: QueueTx<N>,
    {
        <N::Mode as QueueTx<N>>::push(self, queue, payload, not_before)
    }

    pub(crate) fn fixed_push_queue<N>(
        mut self,
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::FixedQueueMode>,
        N::Payload: Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + armour_core::GetType,
    {
        self.queue_depth_admissions.reserve(
            queue.inner.instance_id,
            &queue.inner.live_count,
            queue.inner.config.max_depth,
        )?;
        let now = SystemTime::now();
        let visible_at = super::fixed::FixedQueueInner::<N>::now_ms(not_before.unwrap_or(now))?;
        if visible_at >= (1 << 63) {
            return Err(DbError::Client("queue visibility exceeds 63-bit domain"));
        }
        let id = queue.inner.next_id()?;
        let key = QueueKey::live(visible_at, id);
        let row =
            QueueFixedRow::<N>::new(payload, super::fixed::FixedQueueInner::<N>::now_ms(now)?)?;
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            key.as_bytes().len(),
        )?;
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: key.as_bytes().to_vec(),
            action: Action::Insert(zerocopy::IntoBytes::as_bytes(&row).to_vec()),
        });
        let count = Arc::clone(&queue.inner.live_count);
        let wait = Arc::clone(&queue.inner.wait);
        self.after_commit.push(Box::new(move || {
            count.fetch_add(1, std::sync::atomic::Ordering::Release);
            wait.notify_visible_at(visible_at);
        }));
        Ok(self)
    }

    /// Delete a queue row only if the supplied lease epoch remains current.
    pub fn ack<N>(self, queue: &Arc<Queue<N>>, token: LeaseToken) -> DbResult<Self>
    where
        N: QueueSpec,
        N::Mode: QueueTx<N>,
    {
        <N::Mode as QueueTx<N>>::ack(self, queue, token)
    }

    pub(crate) fn fixed_ack_queue<N>(
        mut self,
        queue: &Queue<N>,
        token: LeaseToken,
    ) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::FixedQueueMode>,
        N::Payload: Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + armour_core::GetType,
    {
        if !token.belongs_to(queue.inner.instance_id) {
            return Err(DbError::LeaseExpired);
        }
        let key = token.key();
        let token_epoch = token.epoch();
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            key.as_bytes().len(),
        )?;
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: key.as_bytes().to_vec(),
            action: Action::Eval(Box::new(move |current| {
                let Some(bytes) = current else {
                    return Err(DbError::LeaseExpired);
                };
                let row = QueueFixedRow::<N>::read_from_bytes(bytes)
                    .map_err(|_| DbError::Internal("queue fixed row size mismatch"))?;
                if row.header().lease_epoch != token_epoch {
                    return Err(DbError::LeaseExpired);
                }
                Ok(crate::Upsert::Delete)
            })),
        });
        let live = Arc::clone(&queue.inner.live_count);
        self.after_commit.push(Box::new(move || {
            live.fetch_sub(1, std::sync::atomic::Ordering::Release);
        }));
        Ok(self)
    }

    /// Reschedule or dead-letter a queue row under the same T5 commit.
    pub fn nack<N>(self, queue: &Arc<Queue<N>>, token: LeaseToken) -> DbResult<Self>
    where
        N: QueueSpec,
        N::Mode: QueueTx<N>,
    {
        <N::Mode as QueueTx<N>>::nack(self, queue, token)
    }

    pub(crate) fn fixed_nack_queue<N>(
        mut self,
        queue: &Queue<N>,
        token: LeaseToken,
    ) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::FixedQueueMode>,
        N::Payload: Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + armour_core::GetType,
    {
        if !token.belongs_to(queue.inner.instance_id) {
            return Err(DbError::LeaseExpired);
        }
        let source = token.key();
        let Some(row) = queue.inner.headers.get(&source) else {
            return Err(DbError::LeaseExpired);
        };
        if row.header().lease_epoch != token.epoch() {
            return Err(DbError::LeaseExpired);
        }
        let now_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| DbError::Client("queue time is before unix epoch"))?
            .as_millis();
        let now_ms = u64::try_from(now_ms)
            .map_err(|_| DbError::Client("queue time exceeds u64 milliseconds"))?;
        let mut header = row.header();
        header.attempts = header
            .attempts
            .checked_add(1)
            .ok_or(DbError::Client("queue attempt count exhausted"))?;
        let final_attempt = header.attempts >= queue.inner.config.max_attempts;
        // Returning a row to Ready must fence the token that just nacked it:
        // otherwise that token can still delete the rescheduled row before a
        // new lease has a chance to advance the epoch. A final nack rekeys
        // into the dead class, so its old token is already fenced by its key.
        if !final_attempt {
            header.lease_epoch = super::next_lease_epoch(header.lease_epoch);
        }
        header.lease_deadline_ms = 0;
        let destination = if final_attempt {
            header.dead_at_ms = queue.inner.config.dead_retention.map_or(0, |retention| {
                now_ms.saturating_add(u64::try_from(retention.as_millis()).unwrap_or(u64::MAX))
            });
            QueueKey::dead(source.visible_at_ms(), source.id())
        } else {
            let index = usize::from(header.attempts.saturating_sub(1))
                .min(queue.inner.config.backoff_secs.len() - 1);
            let retry_ms = queue.inner.config.backoff_secs[index]
                .checked_mul(1000)
                .ok_or(DbError::Client(
                    "queue retry delay exceeds u64 milliseconds",
                ))?;
            let visible_at_ms = now_ms.checked_add(retry_ms).ok_or(DbError::Client(
                "queue retry visibility exceeds u64 milliseconds",
            ))?;
            if visible_at_ms >= (1 << 63) {
                return Err(DbError::Client(
                    "queue retry visibility exceeds 63-bit domain",
                ));
            }
            QueueKey::live(visible_at_ms, source.id())
        };
        let moved = row.with_header(header);
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            source.as_bytes().len(),
        )?;
        check_op_limits(
            self.ops.len() + self.claim_ops.len() + 1,
            self.context.config().max_ops,
            destination.as_bytes().len(),
        )?;
        let epoch = token.epoch();
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: source.as_bytes().to_vec(),
            action: Action::Eval(Box::new(move |current| {
                let Some(bytes) = current else {
                    return Err(DbError::LeaseExpired);
                };
                let row = QueueFixedRow::<N>::read_from_bytes(bytes)
                    .map_err(|_| DbError::Internal("queue fixed row size mismatch"))?;
                if row.header().lease_epoch != epoch {
                    return Err(DbError::LeaseExpired);
                }
                Ok(crate::Upsert::Delete)
            })),
        });
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: destination.as_bytes().to_vec(),
            action: Action::Insert(zerocopy::IntoBytes::as_bytes(&moved).to_vec()),
        });
        if final_attempt {
            let live = Arc::clone(&queue.inner.live_count);
            let dead = Arc::clone(&queue.inner.dead_count);
            self.after_commit.push(Box::new(move || {
                live.fetch_sub(1, std::sync::atomic::Ordering::Release);
                dead.fetch_add(1, std::sync::atomic::Ordering::Release);
            }));
        } else {
            let wait = Arc::clone(&queue.inner.wait);
            let visible_at = destination.visible_at_ms();
            self.after_commit.push(Box::new(move || {
                wait.notify_visible_at(visible_at);
            }));
        }
        Ok(self)
    }
}

impl<N> QueueTx<N> for crate::FixedQueueMode
where
    N: QueueSpec<Mode = crate::FixedQueueMode>,
    N::Payload: Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + armour_core::GetType,
{
    fn push<'d>(
        tx: Tx<'d>,
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Tx<'d>> {
        tx.fixed_push_queue(queue, payload, not_before)
    }

    fn ack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>> {
        tx.fixed_ack_queue(queue, token)
    }

    fn nack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>> {
        tx.fixed_nack_queue(queue, token)
    }
}

#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
impl<'d> Tx<'d> {
    pub(crate) fn var_push_queue<N>(
        mut self,
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::VarQueueMode>,
        N::Payload: rapira::Rapira + armour_core::GetType + Send + Sync,
    {
        queue.inner.retention.process_before_queue_operation();
        let producer_permit = queue.inner.retention.producer_permit();
        self.queue_depth_admissions.reserve(
            queue.inner.instance_id,
            &queue.inner.live_count,
            queue.inner.config.max_depth,
        )?;
        let now = SystemTime::now();
        let visible_at = super::fixed::FixedQueueInner::<N>::now_ms(not_before.unwrap_or(now))?;
        if visible_at >= (1 << 63) {
            return Err(DbError::Client("queue visibility exceeds 63-bit domain"));
        }
        let id = queue.inner.next_id()?;
        let key = QueueKey::live(visible_at, id);
        let payload_key = rend::u64_be::from_native(id);
        let payload = crate::queue::var::QueueVarPayload::<N> {
            bytes: rapira::serialize(&payload),
            _queue: std::marker::PhantomData,
        };
        let mut payload_bytes = Vec::new();
        crate::Codec::encode_to(&crate::RapiraCodec, &payload, &mut payload_bytes)?;
        let header = crate::queue::var::QueueVarHeaderRow::<N>::new(
            super::fixed::FixedQueueInner::<N>::now_ms(now)?,
        );
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            payload_key.as_bytes().len(),
        )?;
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.payloads) as Arc<dyn TxParticipant>,
            key: payload_key.as_bytes().to_vec(),
            action: Action::Insert(payload_bytes),
        });
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            key.as_bytes().len(),
        )?;
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: key.as_bytes().to_vec(),
            action: Action::Insert(zerocopy::IntoBytes::as_bytes(&header).to_vec()),
        });
        let count = Arc::clone(&queue.inner.live_count);
        let wait = Arc::clone(&queue.inner.wait);
        let payload_writes = Arc::clone(&queue.inner.payload_writes);
        self.after_commit.push(Box::new(move || {
            count.fetch_add(1, std::sync::atomic::Ordering::Release);
            payload_writes.fetch_add(1, std::sync::atomic::Ordering::Release);
            wait.notify_visible_at(visible_at);
        }));
        self.queue_producer_permits.push(producer_permit);
        Ok(self)
    }

    pub(crate) fn var_ack_queue<N>(mut self, queue: &Queue<N>, token: LeaseToken) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::VarQueueMode>,
        N::Payload: rapira::Rapira + armour_core::GetType + Send + Sync,
    {
        if !token.belongs_to(queue.inner.instance_id) {
            return Err(DbError::LeaseExpired);
        }
        let key = token.key();
        let epoch = token.epoch();
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            key.as_bytes().len(),
        )?;
        let payload_key = rend::u64_be::from_native(key.id());
        check_op_limits(
            self.ops.len() + self.claim_ops.len() + 1,
            self.context.config().max_ops,
            payload_key.as_bytes().len(),
        )?;
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: key.as_bytes().to_vec(),
            action: Action::Eval(Box::new(move |current| {
                let Some(bytes) = current else {
                    return Err(DbError::LeaseExpired);
                };
                let row = crate::queue::var::QueueVarHeaderRow::<N>::read_from_bytes(bytes)
                    .map_err(|_| DbError::Internal("queue var header size mismatch"))?;
                if row.header().lease_epoch != epoch {
                    return Err(DbError::LeaseExpired);
                }
                Ok(crate::Upsert::Delete)
            })),
        });
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.payloads) as Arc<dyn TxParticipant>,
            key: payload_key.as_bytes().to_vec(),
            action: Action::Delete,
        });
        let live = Arc::clone(&queue.inner.live_count);
        self.after_commit.push(Box::new(move || {
            live.fetch_sub(1, std::sync::atomic::Ordering::Release);
        }));
        Ok(self)
    }

    pub(crate) fn var_nack_queue<N>(mut self, queue: &Queue<N>, token: LeaseToken) -> DbResult<Self>
    where
        N: QueueSpec<Mode = crate::VarQueueMode>,
        N::Payload: rapira::Rapira + armour_core::GetType + Send + Sync,
    {
        if !token.belongs_to(queue.inner.instance_id) {
            return Err(DbError::LeaseExpired);
        }
        // A nack deletes and re-inserts the header at a different key while
        // retaining its payload. Hold the publication gate across the whole
        // transaction so orphan maintenance cannot observe that transient gap.
        queue.inner.retention.process_before_queue_operation();
        let producer_permit = queue.inner.retention.producer_permit();
        let source = token.key();
        let Some(row) = queue.inner.headers.get(&source) else {
            return Err(DbError::LeaseExpired);
        };
        if row.header().lease_epoch != token.epoch() {
            return Err(DbError::LeaseExpired);
        }
        let now_ms = super::fixed::FixedQueueInner::<N>::now_ms(SystemTime::now())?;
        let mut header = row.header();
        header.attempts = header
            .attempts
            .checked_add(1)
            .ok_or(DbError::Client("queue attempt count exhausted"))?;
        let final_attempt = header.attempts >= queue.inner.config.max_attempts;
        // A retry retains the live class and can be addressed by an old token
        // until its next lease, so it must advance the generation here. A final
        // nack changes class/key and therefore fences the token without a wrap.
        if !final_attempt {
            header.lease_epoch = super::next_lease_epoch(header.lease_epoch);
        }
        header.lease_deadline_ms = 0;
        let destination = if final_attempt {
            header.dead_at_ms = queue.inner.config.dead_retention.map_or(0, |retention| {
                now_ms.saturating_add(u64::try_from(retention.as_millis()).unwrap_or(u64::MAX))
            });
            QueueKey::dead(source.visible_at_ms(), source.id())
        } else {
            let index = usize::from(header.attempts.saturating_sub(1))
                .min(queue.inner.config.backoff_secs.len() - 1);
            let delay = queue.inner.config.backoff_secs[index]
                .checked_mul(1000)
                .ok_or(DbError::Client(
                    "queue retry delay exceeds u64 milliseconds",
                ))?;
            let visible_at = now_ms.checked_add(delay).ok_or(DbError::Client(
                "queue retry visibility exceeds u64 milliseconds",
            ))?;
            if visible_at >= (1 << 63) {
                return Err(DbError::Client(
                    "queue retry visibility exceeds 63-bit domain",
                ));
            }
            QueueKey::live(visible_at, source.id())
        };
        let moved = row.with_header(header);
        check_op_limits(
            self.ops.len() + self.claim_ops.len(),
            self.context.config().max_ops,
            source.as_bytes().len(),
        )?;
        check_op_limits(
            self.ops.len() + self.claim_ops.len() + 1,
            self.context.config().max_ops,
            destination.as_bytes().len(),
        )?;
        let epoch = token.epoch();
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: source.as_bytes().to_vec(),
            action: Action::Eval(Box::new(move |current| {
                let Some(bytes) = current else {
                    return Err(DbError::LeaseExpired);
                };
                let row = crate::queue::var::QueueVarHeaderRow::<N>::read_from_bytes(bytes)
                    .map_err(|_| DbError::Internal("queue var header size mismatch"))?;
                if row.header().lease_epoch != epoch {
                    return Err(DbError::LeaseExpired);
                }
                Ok(crate::Upsert::Delete)
            })),
        });
        self.ops.push(Op {
            participant: Arc::clone(&queue.inner.headers) as Arc<dyn TxParticipant>,
            key: destination.as_bytes().to_vec(),
            action: Action::Insert(zerocopy::IntoBytes::as_bytes(&moved).to_vec()),
        });
        if final_attempt {
            let live = Arc::clone(&queue.inner.live_count);
            let dead = Arc::clone(&queue.inner.dead_count);
            self.after_commit.push(Box::new(move || {
                live.fetch_sub(1, std::sync::atomic::Ordering::Release);
                dead.fetch_add(1, std::sync::atomic::Ordering::Release);
            }));
        } else {
            let wait = Arc::clone(&queue.inner.wait);
            let visible_at = destination.visible_at_ms();
            self.after_commit.push(Box::new(move || {
                wait.notify_visible_at(visible_at);
            }));
        }
        self.queue_producer_permits.push(producer_permit);
        Ok(self)
    }
}

#[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
impl<N> QueueTx<N> for crate::VarQueueMode
where
    N: QueueSpec<Mode = crate::VarQueueMode>,
    N::Payload: rapira::Rapira + armour_core::GetType + Send + Sync,
{
    fn push<'d>(
        tx: Tx<'d>,
        queue: &Queue<N>,
        payload: N::Payload,
        not_before: Option<SystemTime>,
    ) -> DbResult<Tx<'d>> {
        tx.var_push_queue(queue, payload, not_before)
    }
    fn ack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>> {
        tx.var_ack_queue(queue, token)
    }
    fn nack<'d>(tx: Tx<'d>, queue: &Queue<N>, token: LeaseToken) -> DbResult<Tx<'d>> {
        tx.var_nack_queue(queue, token)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use armour_core::GetType;
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    use rapira::Rapira;

    use super::*;
    use crate::armour::Db;
    use crate::{DbError, QueueConfig, queue};

    #[derive(
        Clone,
        Copy,
        GetType,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
        zerocopy::KnownLayout,
    )]
    #[repr(transparent)]
    pub struct FixedJob(u64);
    queue!(FixedJobs, payload = FixedJob, name = "queue_tx_epoch_fixed");

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, GetType, Rapira)]
    pub struct VarJob(u64);
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    queue!(VarJobs, payload = VarJob, name = "queue_tx_epoch_var", var);

    fn final_nack_config() -> QueueConfig {
        QueueConfig::balanced()
            .max_attempts(1)
            .backoff_secs(&[0])
            .build()
            .expect("final-nack queue configuration")
    }

    #[test]
    fn fixed_final_nack_at_max_epoch_moves_dead_and_fences_the_old_token() {
        let dir = tempfile::tempdir().expect("temporary database directory");
        let db = Arc::new(Db::open_test(dir.path()).expect("open database"));
        let queue = db
            .open_queue::<FixedJobs>(final_nack_config())
            .expect("open fixed queue");
        queue.push(FixedJob(1), None).expect("push fixed job");
        let leased = queue
            .lease(1, None)
            .expect("lease fixed job")
            .pop()
            .unwrap();
        let token = LeaseToken::new(queue.inner.instance_id, leased.key(), u32::MAX);
        queue
            .inner
            .headers
            .update(&leased.key(), |row| {
                let mut header = row.header();
                header.lease_epoch = u32::MAX;
                row.with_header(header)
            })
            .expect("set fixed lease epoch");

        db.tx()
            .nack(&queue, token)
            .expect("plan final fixed nack")
            .commit()
            .expect("commit final fixed nack");

        assert_eq!(queue.len(), 0);
        assert_eq!(queue.dead_len(), 1);
        assert!(matches!(queue.ack(token), Err(DbError::LeaseExpired)));
        assert!(matches!(queue.nack(token), Err(DbError::LeaseExpired)));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn var_final_nack_at_max_epoch_moves_dead_and_fences_the_old_token() {
        let dir = tempfile::tempdir().expect("temporary database directory");
        let db = Arc::new(Db::open_test(dir.path()).expect("open database"));
        let queue = db
            .open_queue::<VarJobs>(final_nack_config())
            .expect("open variable queue");
        queue.push(VarJob(1), None).expect("push variable job");
        let leased = queue
            .lease(1, None)
            .expect("lease variable job")
            .pop()
            .unwrap();
        let token = LeaseToken::new(queue.inner.instance_id, leased.key(), u32::MAX);
        queue
            .inner
            .headers
            .update(&leased.key(), |row| {
                let mut header = row.header();
                header.lease_epoch = u32::MAX;
                row.with_header(header)
            })
            .expect("set variable lease epoch");

        db.tx()
            .nack(&queue, token)
            .expect("plan final variable nack")
            .commit()
            .expect("commit final variable nack");

        assert_eq!(queue.len(), 0);
        assert_eq!(queue.dead_len(), 1);
        assert!(matches!(queue.ack(token), Err(DbError::LeaseExpired)));
        assert!(matches!(queue.nack(token), Err(DbError::LeaseExpired)));
    }
}