Repository: skytable/skytable (language: rust)

Issue:
Restore pipelines
We removed pipelines in v0.8.0 but will once again add the implementation back in 0.8.1. This issue tracks the implementation.


Discussion / hints:

Fix patch:
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/Cargo.lock b/Cargo.lock
index 6e80763e..98275317 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -108,7 +108,7 @@ checksum = "a507401cad91ec6a857ed5513a2073c82a9b9048762b885bb98655b306964681"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
 ]
 
 [[package]]
@@ -881,7 +881,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
 ]
 
 [[package]]
@@ -1223,7 +1223,7 @@ checksum = "7eb0b34b42edc17f6b7cac84a52a1c5f0e1bb2227e997ca9011ea3dd34e8610b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
 ]
 
 [[package]]
@@ -1365,7 +1365,7 @@ dependencies = [
 [[package]]
 name = "skytable"
 version = "0.8.6"
-source = "git+https://github.com/skytable/client-rust.git#0080df9bddb277eff5f0af704ca0f09aa222b775"
+source = "git+https://github.com/skytable/client-rust.git#21263b5b3875ac185df58475b057c56d35ca5c30"
 dependencies = [
  "async-trait",
  "bb8",
@@ -1422,9 +1422,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.57"
+version = "2.0.58"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11a6ae1e52eb25aab8f3fb9fca13be982a373b8f1157ca14b897a825ba4a2d35"
+checksum = "44cfb93f38070beee36b3fef7d4f5a16f27751d94b187b666a5cc5e9b0d30687"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1489,7 +1489,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
 ]
 
 [[package]]
@@ -1569,7 +1569,7 @@ checksum = "9881bea7cbe687e36c9ab3b778c36cd0487402e270304e8b1296d5085303c1a2"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
 ]
 
 [[package]]
@@ -1611,7 +1611,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
  "wasm-bindgen-shared",
 ]
 
