later 0.0.48

//! Database-backed job delivery with renewable work leases.

use super::{MessageHeaders, MqClient, MqConsumer, MqPayload, MqPublisher};
use std::{collections::VecDeque, time::Duration};

const LEASE_SECONDS: i64 = 30;
const EMPTY_QUEUE_POLL_DELAY: Duration = Duration::from_millis(100);
#[cfg(feature = "postgres")]
const POSTGRES_PREFETCH_SIZE: i64 = 16;
#[cfg(feature = "postgres")]
const POSTGRES_PREFETCH_MINIMUM_BACKLOG: i64 = POSTGRES_PREFETCH_SIZE;
#[cfg(feature = "sqlite")]
const SQLITE_PREFETCH_SIZE: i64 = 16;
#[cfg(feature = "sqlite")]
const SQLITE_PREFETCH_MINIMUM_BACKLOG: i64 = SQLITE_PREFETCH_SIZE;

/// Claims the oldest deliverable rows. `INDEXED BY` pins the order index so
/// `ORDER BY sequence ... LIMIT` stops after `LIMIT` rows instead of sorting
/// the whole backlog - see `later_delivery_queue_order_idx`'s migration, and
/// the plan test below that fails if SQLite would need a sort again.
#[cfg(feature = "sqlite")]
const SQLITE_CLAIM_SQL: &str = "WITH candidates AS ( \
                   SELECT sequence FROM later_delivery_queue \
                     INDEXED BY later_delivery_queue_order_idx \
                   WHERE namespace = ?3 AND queue_name = ?4 AND available_at <= ?5 \
                     AND (lease_until IS NULL OR lease_until <= ?5) \
                   ORDER BY sequence ASC LIMIT ?6 \
                 ), batch AS ( \
                   SELECT sequence FROM candidates ORDER BY sequence \
                   LIMIT CASE \
                     WHEN (SELECT COUNT(*) FROM candidates) >= ?7 THEN ?6 \
                     ELSE 1 \
                   END \
                 ) \
                 UPDATE later_delivery_queue SET lease_owner = ?1, lease_until = ?2, \
                   attempt_count = attempt_count + 1 \
                 WHERE sequence IN (SELECT sequence FROM batch) \
                 RETURNING sequence, payload";

#[cfg(feature = "sqlite")]
mod sqlite {
    use super::*;
    use sqlx::{Row, SqlitePool};

    #[derive(Clone)]
    /// SQLite-backed job delivery shared by workers using one database file.
    pub struct SqliteQueue {
        pool: SqlitePool,
        namespace: String,
    }

    impl SqliteQueue {
        /// Creates a queue client using an initialized Later SQLite database.
        ///
        /// `namespace` must not be empty and should match the state backend.
        pub fn new(pool: SqlitePool, namespace: impl Into<String>) -> anyhow::Result<Self> {
            let namespace = namespace.into();
            if namespace.trim().is_empty() {
                return Err(anyhow::anyhow!("delivery namespace cannot be empty"));
            }
            Ok(Self { pool, namespace })
        }
    }

    struct Publisher {
        pool: SqlitePool,
        namespace: String,
        queue_name: String,
    }

    struct Consumer {
        pool: SqlitePool,
        namespace: String,
        queue_name: String,
        lease_owner: String,
        pending: VecDeque<Box<dyn MqPayload>>,
    }

    struct Payload {
        pool: SqlitePool,
        sequence: i64,
        lease_owner: String,
        data: Vec<u8>,
        renewal: tokio::task::JoinHandle<()>,
    }

    impl Drop for Payload {
        fn drop(&mut self) {
            self.renewal.abort();
        }
    }

    fn now_millis() -> i64 {
        chrono::Utc::now().timestamp_millis()
    }

    #[async_trait::async_trait]
    impl MqClient for SqliteQueue {
        async fn new_consumer(
            &self,
            routing_key: &str,
            worker_id: i32,
        ) -> anyhow::Result<Box<dyn MqConsumer>> {
            Ok(Box::new(Consumer {
                pool: self.pool.clone(),
                namespace: self.namespace.clone(),
                queue_name: routing_key.to_owned(),
                lease_owner: format!("{}-{worker_id}", crate::generate_id()),
                pending: VecDeque::new(),
            }))
        }

