From 2dd6537af05bb605853136bce4db3d27d8dddc13 Mon Sep 17 00:00:00 2001
From: Sayan Nandan <nandansayan@outlook.com>
Date: Thu, 4 Apr 2024 23:03:36 +0530
Subject: [PATCH] net: Add illegal packet escape for pipeline

Additional changes:
- Added changelog entry
---
 CHANGELOG.md                          |  6 ++++
 server/src/engine/net/protocol/mod.rs | 40 +++++++++++----------------
 2 files changed, 22 insertions(+), 24 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e236e454..c4bd8383 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 All changes in this project will be noted in this file.
 
+## Version 0.8.2
+
+### Additions
+
+- Skyhash/2.1: Restored support for pipelines
+
 ## Version 0.8.1
 
 ### Additions
diff --git a/server/src/engine/net/protocol/mod.rs b/server/src/engine/net/protocol/mod.rs
index ace91eb1..8276653e 100644
--- a/server/src/engine/net/protocol/mod.rs
+++ b/server/src/engine/net/protocol/mod.rs
@@ -53,12 +53,14 @@ use {
         },
     },
     super::{IoResult, QueryLoopResult, Socket},
-    crate::engine::{
-        self,
-        core::system_db::VerifyUser,
-        error::{QueryError, QueryResult},
-        fractal::{Global, GlobalInstanceLike},
-        mem::{BufferedScanner, IntegerRepr},
+    crate::{
+        engine::{
+            core::{exec, system_db::VerifyUser},
+            error::{QueryError, QueryResult},
+            fractal::{Global, GlobalInstanceLike},
+            mem::{BufferedScanner, IntegerRepr},
+        },
+        util::compiler,
     },
     bytes::{Buf, BytesMut},
     tokio::io::{AsyncReadExt, AsyncWriteExt, BufWriter},
@@ -348,17 +350,17 @@ async fn exec_simple<S: Socket>(
     global: &Global,
     query: SQuery<'_>,
 ) -> IoResult<()> {
-    write_response(
-        engine::core::exec::dispatch_to_executor(global, cs, query).await,
-        con,
-    )
-    .await
+    write_response(exec::dispatch_to_executor(global, cs, query).await, con).await
 }
 
 /*
     pipeline
+    ---
+    malformed packets
 */
 
+const ILLEGAL_PACKET_ESCAPE: u8 = 0xFF;
+
 async fn exec_pipe<'a, S: Socket>(
     con: &mut BufWriter<S>,
     cs: &mut ClientLocalState,
@@ -368,20 +370,10 @@ async fn exec_pipe<'a, S: Socket>(
     let mut pipe = pipe.into_iter();
     while let Some(query) = pipe.next() {
         match query {
-            Ok(q) => {
-                write_response(
-                    engine::core::exec::dispatch_to_executor(global, cs, q).await,
-                    con,
-                )
-                .await?
-            }
+            Ok(q) => write_response(exec::dispatch_to_executor(global, cs, q).await, con).await?,
             Err(_) => {
-                // respond with error
-                let [a, b] = (QueryError::SysNetworkSystemIllegalClientPacket.value_u8() as u16)
-                    .to_le_bytes();
-                con.write_all(&[ResponseType::Error.value_u8(), a, b])
-                    .await?;
-                return Ok(());
+                return compiler::cold_call(|| async { con.write_u8(ILLEGAL_PACKET_ESCAPE).await })
+                    .await
             }
         }
     }