@@ -1633,7 +1633,7 @@ checksum = "e94f17b526d0a461a191c78ea52bbce64071ed5c04c9ffe424dcb38f74171bb7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.57",
+ "syn 2.0.58",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
diff --git a/server/src/engine/fractal/mgr.rs b/server/src/engine/fractal/mgr.rs
index 09924415..0b8f7190 100644
--- a/server/src/engine/fractal/mgr.rs
+++ b/server/src/engine/fractal/mgr.rs
@@ -430,13 +430,13 @@ impl FractalMgr {
                 _ = sigterm.recv() => {
                     info!("flp: finishing any pending maintenance tasks");
                     let global = global.clone();
-                    tokio::task::spawn_blocking(|| self.general_executor(global)).await.unwrap();
+                    let _ = tokio::task::spawn_blocking(|| self.general_executor(global)).await;
                     info!("flp: exited executor service");
                     break;
                 },
                 _ = tokio::time::sleep(dur) => {
                     let global = global.clone();
-                    tokio::task::spawn_blocking(|| self.general_executor(global)).await.unwrap()
+                    let _ = tokio::task::spawn_blocking(|| self.general_executor(global)).await;
                 }
                 task = lpq.recv() => {
                     let Task { threshold, task } = match task {
diff --git a/server/src/engine/mem/scanner.rs b/server/src/engine/mem/scanner.rs
index 2f2a2f03..c65b38fc 100644
--- a/server/src/engine/mem/scanner.rs
+++ b/server/src/engine/mem/scanner.rs
@@ -66,7 +66,7 @@ impl<'a, T> Scanner<'a, T> {
         self.__cursor
     }
     /// Returns the buffer from the current position
-    pub fn current_buffer(&self) -> &[T] {
+    pub fn current_buffer(&self) -> &'a [T] {
         &self.d[self.__cursor..]
     }
     /// Returns the ptr to the cursor
@@ -86,6 +86,9 @@ impl<'a, T> Scanner<'a, T> {
     pub fn has_left(&self, sizeof: usize) -> bool {
         self.remaining() >= sizeof
     }
+    pub fn remaining_size_is(&self, size: usize) -> bool {
+        self.remaining() == size
+    }
     /// Returns true if the rounded cursor matches the predicate
     pub fn rounded_cursor_matches(&self, f: impl Fn(&T) -> bool) -> bool {
         f(&self.d[self.rounded_cursor()]) & !self.eof()
diff --git a/server/src/engine/net/protocol/exchange.rs b/server/src/engine/net/protocol/exchange.rs
index 7522b97e..ad1e9a39 100644
--- a/server/src/engine/net/protocol/exchange.rs
+++ b/server/src/engine/net/protocol/exchange.rs
@@ -1,5 +1,5 @@
 /*
- * Created on Wed Sep 20 2023
+ * Created on Tue Apr 02 2024
  *
  * This file is a part of Skytable
  * Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
@@ -7,7 +7,7 @@
  * vision to provide flexibility in data modelling without compromising
  * on performance, queryability or scalability.
  *
- * Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
+ * Copyright (c) 2024, Sayan Nandan <nandansayan@outlook.com>
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published by
@@ -24,211 +24,370 @@
  *
 */
 
-use {super::AccumlatorStatus, crate::engine::mem::BufferedScanner};
+use {
+    crate::{engine::mem::BufferedScanner, util::compiler},
+    core::fmt,
+};
 
-#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
-pub struct Resume(usize);
-impl Resume {
-    #[cfg(test)]
-    pub(super) const fn test_new(v: usize) -> Self {
-        Self(v)
-    }
-    #[cfg(test)]
-    pub(super) const fn inner(&self) -> usize {
-        self.0
-    }
-}
-impl Default for Resume {
-    fn default() -> Self {
-        Self(0)
-    }
-}
+/*
+    Skyhash/2.1 Implementation
+    ---
+    This is an implementation of Skyhash/2.1, Skytable's data exchange protocol.
 
-pub(super) unsafe fn resume<'a>(
-    buf: &'a [u8],
-    last_cursor: Resume,
-    last_state: QExchangeState,
-) -> (Resume, QExchangeResult<'a>) {
-    let mut scanner = unsafe {
-        // UNSAFE(@ohsayan): we are the ones who generate the cursor and restore it
-        BufferedScanner::new_with_cursor(buf, last_cursor.0)
-    };
-    let ret = last_state.resume(&mut scanner);
-    (Resume(scanner.cursor()), ret)
-}
+    0. Notes
+    ++++++++++++++++++
+    - 2.1 is fully backwards compatible with 2.0 clients. As such we don't even designate it as a separate version.
+    - The "LF exception" essentially allows `0\n` to be equal to `\n`. It's unimportant to enforce this
+
+    1.1 Query Types
+    ++++++++++++++++++
+    The protocol makes two distinctions, at the protocol-level about the type of queries:
+    a. Simple query
+    b. Pipeline
+
+    1.1.1 Simple Query
+    ++++++++++++++++++
+    A simple query
+*/
 
 /*
-    SQ
+    sq definition
 */
 
 #[derive(Debug, PartialEq)]
 pub struct SQuery<'a> {
-    q: &'a [u8],
+    buf: &'a [u8],
     q_window: usize,
 }
 
 impl<'a> SQuery<'a> {
-    pub(super) fn new(q: &'a [u8], q_window: usize) -> Self {
-        Self { q, q_window }
+    fn new(buf: &'a [u8], q_window: usize) -> Self {
+        Self { buf, q_window }
     }
-    pub fn payload(&self) -> &'a [u8] {
-        self.q
+    #[cfg(test)]
+    pub(super) fn _new(buf: &'a [u8], q_window: usize) -> Self {
+        Self::new(buf, q_window)
     }
-    pub fn q_window(&self) -> usize {
-        self.q_window
+    pub fn query(&self) -> &[u8] {
+        &self.buf[..self.q_window]
     }
-    pub fn query(&self) -> &'a [u8] {
-        &self.payload()[..self.q_window()]
+    pub fn params(&self) -> &[u8] {
+        &self.buf[self.q_window..]
     }
-    pub fn params(&self) -> &'a [u8] {
-        &self.payload()[self.q_window()..]
+}
+
+/*
+    scanint
+*/
+
+fn scan_usize_guaranteed_termination(
+    scanner: &mut BufferedScanner,
+) -> Result<usize, ExchangeError> {
+    let mut ret = 0usize;
+    let mut stop = scanner.rounded_eq(b'\n');
+    while !scanner.eof() & !stop {
+        let this_byte = unsafe {
+            // UNSAFE(@ohsayan): loop invariant
+            scanner.next_byte()
+        };
+        match ret
+            .checked_mul(10)
+            .map(|int| int.checked_add((this_byte & 0x0f) as usize))
+        {
+            Some(Some(int)) if this_byte.is_ascii_digit() => ret = int,
+            _ => return Err(ExchangeError::NotAsciiByteOrOverflow),
+        }
+        stop = scanner.rounded_eq(b'\n');
     }
-    #[cfg(test)]
-    pub fn query_str(&self) -> &str {
-        core::str::from_utf8(self.query()).unwrap()
+    unsafe {
+        // UNSAFE(@ohsayan): scanned stop, but not accounted for yet
+        scanner.incr_cursor_if(stop)
+    }
+    if stop {
+        Ok(ret)
+    } else {
+        Err(ExchangeError::UnterminatedInteger)
+    }
+}
+
+#[derive(Clone, Copy, PartialEq)]
+pub(super) struct Usize {
+    v: isize,
+}
+
+impl Usize {
+    const SHIFT: u32 = isize::BITS - 1;
+    const MASK: isize = 1 << Self::SHIFT;
+    #[inline(always)]
+    const fn new(v: isize) -> Self {
+        Self { v }
+    }
+    #[inline(always)]
+    pub(super) const fn new_unflagged(int: usize) -> Self {
+        Self::new(int as isize)
     }
     #[cfg(test)]
-    pub fn params_str(&self) -> &str {
-        core::str::from_utf8(self.params()).unwrap()
+    pub(super) const fn new_flagged(int: usize) -> Self {
+        Self::new(int as isize | Self::MASK)
+    }
+    #[inline(always)]
+    fn int(&self) -> usize {
+        (self.v & !Self::MASK) as usize
+    }
+    #[inline(always)]
+    fn update(&mut self, new: usize) {
+        self.v = (new as isize) | (self.v & Self::MASK);
+    }
+    #[inline(always)]
+    fn flag(&self) -> bool {
+        (self.v & Self::MASK) != 0
+    }
+    #[inline(always)]
+    fn set_flag_if(&mut self, iff: bool) {
+        self.v |= (iff as isize) << Self::SHIFT;
+    }
+}
+
+impl fmt::Debug for Usize {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Usize")
+            .field("int", &self.int())
+            .field("flag", &self.flag())
+            .finish()
+    }
+}
+
+impl Usize {
+    /// Attempt to "complete" a scan of the integer. Idempotency guarantee: it is guaranteed that calls would not change the state
+    /// of the [`Usize`] or the buffer if the final state has been reached
+    fn update_scanned(&mut self, scanner: &mut BufferedScanner) -> Result<(), ()> {
+        let mut stop = scanner.rounded_eq(b'\n');
+        while !stop & !scanner.eof() & !self.flag() {
+            let byte = unsafe {
+                // UNSAFE(@ohsayan): verified by loop invariant
+                scanner.next_byte()
+            };
+            match (self.int() as isize) // this cast allows us to guarantee that we don't trip the flag
+                .checked_mul(10)
+                .map(|int| int.checked_add((byte & 0x0f) as isize))
+            {
+                Some(Some(int)) if byte.is_ascii_digit() => self.update(int as usize),
+                _ => return Err(()),
+            }
+            stop = scanner.rounded_eq(b'\n');
+        }
+        unsafe {
+            // UNSAFE(@ohsayan): scanned stop byte but did not account for it; the flag check is for cases where the input buffer
+            // has something like [LF][LF] in which case we stopped at the first LF but we would accidentally read the second one
+            // on the second derogatory call
+            scanner.incr_cursor_if(stop & !self.flag())
+        }
+        self.set_flag_if(stop | self.flag()); // if second call we must check the flag
+        Ok(())
     }
 }
 
 /*
-    utils
+    states
 */
 
-#[derive(Debug, PartialEq, Clone, Copy)]
-pub(super) enum QExchangeStateInternal {
+#[derive(Debug, PartialEq)]
+pub enum ExchangeState {
     Initial,
-    PendingMeta1,
-    PendingMeta2,
-    PendingData,
+    Simple(SQState),
+    Pipeline(PipeState),
 }
 
-impl Default for QExchangeStateInternal {
-    fn default() -> Self {
-        Self::Initial
+#[derive(Debug, PartialEq)]
+pub struct SQState {
+    packet_s: Usize,
+}
+
+impl SQState {
+    const fn new(packet_s: Usize) -> Self {
+        Self { packet_s }
+    }
+    #[cfg(test)]
+    pub(super) const fn _new(s: Usize) -> Self {
+        Self::new(s)
     }
 }
 
 #[derive(Debug, PartialEq)]
-pub(super) struct QExchangeState {
-    state: QExchangeStateInternal,
-    target: usize,
-    md_packet_size: u64,
-    md_q_window: u64,
+pub struct PipeState {
+    packet_s: Usize,
+}
+
+impl PipeState {
+    const fn new(packet_s: Usize) -> Self {
+        Self { packet_s }
+    }
+    #[cfg(test)]
+    pub const fn _new(packet_s: Usize) -> Self {
+        Self::new(packet_s)
+    }
 }
 
-impl Default for QExchangeState {
+impl Default for ExchangeState {
     fn default() -> Self {
-        Self::new()
+        Self::Initial
     }
 }
 
 #[derive(Debug, PartialEq)]
-/// Result after attempting to complete (or terminate) a query time exchange
-pub(super) enum QExchangeResult<'a> {
-    /// We completed the exchange and yielded a [`SQuery`]
-    SQCompleted(SQuery<'a>),
-    /// We're changing states
-    ChangeState(QExchangeState),
-    /// We hit an error and need to terminate this exchange
-    Error,
-}
-
-impl QExchangeState {
-    fn _new(
-        state: QExchangeStateInternal,
-        target: usize,
-        md_packet_size: u64,
-        md_q_window: u64,
-    ) -> Self {
-        Self {
-            state,
-            target,
-            md_packet_size,
-            md_q_window,
+pub enum ExchangeResult<'a> {
+    NewState(ExchangeState),
+    Simple(SQuery<'a>),
+    Pipeline(Pipeline<'a>),
+}
+
+#[derive(Debug, PartialEq, Clone, Copy)]
+#[repr(u8)]
+pub enum ExchangeError {
+    UnknownFirstByte,
+    NotAsciiByteOrOverflow,
+    UnterminatedInteger,
+    IncorrectQuerySizeOrMoreBytes,
+}
+
+pub struct Exchange<'a> {
+    scanner: BufferedScanner<'a>,
+}
+
+impl<'a> Exchange<'a> {
+    const MIN_Q_SIZE: usize = "P0\n".len();
+    fn new(scanner: BufferedScanner<'a>) -> Self {
+        Self { scanner }
+    }
+    pub fn try_complete(
+        scanner: BufferedScanner<'a>,
+        state: ExchangeState,
+    ) -> Result<(ExchangeResult, usize), ExchangeError> {
+        Self::new(scanner).complete(state)
+    }
+}
+
+impl<'a> Exchange<'a> {
+    fn complete(
+        mut self,
+        state: ExchangeState,
+    ) -> Result<(ExchangeResult<'a>, usize), ExchangeError> {
+        match state {
+            ExchangeState::Initial => {
+                if compiler::likely(self.scanner.has_left(Self::MIN_Q_SIZE)) {
+                    let first_byte = unsafe {
+                        // UNSAFE(@ohsayan): already verified in above branch
+                        self.scanner.next_byte()
+                    };
+                    match first_byte {
+                        b'S' => self.process_simple(SQState::new(Usize::new_unflagged(0))),
+                        b'P' => self.process_pipe(PipeState::new(Usize::new_unflagged(0))),
+                        _ => return Err(ExchangeError::UnknownFirstByte),
+                    }
+                } else {
+                    Ok(ExchangeResult::NewState(state))
+                }
+            }
+            ExchangeState::Simple(sq_s) => self.process_simple(sq_s),
+            ExchangeState::Pipeline(pipe_s) => self.process_pipe(pipe_s),
         }
+        .map(|ret| (ret, self.scanner.cursor()))
     }
-    #[cfg(test)]
-    pub(super) fn new_test(
-        state: QExchangeStateInternal,
-        target: usize,
-        md_packet_size: u64,
-        md_q_window: u64,
-    ) -> Self {
-        Self::_new(state, target, md_packet_size, md_q_window)
-    }
-}
-
-impl QExchangeState {
-    pub const MIN_READ: usize = b"S\x00\n\x00\n".len();
-    pub fn new() -> Self {
-        Self::_new(QExchangeStateInternal::Initial, Self::MIN_READ, 0, 0)
-    }
-    pub fn has_reached_target(&self, new_buffer: &[u8]) -> bool {
-        new_buffer.len() >= self.target
-    }
-    fn resume<'a>(self, scanner: &mut BufferedScanner<'a>) -> QExchangeResult<'a> {
-        debug_assert!(scanner.has_left(Self::MIN_READ));
-        match self.state {
-            QExchangeStateInternal::Initial => self.start_initial(scanner),
-            QExchangeStateInternal::PendingMeta1 => self.resume_at_md1(scanner),
-            QExchangeStateInternal::PendingMeta2 => self.resume_at_md2(scanner),
-            QExchangeStateInternal::PendingData => self.resume_data(scanner),
+    fn process_simple(
+        &mut self,
+        mut sq_state: SQState,
+    ) -> Result<ExchangeResult<'a>, ExchangeError> {
+        // try to complete the packet size if needed
+        sq_state
+            .packet_s
+            .update_scanned(&mut self.scanner)
+            .map_err(|_| ExchangeError::NotAsciiByteOrOverflow)?;
+        if sq_state.packet_s.flag() & self.scanner.has_left(sq_state.packet_s.int()) {
+            // we have the full packet size and the required data
+            // scan the query window
+            let start = self.scanner.cursor();
+            let q_window = scan_usize_guaranteed_termination(&mut self.scanner)?;
+            let stop = self.scanner.cursor();
+            // now compute remaining buffer length and nonzero condition
+            let expected_remaining_buffer = sq_state.packet_s.int() - (stop - start);
+            let nonzero = (q_window != 0) & (sq_state.packet_s.int() != 0);
+            // validate and return
+            if compiler::likely(self.scanner.remaining_size_is(expected_remaining_buffer) & nonzero)
+            {
+                // this check is important because the client might have given us an incorrect q size
+                let block = unsafe {
+                    // UNSAFE(@ohsayan): just verified earlier
+                    self.scanner.next_chunk_variable(expected_remaining_buffer)
+                };
+                Ok(ExchangeResult::Simple(SQuery::new(block, q_window)))
+            } else {
+                Err(ExchangeError::IncorrectQuerySizeOrMoreBytes)
+            }
+        } else {
+            Ok(ExchangeResult::NewState(ExchangeState::Simple(sq_state)))
         }
     }
-    fn start_initial<'a>(self, scanner: &mut BufferedScanner<'a>) -> QExchangeResult<'a> {
-        if unsafe { scanner.next_byte() } != b'S' {
-            // has to be a simple query!
-            return QExchangeResult::Error;
+    fn process_pipe(&mut self, mut pipe_s: PipeState) -> Result<ExchangeResult<'a>, ExchangeError> {
+        // try to complete the packet size if needed
+        pipe_s
+            .packet_s
+            .update_scanned(&mut self.scanner)
+            .map_err(|_| ExchangeError::NotAsciiByteOrOverflow)?;
+        if pipe_s.packet_s.flag() & self.scanner.remaining_size_is(pipe_s.packet_s.int()) {
+            // great, we have the entire packet
+            let block = unsafe {
+                // UNSAFE(@ohsayan): just verified earlier
+                self.scanner.next_chunk_variable(pipe_s.packet_s.int())
+            };
+            Ok(ExchangeResult::Pipeline(Pipeline::new(block)))
+        } else {
+            Ok(ExchangeResult::NewState(ExchangeState::Pipeline(pipe_s)))
         }
-        self.resume_at_md1(scanner)
-    }
-    fn resume_at_md1<'a>(mut self, scanner: &mut BufferedScanner<'a>) -> QExchangeResult<'a> {
-        let packet_size = match super::scan_int(scanner, self.md_packet_size) {
-            Ok(AccumlatorStatus::Completed(v)) => v,
-            Ok(AccumlatorStatus::Pending(p)) => {
-                // if this is the first run, we read 5 bytes and need atleast one more; if this is a resume we read one or more bytes and
-                // need atleast one more
-                self.target += 1;
-                self.md_packet_size = p;
-                self.state = QExchangeStateInternal::PendingMeta1;
-                return QExchangeResult::ChangeState(self);
-            }
-            Err(()) => return QExchangeResult::Error,
-        };
-        self.md_packet_size = packet_size;
-        self.target = scanner.cursor() + packet_size as usize;
-        // hand over control to md2
-        self.resume_at_md2(scanner)
-    }
-    fn resume_at_md2<'a>(mut self, scanner: &mut BufferedScanner<'a>) -> QExchangeResult<'a> {
-        let q_window = match super::scan_int(scanner, self.md_q_window) {
-            Ok(AccumlatorStatus::Completed(v)) => v,
-            Ok(AccumlatorStatus::Pending(p)) => {
-                self.md_q_window = p;
-                self.state = QExchangeStateInternal::PendingMeta2;
-                return QExchangeResult::ChangeState(self);
-            }
-            Err(()) => return QExchangeResult::Error,
-        };
-        self.md_q_window = q_window;
-        // hand over control to data
-        self.resume_data(scanner)
-    }
-    fn resume_data<'a>(mut self, scanner: &mut BufferedScanner<'a>) -> QExchangeResult<'a> {
-        let df_size = self.target - scanner.cursor();
-        if scanner.remaining() == df_size {
-            unsafe {
-                QExchangeResult::SQCompleted(SQuery::new(
-                    scanner.next_chunk_variable(df_size),
-                    self.md_q_window as usize,
-                ))
-            }
+    }
+}
+
+/*
+    pipeline
+*/
+
+#[derive(Debug, PartialEq)]
+pub struct Pipeline<'a> {
+    scanner: BufferedScanner<'a>,
+}
+
+impl<'a> Pipeline<'a> {
+    fn new(buf: &'a [u8]) -> Self {
+        Self {
+            scanner: BufferedScanner::new(buf),
+        }
+    }
+}
+
+impl<'a> Iterator for Pipeline<'a> {
+    type Item = Result<SQuery<'a>, ExchangeError>;
+    fn next(&mut self) -> Option<Self::Item> {
+        let nonzero = self.scanner.buffer_len() != 0;
+        if self.scanner.eof() & nonzero {
+            None
         } else {
-            self.state = QExchangeStateInternal::PendingData;
-            QExchangeResult::ChangeState(self)
+            let mut ret = || {
+                let query_size = scan_usize_guaranteed_termination(&mut self.scanner)?;
+                let param_size = scan_usize_guaranteed_termination(&mut self.scanner)?;
+                let (full_size, overflow) = param_size.overflowing_add(query_size);
+                if compiler::likely(self.scanner.has_left(full_size) & !overflow) {
+                    let block = unsafe {
+                        // UNSAFE(@ohsayan): checked in above branch
+                        self.scanner.next_chunk_variable(full_size)
+                    };
+                    Ok(SQuery {
+                        buf: block,
+                        q_window: query_size,
+                    })
+                } else {
+                    Err(ExchangeError::IncorrectQuerySizeOrMoreBytes)
+                }
+            };
+            Some(ret())
         }
     }
 }
diff --git a/server/src/engine/net/protocol/handshake.rs b/server/src/engine/net/protocol/handshake.rs
index 29a798de..7dd6ec22 100644
--- a/server/src/engine/net/protocol/handshake.rs
+++ b/server/src/engine/net/protocol/handshake.rs
@@ -210,12 +210,7 @@ pub enum HandshakeResult<'a> {
     /// Finished handshake
     Completed(CHandshake<'a>),
     /// Update handshake state
-    ///
-    /// **NOTE:** expect does not take into account the current amount of buffered data (hence the unbuffered part must be computed!)
-    ChangeState {
-        new_state: HandshakeState,
-        expect: usize,
-    },
+    ChangeState { new_state: HandshakeState },
     /// An error occurred
     Error(ProtocolError),
 }
@@ -271,17 +266,15 @@ impl<'a> CHandshake<'a> {
     /// Resume from the initial state (nothing buffered yet)
     fn resume_initial(scanner: &mut BufferedScanner<'a>) -> HandshakeResult<'a> {
         // get our block
-        if cfg!(debug_assertions) {
-            if scanner.remaining() < Self::INITIAL_READ {
-                return HandshakeResult::ChangeState {
-                    new_state: HandshakeState::Initial,
-                    expect: Self::INITIAL_READ,
-                };
-            }
-        } else {
-            assert!(scanner.remaining() >= Self::INITIAL_READ);
+        if scanner.remaining() < Self::INITIAL_READ {
+            return HandshakeResult::ChangeState {
+                new_state: HandshakeState::Initial,
+            };
         }
-        let buf: [u8; CHandshake::INITIAL_READ] = unsafe { scanner.next_chunk() };
+        let buf: [u8; CHandshake::INITIAL_READ] = unsafe {
+            // UNSAFE(@ohsayan): validated in earlier branch
+            scanner.next_chunk()
+        };
         let invalid_first_byte = buf[0] != Self::CLIENT_HELLO;
         let invalid_hs_version = buf[1] > HandshakeVersion::MAX_DSCR;
         let invalid_proto_version = buf[2] > ProtocolVersion::MAX_DSCR;
@@ -350,7 +343,6 @@ impl<'a> CHandshake<'a> {
                 uname_l,
                 pwd_l,
             },
-            expect: (uname_l + pwd_l),
         }
     }
 }
@@ -366,7 +358,6 @@ impl<'a> CHandshake<'a> {
             // we need more data
             return HandshakeResult::ChangeState {
                 new_state: HandshakeState::StaticBlock(static_header),
-                expect: static_header.auth_mode.min_payload_bytes(),
             };
         }
         // we seem to have enough data for this auth mode
@@ -379,7 +370,6 @@ impl<'a> CHandshake<'a> {
             ScannerDecodeResult::NeedMore => {
                 return HandshakeResult::ChangeState {
                     new_state: HandshakeState::StaticBlock(static_header),
-                    expect: AuthMode::Password.min_payload_bytes(), // 2 for uname_l and 2 for pwd_l
                 };
             }
             ScannerDecodeResult::Value(v) => v as usize,
@@ -402,7 +392,6 @@ impl<'a> CHandshake<'a> {
                 // newline missing (or maybe there's more?)
                 return HandshakeResult::ChangeState {
                     new_state: HandshakeState::ExpectingMetaForVariableBlock { static_hs, uname_l },
-                    expect: uname_l + 2, // space for username + password len
                 };
             }
             ScannerDecodeResult::Error => {
diff --git a/server/src/engine/net/protocol/mod.rs b/server/src/engine/net/protocol/mod.rs
index c1745eae..6017d8ce 100644
--- a/server/src/engine/net/protocol/mod.rs
+++ b/server/src/engine/net/protocol/mod.rs
@@ -44,55 +44,34 @@ mod handshake;
 #[cfg(test)]
 mod tests;
 
-// re-export
-pub use exchange::SQuery;
-
-use crate::engine::core::system_db::VerifyUser;
-
 use {
     self::{
-        exchange::{QExchangeResult, QExchangeState},
+        exchange::{Exchange, ExchangeResult, ExchangeState, Pipeline},
         handshake::{
-            AuthMode, CHandshake, DataExchangeMode, HandshakeResult, HandshakeState,
-            HandshakeVersion, ProtocolError, ProtocolVersion, QueryMode,
+            AuthMode, DataExchangeMode, HandshakeResult, HandshakeState, HandshakeVersion,
+            ProtocolError, ProtocolVersion, QueryMode,
         },
     },
     super::{IoResult, QueryLoopResult, Socket},
-    crate::engine::{
-        self,
-        error::QueryError,
-        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},
 };
 
-#[repr(u8)]
-#[derive(sky_macros::EnumMethods, Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
-#[allow(unused)]
-pub enum ResponseType {
-    Null = 0x00,
-    Bool = 0x01,
-    UInt8 = 0x02,
-    UInt16 = 0x03,
-    UInt32 = 0x04,
-    UInt64 = 0x05,
-    SInt8 = 0x06,
-    SInt16 = 0x07,
-    SInt32 = 0x08,
-    SInt64 = 0x09,
-    Float32 = 0x0A,
-    Float64 = 0x0B,
-    Binary = 0x0C,
-    String = 0x0D,
-    List = 0x0E,
-    Dict = 0x0F,
-    Error = 0x10,
-    Row = 0x11,
-    Empty = 0x12,
-    MultiRow = 0x13,
-}
+// re-export
+pub use self::exchange::SQuery;
+
+/*
+    connection state
+*/
 
 #[derive(Debug, PartialEq)]
 pub struct ClientLocalState {
@@ -128,111 +107,43 @@ impl ClientLocalState {
     }
 }
 
-#[derive(Debug, PartialEq)]
-pub enum Response {
-    Empty,
-    Null,
-    Serialized {
-        ty: ResponseType,
-        size: usize,
-        data: Vec<u8>,
-    },
-    Bool(bool),
-}
+/*
+    read loop
+*/
 
-pub(super) async fn query_loop<S: Socket>(
-    con: &mut BufWriter<S>,
-    buf: &mut BytesMut,
-    global: &Global,
-) -> IoResult<QueryLoopResult> {
-    // handshake
-    let mut client_state = match do_handshake(con, buf, global).await? {
-        PostHandshake::Okay(hs) => hs,
-        PostHandshake::ConnectionClosedFin => return Ok(QueryLoopResult::Fin),
-        PostHandshake::ConnectionClosedRst => return Ok(QueryLoopResult::Rst),
-        PostHandshake::Error(e) => {
-            // failed to handshake; we'll close the connection
-            let hs_err_packet = [b'H', 0, 1, e.value_u8()];
-            con.write_all(&hs_err_packet).await?;
-            return Ok(QueryLoopResult::HSFailed);
-        }
-    };
-    // done handshaking
-    con.write_all(b"H\x00\x00\x00").await?;
-    con.flush().await?;
-    let mut state = QExchangeState::default();
-    let mut cursor = Default::default();
-    loop {
-        if con.read_buf(buf).await? == 0 {
-            if buf.is_empty() {
-                return Ok(QueryLoopResult::Fin);
-            } else {
-                return Ok(QueryLoopResult::Rst);
-            }
-        }
-        if !state.has_reached_target(buf) {
-            // we haven't buffered sufficient bytes; keep working
-            continue;
-        }
-        let sq = match unsafe {
-            // UNSAFE(@ohsayan): as the resume cursor is private, we can't access this anyways
-            exchange::resume(buf, cursor, state)
-        } {
-            (_, QExchangeResult::SQCompleted(sq)) => sq,
-            (new_cursor, QExchangeResult::ChangeState(new_state)) => {
-                cursor = new_cursor;
-                state = new_state;
-                continue;
-            }
-            (_, QExchangeResult::Error) => {
-                // 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?;
-                con.flush().await?;
-                // reset buffer, cursor and state
-                buf.clear();
-                cursor = Default::default();
-                state = QExchangeState::default();
-                continue;
-            }
-        };
-        // now execute query
-        match engine::core::exec::dispatch_to_executor(global, &mut client_state, sq).await {
-            Ok(Response::Empty) => {
-                con.write_all(&[ResponseType::Empty.value_u8()]).await?;
-            }
-            Ok(Response::Serialized { ty, size, data }) => {
-                con.write_u8(ty.value_u8()).await?;
-                let mut irep = IntegerRepr::new();
-                con.write_all(irep.as_bytes(size as u64)).await?;
-                con.write_u8(b'\n').await?;
-                con.write_all(&data).await?;
-            }
-            Ok(Response::Bool(b)) => {
-                con.write_all(&[ResponseType::Bool.value_u8(), b as u8])
-                    .await?
-            }
-            Ok(Response::Null) => con.write_u8(ResponseType::Null.value_u8()).await?,
-            Err(e) => {
-                let [a, b] = (e.value_u8() as u16).to_le_bytes();
-                con.write_all(&[ResponseType::Error.value_u8(), a, b])
-                    .await?;
+macro_rules! read_loop {
+    ($con:expr, $buf:expr, $conn_closed:expr, $conn_reset:expr, $body:block) => {
+        loop {
+            let read_many = $con.read_buf($buf).await?;
+            if read_many == 0 {
+                if $buf.is_empty() {
+                    return $conn_closed;
+                } else {
+                    return $conn_reset;
+                }
             }
+            $body
         }
-        con.flush().await?;
-        // reset buffer, cursor and state
-        buf.clear();
-        cursor = Default::default();
-        state = QExchangeState::default();
-    }
+    };
+}
+
+/*
+    handshake
+*/
+
+#[inline(always)]
+async fn write_handshake_error<S: Socket>(
+    con: &mut BufWriter<S>,
+    e: ProtocolError,
+) -> IoResult<()> {
+    let hs_err_packet = [b'H', 0, 1, e.value_u8()];
+    con.write_all(&hs_err_packet).await
 }
 
 #[derive(Debug, PartialEq)]
-enum PostHandshake {
+enum HandshakeCompleteResult {
     Okay(ClientLocalState),
-    Error(ProtocolError),
+    Error,
     ConnectionClosedFin,
     ConnectionClosedRst,
 }
@@ -241,40 +152,34 @@ async fn do_handshake<S: Socket>(
     con: &mut BufWriter<S>,
     buf: &mut BytesMut,
     global: &Global,
-) -> IoResult<PostHandshake> {
-    let mut expected = CHandshake::INITIAL_READ;
+) -> IoResult<HandshakeCompleteResult> {
     let mut state = HandshakeState::default();
     let mut cursor = 0;
     let handshake;
-    loop {
-        let read_many = con.read_buf(buf).await?;
-        if read_many == 0 {
-            if buf.is_empty() {
-                return Ok(PostHandshake::ConnectionClosedFin);
-            } else {
-                return Ok(PostHandshake::ConnectionClosedRst);
-            }
-        }
-        if buf.len() < expected {
-            continue;
-        }
-        let mut scanner = unsafe { BufferedScanner::new_with_cursor(buf, cursor) };
-        match handshake::CHandshake::resume_with(&mut scanner, state) {
-            HandshakeResult::Completed(hs) => {
-                handshake = hs;
-                cursor = scanner.cursor();
-                break;
-            }
-            HandshakeResult::ChangeState { new_state, expect } => {
-                expected = expect;
-                state = new_state;
-                cursor = scanner.cursor();
-            }
-            HandshakeResult::Error(e) => {
-                return Ok(PostHandshake::Error(e));
+    read_loop!(
+        con,
+        buf,
+        Ok(HandshakeCompleteResult::ConnectionClosedFin),
+        Ok(HandshakeCompleteResult::ConnectionClosedRst),
+        {
+            let mut scanner = unsafe { BufferedScanner::new_with_cursor(buf, cursor) };
+            match handshake::CHandshake::resume_with(&mut scanner, state) {
+                HandshakeResult::Completed(hs) => {
+                    handshake = hs;
+                    cursor = scanner.cursor();
+                    break;
+                }
+                HandshakeResult::ChangeState { new_state } => {
+                    state = new_state;
+                    cursor = scanner.cursor();
+                }
+                HandshakeResult::Error(e) => {
+                    write_handshake_error(con, e).await?;
+                    return Ok(HandshakeCompleteResult::Error);
+                }
             }
         }
-    }
+    );
     // check handshake
     if cfg!(debug_assertions) {
         assert_eq!(
@@ -299,7 +204,7 @@ async fn do_handshake<S: Socket>(
             {
                 okay @ (VerifyUser::Okay | VerifyUser::OkayRoot) => {
                     let hs = handshake.hs_static();
-                    let ret = Ok(PostHandshake::Okay(ClientLocalState::new(
+                    let ret = Ok(HandshakeCompleteResult::Okay(ClientLocalState::new(
                         uname.into(),
                         okay.is_root(),
                         hs,
@@ -312,44 +217,186 @@ async fn do_handshake<S: Socket>(
         }
         Err(_) => {}
     };
-    Ok(PostHandshake::Error(ProtocolError::RejectAuth))
+    write_handshake_error(con, ProtocolError::RejectAuth).await?;
+    Ok(HandshakeCompleteResult::Error)
 }
 
-#[derive(Debug, PartialEq, Clone, Copy)]
-enum AccumlatorStatus {
-    Pending(u64),
-    Completed(u64),
+/*
+    exec event loop
+*/
+
+async fn cleanup_for_next_query<S: Socket>(
+    con: &mut BufWriter<S>,
+    buf: &mut BytesMut,
+) -> IoResult<(ExchangeState, usize)> {
+    con.flush().await?; // flush write buffer
+    buf.clear(); // clear read buffer
+    Ok((ExchangeState::default(), 0))
 }
 
-/// Scan an integer
-///
-/// Allowed sequences:
-/// - < int >\n
-/// - \n ; FIXME(@ohsayan): a LF only sequence is allowed. should it be removed?
-fn scan_int(s: &mut BufferedScanner, acc: u64) -> Result<AccumlatorStatus, ()> {
-    let mut acc = acc;
-    let mut okay = true;
-    let mut end = s.rounded_eq(b'\n');
-    while okay & !end & !s.eof() {
-        let d = unsafe { s.next_byte() };
-        okay &= d.is_ascii_digit();
-        match acc.checked_mul(10).map(|v| v.checked_add((d & 0x0f) as _)) {
-            Some(Some(v)) => acc = v,
-            _ => okay = false,
+pub(super) async fn query_loop<S: Socket>(
+    con: &mut BufWriter<S>,
+    buf: &mut BytesMut,
+    global: &Global,
+) -> IoResult<QueryLoopResult> {
+    // handshake
+    let mut client_state = match do_handshake(con, buf, global).await? {
+        HandshakeCompleteResult::Okay(hs) => hs,
+        HandshakeCompleteResult::ConnectionClosedFin => return Ok(QueryLoopResult::Fin),
+        HandshakeCompleteResult::ConnectionClosedRst => return Ok(QueryLoopResult::Rst),
+        HandshakeCompleteResult::Error => return Ok(QueryLoopResult::HSFailed),
+    };
+    // done handshaking
+    con.write_all(b"H\x00\x00\x00").await?;
+    con.flush().await?;
+    let mut state = ExchangeState::default();
+    let mut cursor = 0;
+    read_loop!(
+        con,
+        buf,
+        Ok(QueryLoopResult::Fin),
+        Ok(QueryLoopResult::Rst),
+        {
+            match Exchange::try_complete(
+                unsafe {
+                    // UNSAFE(@ohsayan): the cursor is either 0 or returned by the exchange impl
+                    BufferedScanner::new_with_cursor(&buf, cursor)
+                },
+                state,
+            ) {
+                Ok((result, new_cursor)) => match result {
+                    ExchangeResult::NewState(new_state) => {
+                        state = new_state;
+                        cursor = new_cursor;
+                    }
+                    ExchangeResult::Simple(query) => {
+                        exec_simple(con, &mut client_state, global, query).await?;
+                        (state, cursor) = cleanup_for_next_query(con, buf).await?;
+                    }
+                    ExchangeResult::Pipeline(pipe) => {
+                        exec_pipe(con, &mut client_state, global, pipe).await?;
+                        (state, cursor) = cleanup_for_next_query(con, buf).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?;
+                    (state, cursor) = cleanup_for_next_query(con, buf).await?;
+                }
+            }
+        }
+    );
+}
+
+/*
+    responses
+*/
+
+#[repr(u8)]
+#[derive(sky_macros::EnumMethods, Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+#[allow(unused)]
+pub enum ResponseType {
+    Null = 0x00,
+    Bool = 0x01,
+    UInt8 = 0x02,
+    UInt16 = 0x03,
+    UInt32 = 0x04,
+    UInt64 = 0x05,
+    SInt8 = 0x06,
+    SInt16 = 0x07,
+    SInt32 = 0x08,
+    SInt64 = 0x09,
+    Float32 = 0x0A,
+    Float64 = 0x0B,
+    Binary = 0x0C,
+    String = 0x0D,
+    List = 0x0E,
+    Dict = 0x0F,
+    Error = 0x10,
+    Row = 0x11,
+    Empty = 0x12,
+    MultiRow = 0x13,
+}
+
+#[derive(Debug, PartialEq)]
+pub enum Response {
+    Empty,
+    Null,
+    Serialized {
+        ty: ResponseType,
+        size: usize,
+        data: Vec<u8>,
+    },
+    Bool(bool),
+}
+
+async fn write_response<S: Socket>(
+    resp: QueryResult<Response>,
+    con: &mut BufWriter<S>,
+) -> IoResult<()> {
+    match resp {
+        Ok(Response::Empty) => con.write_all(&[ResponseType::Empty.value_u8()]).await,
+        Ok(Response::Serialized { ty, size, data }) => {
+            con.write_u8(ty.value_u8()).await?;
+            let mut irep = IntegerRepr::new();
+            con.write_all(irep.as_bytes(size as u64)).await?;
+            con.write_u8(b'\n').await?;
+            con.write_all(&data).await
+        }
+        Ok(Response::Bool(b)) => {
+            con.write_all(&[ResponseType::Bool.value_u8(), b as u8])
+                .await
+        }
+        Ok(Response::Null) => con.write_u8(ResponseType::Null.value_u8()).await,
+        Err(e) => {
+            let [a, b] = (e.value_u8() as u16).to_le_bytes();
+            con.write_all(&[ResponseType::Error.value_u8(), a, b]).await
         }
-        end = s.rounded_eq(b'\n');
-    }
-    unsafe {
-        // UNSAFE(@ohsayan): if we hit an LF, then we still have space until EOA
-        s.incr_cursor_if(end)
     }
-    if okay & end {
-        Ok(AccumlatorStatus::Completed(acc))
-    } else {
-        if okay {
-            Ok(AccumlatorStatus::Pending(acc))
-        } else {
-            Err(())
+}
+
+/*
+    simple query
+*/
+
+async fn exec_simple<S: Socket>(
+    con: &mut BufWriter<S>,
+    cs: &mut ClientLocalState,
+    global: &Global,
+    query: SQuery<'_>,
+) -> IoResult<()> {
+    write_response(exec::dispatch_to_executor(global, cs, query).await, con).await
+}
+
+/*
+    pipeline
+    ---
+    notes:
+    - executed without lookahead
+    - hence, malformed packets will need a special escape
+*/
+
+const ILLEGAL_PACKET_ESCAPE: u8 = 0xFF;
+
+async fn exec_pipe<'a, S: Socket>(
+    con: &mut BufWriter<S>,
+    cs: &mut ClientLocalState,
+    global: &Global,
+    pipe: Pipeline<'a>,
+) -> IoResult<()> {
+    let mut pipe = pipe.into_iter();
+    while let Some(query) = pipe.next() {
+        match query {
+            Ok(q) => write_response(exec::dispatch_to_executor(global, cs, q).await, con).await?,
+            Err(_) => {
+                return compiler::cold_call(|| async { con.write_u8(ILLEGAL_PACKET_ESCAPE).await })
+                    .await
+            }
         }
     }
+    Ok(())
 }
diff --git a/server/src/util/macros.rs b/server/src/util/macros.rs
index cc8b014a..aa72e22b 100644
--- a/server/src/util/macros.rs
+++ b/server/src/util/macros.rs
@@ -132,81 +132,6 @@ macro_rules! assert_hmeq {
     };
 }
 
-#[macro_export]
-/// ## The action macro
-///
-/// A macro for adding all the _fuss_ to an action. Implementing actions should be simple
-/// and should not require us to repeatedly specify generic paramters and/or trait bounds.
-/// This is exactly what this macro does: does all the _magic_ behind the scenes for you,
-/// including adding generic parameters, handling docs (if any), adding the correct
-/// trait bounds and finally making your function async. Rest knowing that all your
-/// action requirements have been happily addressed with this macro and that you don't have
-/// to write a lot of code to do the exact same thing
-///
-///
-/// ## Limitations
-///
-/// This macro can only handle mutable parameters for a fixed number of arguments (three)
-///
-macro_rules! action {
-    (
-        $($(#[$attr:meta])*
-        fn $fname:ident($($argname:ident: $argty:ty),* $(,)?)
-        $block:block)*
-    ) => {
-            $($(#[$attr])*
-            pub async fn $fname<
-                'a,
-                C: 'a + $crate::dbnet::BufferedSocketStream,
-                P: $crate::protocol::interface::ProtocolSpec,
-            > (
-                $($argname: $argty,)*
-            ) -> $crate::actions::ActionResult<()>
-            $block)*
-    };
-    (
-        $($(#[$attr:meta])*
-        fn $fname:ident(
-            $argone:ident: $argonety:ty,
-            $argtwo:ident: $argtwoty:ty,
-            mut $argthree:ident: $argthreety:ty $(,)?
-        ) $block:block)*
-    ) => {
-            $($(#[$attr])*
-            pub async fn $fname<
-            'a,
-                C: 'a + $crate::dbnet::BufferedSocketStream,
-                P: $crate::protocol::interface::ProtocolSpec,
-            >(
-                $argone: $argonety,
-                $argtwo: $argtwoty,
-                mut $argthree: $argthreety
-            ) -> $crate::actions::ActionResult<()>
-            $block)*
-    };
-    (
-        $($(#[$attr:meta])*
-        fn $fname:ident(
-            $argone:ident: $argonety:ty,
-            $argtwo:ident: $argtwoty:ty,
-            $argthree:ident: $argthreety:ty
-        ) $block:block)*
-    ) => {
-            $($(#[$attr])*
-            pub async fn $fname<
-                'a,
-                T: 'a + $crate::dbnet::connection::ClientConnection<P, Strm>,
-                Strm: $crate::dbnet::connection::Stream,
-                P: $crate::protocol::interface::ProtocolSpec
-            >(
-                $argone: $argonety,
-                $argtwo: $argtwoty,
-                $argthree: $argthreety
-            ) -> $crate::actions::ActionResult<()>
-            $block)*
-    };
-}
-
 #[macro_export]
 macro_rules! byt {
     ($f:expr) => {


Test patch:
diff --git a/server/src/engine/net/protocol/tests.rs b/server/src/engine/net/protocol/tests.rs
index 811a0569..ed3b5448 100644
--- a/server/src/engine/net/protocol/tests.rs
+++ b/server/src/engine/net/protocol/tests.rs
@@ -24,52 +24,25 @@
  *
 */
 
-use crate::engine::net::protocol::exchange::Resume;
-
 use {
     super::{
-        exchange::{self, QExchangeResult, QExchangeState},
+        exchange::{Exchange, ExchangeError, ExchangeResult, ExchangeState},
         handshake::ProtocolError,
-        SQuery,
     },
-    crate::{
-        engine::{
-            mem::BufferedScanner,
-            net::protocol::{
-                handshake::{
-                    AuthMode, CHandshake, CHandshakeAuth, CHandshakeStatic, DataExchangeMode,
-                    HandshakeResult, HandshakeState, HandshakeVersion, ProtocolVersion, QueryMode,
-                },
-                scan_int, AccumlatorStatus,
+    crate::engine::{
+        mem::BufferedScanner,
+        net::protocol::{
+            exchange::{PipeState, SQState, Usize},
+            handshake::{
+                AuthMode, CHandshake, CHandshakeAuth, CHandshakeStatic, DataExchangeMode,
+                HandshakeResult, HandshakeState, HandshakeVersion, ProtocolVersion, QueryMode,
             },
+            SQuery,
         },
-        util::test_utils,
     },
-    rand::Rng,
+    std::ops::Range,
 };
 
-pub(super) fn create_simple_query<const N: usize>(query: &str, params: [&str; N]) -> Vec<u8> {
-    let mut buf = vec![];
-    let query_size_as_string = query.len().to_string();
-    let total_packet_size = query.len()
-        + params.iter().map(|l| l.len()).sum::<usize>()
-        + query_size_as_string.len()
-        + 1;
-    // segment 1
-    buf.push(b'S');
-    buf.extend(total_packet_size.to_string().as_bytes());
-    buf.push(b'\n');
-    // segment
-    buf.extend(query_size_as_string.as_bytes());
-    buf.push(b'\n');
-    // dataframe
-    buf.extend(query.as_bytes());
-    params
-        .into_iter()
-        .for_each(|param| buf.extend(param.as_bytes()));
-    buf
-}
-
 /*
     client handshake
 */
@@ -101,7 +74,6 @@ fn parse_staged_with_auth() {
                     result,
                     HandshakeResult::ChangeState {
                         new_state: HandshakeState::Initial,
-                        expect: CHandshake::INITIAL_READ
                     }
                 );
             }
@@ -112,7 +84,6 @@ fn parse_staged_with_auth() {
                     result,
                     HandshakeResult::ChangeState {
                         new_state: HandshakeState::StaticBlock(STATIC_HANDSHAKE_WITH_AUTH),
-                        expect: 4
                     }
                 );
             }
@@ -125,7 +96,6 @@ fn parse_staged_with_auth() {
                             uname_l: 5,
                             pwd_l: 8
                         },
-                        expect: 13,
                     }
                 );
             }
@@ -151,15 +121,13 @@ fn run_state_changes_return_rounds(src: &[u8], expected_final_handshake: CHandsh
     let mut rounds = 0;
     let mut state = HandshakeState::default();
     let mut cursor = 0;
-    let mut expect_many = CHandshake::INITIAL_READ;
     loop {
         rounds += 1;
-        let buf = &src[..cursor + expect_many];
+        let buf = &src[..rounds];
         let mut scanner = unsafe { BufferedScanner::new_with_cursor(buf, cursor) };
         match CHandshake::resume_with(&mut scanner, state) {
-            HandshakeResult::ChangeState { new_state, expect } => {
+            HandshakeResult::ChangeState { new_state } => {
                 state = new_state;
-                expect_many = expect;
                 cursor = scanner.cursor();
             }
             HandshakeResult::Completed(hs) => {
@@ -181,7 +149,7 @@ fn parse_auth_with_state_updates() {
             CHandshakeAuth::new(b"sayan", b"pass1234"),
         ),
     );
-    assert_eq!(rounds, 3); // r1 = initial read, r2 = lengths, r3 = items
+    assert_eq!(rounds, FULL_HANDSHAKE_WITH_AUTH.len())
 }
 
 const HS_BAD_PACKET: [u8; 6] = *b"I\x00\0\0\0\0";
@@ -285,210 +253,597 @@ fn hs_bad_auth_mode() {
 }
 
 /*
-    QT-DEX/SQ
+    QT-DEX
 */
 
-const SQ: &str = "select * from myspace.mymodel where username = ?";
+fn iterate_payload(payload: impl AsRef<[u8]>, start: usize, f: impl Fn(usize, &[u8])) {
+    let payload = payload.as_ref();
+    for i in start..=payload.len() {
+        f(i, &payload[..i])
+    }
+}
 
-fn parse_staged<const N: usize>(
-    query: &str,
-    params: [&str; N],
-    eq: impl Fn(SQuery),
-    rng: &mut impl Rng,
+fn iterate_exchange_payload(
+    payload: impl AsRef<[u8]>,
+    start: usize,
+    f: impl Fn(usize, Result<(ExchangeResult, usize), ExchangeError>),
 ) {
-    let __query_buffer = create_simple_query(query, params);
-    for _ in 0..__query_buffer.len() {
-        let mut __read_total = 0;
-        let mut cursor = Default::default();
-        let mut state = QExchangeState::default();
-        loop {
-            let remaining = __query_buffer.len() - __read_total;
-            let read_this_time = {
-                let mut cnt = 0;
-                if remaining == 1 {
-                    1
+    iterate_payload(payload, start, |i, bytes| {
+        let scanner = BufferedScanner::new(bytes);
+        f(i, Exchange::try_complete(scanner, ExchangeState::default()))
+    })
+}
+
+fn iterate_exchange_payload_from_zero(
+    payload: impl AsRef<[u8]>,
+    f: impl Fn(usize, Result<(ExchangeResult, usize), ExchangeError>),
+) {
+    iterate_exchange_payload(payload, 0, f)
+}
+
+/*
+    corner cases
+*/
+
+#[test]
+fn zero_sized_packet() {
+    for payload in [
+        "S0\n",    // zero packet
+        "S2\n0\n", // zero query
+        "S1\n\n",  // zero query
+    ] {
+        iterate_exchange_payload_from_zero(payload, |size, result| {
+            if size == payload.len() {
+                // we got the full payload
+                if payload.len() == 3 {
+                    assert_eq!(result, Err(ExchangeError::UnterminatedInteger))
                 } else {
-                    let mut last = test_utils::random_number(1, remaining, rng);
-                    loop {
-                        if cnt >= 10 {
-                            break last;
-                        }
-                        // if we're reading exact, try to keep it low
-                        if last == remaining {
-                            cnt += 1;
-                            last = test_utils::random_number(1, remaining, rng);
-                        } else {
-                            break last;
-                        }
-                    }
+                    assert_eq!(result, Err(ExchangeError::IncorrectQuerySizeOrMoreBytes))
                 }
-            };
-            __read_total += read_this_time;
-            let buffered = &__query_buffer[..__read_total];
-            if !state.has_reached_target(buffered) {
-                continue;
-            }
-            match unsafe { exchange::resume(buffered, cursor, state) } {
-                (new_cursor, QExchangeResult::ChangeState(new_state)) => {
-                    cursor = new_cursor;
-                    state = new_state;
-                    continue;
-                }
-                (_, QExchangeResult::SQCompleted(q)) => {
-                    eq(q);
-                    break;
+            } else {
+                // we don't have the full payload
+                if size < 3 {
+                    assert_eq!(
+                        result,
+                        Ok((ExchangeResult::NewState(ExchangeState::Initial), 0))
+                    )
+                } else {
+                    assert!(
+                        matches!(
+                            result,
+                            Ok((ExchangeResult::NewState(ExchangeState::Simple(_)), _))
+                        ),
+                        "failed for {:?}, result is {:?}",
+                        &payload[..size],
+                        result,
+                    );
                 }
-                _ => panic!(),
             }
-        }
+        });
     }
 }
 
 #[test]
-fn staged_randomized() {
-    let mut rng = test_utils::rng();
-    parse_staged(
-        SQ,
-        ["sayan"],
-        |q| {
-            assert_eq!(q.query_str(), SQ);
-            assert_eq!(q.params_str(), "sayan");
-        },
-        &mut rng,
-    );
+fn invalid_first_byte() {
+    for payload in ["A1\n\n", "B7\n5\nsayan"] {
+        iterate_exchange_payload(payload, 1, |size, result| {
+            if size >= 3 {
+                assert_eq!(result, Err(ExchangeError::UnknownFirstByte))
+            } else {
+                assert_eq!(
+                    result,
+                    Ok((ExchangeResult::NewState(ExchangeState::Initial), 0))
+                )
+            }
+        })
+    }
+}
+
+pub struct EQuery {
+    // payload
+    payload: String,
+    variable_range: Range<usize>,
+    // query
+    query: String,
+    query_range: Range<usize>,
+    // params
+    params: &'static [&'static str],
+    param_range: Range<usize>,
+    param_indices: Vec<Range<usize>>,
+}
+
+impl EQuery {
+    fn new(query: String, params: &'static [&'static str]) -> Self {
+        var!(let variable_start, variable_end, query_start, query_end, param_start);
+        /*
+            prepare the "back" of the payload
+        */
+        let total_size = query.len() + params.iter().map(|p| p.len()).sum::<usize>();
+        let total_size_string = format!("{total_size}\n");
+
+        /*
+            compute offsets
+        */
+
+        let packet_size = total_size_string.len() + total_size;
+        let mut buffer = String::new();
+        buffer.push('S');
+        buffer.push_str(&format!("{packet_size}\n"));
+
+        // record start of variable block
+        variable_start = buffer.len();
+
+        buffer.push_str(&query.len().to_string());
+        buffer.push('\n');
+
+        // record start of query
+        query_start = buffer.len();
+        buffer.push_str(&query);
+        query_end = buffer.len();
+
+        // record start of params
+        param_start = buffer.len();
+        let mut param_indices = Vec::new();
+        for param in params {
+            let start = buffer.len();
+            buffer.push_str(param);
+            param_indices.push(start..buffer.len());
+        }
+
+        variable_end = buffer.len();
+        Self {
+            payload: buffer,
+            variable_range: variable_start..variable_end,
+            query,
+            query_range: query_start..query_end,
+            params,
+            param_range: param_start..variable_end,
+            param_indices,
+        }
+    }
 }
 
 #[test]
-fn stages_manual() {
-    let query = create_simple_query("select * from mymodel where username = ?", ["sayan"]);
-    assert_eq!(
-        unsafe {
-            exchange::resume(
-                &query[..QExchangeState::MIN_READ],
-                Default::default(),
-                Default::default(),
-            )
-        },
-        (
-            Resume::test_new(5),
-            QExchangeResult::ChangeState(QExchangeState::new_test(
-                exchange::QExchangeStateInternal::PendingMeta2,
-                52,
-                48,
-                4
-            ))
-        )
-    );
+fn ext_query() {
+    let ext_query = EQuery::new("create space myspace".to_owned(), &["sayan", "pass", ""]);
+    let query_starts_at = ext_query.payload[ext_query.variable_range.clone()]
+        .find('\n')
+        .unwrap()
+        + 1;
     assert_eq!(
-        unsafe {
-            exchange::resume(
-                &query[..QExchangeState::MIN_READ + 1],
-                Default::default(),
-                Default::default(),
-            )
-        },
-        (
-            Resume::test_new(6),
-            QExchangeResult::ChangeState(QExchangeState::new_test(
-                exchange::QExchangeStateInternal::PendingMeta2,
-                52,
-                48,
-                40
-            ))
-        )
+        &ext_query.payload[ext_query.variable_range.clone()]
+            [query_starts_at..query_starts_at + ext_query.query.len()],
+        ext_query.query
     );
+    assert_eq!(ext_query.query, &ext_query.payload[ext_query.query_range]);
+    assert_eq!("sayanpass", &ext_query.payload[ext_query.param_range]);
+    for (param_indices, real_param) in ext_query.param_indices.iter().zip(ext_query.params) {
+        assert_eq!(*real_param, &ext_query.payload[param_indices.clone()]);
+    }
+}
+
+/*
+    simple queries
+*/
+
+const fn dig_count(real: usize) -> usize {
+    // count the number of digits
+    let mut dig_count = 0;
+    let mut real_ = real;
+    while real_ != 0 {
+        dig_count += 1;
+        real_ /= 10;
+    }
+    // account for a `0`
+    dig_count += (real == 0) as usize;
+    dig_count
+}
+
+const fn nth_position_value(mut real: usize, mut pos: usize) -> usize {
+    let digits = dig_count(real);
+    while digits != pos {
+        real /= 10;
+        pos += 1;
+    }
+    real
+}
+
+#[test]
+fn simple_query() {
+    for query in [
+        // small query without params
+        EQuery::new("small query".to_owned(), &[]),
+        // small query with params
+        EQuery::new("small query".to_owned(), &["hello", "world"]),
+        // giant query without params
+        EQuery::new(
+            "abcdefghijklmnopqrstuvwxyz 123456789 ABCDEFGHIJKLMNOPQRSTUVWXYZ".repeat(1000),
+            &[],
+        ),
+        // giant query with params
+        EQuery::new(
+            "abcdefghijklmnopqrstuvwxyz 123456789 ABCDEFGHIJKLMNOPQRSTUVWXYZ".repeat(1000),
+            &["hello", "world"],
+        ),
+    ] {
+        iterate_exchange_payload_from_zero(query.payload.as_bytes(), |read_position, result| {
+            /*
+                S<packet size>\n<query window>\n<query><param>
+                                ^ variable      ^query ^param
+                                range start     start   start
+
+                - if before (variable range start - 1) then depending on the position from the first byte we will have, say the query size is 123
+                then we will have wrt distance from first byte (i.e position - 1) [1], [12], [123]
+                - if at (variable range start - 1) then we will have the exact size at [123] and in completed state
+                - if >= query start, then we will continue to issue changes of state until we have the full size which will be caught in a  different branch
+            */
+            if read_position < 3 {
+                // didn't reach minimum threshold
+                assert_eq!(
+                    result,
+                    Ok((ExchangeResult::NewState(ExchangeState::Initial), 0))
+                )
+            } else if read_position <= query.variable_range.start - 1 {
+                let index = read_position - 1;
+                assert_eq!(
+                    result,
+                    Ok((
+                        ExchangeResult::NewState(ExchangeState::Simple(SQState::_new(
+                            Usize::new_unflagged(nth_position_value(
+                                query.variable_range.len(),
+                                index
+                            ))
+                        ))),
+                        read_position
+                    ))
+                )
+            } else if read_position >= query.variable_range.start {
+                if read_position == query.payload.len() {
+                    let (result, cursor) = result.unwrap();
+                    assert_eq!(cursor, query.payload.len());
+                    assert_eq!(
+                        result,
+                        ExchangeResult::Simple(SQuery::_new(
+                            query.payload[query.query_range.start..].as_bytes(),
+                            query.query_range.len()
+                        ))
+                    );
+                } else {
+                    assert_eq!(
+                        result,
+                        Ok((
+                            ExchangeResult::NewState(ExchangeState::Simple(SQState::_new(
+                                Usize::new_flagged(query.variable_range.len())
+                            ))),
+                            query.variable_range.start // the cursor will not go ahead until the full query is read
+                        ))
+                    )
+                }
+            } else {
+                unreachable!()
+            }
+        })
+    }
+}
+
+/*
+    pipeline
+*/
+
+struct EPipeQuery {
+    payload: String,
+    q: String,
+    p: &'static [&'static str],
+    q_range: Range<usize>,
+    p_range: Range<usize>,
+}
+
+impl EPipeQuery {
+    fn new_string(q: &str, p: &'static [&'static str]) -> String {
+        Self::new(q.to_owned(), p).payload
+    }
+    fn new(q: String, p: &'static [&'static str]) -> Self {
+        let mut buffer = String::new();
+        buffer.push_str(&q.len().to_string());
+        buffer.push('\n');
+        buffer.push_str(&p.iter().map(|s| s.len()).sum::<usize>().to_string());
+        buffer.push('\n');
+        let q_start = buffer.len();
+        buffer.push_str(q.as_str());
+        let q_stop = buffer.len();
+        p.iter().for_each(|p_| buffer.push_str(p_));
+        let p_stop = buffer.len();
+        Self {
+            payload: buffer,
+            q,
+            p,
+            q_range: q_start..q_stop,
+            p_range: q_stop..p_stop,
+        }
+    }
+}
+
+fn pipe<const N: usize>(queries: [String; N]) -> String {
+    let packed_queries = queries.concat();
+    format!("P{}\n{packed_queries}", packed_queries.len())
+}
+
+fn get_pipeline_from_result(pipeline: ExchangeResult) -> Vec<SQuery> {
+    let pipeline: Vec<SQuery<'_>> = match pipeline {
+        ExchangeResult::Pipeline(p) => p.into_iter().map(Result::unwrap).collect(),
+        _ => panic!("expected pipeline got: {:?}", pipeline),
+    };
+    pipeline
+}
+
+#[test]
+fn full_pipe_scan() {
+    let pipeline_buffer = pipe([
+        EPipeQuery::new_string("create space myspace", &[]),
+        EPipeQuery::new_string(
+            "create model myspace.mymodel(username: string, password: string)",
+            &[],
+        ),
+        EPipeQuery::new_string("insert into myspace.mymodel(?, ?)", &["sayan", "cake"]),
+    ]);
+    let (pipeline, cursor) = Exchange::try_complete(
+        BufferedScanner::new(pipeline_buffer.as_bytes()),
+        ExchangeState::default(),
+    )
+    .unwrap();
+    assert_eq!(cursor, pipeline_buffer.len());
+    let pipeline = get_pipeline_from_result(pipeline);
     assert_eq!(
-        unsafe {
-            exchange::resume(
-                &query[..QExchangeState::MIN_READ + 2],
-                Default::default(),
-                Default::default(),
+        pipeline,
+        vec![
+            SQuery::_new(b"create space myspace", "create space myspace".len()),
+            SQuery::_new(
+                b"create model myspace.mymodel(username: string, password: string)",
+                "create model myspace.mymodel(username: string, password: string)".len()
+            ),
+            SQuery::_new(
+                b"insert into myspace.mymodel(?, ?)sayancake",
+                "insert into myspace.mymodel(?, ?)".len()
             )
-        },
-        (
-            Resume::test_new(7),
-            QExchangeResult::ChangeState(QExchangeState::new_test(
-                exchange::QExchangeStateInternal::PendingData,
-                52,
-                48,
-                40
-            ))
-        )
+        ]
     );
-    // the cursor should never change
-    for upper_bound in QExchangeState::MIN_READ + 2..query.len() {
-        assert_eq!(
-            unsafe {
-                exchange::resume(
-                    &query[..upper_bound],
-                    Default::default(),
-                    Default::default(),
+}
+
+struct EPipe {
+    payload: String,
+    variable_range: Range<usize>,
+    queries: Vec<EPipeQuery>,
+}
+
+impl EPipe {
+    fn new<const N: usize>(queries: [EPipeQuery; N]) -> Self {
+        let packed_queries_len = queries.iter().map(|epq| epq.payload.len()).sum::<usize>();
+        let mut buffer = format!("P{packed_queries_len}\n");
+        let variable_start = buffer.len();
+        for query in queries.iter() {
+            buffer.push_str(query.payload.as_str());
+        }
+        let variable_end = buffer.len();
+        Self {
+            payload: buffer,
+            variable_range: variable_start..variable_end,
+            queries: queries.into_iter().collect(),
+        }
+    }
+}
+
+#[test]
+fn pipeline() {
+    for pipe in [
+        EPipe::new([
+            // small query with no params
+            EPipeQuery::new("create space myspace".to_owned(), &[]),
+            // small query with params
+            EPipeQuery::new(
+                "insert into myspace.mymodel(?, ?)".to_owned(),
+                &["sayan", "elx"],
+            ),
+            // giant query with no params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &[],
+            ),
+            // giant query with params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &["hello", "world"],
+            ),
+        ]),
+        EPipe::new([
+            // giant query with no params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &[],
+            ),
+            // giant query with params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &["hello", "world"],
+            ),
+            // small query with no params
+            EPipeQuery::new("create space myspace".to_owned(), &[]),
+            // small query with params
+            EPipeQuery::new(
+                "insert into myspace.mymodel(?, ?)".to_owned(),
+                &["sayan", "elx"],
+            ),
+        ]),
+    ] {
+        iterate_exchange_payload_from_zero(pipe.payload.as_bytes(), |buffer_position, result| {
+            if buffer_position < 3 {
+                // we didn't read enough to pass the initial bounds check
+                assert_eq!(
+                    result,
+                    Ok((ExchangeResult::NewState(ExchangeState::Initial), 0))
                 )
-            },
-            (
-                Resume::test_new(7),
-                QExchangeResult::ChangeState(QExchangeState::new_test(
-                    exchange::QExchangeStateInternal::PendingData,
-                    52,
-                    48,
-                    40
-                ))
-            )
-        );
+            } else {
+                if buffer_position <= pipe.variable_range.start - 1 {
+                    // in this case we're before completion of the packet size. we succesively read [1], [12], [123]
+                    assert_eq!(
+                        result,
+                        Ok((
+                            ExchangeResult::NewState(ExchangeState::Pipeline(PipeState::_new(
+                                Usize::new_unflagged(nth_position_value(
+                                    pipe.variable_range.len(),
+                                    buffer_position - 1
+                                ))
+                            ))),
+                            buffer_position
+                        ))
+                    )
+                } else if buffer_position == pipe.variable_range.start {
+                    // in this case we passed the newline for the packet size so we have everything in order
+                    assert_eq!(
+                        result,
+                        Ok((
+                            ExchangeResult::NewState(ExchangeState::Pipeline(PipeState::_new(
+                                Usize::new_flagged(pipe.variable_range.len())
+                            ))),
+                            pipe.variable_range.start
+                        ))
+                    )
+                } else if buffer_position >= pipe.variable_range.start {
+                    // in this case we have more bytes than the variable range start
+                    if buffer_position == pipe.payload.len() {
+                        // if we're here, we've read the full packet
+                        let (status, cursor) = result.unwrap();
+                        assert_eq!(cursor, pipe.variable_range.end); // the whole chunk is obtained
+                        let pipeline = get_pipeline_from_result(status);
+                        pipeline
+                            .into_iter()
+                            .zip(pipe.queries.iter())
+                            .for_each(|(spq, epq)| {
+                                let decoded_query = core::str::from_utf8(spq.query()).unwrap();
+                                let decoded_params = core::str::from_utf8(spq.params()).unwrap();
+                                assert_eq!(decoded_query, epq.q);
+                                assert_eq!(decoded_params, epq.p.concat());
+                                assert_eq!(decoded_query, &epq.payload[epq.q_range.clone()]);
+                                assert_eq!(decoded_params, &epq.payload[epq.p_range.clone()]);
+                            })
+                    } else {
+                        // if we're here we haven't read the full packet because we're waiting for it to complete
+                        let (status, cursor) = result.unwrap();
+                        assert_eq!(cursor, pipe.variable_range.start); // the cursor won't move until
+                        assert_eq!(
+                            status,
+                            ExchangeResult::NewState(ExchangeState::Pipeline(PipeState::_new(
+                                Usize::new_flagged(pipe.variable_range.len())
+                            )))
+                        )
+                    }
+                } else {
+                    unreachable!()
+                }
+            }
+        })
     }
-    match unsafe { exchange::resume(&query, Default::default(), Default::default()) } {
-        (l, QExchangeResult::SQCompleted(q)) if l.inner() == query.len() => {
-            assert_eq!(q.query_str(), "select * from mymodel where username = ?");
-            assert_eq!(q.params_str(), "sayan");
+}
+
+fn run_staged(full_payload: &[u8], f: impl Fn(ExchangeResult)) {
+    let mut read_amount = 0;
+    let mut state = ExchangeState::default();
+    let mut cursor = 0;
+    loop {
+        let buffer = &full_payload[..read_amount];
+        let scanner = unsafe { BufferedScanner::new_with_cursor(buffer, cursor) };
+        match Exchange::try_complete(scanner, state).unwrap() {
+            (result, new_cursor) => match result {
+                ExchangeResult::NewState(new_state) => {
+                    state = new_state;
+                    cursor = new_cursor;
+                }
+                state => {
+                    f(state);
+                    break;
+                }
+            },
         }
-        e => panic!("expected end, got {e:?}"),
+        read_amount += 1;
     }
 }
 
 #[test]
-fn exchange_bad_segment_metadata() {
-    let exchange_packets = [
-        (
-            b"S4A\n32\ndelete from dbs where dbname = ?sillydb".as_slice(),
-            4,
-            "incorrect packet size",
-        ),
-        (
-            b"S42\n3A\ndelete from dbs where dbname = ?sillydb",
-            7,
-            "incorrect q window",
+fn staged_simple_query() {
+    for eq in [
+        EQuery::new(
+            "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+            &[],
         ),
-        (
-            b"S4A\n3A\ndelete from dbs where dbname = ?sillydb",
-            4,
-            "incorrect packet size and q window",
+        EQuery::new(
+            "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+            &["hello", "world"],
         ),
-    ];
-    for (packet, cursor, description) in exchange_packets {
-        assert_eq!(
-            unsafe { exchange::resume(packet, Resume::test_new(0), Default::default()) },
-            (Resume::test_new(cursor), QExchangeResult::Error),
-            "failed for `{description}`"
-        )
+    ] {
+        run_staged(eq.payload.as_bytes(), |state| match state {
+            ExchangeResult::NewState(_) => unreachable!(),
+            ExchangeResult::Simple(s) => {
+                assert_eq!(s.query(), eq.query.as_bytes());
+                assert_eq!(s.params(), eq.payload[eq.param_range.clone()].as_bytes());
+            }
+            ExchangeResult::Pipeline(_) => unreachable!(),
+        });
     }
 }
 
 #[test]
-fn num_accumulate() {
-    let x = [
-        (b"1".as_slice(), Ok(AccumlatorStatus::Pending(1)), 1usize),
-        (b"12", Ok(AccumlatorStatus::Pending(12)), 2),
-        (b"123", Ok(AccumlatorStatus::Pending(123)), 3),
-        (b"123\n", Ok(AccumlatorStatus::Completed(123)), 4),
-        (b"\n", Ok(AccumlatorStatus::Completed(0)), 1),
-        (b"A", Err(()), 1),
-        (b"A\n", Err(()), 2),
-        (b"1A", Err(()), 2),
-        (b"1A\n", Err(()), 3),
-    ];
-    for (buf, res, cursor) in x {
-        let mut bs = BufferedScanner::new(buf);
-        assert_eq!(scan_int(&mut bs, 0), res);
-        assert_eq!(bs.cursor(), cursor);
+fn staged_pipeline() {
+    for epipe in [
+        EPipe::new([
+            // small query with no params
+            EPipeQuery::new("create space myspace".to_owned(), &[]),
+            // small query with params
+            EPipeQuery::new(
+                "insert into myspace.mymodel(?, ?)".to_owned(),
+                &["sayan", "elx"],
+            ),
+            // giant query with no params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &[],
+            ),
+            // giant query with params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &["hello", "world"],
+            ),
+        ]),
+        EPipe::new([
+            // giant query with no params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &[],
+            ),
+            // giant query with params
+            EPipeQuery::new(
+                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789".repeat(1000),
+                &["hello", "world"],
+            ),
+            // small query with no params
+            EPipeQuery::new("create space myspace".to_owned(), &[]),
+            // small query with params
+            EPipeQuery::new(
+                "insert into myspace.mymodel(?, ?)".to_owned(),
+                &["sayan", "elx"],
+            ),
+        ]),
+    ] {
+        run_staged(epipe.payload.as_bytes(), |result| match result {
+            ExchangeResult::NewState(_) => unreachable!(),
+            ExchangeResult::Simple(_) => unreachable!(),
+            ExchangeResult::Pipeline(p) => {
+                p.into_iter()
+                    .zip(epipe.queries.iter())
+                    .for_each(|(spq, epq)| {
+                        let spq = spq.unwrap();
+                        assert_eq!(spq.query(), epq.q.as_bytes());
+                        assert_eq!(spq.params(), epq.payload[epq.p_range.clone()].as_bytes());
+                    })
+            }
+        })
     }
 }
diff --git a/server/src/engine/tests/client/mod.rs b/server/src/engine/tests/client/mod.rs
index 2ac2be22..b1ff7d7d 100644
--- a/server/src/engine/tests/client/mod.rs
+++ b/server/src/engine/tests/client/mod.rs
@@ -23,3 +23,59 @@
  * along with this program. If not, see <https://www.gnu.org/licenses/>.
  *
 */
+
+use skytable::{
+    query,
+    query::Pipeline,
+    response::{Response, Value},
+};
+
+const PIPE_RUNS: usize = 20;
+
+#[sky_macros::dbtest]
+fn pipe() {
+    let mut db = db!();
+    let mut pipe = Pipeline::new();
+    for _ in 0..PIPE_RUNS {
+        pipe.push(&query!("sysctl report status"));
+    }
+    assert_eq!(
+        db.execute_pipeline(&pipe).unwrap(),
+        vec![Response::Empty; PIPE_RUNS]
+    );
+}
+
+#[sky_macros::dbtest]
+fn pipe_params() {
+    let mut db = db!();
+    let pipe = Pipeline::new()
+        .add(&query!("create space pipe_params"))
+        .add(&query!(
+            "create model pipe_params.pipe_model(username: string, pipes_per_day: uint64)"
+        ))
+        .add(&query!(
+            "insert into pipe_params.pipe_model(?,?)",
+            "sayan",
+            0u64
+        ))
+        .add(&query!(
+            "select * from pipe_params.pipe_model where username = ?",
+            "sayan"
+        ))
+        .add(&query!("drop space allow not empty pipe_params"));
+    let result = db.execute_pipeline(&pipe).unwrap();
+    assert_eq!(
+        &result[..3],
+        vec![Response::Empty, Response::Empty, Response::Empty]
+    );
+    match &result[3] {
+        Response::Row(r) => {
+            assert_eq!(
+                r.values(),
+                [Value::String("sayan".into()), Value::UInt64(0)]
+            )
+        }
+        unknown => panic!("expected row, got {unknown:?}"),
+    }
+    assert_eq!(result[4], Response::Empty);
+}
diff --git a/server/src/engine/tests/client_misc/sec/mod.rs b/server/src/engine/tests/client_misc/sec/mod.rs
index ff3df078..9c57c5c9 100644
--- a/server/src/engine/tests/client_misc/sec/mod.rs
+++ b/server/src/engine/tests/client_misc/sec/mod.rs
@@ -40,6 +40,7 @@ use {
 const INVALID_SYNTAX_ERR: u16 = QueryError::QLInvalidSyntax.value_u8() as u16;
 const EXPECTED_STATEMENT_ERR: u16 = QueryError::QLExpectedStatement.value_u8() as u16;
 const UNKNOWN_STMT_ERR: u16 = QueryError::QLUnknownStatement.value_u8() as u16;
+const ILLEGAL_PACKET: u16 = QueryError::SysNetworkSystemIllegalClientPacket.value_u8() as u16;
 
 #[dbtest]
 fn deny_unknown_tokens() {
@@ -48,11 +49,17 @@ fn deny_unknown_tokens() {
         "model", "space", "where", "force", "into", "from", "with", "set", "add", "remove", "*",
         ",", "",
     ] {
-        assert_err_eq!(
-            db.query_parse::<()>(&query!(token)),
-            Error::ServerError(EXPECTED_STATEMENT_ERR),
-            "{token}",
-        );
+        let result = db.query_parse::<()>(&query!(token));
+        if token.is_empty() {
+            // the server will reject empty queries
+            assert_err_eq!(result, Error::ServerError(ILLEGAL_PACKET), "{token}")
+        } else {
+            assert_err_eq!(
+                result,
+                Error::ServerError(EXPECTED_STATEMENT_ERR),
+                "{token}",
+            );
+        }
     }
 }