        async fn new_publisher(&self, routing_key: &str) -> anyhow::Result<Box<dyn MqPublisher>> {
            Ok(Box::new(Publisher {
                pool: self.pool.clone(),
                namespace: self.namespace.clone(),
                queue_name: routing_key.to_owned(),
            }))
        }

        async fn queue_depth(&self, routing_key: &str) -> anyhow::Result<usize> {
            let count: i64 = sqlx::query_scalar(
                "SELECT COUNT(*) FROM later_delivery_queue \
                 WHERE namespace = ?1 AND queue_name = ?2 AND available_at <= ?3 \
                   AND (lease_until IS NULL OR lease_until <= ?3)",
            )
            .bind(&self.namespace)
            .bind(routing_key)
            .bind(now_millis())
            .fetch_one(&self.pool)
            .await?;
            Ok(usize::try_from(count)?)
        }
    }

    #[async_trait::async_trait]
    impl MqPublisher for Publisher {
        async fn publish(&self, payload: &[u8]) -> anyhow::Result<()> {
            sqlx::query(
                "INSERT INTO later_delivery_queue \
                 (namespace, queue_name, payload, available_at) VALUES (?1, ?2, ?3, ?4)",
            )
            .bind(&self.namespace)
            .bind(&self.queue_name)
            .bind(payload)
            .bind(now_millis())
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        async fn ensure_consumer(&self) -> anyhow::Result<()> {
            Ok(())
        }

        async fn has_consumer(&self) -> anyhow::Result<bool> {
            Ok(true)
        }
    }

    impl Consumer {
        async fn claim(&self) -> anyhow::Result<VecDeque<Box<dyn MqPayload>>> {
            // A shallow queue is claimed one message at a time to keep work
            // evenly shared by consumers that start at different moments. A
            // sustained backlog amortizes the SQLite round trip by leasing a
            // bounded batch instead. Every claimed payload keeps its own
            // renewal task, including while buffered locally, so a slow
            // handler cannot let a prefetched delivery expire.
            let now = now_millis();
            let lease_until = now
                .checked_add(LEASE_SECONDS * 1_000)
                .ok_or_else(|| anyhow::anyhow!("SQLite delivery lease is out of range"))?;
            let rows = sqlx::query(SQLITE_CLAIM_SQL)
                .bind(&self.lease_owner)
                .bind(lease_until)
                .bind(&self.namespace)
                .bind(&self.queue_name)
                .bind(now)
                .bind(SQLITE_PREFETCH_SIZE)
                .bind(SQLITE_PREFETCH_MINIMUM_BACKLOG)
                .fetch_all(&self.pool)
                .await?;

            rows.into_iter()
                .map(|row| {
                    let sequence = row.try_get("sequence")?;
                    let pool = self.pool.clone();
                    let lease_owner = self.lease_owner.clone();
                    let renewal = tokio::spawn(async move {
                        loop {
                            tokio::time::sleep(Duration::from_secs(10)).await;
                            let lease_until = now_millis().saturating_add(LEASE_SECONDS * 1_000);
                            match sqlx::query(
                                "UPDATE later_delivery_queue SET lease_until = ?1 \
                                 WHERE sequence = ?2 AND lease_owner = ?3",
                            )
                            .bind(lease_until)
                            .bind(sequence)
                            .bind(&lease_owner)
                            .execute(&pool)
                            .await
                            {
                                Ok(result) if result.rows_affected() == 1 => {}
                                Ok(_) => return,
                                Err(error) => {
                                    tracing::warn!(%error, sequence, "Failed to renew SQLite delivery lease");
                                }
                            }
                        }
                    });
                    Ok(Box::new(Payload {
                        pool: self.pool.clone(),
                        sequence,
                        lease_owner: self.lease_owner.clone(),
                        data: row.try_get("payload")?,
                        renewal,
                    }) as Box<dyn MqPayload>)
                })
                .collect::<Result<VecDeque<_>, sqlx::Error>>()
                .map_err(anyhow::Error::from)
        }
    }

    #[async_trait::async_trait]
    impl MqConsumer for Consumer {
        async fn next(&mut self) -> Option<anyhow::Result<Box<dyn MqPayload>>> {
            loop {
                if let Some(payload) = self.pending.pop_front() {
                    return Some(Ok(payload));
                }
                match self.claim().await {
                    Ok(payloads) if !payloads.is_empty() => self.pending = payloads,
                    Ok(_) => tokio::time::sleep(EMPTY_QUEUE_POLL_DELAY).await,
                    Err(error) => return Some(Err(error)),
                }
            }
        }
    }

