Add 'set_ack_deadline' to Delivery, and implement for SQS/GCP

This allows changing the 'ack deadline' (e.g. the visibility
timeout in SQS) for an existing message, which allows the consumer
to spend more time processing the message before calling 'ack()'.

At the moment, this is only implemented for SQS and GCP.
To support this, I've added back the `QueueError::Unsupported`
enum variant (which is a breaking change).

This is gated behind an off-by-default 'beta' feature flag at
the moment

omniqueue/Cargo.toml

@@ -50,3 +50,4 @@ redis = ["dep:bb8", "dep:bb8-redis", "dep:redis", "dep:svix-ksuid"]
 redis_cluster = ["redis", "redis/cluster-async"]
 sqs = ["dep:aws-config", "dep:aws-sdk-sqs"]
 azure_queue_storage = ["dep:azure_storage", "dep:azure_storage_queues"]
+beta = []

omniqueue/src/backends/azure_queue_storage.rs

@@ -170,6 +170,12 @@ impl Acker for AqsAcker {
     async fn nack(&mut self) -> Result<()> {
         Ok(())
     }
+
+    async fn set_ack_deadline(&mut self, _duration: Duration) -> Result<()> {
+        Err(QueueError::Unsupported(
+            "set_ack_deadline is not yet supported by InMemoryBackend",
+        ))
+    }
 }
 
 impl AqsConsumer {

omniqueue/src/backends/gcp_pubsub.rs

@@ -281,4 +281,17 @@ impl Acker for GcpPubSubAcker {
     async fn nack(&mut self) -> Result<()> {
         self.recv_msg.nack().await.map_err(QueueError::generic)
     }
+
+    async fn set_ack_deadline(&mut self, duration: Duration) -> Result<()> {
+        let duration_secs = duration.as_secs().try_into().map_err(|e| {
+            QueueError::Generic(Box::<dyn std::error::Error + Send + Sync>::from(format!(
+                "set_ack_deadline duration {duration:?} is too large: {e:?}"
+            )))
+        })?;
+
+        self.recv_msg
+            .modify_ack_deadline(duration_secs)
+            .await
+            .map_err(QueueError::generic)
+    }
 }

omniqueue/src/backends/in_memory.rs

@@ -176,6 +176,12 @@ impl Acker for InMemoryAcker {
                 .map_err(QueueError::generic)
         }
     }
+
+    async fn set_ack_deadline(&mut self, _duration: Duration) -> Result<()> {
+        Err(QueueError::Unsupported(
+            "set_ack_deadline is not yet supported by InMemoryBackend",
+        ))
+    }
 }
 
 #[cfg(test)]

omniqueue/src/backends/rabbitmq.rs

@@ -301,4 +301,10 @@ impl Acker for RabbitMqAcker {
             .map(|_| ())
             .map_err(QueueError::generic)
     }
+
+    async fn set_ack_deadline(&mut self, _duration: Duration) -> Result<()> {
+        Err(QueueError::Unsupported(
+            "set_ack_deadline is not supported by RabbitMQ",
+        ))
+    }
 }

omniqueue/src/backends/redis/fallback.rs

@@ -116,6 +116,12 @@ impl<R: RedisConnection> Acker for RedisFallbackAcker<R> {
 
         Ok(())
     }
+
+    async fn set_ack_deadline(&mut self, _duration: Duration) -> Result<()> {
+        Err(QueueError::Unsupported(
+            "set_ack_deadline is not yet supported by redis fallback backend",
+        ))
+    }
 }
 
 pub(super) async fn add_to_main_queue(

omniqueue/src/backends/redis/streams.rs

@@ -163,6 +163,12 @@ impl<R: RedisConnection> Acker for RedisStreamsAcker<R> {
 
         Ok(())
     }
+
+    async fn set_ack_deadline(&mut self, _duration: Duration) -> Result<()> {
+        Err(QueueError::Unsupported(
+            "set_ack_deadline is not yet supported by redis streams backend",
+        ))
+    }
 }
 
 pub(super) async fn add_to_main_queue(

omniqueue/src/backends/sqs.rs

@@ -225,6 +225,34 @@ impl Acker for SqsAcker {
     async fn nack(&mut self) -> Result<()> {
         Ok(())
     }
+
+    async fn set_ack_deadline(&mut self, duration: Duration) -> Result<()> {
+        if let Some(receipt_handle) = &self.receipt_handle {
+            let duration_secs = duration.as_secs().try_into().map_err(|e| {
+                QueueError::Generic(Box::<dyn std::error::Error + Send + Sync>::from(format!(
+                    "set_ack_deadline duration {duration:?} is too large: {e:?}"
+                )))
+            })?;
+            self.ack_client
+                .change_message_visibility()
+                .set_visibility_timeout(Some(duration_secs))
+                .queue_url(&self.queue_dsn)
+                .receipt_handle(receipt_handle)
+                .send()
+                .await
+                .map_err(aws_to_queue_error)?;
+
+            Ok(())
+        } else {
+            Err(QueueError::generic(
+                DeleteMessageError::ReceiptHandleIsInvalid(
+                    ReceiptHandleIsInvalid::builder()
+                        .message("receipt handle must be Some to set ack deadline")
+                        .build(),
+                ),
+            ))
+        }
+    }
 }
 
 pub struct SqsProducer {

omniqueue/src/lib.rs

@@ -134,6 +134,9 @@ pub enum QueueError {
 
     #[error("{0}")]
     Generic(Box<dyn std::error::Error + Send + Sync>),
+
+    #[error("{0}")]
+    Unsupported(&'static str),
 }
 
 impl QueueError {

omniqueue/src/queue/mod.rs

@@ -1,4 +1,4 @@
-use std::{fmt, future::Future};
+use std::{fmt, future::Future, time::Duration};
 
 use async_trait::async_trait;
 use serde::de::DeserializeOwned;
@@ -59,6 +59,19 @@ impl Delivery {
         self.acker.ack().await.map_err(|e| (e, self))
     }
 
+    #[cfg(feature = "beta")]
+    /// Sets the deadline for acknowledging this [`Delivery`] to `duration`,
+    /// starting from the time this method is called.
+    ///
+    /// The exact nature of this will vary per backend, but usually ensures
+    /// that the same message will not be reprocessed if `ack()` is called
+    /// within an interval of `duration` from the time this method is
+    /// called. For example, this corresponds to the 'visibility timeout' in
+    /// SQS, and the 'ack deadline' in GCP
+    pub async fn set_ack_deadline(&mut self, duration: Duration) -> Result<(), QueueError> {
+        self.acker.set_ack_deadline(duration).await
+    }
+
     /// Explicitly does not Acknowledge the successful processing of this
     /// [`Delivery`].
     ///
@@ -102,4 +115,6 @@ impl fmt::Debug for Delivery {
 pub(crate) trait Acker: Send + Sync {
     async fn ack(&mut self) -> Result<()>;
     async fn nack(&mut self) -> Result<()>;
+    #[cfg_attr(not(feature = "beta"), allow(dead_code))]
+    async fn set_ack_deadline(&mut self, duration: Duration) -> Result<()>;
 }