feat(cubestore): Queue - return success marker on ack (set result)

Author: ovr

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -8274,10 +8274,14 @@ async fn queue_full_workflow_v1(service: Box<dyn SqlClient>) {
         let ack = async move {
             tokio::time::sleep(Duration::from_millis(1000)).await;
 
-            service_to_move
+            let ack_result = service_to_move
                 .exec_query(r#"QUEUE ACK "STANDALONE#queue:3" "result:3""#)
                 .await
-                .unwrap()
+                .unwrap();
+            assert_eq!(
+                ack_result.get_rows(),
+                &vec![Row::new(vec![TableValue::Boolean(true)])]
+            )
         };
 
         let (blocking_res, _ack_res) = join!(blocking, ack);
@@ -8439,10 +8443,38 @@ async fn queue_ack_then_result(service: Box<dyn SqlClient>) {
         .await
         .unwrap();
 
-    service
+    let ack_result = service
         .exec_query(r#"QUEUE ACK "STANDALONE#queue:5555" "result:5555""#)
         .await
         .unwrap();
+    assert_eq!(
+        ack_result.get_rows(),
+        &vec![Row::new(vec![TableValue::Boolean(true)])]
+    );
+
+    // double ack for result
+    {
+        let ack_result = service
+            .exec_query(r#"QUEUE ACK "STANDALONE#queue:5555" "result:5555""#)
+            .await
+            .unwrap();
+        assert_eq!(
+            ack_result.get_rows(),
+            &vec![Row::new(vec![TableValue::Boolean(false)])]
+        );
+    }
+
+    // ack on unknown queue item
+    {
+        let ack_result = service
+            .exec_query(r#"QUEUE ACK "STANDALONE#queue:123456" "result:5555""#)
+            .await
+            .unwrap();
+        assert_eq!(
+            ack_result.get_rows(),
+            &vec![Row::new(vec![TableValue::Boolean(false)])]
+        );
+    }
 
     let result = service
         .exec_query(r#"QUEUE RESULT "STANDALONE#queue:5555""#)
@@ -8737,10 +8769,14 @@ async fn queue_multiple_result_blocking(service: Box<dyn SqlClient>) {
         let ack = async move {
             tokio::time::sleep(Duration::from_millis(1000)).await;
 
-            service_to_move
+            let ack_result = service_to_move
                 .exec_query(r#"QUEUE ACK "STANDALONE#queue:12345" "result:12345""#)
                 .await
-                .unwrap()
+                .unwrap();
+            assert_eq!(
+                ack_result.get_rows(),
+                &vec![Row::new(vec![TableValue::Boolean(true)])]
+            )
         };
 
         let (blocking1_res, blocking2_res, _ack_res) = join!(blocking1, blocking2, ack);

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -405,7 +405,7 @@ pub trait CacheStore: DIService + Send + Sync {
         &self,
         path: String,
         result: Option<String>,
-    ) -> Result<(), CubeError>;
+    ) -> Result<bool, CubeError>;
     async fn queue_result_by_path(
         &self,
         path: String,
@@ -811,7 +811,7 @@ impl CacheStore for RocksCacheStore {
         &self,
         path: String,
         result: Option<String>,
-    ) -> Result<(), CubeError> {
+    ) -> Result<bool, CubeError> {
         self.store
             .write_operation(move |db_ref, batch_pipe| {
                 let queue_schema = QueueItemRocksTable::new(db_ref.clone());
@@ -841,11 +841,11 @@ impl CacheStore for RocksCacheStore {
                         }));
                     }
 
-                    Ok(())
+                    Ok(true)
                 } else {
                     warn!("Unable to ack queue, unknown path: {}", path);
 
-                    Ok(())
+                    Ok(false)
                 }
             })
             .await
@@ -1078,7 +1078,7 @@ impl CacheStore for ClusterCacheStoreClient {
         &self,
         _path: String,
         _result: Option<String>,
-    ) -> Result<(), CubeError> {
+    ) -> Result<bool, CubeError> {
         panic!("CacheStore cannot be used on the worker node! queue_ack_by_path was used.")
     }
 

rust/cubestore/cubestore/src/cachestore/lazy.rs

@@ -271,7 +271,7 @@ impl CacheStore for LazyRocksCacheStore {
         &self,
         path: String,
         result: Option<String>,
-    ) -> Result<(), CubeError> {
+    ) -> Result<bool, CubeError> {
         self.init().await?.queue_ack_by_path(path, result).await
     }
 

rust/cubestore/cubestore/src/queryplanner/test_utils.rs

@@ -808,7 +808,7 @@ impl CacheStore for CacheStoreMock {
         &self,
         _path: String,
         _result: Option<String>,
-    ) -> Result<(), CubeError> {
+    ) -> Result<bool, CubeError> {
         panic!("CacheStore mock!")
     }
 

rust/cubestore/cubestore/src/sql/cachestore.rs

@@ -229,9 +229,15 @@ impl CacheStoreSqlService {
                 (Arc::new(DataFrame::new(vec![], vec![])), true)
             }
             QueueCommand::Ack { key, result } => {
-                self.cachestore.queue_ack_by_path(key.value, result).await?;
+                let success = self.cachestore.queue_ack_by_path(key.value, result).await?;
 
-                (Arc::new(DataFrame::new(vec![], vec![])), true)
+                (
+                    Arc::new(DataFrame::new(
+                        vec![Column::new("success".to_string(), ColumnType::Boolean, 0)],
+                        vec![Row::new(vec![TableValue::Boolean(success)])],
+                    )),
+                    true,
+                )
             }
             QueueCommand::Get { key } => {
                 let result = self.cachestore.queue_get_by_path(key.value).await?;