Merge pull request #80 from c410-f3r/ci

Fix CI

.github/workflows/ci.yaml

@@ -14,7 +14,7 @@ jobs:
         with:
           override: true
           profile: minimal
-          toolchain: nightly-2024-01-10
+          toolchain: nightly-2023-10-15
       - uses: Swatinem/rust-cache@v2
       - run: .scripts/autobahn-fuzzingclient.sh ci
 
@@ -26,7 +26,7 @@ jobs:
         with:
           override: true
           profile: minimal
-          toolchain: nightly-2024-01-10
+          toolchain: nightly-2023-10-15
       - uses: Swatinem/rust-cache@v2
       - run: .scripts/autobahn-fuzzingserver.sh ci
 
@@ -38,7 +38,7 @@ jobs:
         with:
           override: true
           profile: minimal
-          toolchain: nightly-2024-01-10
+          toolchain: nightly-2023-10-15
       - uses: actions-rs/[email protected]
         with:
           crate: cargo-fuzz
@@ -53,7 +53,7 @@ jobs:
         with:
           override: true
           profile: minimal
-          toolchain: nightly-2024-01-10
+          toolchain: nightly-2023-10-15
       - uses: Swatinem/rust-cache@v2
       - run: docker-compose -f .test-utils/docker-compose.yml up -d
       - run: sleep 30
@@ -69,6 +69,6 @@ jobs:
           components: clippy, rustfmt
           override: true
           profile: minimal
-          toolchain: nightly-2024-01-10
+          toolchain: nightly-2023-10-15
       - uses: Swatinem/rust-cache@v2
       - run: .scripts/internal-tests.sh

rust-toolchain

@@ -1,4 +1,4 @@
 [toolchain]
-channel = "nightly-2024-01-10"
+channel = "nightly-2023-10-15"
 components = ["clippy", "rustfmt"]
 profile = "minimal"

wtx/Cargo.toml

@@ -93,7 +93,7 @@ simdutf8 = { default-features = false, features = ["aarch64_neon"], optional = t
 smallvec = { default-features = false, features = ["const_generics", "union"], optional = true, version = "1.0" }
 smol = { default-features = false, optional = true, version = "2.0" }
 test-strategy = { default-features = false, optional = true, version = "0.3" }
-tokio = { default-features = false, features = ["io-util", "net", "time"], optional = true, version = "1.35" }
+tokio = { default-features = false, features = ["io-util", "net", "sync", "time"], optional = true, version = "1.35" }
 tokio-rustls = { default-features = false, features = ["ring"], optional = true, version = "0.25" }
 tracing = { default-features = false, features = ["attributes"], optional = true, version = "0.1" }
 tracing-subscriber = { default-features = false, features = ["env-filter", "fmt"], optional = true, version = "0.3" }

wtx/src/database.rs

@@ -27,7 +27,7 @@ pub use from_records::FromRecords;
 pub use record::Record;
 pub use record_values::RecordValues;
 pub use records::Records;
-pub use stmt::Stmt;
+pub use stmt::StmtCmd;
 pub use transaction_manager::TransactionManager;
 pub use value_ident::ValueIdent;
 

wtx/src/database/client/postgres.rs

@@ -36,7 +36,7 @@ pub use records::Records;
 pub use sql_state::SqlState;
 pub use statements::Statements;
 pub use transaction_manager::TransactionManager;
-pub use ty::{CustomTy, Ty, TyKind};
+pub use ty::Ty;
 
 pub(crate) type Oid = u32;
 

wtx/src/database/client/postgres/executor.rs

@@ -12,7 +12,7 @@ use crate::{
       executor_buffer::{ExecutorBuffer, ExecutorBufferPartsMut},
       initial_conn_msg, Config, MessageTy, Postgres, Record, Records, TransactionManager,
     },
-    Database, RecordValues, Stmt, TransactionManager as _,
+    Database, RecordValues, StmtCmd, TransactionManager as _,
   },
   misc::{FilledBufferWriter, Stream, TlsStream},
   rng::Rng,
@@ -128,32 +128,29 @@ where
   }
 
   #[inline]