    #[async_trait::async_trait]
    impl MqPayload for Payload {
        async fn ack(&self) -> anyhow::Result<()> {
            self.renewal.abort();
            sqlx::query(
                "DELETE FROM later_delivery_queue WHERE sequence = ?1 AND lease_owner = ?2",
            )
            .bind(self.sequence)
            .bind(&self.lease_owner)
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        async fn nack_requeue(&self) -> anyhow::Result<()> {
            self.renewal.abort();
            sqlx::query(
                "UPDATE later_delivery_queue SET lease_owner = NULL, lease_until = NULL, \
                   available_at = ?1 WHERE sequence = ?2 AND lease_owner = ?3",
            )
            .bind(now_millis())
            .bind(self.sequence)
            .bind(&self.lease_owner)
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        fn get_headers(&self) -> Option<MessageHeaders> {
            None
        }

        fn data(&self) -> &[u8] {
            &self.data
        }
    }
}

#[cfg(feature = "sqlite")]
pub use sqlite::SqliteQueue;

#[cfg(feature = "postgres")]
mod postgres {
    use super::*;
    use sqlx::{PgPool, Row};

    #[derive(Clone)]
    /// PostgreSQL-backed job delivery shared by workers using one database.
    pub struct PostgresQueue {
        pool: PgPool,
        namespace: String,
    }

    impl PostgresQueue {
        /// Creates a queue client using an initialized Later PostgreSQL database.
        ///
        /// `namespace` must not be empty and should match the state backend.
        pub fn new(pool: PgPool, namespace: impl Into<String>) -> anyhow::Result<Self> {
            let namespace = namespace.into();
            if namespace.trim().is_empty() {
                return Err(anyhow::anyhow!("delivery namespace cannot be empty"));
            }
            Ok(Self { pool, namespace })
        }
    }

    struct Publisher {
        pool: PgPool,
        namespace: String,
        queue_name: String,
    }

    struct Consumer {
        pool: PgPool,
        namespace: String,
        queue_name: String,
        lease_owner: String,
        pending: VecDeque<Box<dyn MqPayload>>,
    }

    struct Payload {
        pool: PgPool,
        sequence: i64,
        lease_owner: String,
        data: Vec<u8>,
        renewal: tokio::task::JoinHandle<()>,
    }

    impl Drop for Payload {
        fn drop(&mut self) {
            self.renewal.abort();
        }
    }

    #[async_trait::async_trait]
    impl MqClient for PostgresQueue {
        async fn new_consumer(
            &self,
            routing_key: &str,
            worker_id: i32,
        ) -> anyhow::Result<Box<dyn MqConsumer>> {
            Ok(Box::new(Consumer {
                pool: self.pool.clone(),
                namespace: self.namespace.clone(),
                queue_name: routing_key.to_owned(),
                lease_owner: format!("{}-{worker_id}", crate::generate_id()),
                pending: VecDeque::new(),
            }))
        }

        async fn new_publisher(&self, routing_key: &str) -> anyhow::Result<Box<dyn MqPublisher>> {
            Ok(Box::new(Publisher {
                pool: self.pool.clone(),
                namespace: self.namespace.clone(),
                queue_name: routing_key.to_owned(),
            }))
        }

        async fn queue_depth(&self, routing_key: &str) -> anyhow::Result<usize> {
            let count: i64 = sqlx::query_scalar(
                "SELECT COUNT(*) FROM later_delivery_queue \
                 WHERE namespace = $1 AND queue_name = $2 AND available_at <= NOW() \
                   AND (lease_until IS NULL OR lease_until <= NOW())",
            )
            .bind(&self.namespace)
            .bind(routing_key)
            .fetch_one(&self.pool)
            .await?;
            Ok(usize::try_from(count)?)
        }
    }

    #[async_trait::async_trait]
    impl MqPublisher for Publisher {
        async fn publish(&self, payload: &[u8]) -> anyhow::Result<()> {
            sqlx::query(
                "INSERT INTO later_delivery_queue (namespace, queue_name, payload) \
                 VALUES ($1, $2, $3)",
            )
            .bind(&self.namespace)
            .bind(&self.queue_name)
            .bind(payload)
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        async fn ensure_consumer(&self) -> anyhow::Result<()> {
            Ok(())
        }

        async fn has_consumer(&self) -> anyhow::Result<bool> {
            Ok(true)
        }
    }

