diff --git a/Cargo.lock b/Cargo.lock
index 47f9ebf1d0..07a8dbd526 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1252,7 +1252,6 @@ dependencies = [
 name = "database-utils"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "clap 4.3.2",
  "deadpool-postgres",
  "derive_more",
@@ -1585,14 +1584,14 @@ dependencies = [
 
 [[package]]
 name = "enum_dispatch"
-version = "0.3.8"
+version = "0.3.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0eb359f1476bf611266ac1f5355bc14aeca37b299d0ebccc038ee7058891c9cb"
+checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd"
 dependencies = [
  "once_cell",
  "proc-macro2 1.0.69",
  "quote 1.0.33",
- "syn 1.0.109",
+ "syn 2.0.39",
 ]
 
 [[package]]
@@ -2923,7 +2922,6 @@ dependencies = [
 name = "mysql-srv"
 version = "0.8.8"
 dependencies = [
- "async-trait",
  "byteorder",
  "chrono",
  "futures",
@@ -4005,7 +4003,6 @@ dependencies = [
 name = "psql-srv"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "base64 0.21.0",
  "bit-vec",
  "bytes",
@@ -4276,7 +4273,6 @@ name = "readyset"
 version = "1.3.0"
 dependencies = [
  "anyhow",
- "async-trait",
  "chrono",
  "clap 4.3.2",
  "crossbeam-skiplist",
diff --git a/Cargo.toml b/Cargo.toml
index 4be2086470..4e98e58331 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -113,7 +113,7 @@ derive_more = "0.99.17"
 diff = "0.1.10"
 enum-display-derive = "0.1.1"
 enum-kinds = "0.5.1"
-enum_dispatch = "0.3.7"
+enum_dispatch = "0.3.13"
 envy = "0.4"
 fail = "0.5.0"
 fallible-iterator = "0.2.0"
diff --git a/benchmarks/src/benchmark.rs b/benchmarks/src/benchmark.rs
index 10d5d155bf..7d04266aa6 100644
--- a/benchmarks/src/benchmark.rs
+++ b/benchmarks/src/benchmark.rs
@@ -14,7 +14,6 @@ use std::collections::HashMap;
 use std::str::FromStr;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseType, DatabaseURL};
 use enum_dispatch::enum_dispatch;
@@ -265,7 +264,8 @@ impl BenchmarkOutput {
 
 /// The set of control functions needed to execute the benchmark in
 /// the `BenchmarkRunner`.
-#[async_trait]
+// Only used internally
+#[allow(async_fn_in_trait)]
 #[enum_dispatch]
 pub trait BenchmarkControl {
     /// Any code required to perform setup of the benchmark goes here. This
diff --git a/benchmarks/src/cache_hit_benchmark.rs b/benchmarks/src/cache_hit_benchmark.rs
index a12b81f3ae..0400c4e066 100644
--- a/benchmarks/src/cache_hit_benchmark.rs
+++ b/benchmarks/src/cache_hit_benchmark.rs
@@ -4,7 +4,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -36,7 +35,6 @@ pub struct CacheHitBenchmark {
     num_cache_misses: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for CacheHitBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
diff --git a/benchmarks/src/eviction_benchmark.rs b/benchmarks/src/eviction_benchmark.rs
index 4a69fdbe43..a69d72d6d5 100644
--- a/benchmarks/src/eviction_benchmark.rs
+++ b/benchmarks/src/eviction_benchmark.rs
@@ -17,7 +17,6 @@ use std::sync::atomic::{AtomicBool, AtomicU64};
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -83,7 +82,6 @@ pub struct EvictionBenchmarkParams {
     target_hit_rate: f64,
 }
 
-#[async_trait]
 impl BenchmarkControl for EvictionBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -335,7 +333,6 @@ async fn metrics_task(
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for EvictionBenchmark {
     type BenchmarkResult = EvictionBenchmarkResultBatch;
     type Parameters = EvictionBenchmarkParams;
diff --git a/benchmarks/src/lib.rs b/benchmarks/src/lib.rs
index ebc5a8d419..be956c5e53 100644
--- a/benchmarks/src/lib.rs
+++ b/benchmarks/src/lib.rs
@@ -23,7 +23,6 @@
 //!     row_count: u32,
 //! }
 //!
-//! #[async_trait::async_trait]
 //! impl BenchmarkControl for MyBenchmark {
 //!     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
 //!         let mut conn = deployment.connect_to_setup().await?;
diff --git a/benchmarks/src/migration_benchmark.rs b/benchmarks/src/migration_benchmark.rs
index c2cdd6c13d..cb94bfcb2f 100644
--- a/benchmarks/src/migration_benchmark.rs
+++ b/benchmarks/src/migration_benchmark.rs
@@ -4,7 +4,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::DatabaseURL;
 use metrics::Unit;
@@ -33,7 +32,6 @@ pub struct MigrationBenchmark {
     num_migrations: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for MigrationBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
diff --git a/benchmarks/src/query_benchmark.rs b/benchmarks/src/query_benchmark.rs
index 8c9adfc441..68e5603895 100644
--- a/benchmarks/src/query_benchmark.rs
+++ b/benchmarks/src/query_benchmark.rs
@@ -8,7 +8,6 @@ use std::str::FromStr;
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -60,7 +59,6 @@ pub struct QueryBenchmarkThreadParams {
     query: ArbitraryQueryParameters,
 }
 
-#[async_trait]
 impl BenchmarkControl for QueryBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -155,7 +153,6 @@ impl QueryBenchmarkResultBatch {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for QueryBenchmark {
     type BenchmarkResult = QueryBenchmarkResultBatch;
     type Parameters = QueryBenchmarkThreadParams;
diff --git a/benchmarks/src/read_write_benchmark.rs b/benchmarks/src/read_write_benchmark.rs
index a1e10ae3e4..2fa846962d 100644
--- a/benchmarks/src/read_write_benchmark.rs
+++ b/benchmarks/src/read_write_benchmark.rs
@@ -5,7 +5,6 @@ use std::str::FromStr;
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -84,7 +83,6 @@ pub struct ReadWriteBenchmarkThreadParams {
     upstream_conn_str: String,
 }
 
-#[async_trait]
 impl BenchmarkControl for ReadWriteBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
@@ -190,7 +188,6 @@ impl ReadWriteBenchmarkResultBatch {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for ReadWriteBenchmark {
     type BenchmarkResult = ReadWriteBenchmarkResultBatch;
     type Parameters = ReadWriteBenchmarkThreadParams;
diff --git a/benchmarks/src/scale_connections.rs b/benchmarks/src/scale_connections.rs
index bc8e256996..0e567cf0b7 100644
--- a/benchmarks/src/scale_connections.rs
+++ b/benchmarks/src/scale_connections.rs
@@ -3,7 +3,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::DatabaseURL;
 use metrics::Unit;
@@ -27,7 +26,6 @@ pub struct ScaleConnections {
     parallel: bool,
 }
 
-#[async_trait]
 impl BenchmarkControl for ScaleConnections {
     async fn setup(&self, _: &DeploymentParameters) -> Result<()> {
         Ok(())
diff --git a/benchmarks/src/scale_views.rs b/benchmarks/src/scale_views.rs
index 0a86c432e4..5e6a49953b 100644
--- a/benchmarks/src/scale_views.rs
+++ b/benchmarks/src/scale_views.rs
@@ -11,7 +11,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::{bail, Result};
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseURL, QueryableConnection};
 use itertools::Itertools;
@@ -53,7 +52,6 @@ fn get_columns(num_views: usize, param_count: usize) -> Vec<String> {
     (0..num_columns).map(|i| format!("c{}", i)).collect()
 }
 
-#[async_trait]
 impl BenchmarkControl for ScaleViews {
     /// Creates a table with enough columns that we can create `num_views` off a
     /// combination of the columns.
diff --git a/benchmarks/src/single_query_benchmark.rs b/benchmarks/src/single_query_benchmark.rs
index 9acfa1ce24..e5405fb315 100644
--- a/benchmarks/src/single_query_benchmark.rs
+++ b/benchmarks/src/single_query_benchmark.rs
@@ -6,7 +6,6 @@ use std::str::FromStr;
 use std::time::Instant;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::{DatabaseConnection, DatabaseURL, QueryableConnection};
 use metrics::Unit;
@@ -37,7 +36,6 @@ pub struct SingleQueryBenchmark {
     ad_hoc: bool,
 }
 
-#[async_trait]
 impl BenchmarkControl for SingleQueryBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
diff --git a/benchmarks/src/template.rs b/benchmarks/src/template.rs
index d6695213ec..e2edce8faf 100644
--- a/benchmarks/src/template.rs
+++ b/benchmarks/src/template.rs
@@ -1,7 +1,6 @@
 use std::collections::HashMap;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::Parser;
 use metrics::Unit;
 use serde::{Deserialize, Serialize};
@@ -16,7 +15,6 @@ pub struct Template {
     // Benchmark specific parameters go here.
 }
 
-#[async_trait]
 impl BenchmarkControl for Template {
     async fn setup(&self, _: &DeploymentParameters) -> Result<()> {
         // Any code required to setup the benchmark goes. This may include
diff --git a/benchmarks/src/utils/multi_thread.rs b/benchmarks/src/utils/multi_thread.rs
index 78a790f49e..7c35aa2d38 100644
--- a/benchmarks/src/utils/multi_thread.rs
+++ b/benchmarks/src/utils/multi_thread.rs
@@ -1,7 +1,7 @@
+use std::future::Future;
 use std::time::Duration;
 
 use anyhow::Result;
-use async_trait::async_trait;
 use futures::stream::futures_unordered::FuturesUnordered;
 use futures::StreamExt;
 use tokio::select;
@@ -14,7 +14,6 @@ use crate::benchmark::BenchmarkResults;
 /// A group of methods that facilitate executing a single benchmark from multiple
 /// threads. This should be used in conjunction with `run_multithread_benchmark`
 /// to spawn the threads to run the benchmark.
-#[async_trait]
 pub(crate) trait MultithreadBenchmark {
     /// The result messages passed to the result's thread via an UnboundedSender.
     type BenchmarkResult: Send;
@@ -22,18 +21,18 @@ pub(crate) trait MultithreadBenchmark {
     type Parameters: Clone;
     /// Process a batch of benchmark results collected over `interval`. This aggregates
     /// all updates send on the `sender` parameter fo `benchmark_thread`.
-    async fn handle_benchmark_results(
+    fn handle_benchmark_results(
         results: Vec<Self::BenchmarkResult>,
         interval: Duration,
         results: &mut BenchmarkResults,
-    ) -> Result<()>;
+    ) -> impl Future<Output = Result<()>> + Send;
 
     /// Benchmarking code that is initialized using `params` that sends `BenchmarkResult`
     /// to be batched along `sender`.
-    async fn benchmark_thread(
+    fn benchmark_thread(
         params: Self::Parameters,
         sender: UnboundedSender<Self::BenchmarkResult>,
-    ) -> Result<()>;
+    ) -> impl Future<Output = Result<()>> + Send;
 }
 
 /// Returns after `duration` if it is Some, otherwise, never returns. Useful
diff --git a/benchmarks/src/workload_emulator.rs b/benchmarks/src/workload_emulator.rs
index 38958ac114..1f8e9836ac 100644
--- a/benchmarks/src/workload_emulator.rs
+++ b/benchmarks/src/workload_emulator.rs
@@ -15,7 +15,6 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
 use database_utils::{DatabaseConnection, DatabaseStatement, QueryableConnection};
 use metrics::Unit;
@@ -178,7 +177,6 @@ impl WorkloadResultBatch {
     }
 }
 
-#[async_trait]
 impl BenchmarkControl for WorkloadEmulator {
     async fn setup(&self, deployment: &DeploymentParameters) -> anyhow::Result<()> {
         if let Some(ref data_generator) = self.data_generator {
@@ -472,7 +470,6 @@ fn ttl_jitter(ttl_secs: u32) -> u32 {
     ttl_secs + (ttl_secs as f32 * variance) as u32
 }
 
-#[async_trait]
 impl MultithreadBenchmark for WorkloadEmulator {
     type BenchmarkResult = WorkloadResultBatch;
     type Parameters = WorkloadThreadParams;
diff --git a/benchmarks/src/write_benchmark.rs b/benchmarks/src/write_benchmark.rs
index 4454a5b33f..dedfdfa3b8 100644
--- a/benchmarks/src/write_benchmark.rs
+++ b/benchmarks/src/write_benchmark.rs
@@ -6,7 +6,6 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use anyhow::{bail, Result};
-use async_trait::async_trait;
 use clap::{Parser, ValueHint};
 use database_utils::{DatabaseConnection, DatabaseError, DatabaseURL, QueryableConnection};
 use itertools::Itertools;
@@ -167,7 +166,6 @@ async fn create_indices(
     Ok(())
 }
 
-#[async_trait]
 impl BenchmarkControl for WriteBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         let mut conn = DatabaseURL::from_str(&deployment.target_conn_str)?
@@ -224,7 +222,6 @@ impl BenchmarkControl for WriteBenchmark {
     }
 }
 
-#[async_trait]
 impl MultithreadBenchmark for WriteBenchmark {
     type BenchmarkResult = u128;
     type Parameters = WriteBenchmarkThreadData;
diff --git a/benchmarks/src/write_latency_benchmark.rs b/benchmarks/src/write_latency_benchmark.rs
index 5aec2b1157..b789a9bc5b 100644
--- a/benchmarks/src/write_latency_benchmark.rs
+++ b/benchmarks/src/write_latency_benchmark.rs
@@ -3,7 +3,6 @@ use std::convert::TryFrom;
 use std::time::Instant;
 
 use anyhow::{anyhow, bail, Result};
-use async_trait::async_trait;
 use clap::Parser;
 use database_utils::QueryableConnection;
 use metrics::Unit;
@@ -34,7 +33,6 @@ pub struct WriteLatencyBenchmark {
     updates: u32,
 }
 
-#[async_trait]
 impl BenchmarkControl for WriteLatencyBenchmark {
     async fn setup(&self, deployment: &DeploymentParameters) -> Result<()> {
         self.data_generator
diff --git a/database-utils/Cargo.toml b/database-utils/Cargo.toml
index faad81a580..2500fbe627 100644
--- a/database-utils/Cargo.toml
+++ b/database-utils/Cargo.toml
@@ -18,7 +18,6 @@ clap = { workspace = true, features = ["derive","env"] }
 serde = { workspace = true, features = ["derive"] }
 readyset-util = { path = "../readyset-util" }
 readyset-errors = { path = "../readyset-errors" }
-async-trait = { workspace = true }
 nom-sql = { path = "../nom-sql" }
 deadpool-postgres = { workspace = true }
 num_cpus = { workspace = true }
diff --git a/database-utils/src/connection.rs b/database-utils/src/connection.rs
index 160c908367..9bc3f89699 100644
--- a/database-utils/src/connection.rs
+++ b/database-utils/src/connection.rs
@@ -3,7 +3,6 @@ use std::hash::{Hash, Hasher};
 use std::marker::{Send, Sync};
 use std::str;
 
-use async_trait::async_trait;
 use derive_more::From;
 use futures::TryStreamExt;
 use mysql::prelude::AsQuery;
@@ -15,7 +14,8 @@ use {mysql_async as mysql, tokio_postgres as pgsql};
 
 use crate::error::{ConnectionType, DatabaseError};
 
-#[async_trait]
+// Only used internally
+#[allow(async_fn_in_trait)]
 pub trait QueryableConnection: Send {
     /// Executes query_drop for either mysql or postgres, whichever is the underlying
     /// connection variant.
@@ -121,7 +121,6 @@ pub enum DatabaseConnection {
     PostgreSQLPool(deadpool_postgres::Client),
 }
 
-#[async_trait]
 impl QueryableConnection for DatabaseConnection {
     async fn query_drop<Q>(&mut self, stmt: Q) -> Result<(), DatabaseError>
     where
@@ -316,7 +315,6 @@ impl DatabaseConnectionPool {
     }
 }
 
-#[async_trait]
 impl QueryableConnection for DatabaseConnectionPool {
     async fn query_drop<Q>(&mut self, stmt: Q) -> Result<(), DatabaseError>
     where
@@ -670,7 +668,6 @@ pub enum Transaction<'a> {
     PostgresPool(deadpool_postgres::Transaction<'a>),
 }
 
-#[async_trait]
 impl<'a> QueryableConnection for Transaction<'a> {
     async fn query_drop<Q>(&mut self, stmt: Q) -> Result<(), DatabaseError>
     where
diff --git a/mysql-srv/Cargo.toml b/mysql-srv/Cargo.toml
index 41cba83cc1..340e1f0bc2 100644
--- a/mysql-srv/Cargo.toml
+++ b/mysql-srv/Cargo.toml
@@ -19,7 +19,6 @@ byteorder = { workspace = true }
 chrono = { workspace = true }
 time = { workspace = true }
 getrandom = { workspace = true }
-async-trait = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 thiserror = { workspace = true }
 sha-1 = { workspace = true }
diff --git a/mysql-srv/src/lib.rs b/mysql-srv/src/lib.rs
index e449c6cafa..8c84ede2bc 100644
--- a/mysql-srv/src/lib.rs
+++ b/mysql-srv/src/lib.rs
@@ -22,14 +22,12 @@
 //! use std::collections::HashMap;
 //! use std::iter;
 //!
-//! use async_trait::async_trait;
 //! use mysql::prelude::*;
 //! use mysql_srv::*;
 //! use readyset_adapter_types::DeallocateId;
 //! use tokio::io::AsyncWrite;
 //!
 //! struct Backend;
-//! #[async_trait]
 //! impl<W: AsyncWrite + Unpin + Send + 'static> MySqlShim<W> for Backend {
 //!     async fn on_prepare(
 //!         &mut self,
@@ -171,7 +169,6 @@ use std::collections::HashMap;
 use std::io;
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use constants::{CLIENT_PLUGIN_AUTH, CONNECT_WITH_DB, PROTOCOL_41, RESERVED, SECURE_CONNECTION};
 use error::{other_error, OtherErrorKind};
 use mysql_common::constants::CapabilityFlags;
@@ -250,7 +247,8 @@ pub enum QueryResultsResponse {
 }
 
 /// Implementors of this trait can be used to drive a MySQL-compatible database backend.
-#[async_trait]
+// Only used internally
+#[allow(async_fn_in_trait)]
 pub trait MySqlShim<W: AsyncWrite + Unpin + Send> {
     /// Called when the client issues a request to prepare `query` for later execution.
     ///
diff --git a/mysql-srv/tests/main.rs b/mysql-srv/tests/main.rs
index a77ebb6862..b873b98bb9 100644
--- a/mysql-srv/tests/main.rs
+++ b/mysql-srv/tests/main.rs
@@ -14,7 +14,6 @@ use std::marker::PhantomData;
 use std::pin::Pin;
 use std::{io, net, thread};
 
-use async_trait::async_trait;
 use mysql::prelude::Queryable;
 use mysql::Row;
 use mysql_srv::{
@@ -38,7 +37,6 @@ struct TestingShim<Q, P, E, I, CU, W> {
     _phantom: PhantomData<W>,
 }
 
-#[async_trait]
 impl<Q, P, E, I, CU, W> MySqlShim<W> for TestingShim<Q, P, E, I, CU, W>
 where
     Q: for<'a> FnMut(
@@ -118,7 +116,7 @@ where
     ) -> QueryResultsResponse {
         if query.starts_with("SELECT @@") || query.starts_with("select @@") {
             let var = &query.get(b"SELECT @@".len()..);
-            return match var {
+            match var {
                 Some("max_allowed_packet") => {
                     let cols = &[Column {
                         table: String::new(),
@@ -133,7 +131,7 @@ where
                     QueryResultsResponse::IoResult(w.finish().await)
                 }
                 _ => QueryResultsResponse::IoResult(results.completed(0, 0, None).await),
-            };
+            }
         } else {
             QueryResultsResponse::IoResult((self.on_q)(query, results).await)
         }
diff --git a/psql-srv/Cargo.toml b/psql-srv/Cargo.toml
index 93fee6ffbf..ca46a9e213 100644
--- a/psql-srv/Cargo.toml
+++ b/psql-srv/Cargo.toml
@@ -9,7 +9,6 @@ description = "Bindings for emulating a PostgreSQL server"
 readme = "README.md"
 
 [dependencies]
-async-trait = { workspace = true }
 base64 = "0.21"
 bit-vec = { workspace = true, features = ["serde"] }
 bytes = { workspace = true }
diff --git a/psql-srv/src/lib.rs b/psql-srv/src/lib.rs
index f982b190ac..325435e47e 100644
--- a/psql-srv/src/lib.rs
+++ b/psql-srv/src/lib.rs
@@ -29,7 +29,6 @@ mod value;
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use futures::Stream;
 use nom_sql::SqlIdentifier;
 use postgres::SimpleQueryMessage;
@@ -62,7 +61,8 @@ pub enum Credentials<'a> {
 
 /// A trait for implementing a SQL backend that produces responses to SQL query statements. This
 /// trait is the primary interface for the `psql-srv` crate.
-#[async_trait]
+// Only used internally
+#[allow(async_fn_in_trait)]
 pub trait PsqlBackend {
     /// An associated type representing a resultset returned by a SQL query, which can be iterated
     /// to produce `Self::Row`s.
diff --git a/psql-srv/src/protocol.rs b/psql-srv/src/protocol.rs
index 020589cd22..900988ef33 100644
--- a/psql-srv/src/protocol.rs
+++ b/psql-srv/src/protocol.rs
@@ -1076,7 +1076,6 @@ mod tests {
     use std::task::Poll;
     use std::{io, vec};
 
-    use async_trait::async_trait;
     use bytes::BytesMut;
     use futures::task::Context;
     use futures::{stream, TryStreamExt};
@@ -1134,7 +1133,6 @@ mod tests {
         }
     }
 
-    #[async_trait]
     impl PsqlBackend for Backend {
         type Resultset = stream::Iter<vec::IntoIter<Result<PsqlSrvRow, Error>>>;
 
diff --git a/psql-srv/tests/authentication.rs b/psql-srv/tests/authentication.rs
index b74afcd8f2..4652a6ba8b 100644
--- a/psql-srv/tests/authentication.rs
+++ b/psql-srv/tests/authentication.rs
@@ -2,7 +2,6 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::vec;
 
-use async_trait::async_trait;
 use futures::stream;
 use postgres::config::{ChannelBinding, SslMode};
 use postgres::error::SqlState;
@@ -34,7 +33,6 @@ struct ScramSha256Backend {
     password: &'static str,
 }
 
-#[async_trait]
 impl PsqlBackend for ScramSha256Backend {
     type Resultset = stream::Iter<vec::IntoIter<Result<PsqlSrvRow, psql_srv::Error>>>;
 
diff --git a/psql-srv/tests/errors.rs b/psql-srv/tests/errors.rs
index e028900eed..b1e983971f 100644
--- a/psql-srv/tests/errors.rs
+++ b/psql-srv/tests/errors.rs
@@ -1,7 +1,6 @@
 use std::collections::HashMap;
 use std::vec;
 
-use async_trait::async_trait;
 use futures::{stream, Future};
 use postgres::NoTls;
 use postgres_protocol::Oid;
@@ -27,7 +26,6 @@ enum ErrorPosition {
 #[derive(Clone, Copy)]
 struct ErrorBackend(ErrorPosition);
 
-#[async_trait]
 impl PsqlBackend for ErrorBackend {
     type Resultset = stream::Iter<vec::IntoIter<Result<PsqlSrvRow, psql_srv::Error>>>;
 
diff --git a/psql-srv/tests/tls.rs b/psql-srv/tests/tls.rs
index 9e332f489e..ea320a0525 100644
--- a/psql-srv/tests/tls.rs
+++ b/psql-srv/tests/tls.rs
@@ -2,7 +2,6 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::vec;
 
-use async_trait::async_trait;
 use database_utils::{DatabaseURL, QueryableConnection};
 use futures::stream;
 use postgres_protocol::Oid;
@@ -28,7 +27,6 @@ impl TryFrom<TestValue> for psql_srv::PsqlValue {
     }
 }
 
-#[async_trait]
 impl PsqlBackend for TestBackend {
     type Resultset = stream::Iter<vec::IntoIter<Result<PsqlSrvRow, Error>>>;
 
diff --git a/readyset-adapter/src/upstream_database.rs b/readyset-adapter/src/upstream_database.rs
index 9fe6e57eaf..d93d50a0bd 100644
--- a/readyset-adapter/src/upstream_database.rs
+++ b/readyset-adapter/src/upstream_database.rs
@@ -357,11 +357,11 @@ where
         self.upstream().await?.start_tx(stmt).await
     }
 
-    async fn commit<'a>(&'a mut self) -> Result<Self::QueryResult<'a>, Self::Error> {
+    async fn commit(&mut self) -> Result<Self::QueryResult<'_>, Self::Error> {
         self.upstream().await?.commit().await
     }
 
-    async fn rollback<'a>(&'a mut self) -> Result<Self::QueryResult<'a>, Self::Error> {
+    async fn rollback(&mut self) -> Result<Self::QueryResult<'_>, Self::Error> {
         self.upstream().await?.rollback().await
     }
 
diff --git a/readyset-mysql/src/backend.rs b/readyset-mysql/src/backend.rs
index 9660bcbc8c..1fce235982 100644
--- a/readyset-mysql/src/backend.rs
+++ b/readyset-mysql/src/backend.rs
@@ -5,7 +5,6 @@ use std::convert::TryFrom;
 use std::fmt::Formatter;
 use std::ops::{Deref, DerefMut};
 
-use async_trait::async_trait;
 use futures_util::StreamExt;
 use itertools::{izip, Itertools};
 use mysql_async::consts::StatusFlags;
@@ -498,7 +497,6 @@ where
     }
 }
 
-#[async_trait]
 impl<W> MySqlShim<W> for Backend
 where
     W: AsyncWrite + Unpin + Send + 'static,
@@ -600,7 +598,7 @@ where
             Err(e) => info.error(e.error_kind(), e.to_string().as_bytes()).await,
         };
 
-        Ok(res?)
+        res
     }
 
     async fn on_execute(
@@ -741,7 +739,7 @@ where
         }
 
         let query_result = self.query(query).await;
-        return handle_query_result(query_result, results).await;
+        handle_query_result(query_result, results).await
     }
 
     fn password_for_username(&self, username: &str) -> Option<Vec<u8>> {
diff --git a/readyset-psql/benches/proxy.rs b/readyset-psql/benches/proxy.rs
index 4afd5012e5..6f163b9b15 100644
--- a/readyset-psql/benches/proxy.rs
+++ b/readyset-psql/benches/proxy.rs
@@ -18,7 +18,6 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::{io, vec};
 
-use async_trait::async_trait;
 use criterion::{criterion_group, criterion_main, Criterion};
 use futures::future::{try_select, Either};
 use futures::stream::Peekable;
@@ -128,7 +127,6 @@ impl Backend {
     }
 }
 
-#[async_trait]
 impl PsqlBackend for Backend {
     type Resultset = ResultStream;
 
diff --git a/readyset-psql/src/backend.rs b/readyset-psql/src/backend.rs
index 35861f8b26..b81be12562 100644
--- a/readyset-psql/src/backend.rs
+++ b/readyset-psql/src/backend.rs
@@ -4,7 +4,6 @@ use std::ops::Deref;
 use std::str::FromStr;
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use clap::ValueEnum;
 use eui48::MacAddressFormat;
 use postgres_types::{Oid, Type};
@@ -121,7 +120,6 @@ impl Backend {
     }
 }
 
-#[async_trait]
 impl ps::PsqlBackend for Backend {
     type Resultset = Resultset;
 
diff --git a/readyset/Cargo.toml b/readyset/Cargo.toml
index 778247e2af..4e7cf6e809 100644
--- a/readyset/Cargo.toml
+++ b/readyset/Cargo.toml
@@ -11,7 +11,6 @@ A real-time SQL caching engine for Postgres and MySQL."""
 
 [dependencies]
 anyhow = { workspace = true }
-async-trait = { workspace = true }
 clap = { workspace = true, features = ["derive","env"] }
 futures-util = { workspace = true }
 fail = { workspace = true }
diff --git a/readyset/src/lib.rs b/readyset/src/lib.rs
index 8265092b8c..876b24808d 100644
--- a/readyset/src/lib.rs
+++ b/readyset/src/lib.rs
@@ -6,6 +6,7 @@ pub mod psql;
 mod query_logger;
 use std::collections::HashMap;
 use std::fs::remove_dir_all;
+use std::future::Future;
 use std::io;
 use std::marker::Send;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr, ToSocketAddrs};
@@ -16,7 +17,6 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
 
 use anyhow::{anyhow, bail};
-use async_trait::async_trait;
 use clap::builder::NonEmptyStringValueParser;
 use clap::{ArgGroup, Parser, ValueEnum};
 use crossbeam_skiplist::SkipSet;
@@ -74,19 +74,22 @@ const UPSTREAM_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
 /// Retry interval to use when attempting to connect to the upstream database
 const UPSTREAM_CONNECTION_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 
-#[async_trait]
 pub trait ConnectionHandler {
     type UpstreamDatabase: UpstreamDatabase;
     type Handler: QueryHandler;
 
-    async fn process_connection(
+    fn process_connection(
         &mut self,
         stream: net::TcpStream,
         backend: Backend<Self::UpstreamDatabase, Self::Handler>,
-    );
+    ) -> impl Future<Output = ()> + Send;
 
     /// Return an immediate error to a newly-established connection, then immediately disconnect
-    async fn immediate_error(self, stream: net::TcpStream, error_message: String);
+    fn immediate_error(
+        self,
+        stream: net::TcpStream,
+        error_message: String,
+    ) -> impl Future<Output = ()> + Send;
 }
 
 /// How to behave when receiving unsupported `SET` statements.
diff --git a/readyset/src/mysql.rs b/readyset/src/mysql.rs
index 60017f1af3..6438ecfbd6 100644
--- a/readyset/src/mysql.rs
+++ b/readyset/src/mysql.rs
@@ -1,4 +1,3 @@
-use async_trait::async_trait;
 use mysql_srv::MySqlIntermediary;
 use readyset_adapter::upstream_database::LazyUpstream;
 use readyset_mysql::{MySqlQueryHandler, MySqlUpstream};
@@ -13,7 +12,6 @@ pub struct MySqlHandler {
     pub enable_statement_logging: bool,
 }
 
-#[async_trait]
 impl ConnectionHandler for MySqlHandler {
     type UpstreamDatabase = LazyUpstream<MySqlUpstream>;
     type Handler = MySqlQueryHandler;
diff --git a/readyset/src/psql.rs b/readyset/src/psql.rs
index fa4aff04d3..986a265052 100644
--- a/readyset/src/psql.rs
+++ b/readyset/src/psql.rs
@@ -1,7 +1,6 @@
 use std::io::Read;
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use clap::Parser;
 use readyset_adapter::upstream_database::LazyUpstream;
 use readyset_errors::ReadySetResult;
@@ -92,7 +91,6 @@ impl PsqlHandler {
     }
 }
 
-#[async_trait]
 impl ConnectionHandler for PsqlHandler {
     type UpstreamDatabase = LazyUpstream<PostgreSqlUpstream>;
     type Handler = PostgreSqlQueryHandler;