-  async fn execute_with_stmt<STMT, RV>(
+  async fn execute_with_stmt<SC, RV>(
     &mut self,
-    stmt: STMT,
+    sc: SC,
     rv: RV,
   ) -> Result<u64, <Self::Database as Database>::Error>
   where
     RV: RecordValues<Self::Database>,
-    STMT: Stmt,
+    SC: StmtCmd,
   {
-    let ExecutorBufferPartsMut { ftb, nb, rb, stmts, tb, vb, .. } =
-      self.eb.borrow_mut().parts_mut();
+    let ExecutorBufferPartsMut { nb, rb, stmts, vb, .. } = self.eb.borrow_mut().parts_mut();
     ExecutorBuffer::clear_cmd_buffers(nb, rb, vb);
     let mut rows = 0;
     let mut fwsc = FetchWithStmtCommons {
-      ftb,
       is_closed: &mut self.is_closed,
       rb,
       stream: &mut self.stream,
-      tb,
       tys: &[],
     };
     let (_, stmt_id_str, stmt) =
-      Self::write_send_await_stmt_prot(&mut fwsc, nb, stmt, stmts, vb).await?;
+      Self::write_send_await_stmt_prot(&mut fwsc, nb, sc, stmts, vb).await?;
     Self::write_send_await_stmt_initial(&mut fwsc, nb, rv, &stmt, &stmt_id_str).await?;
     loop {
-      let msg = Self::fetch_msg_from_stream(&mut fwsc.is_closed, nb, fwsc.stream).await?;
+      let msg = Self::fetch_msg_from_stream(fwsc.is_closed, nb, fwsc.stream).await?;
       match msg.ty {
         MessageTy::CommandComplete(local_rows) => {
           rows = local_rows;
@@ -167,59 +164,53 @@ where
   }
 
   #[inline]
-  async fn fetch_with_stmt<STMT, RV>(
+  async fn fetch_with_stmt<SC, RV>(
     &mut self,
-    stmt: STMT,
+    sc: SC,
     rv: RV,
   ) -> Result<<Self::Database as Database>::Record<'_>, E>
   where
     RV: RecordValues<Self::Database>,
-    STMT: Stmt,
+    SC: StmtCmd,
   {
-    let ExecutorBufferPartsMut { ftb, nb, rb, stmts, tb, vb, .. } =
-      self.eb.borrow_mut().parts_mut();
+    let ExecutorBufferPartsMut { nb, rb, stmts, vb, .. } = self.eb.borrow_mut().parts_mut();
     Self::write_send_await_fetch_with_stmt(
       &mut FetchWithStmtCommons {
-        ftb,
         is_closed: &mut self.is_closed,
         rb,
         stream: &mut self.stream,
-        tb,
         tys: &[],
       },
       nb,
       rv,
-      stmt,
+      sc,
       stmts,
       vb,
     )
     .await
   }
 
   #[inline]
-  async fn fetch_many_with_stmt<'this, STMT, RV>(
+  async fn fetch_many_with_stmt<'this, SC, RV>(
     &'this mut self,
-    stmt: STMT,
+    sc: SC,
     rv: RV,
     mut cb: impl FnMut(&<Self::Database as Database>::Record<'_>) -> Result<(), E> + 'this,
   ) -> Result<<Self::Database as Database>::Records<'_>, E>
   where
     RV: RecordValues<Self::Database>,
-    STMT: Stmt,
+    SC: StmtCmd,
   {
-    let ExecutorBufferPartsMut { ftb, nb, rb, stmts, tb, vb, .. } =
-      self.eb.borrow_mut().parts_mut();
+    let ExecutorBufferPartsMut { nb, rb, stmts, vb, .. } = self.eb.borrow_mut().parts_mut();
     ExecutorBuffer::clear_cmd_buffers(nb, rb, vb);
     let mut fwsc = FetchWithStmtCommons {
-      ftb,
       is_closed: &mut self.is_closed,
       rb,
       stream: &mut self.stream,
-      tb,
       tys: &[],
     };
     let (_, stmt_id_str, stmt) =
-      Self::write_send_await_stmt_prot(&mut fwsc, nb, stmt, stmts, vb).await?;
+      Self::write_send_await_stmt_prot(&mut fwsc, nb, sc, stmts, vb).await?;
     Self::write_send_await_stmt_initial(&mut fwsc, nb, rv, &stmt, &stmt_id_str).await?;
     let begin = nb._current_end_idx();
     let begin_data = nb._current_end_idx().wrapping_add(7);
@@ -261,17 +252,14 @@ where
 
   #[inline]
   async fn prepare(&mut self, cmd: &str) -> Result<u64, E> {
-    let ExecutorBufferPartsMut { ftb, nb, rb, stmts, tb, vb, .. } =
-      self.eb.borrow_mut().parts_mut();
+    let ExecutorBufferPartsMut { nb, rb, stmts, vb, .. } = self.eb.borrow_mut().parts_mut();
     ExecutorBuffer::clear_cmd_buffers(nb, rb, vb);
     Ok(
       Self::write_send_await_stmt_prot(
         &mut FetchWithStmtCommons {
-          ftb,
           is_closed: &mut self.is_closed,
           rb,
           stream: &mut self.stream,
-          tb,
           tys: &[],
         },
         nb,

wtx/src/database/client/postgres/executor/commons.rs

@@ -1,12 +1,9 @@
 use crate::database::client::postgres::Ty;
 use alloc::vec::Vec;
-use hashbrown::HashMap;
 
 pub(crate) struct FetchWithStmtCommons<'others, S> {
-  pub(crate) ftb: &'others mut Vec<(usize, u32)>,
   pub(crate) is_closed: &'others mut bool,
   pub(crate) rb: &'others mut Vec<usize>,
   pub(crate) stream: &'others mut S,
-  pub(crate) tb: &'others mut HashMap<u32, Ty>,
   pub(crate) tys: &'others [Ty],
 }

wtx/src/database/client/postgres/executor/fetch.rs

@@ -4,7 +4,7 @@ use crate::{
       executor::commons::FetchWithStmtCommons, message::Message, statements::Statement, Executor,
       ExecutorBuffer, MessageTy, Postgres, Record, Statements,
     },
-    RecordValues, Stmt,
+    RecordValues, StmtCmd,
   },
   misc::{PartitionedFilledBuffer, Stream, _read_until},
 };
@@ -16,21 +16,20 @@ where
   EB: BorrowMut<ExecutorBuffer>,
   S: Stream,
 {
-  pub(crate) async fn write_send_await_fetch_with_stmt<'rec, STMT, RV>(
+  pub(crate) async fn write_send_await_fetch_with_stmt<'rec, SC, RV>(
     fwsc: &mut FetchWithStmtCommons<'_, S>,
     nb: &'rec mut PartitionedFilledBuffer,
     rv: RV,
-    stmt: STMT,
+    sc: SC,
     stmts: &'rec mut Statements,
     vb: &'rec mut Vec<(bool, Range<usize>)>,
   ) -> Result<Record<'rec, E>, E>
   where
     E: From<crate::Error>,
     RV: RecordValues<Postgres<E>>,
-    STMT: Stmt,
+    SC: StmtCmd,
   {
-    let (_, stmt_id_str, stmt) =
-      Self::write_send_await_stmt_prot(fwsc, nb, stmt, stmts, vb).await?;
+    let (_, stmt_id_str, stmt) = Self::write_send_await_stmt_prot(fwsc, nb, sc, stmts, vb).await?;
     Self::write_send_await_fetch_with_stmt_wo_prot(fwsc, nb, rv, stmt, &stmt_id_str, vb).await
   }
 
@@ -46,7 +45,7 @@ where
     E: From<crate::Error>,
     RV: RecordValues<Postgres<E>>,
   {
-    Self::write_send_await_stmt_initial(fwsc, nb, rv, &stmt, &stmt_id_str).await?;
+    Self::write_send_await_stmt_initial(fwsc, nb, rv, &stmt, stmt_id_str).await?;
     let mut data_row_msg_range = None;
     loop {
       let msg = Self::fetch_msg_from_stream(fwsc.is_closed, nb, fwsc.stream).await?;