    impl Consumer {
        async fn claim(&self) -> anyhow::Result<VecDeque<Box<dyn MqPayload>>> {
            // A shallow queue is claimed one message at a time to keep work
            // evenly shared by consumers that start at different moments. A
            // sustained backlog amortizes PostgreSQL client/server round
            // trips by leasing a bounded batch. Every claimed payload keeps
            // its own renewal task, including while buffered locally, so a
            // slow handler cannot let a prefetched delivery expire.
            let rows = sqlx::query(
                "WITH candidates AS MATERIALIZED ( \
                   SELECT sequence FROM later_delivery_queue \
                   WHERE namespace = $3 AND queue_name = $4 AND available_at <= NOW() \
                     AND (lease_until IS NULL OR lease_until <= NOW()) \
                   ORDER BY sequence ASC FOR UPDATE SKIP LOCKED LIMIT $5 \
                 ), batch AS ( \
                   SELECT sequence FROM candidates ORDER BY sequence \
                   LIMIT CASE \
                     WHEN (SELECT COUNT(*) FROM candidates) >= $6 THEN $5 \
                     ELSE 1 \
                   END \
                 ) \
                 UPDATE later_delivery_queue queue SET lease_owner = $1, \
                   lease_until = NOW() + make_interval(secs => $2), \
                   attempt_count = queue.attempt_count + 1 \
                 FROM batch WHERE queue.sequence = batch.sequence \
                 RETURNING queue.sequence, queue.payload",
            )
            .bind(&self.lease_owner)
            .bind(LEASE_SECONDS)
            .bind(&self.namespace)
            .bind(&self.queue_name)
            .bind(POSTGRES_PREFETCH_SIZE)
            .bind(POSTGRES_PREFETCH_MINIMUM_BACKLOG)
            .fetch_all(&self.pool)
            .await?;

            rows.into_iter()
                .map(|row| {
                    let sequence = row.try_get("sequence")?;
                    let data = row.try_get("payload")?;
                    let pool = self.pool.clone();
                    let lease_owner = self.lease_owner.clone();
                    let renewal = tokio::spawn(async move {
                        loop {
                            tokio::time::sleep(Duration::from_secs(10)).await;
                            match sqlx::query(
                                "UPDATE later_delivery_queue SET \
                               lease_until = NOW() + make_interval(secs => $1) \
                             WHERE sequence = $2 AND lease_owner = $3",
                            )
                            .bind(LEASE_SECONDS)
                            .bind(sequence)
                            .bind(&lease_owner)
                            .execute(&pool)
                            .await
                            {
                                Ok(result) if result.rows_affected() == 1 => {}
                                Ok(_) => return,
                                Err(error) => {
                                    tracing::warn!(%error, sequence, "Failed to renew Postgres delivery lease");
                                }
                            }
                        }
                    });
                    Ok(Box::new(Payload {
                        pool: self.pool.clone(),
                        sequence,
                        lease_owner: self.lease_owner.clone(),
                        data,
                        renewal,
                    }) as Box<dyn MqPayload>)
                })
                .collect::<Result<VecDeque<_>, sqlx::Error>>()
                .map_err(anyhow::Error::from)
        }
    }

    #[async_trait::async_trait]
    impl MqConsumer for Consumer {
        async fn next(&mut self) -> Option<anyhow::Result<Box<dyn MqPayload>>> {
            loop {
                if let Some(payload) = self.pending.pop_front() {
                    return Some(Ok(payload));
                }
                match self.claim().await {
                    Ok(payloads) if !payloads.is_empty() => self.pending = payloads,
                    Ok(_) => tokio::time::sleep(EMPTY_QUEUE_POLL_DELAY).await,
                    Err(error) => return Some(Err(error)),
                }
            }
        }
    }

