redis: Only set already_acked_or_nacked after ack succeeded

svix · Aug 13, 2024 · 8c132ca · 8c132ca
1 parent 1e1cca5
commit 8c132ca
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/omniqueue/src/backends/redis/fallback.rs b/omniqueue/src/backends/redis/fallback.rs
@@ -92,8 +92,6 @@ impl<R: RedisConnection> Acker for RedisFallbackAcker<R> {
             return Err(QueueError::CannotAckOrNackTwice);
         }
 
-        self.already_acked_or_nacked = true;
-
         let _: () = self
             .redis
             .get()
@@ -103,6 +101,8 @@ impl<R: RedisConnection> Acker for RedisFallbackAcker<R> {
             .await
             .map_err(QueueError::generic)?;
 
+        self.already_acked_or_nacked = true;
+
         Ok(())
     }
 

diff --git a/omniqueue/src/backends/redis/streams.rs b/omniqueue/src/backends/redis/streams.rs
@@ -139,17 +139,19 @@ impl<R: RedisConnection> Acker for RedisStreamsAcker<R> {
             return Err(QueueError::CannotAckOrNackTwice);
         }
 
-        self.already_acked_or_nacked = true;
-
         let mut pipeline = redis::pipe();
         pipeline.xack(&self.queue_key, &self.consumer_group, &[&self.entry_id]);
         pipeline.xdel(&self.queue_key, &[&self.entry_id]);
 
         let mut conn = self.redis.get().await.map_err(QueueError::generic)?;
-        pipeline
+        let _: () = pipeline
             .query_async(&mut *conn)
             .await
-            .map_err(QueueError::generic)
+            .map_err(QueueError::generic)?;
+
+        self.already_acked_or_nacked = true;
+
+        Ok(())
     }
 
     async fn nack(&mut self) -> Result<()> {