Fix pipeline with pedantic execution for stages

CHANGELOG.md

@@ -23,6 +23,7 @@ All changes in this project will be noted in this file.
 - (skyd) Fixed LF check in protocol impl
 - (skyd) Fixed new instance detection (now checks if data directory is empty or not)
 - (skyd) Fixed panic resulting from corrupted metadata in `PARTMAP`
+- (skyd) Fixed invalid pipeline response when action error is propagated from a single stage
 
 ## Version 0.7.4
 

server/src/queryengine/mod.rs

@@ -26,7 +26,7 @@
 
 //! # The Query Engine
 
-use crate::actions::ActionResult;
+use crate::actions::{ActionError, ActionResult};
 use crate::auth;
 use crate::corestore::Corestore;
 use crate::dbnet::connection::prelude::*;
@@ -198,6 +198,26 @@ action! {
     }
 }
 
+/// Execute a stage **completely**. This means that action errors are never propagated
+/// over the try operator
+async fn execute_stage_pedantic<'a, T: ClientConnection<Strm> + 'a, Strm: Stream + 'a>(
+    handle: &mut Corestore,
+    con: &mut T,
+    auth: &mut AuthProviderHandle<'_, T, Strm>,
+    stage: &UnsafeElement,
+) -> crate::IoResult<()> {
+    let ret = async {
+        ensure_cond_or_err(stage.is_any_array(), groups::WRONGTYPE_ERR)?;
+        self::execute_stage(handle, con, auth, stage).await?;
+        Ok(())
+    };
+    match ret.await {
+        Ok(()) => Ok(()),
+        Err(ActionError::ActionError(e)) => con.write_response(e).await,
+        Err(ActionError::IoError(ioe)) => Err(ioe),
+    }
+}
+
 action! {
     /// Execute a basic pipelined query
     fn execute_pipeline(
@@ -207,8 +227,7 @@ action! {
         pipeline: PipelineQuery
     ) {
         for stage in pipeline.iter() {
-            ensure_cond_or_err(stage.is_any_array(), groups::WRONGTYPE_ERR)?;
-            self::execute_stage(handle, con, auth, stage).await?;
+            self::execute_stage_pedantic(handle, con, auth, stage).await?;
         }
         Ok(())
     }

server/src/tests/mod.rs

@@ -55,15 +55,6 @@ mod sys {
     use libsky::VERSION;
     use sky_macros::dbtest_func as dbtest;
     use skytable::{query, Element, RespCode};
-    const UNKNOWN_ACTION: &str = "Unknown action";
-    #[dbtest]
-    async fn test_sys_unknown_action() {
-        runeq!(
-            con,
-            query!("sys", "what"),
-            Element::RespCode(RespCode::ErrorString(UNKNOWN_ACTION.to_owned()))
-        );
-    }
     #[dbtest]
     async fn test_sys_info_aerr() {
         runeq!(

server/src/tests/pipeline.rs

@@ -26,7 +26,7 @@
 
 #[sky_macros::dbtest_module]
 mod tests {
-    use skytable::{query, Element, Pipeline, RespCode};
+    use skytable::{query, types::Array, Element, Pipeline, RespCode};
     async fn test_pipeline_heya_echo() {
         let pipe = Pipeline::new()
             .append(query!("heya", "first"))
@@ -67,4 +67,19 @@ mod tests {
             ]
         );
     }
+    async fn test_pipeline_with_multiple_error() {
+        let pipe = Pipeline::new()
+            .append(query!("mset", "x", "y", "z"))
+            .append(query!("mget", "x", "y", "z"))
+            .append(query!("heya", "finally"));
+        let ret = con.run_pipeline(pipe).await.unwrap();
+        assert_eq!(
+            ret,
+            vec![
+                Element::RespCode(RespCode::ActionError),
+                Element::Array(Array::Str(vec![None, None, None])),
+                Element::String("finally".to_owned())
+            ]
+        )
+    }
 }