    #[async_trait::async_trait]
    impl MqPayload for Payload {
        async fn ack(&self) -> anyhow::Result<()> {
            self.renewal.abort();
            sqlx::query(
                "DELETE FROM later_delivery_queue WHERE sequence = $1 AND lease_owner = $2",
            )
            .bind(self.sequence)
            .bind(&self.lease_owner)
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        async fn nack_requeue(&self) -> anyhow::Result<()> {
            self.renewal.abort();
            sqlx::query(
                "UPDATE later_delivery_queue SET lease_owner = NULL, lease_until = NULL, \
                   available_at = NOW() WHERE sequence = $1 AND lease_owner = $2",
            )
            .bind(self.sequence)
            .bind(&self.lease_owner)
            .execute(&self.pool)
            .await?;
            Ok(())
        }

        fn get_headers(&self) -> Option<MessageHeaders> {
            None
        }

        fn data(&self) -> &[u8] {
            &self.data
        }
    }
}

#[cfg(feature = "postgres")]
pub use postgres::PostgresQueue;

#[cfg(all(test, feature = "sqlite"))]
mod tests {
    use super::{MqClient, SqliteQueue};
    use crate::storage::Sqlite;
    use std::time::Duration;

    #[tokio::test]
    async fn sqlite_queues_distribute_across_pools_and_isolate_namespaces() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("delivery.db");
        let connection_string = format!("sqlite://{}", path.display());
        let first_storage = Sqlite::new(&connection_string).await?;
        let second_storage = Sqlite::new(&connection_string).await?;
        let first_queue = SqliteQueue::new(first_storage.pool().clone(), "shared")?;
        let second_queue = SqliteQueue::new(second_storage.pool().clone(), "shared")?;
        let isolated_queue = SqliteQueue::new(second_storage.pool().clone(), "isolated")?;
        let publisher = first_queue.new_publisher("jobs").await?;

        publisher.publish(b"first").await?;
        publisher.publish(b"second").await?;

        let mut first_consumer = first_queue.new_consumer("jobs", 1).await?;
        let mut second_consumer = second_queue.new_consumer("jobs", 2).await?;
        let first = first_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("first SQLite consumer stopped"))??;
        let second = second_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("second SQLite consumer stopped"))??;
        assert_eq!(
            [first.data(), second.data()],
            [b"first".as_slice(), b"second".as_slice()]
        );

        first.nack_requeue().await?;
        second.ack().await?;
        let retried = second_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("SQLite retry consumer stopped"))??;
        assert_eq!(retried.data(), b"first");
        retried.ack().await?;

        publisher.publish(b"shared-only").await?;
        let mut isolated_consumer = isolated_queue.new_consumer("jobs", 3).await?;
        assert!(
            tokio::time::timeout(Duration::from_millis(250), isolated_consumer.next())
                .await
                .is_err()
        );
        let shared_only = first_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("shared SQLite consumer stopped"))??;
        assert_eq!(shared_only.data(), b"shared-only");
        shared_only.ack().await?;
        Ok(())
    }

    #[tokio::test]
    async fn sqlite_claim_walks_the_order_index_without_sorting() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("delivery.db");
        let storage = Sqlite::new(&format!("sqlite://{}", path.display())).await?;
        // Only the candidates scan matters: the later `batch` step re-sorts
        // at most `SQLITE_PREFETCH_SIZE` already-selected rows, which is cheap.
        let start = super::SQLITE_CLAIM_SQL
            .find("SELECT sequence FROM later_delivery_queue")
            .ok_or_else(|| anyhow::anyhow!("claim SQL changed shape"))?;
        let end = super::SQLITE_CLAIM_SQL
            .find("), batch AS")
            .ok_or_else(|| anyhow::anyhow!("claim SQL changed shape"))?;
        let candidates = &super::SQLITE_CLAIM_SQL[start..end];
        let details = sqlx::query(&format!("EXPLAIN QUERY PLAN {candidates}"))
            .bind("")
            .bind("")
            .bind("ns")
            .bind("jobs")
            .bind(0_i64)
            .bind(super::SQLITE_PREFETCH_SIZE)
            .fetch_all(storage.pool())
            .await?;
        let details: Vec<String> = details
            .iter()
            .map(|row| sqlx::Row::try_get::<String, _>(row, "detail"))
            .collect::<Result<_, _>>()?;
        assert!(
            details
                .iter()
                .any(|d| d.contains("later_delivery_queue_order_idx")),
            "claim must use the order index: {details:?}"
        );
        assert!(
            !details.iter().any(|d| d.contains("TEMP B-TREE")),
            "claim must not sort the backlog: {details:?}"
        );
        Ok(())
    }

    #[tokio::test]
    async fn sqlite_consumer_prefetches_a_deep_backlog() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("delivery.db");
        let connection_string = format!("sqlite://{}", path.display());
        let storage = Sqlite::new(&connection_string).await?;
        let queue = SqliteQueue::new(storage.pool().clone(), "prefetch")?;
        let publisher = queue.new_publisher("jobs").await?;
        for number in 0..super::SQLITE_PREFETCH_SIZE {
            publisher.publish(number.to_string().as_bytes()).await?;
        }

        let mut consumer = queue.new_consumer("jobs", 1).await?;
        let first = consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("SQLite consumer stopped"))??;
        let leased: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM later_delivery_queue \
             WHERE queue_name = 'jobs' AND lease_owner IS NOT NULL",
        )
        .fetch_one(storage.pool())
        .await?;
        assert_eq!(leased, super::SQLITE_PREFETCH_SIZE);
        first.ack().await?;

        for _ in 1..super::SQLITE_PREFETCH_SIZE {
            let payload = consumer
                .next()
                .await
                .ok_or_else(|| anyhow::anyhow!("SQLite consumer stopped"))??;
            payload.ack().await?;
        }
        assert!(
            tokio::time::timeout(Duration::from_millis(250), consumer.next())
                .await
                .is_err()
        );
        Ok(())
    }
}

#[cfg(all(test, feature = "postgres"))]
mod postgres_tests {
    use super::{MqClient, PostgresQueue};
    use crate::storage::Postgres;

    #[tokio::test]
    async fn postgres_queues_distribute_across_pools() -> anyhow::Result<()> {
        let connection_string = std::env::var("LATER_POSTGRES_TEST_URL")
            .unwrap_or_else(|_| "postgres://test:test@127.0.0.1:55432/later_test".to_owned());
        let first_storage = Postgres::new(&connection_string).await?;
        let second_storage = Postgres::new(&connection_string).await?;
        let namespace = format!("postgres-delivery-{}", crate::generate_id());
        let first_queue = PostgresQueue::new(first_storage.pool().clone(), namespace.clone())?;
        let second_queue = PostgresQueue::new(second_storage.pool().clone(), namespace)?;
        let publisher = first_queue.new_publisher("jobs").await?;

        publisher.publish(b"first").await?;
        publisher.publish(b"second").await?;

        let mut first_consumer = first_queue.new_consumer("jobs", 1).await?;
        let mut second_consumer = second_queue.new_consumer("jobs", 2).await?;
        let first = first_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("first Postgres consumer stopped"))??;
        let second = second_consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("second Postgres consumer stopped"))??;
        assert_eq!(
            [first.data(), second.data()],
            [b"first".as_slice(), b"second".as_slice()]
        );

        first.ack().await?;
        second.ack().await?;
        Ok(())
    }

    #[tokio::test]
    async fn postgres_consumer_prefetches_a_deep_backlog() -> anyhow::Result<()> {
        let connection_string = std::env::var("LATER_POSTGRES_TEST_URL")
            .unwrap_or_else(|_| "postgres://test:test@127.0.0.1:55432/later_test".to_owned());
        let storage = Postgres::new(&connection_string).await?;
        let namespace = format!("postgres-prefetch-{}", crate::generate_id());
        let queue = PostgresQueue::new(storage.pool().clone(), namespace.clone())?;
        let publisher = queue.new_publisher("jobs").await?;
        for number in 0..super::POSTGRES_PREFETCH_SIZE {
            publisher.publish(number.to_string().as_bytes()).await?;
        }

        let mut consumer = queue.new_consumer("jobs", 1).await?;
        let first = consumer
            .next()
            .await
            .ok_or_else(|| anyhow::anyhow!("Postgres consumer stopped"))??;
        let leased: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM later_delivery_queue \
             WHERE namespace = $1 AND queue_name = 'jobs' AND lease_owner IS NOT NULL",
        )
        .bind(&namespace)
        .fetch_one(storage.pool())
        .await?;
        assert_eq!(leased, super::POSTGRES_PREFETCH_SIZE);
        first.ack().await?;

        for _ in 1..super::POSTGRES_PREFETCH_SIZE {
            let payload = consumer
                .next()
                .await
                .ok_or_else(|| anyhow::anyhow!("Postgres consumer stopped"))??;
            payload.ack().await?;
        }
        Ok(())
    }
}