Here is PoC for connection pooling based on deadpool create. It mostly mirrors r2d2 code from main diesel repo. If you have an interest to contribution i can finalise PR and provide tests.

The code in theory can be used outside of the repo, but it will require wrapping of connections to be able implement traits and there is enough traits magic to make it not so simple task.

diesel-async + AsyncMysqlConnection gives error after executing more than 16382 inserts, while diesel does not

Can't create more than max_prepared_stmt_count statements (current value: 16382)

Can't create more than max_prepared_stmt_count statements (current value: 16382)

Setup

Versions

• Rust: stable rustc 1.64.0 (a55dd71d5 2022-09-19)
• Diesel: 2
• Diesel_async: 0.1.0
• Database: mysql
• Operating System windows/linux

Feature Flags

• diesel: ["mysql", "chrono"]
• diesel_async: ["bb8", "mysql", "tokio"]

What are you trying to accomplish?

Since diesel + multithread does not have a problem, I expect diesel-async to have the same ability.

What is the actual output?

Can't create more than max_prepared_stmt_count statements

Are you seeing any additional errors?

No

Steps to reproduce

migration

CREATE TABLE `user` (
    `id`    bigint unsigned     NOT NULL AUTO_INCREMENT,
    `name`  varchar(255)        NOT NULL,
    `email` varchar(255)        NOT NULL,
    `created_at` datetime(3) NOT NULL DEFAULT now(3),
    `updated_at` datetime(3) NOT NULL DEFAULT now(3) ON UPDATE now(3),
    PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

main.rs

use diesel_async::{AsyncConnection, AsyncMysqlConnection, RunQueryDsl};

mod schema {
    diesel::table! {
        user (id) {
            id -> Unsigned<Bigint>,
            name -> Varchar,
            email -> Varchar,
            created_at -> Datetime,
            updated_at -> Datetime,
        }
    }
}

#[derive(diesel::Insertable)]
#[diesel(table_name = schema::user)]
pub struct NewUser<'a> {
    pub name: &'a str,
    pub email: &'a str,
}
impl<'a> NewUser<'a> {
    pub async fn insert(&self, conn: &mut AsyncMysqlConnection) -> anyhow::Result<()> {
        diesel::insert_into(schema::user::dsl::user)
            .values(self)
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let url = dotenv::var("DATABASE_URL")?;

    let mut conn = diesel_async::AsyncMysqlConnection::establish(&url).await?;

    for i in 0..16382 + 10 {
        NewUser {
            name: &format!("name-{}", i),
            email: &format!("email-{}", i),
        }
        .insert(&mut conn)
        .await?;
        if i % 100 == 0 {
            println!("{}", i);
        }
    }
    Ok(())
}

cargo.toml

[dependencies]
anyhow = "1.0.65"
diesel = { version = "2.0.0", features = ["mysql", "chrono"] }
diesel-async = { version = "0.1.0", features = ["bb8", "mysql", "tokio"], optional = true }
dotenv = "0.15.0"
tokio = { version = "1.21.2", features = ["full"] }

git clone https://github.com/gwy15/diesel-async-example
cd diesel-async-example
docker-compose up -d
diesel migration run
cargo t --no-default-features --features async

Checklist

• [x] I have already looked over the issue tracker for similar possible closed issues.

• [x] This issue can be reproduced on Rust's stable channel. (Your issue will be closed if this is not the case)
• [x] This issue can be reproduced without requiring a third party crate

I'm trying to get diesel-async + bb8 + postgres working, but its not able to convert the pooled connection -> AsyncPgConnection to work with diesel-async. I also can't find any unit tests for any of the connection pools.

Example:

use diesel_async::pooled_connection::bb8::Pool;
use diesel_async::pg::AsyncPgConnection;
pub type DbPool = Pool<AsyncPgConnection>;

pub async fn delete_for_community(
  pool: &DbPool,
  for_community_id: CommunityId,
) -> Result<usize, Error> {
  let mut conn = pool.get().await.unwrap();
  diesel::delete(community_moderator.filter(community_id.eq(for_community_id))).execute(&mut conn); // <--- Error here
}

Error:

diesel::delete(community_moderator.filter(community_id.eq(for_community_id))).execute(&mut conn);          |                                                                                           ^^^^^^^ the trait
`Connection` is not implemented for `bb8::api::PooledConnection<'_, AsyncDieselConnectionManager<AsyncPgConnection>>`               |                                                                                                                        =
help: the following other types implement trait `Connection`:                                                                    PooledConnection<M>                                                                                                      diesel::PgConnection

note: required because of the requirements on the impl of `diesel::query_dsl::methods::ExecuteDsl<bb8::api::PooledConnection<'_, AsyncDieselConnectionManager<AsyncPgConnection>>, Pg>` for `DeleteStatement<schema::community_moderator::table, query_builder::where_clause::WhereClause<diesel::expression::grouped::Grouped<diesel::expression::operators::Eq<schema::community_moderator::columns::community_id, diesel::expression::bound::Bound<diesel::sql_types::Integer, i32>>>>>`  note: required by a bound in `diesel::RunQueryDsl::execute`                                                                  --> /home/xxx/.cargo/registry/src/github.com-1ecc6299db9ec823/diesel-2.0.0/src/query_dsl/mod.rs:1401:15                 |                                                                                                                   1401 |         Self: methods::ExecuteDsl<Conn>,                                                                               |               ^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `diesel::RunQueryDsl::execute`

• Tokio updated to use "1" for compatibility + make rt-multi-thread dev-dependency
• Diesel dev-dependency set as 2.0.0 (erroneously 2.0.0-RC-1)
• Remove tokio/rt-multi-thread from postgres feature (needed only for testing)
• Alpha sort dependencies

First, thank you for this crate and also the latest changes (new_with_setup), which enabled me to use it with rustls.

This change is needed to use the #[instrument] macro of tracing, which seems to require Debug impls for types being part of the function; since they get printed as part of the logs.

Example: https://github.com/asaaki/capture-the-ip/blob/main/cti_core/src/app/api.rs#L95-L97

The state is a pool represented by the above mentioned types. The tracing crate is very unhappy if those types are not Debug.

I kept the implementations very simple and minimal; the connection would only display its name, the manager also includes the connection URL. All other fields left untouched to not infect everything.

Setup

Versions

• Rust: 1.64
• Diesel: 2.0
• Diesel_async: 0.1
• Database: tested using postgres but should be database-agnostic since it's a compile-time issue
• Operating System: Linux

Feature Flags

• diesel: postgres
• diesel_async: postgres

Problem Description

When capturing data in the closure + future passed to AsyncConnection::transaction(), lifetimes makes the code not compile.

The actual error given out of the box is the following:

error[E0597]: `user_id` does not live long enough
  --> src/main.rs:30:66
   |
30 |         .transaction(|conn| Box::pin(async { users::table.find(*&user_id).first(conn).await }))
   |                      ------ -------------------------------------^^^^^^^----------------------
   |                      |      |                                    |
   |                      |      |                                    borrowed value does not live long enough
   |                      |      returning this value requires that `user_id` is borrowed for `'static`
   |                      value captured here
...
40 | }
   | - `user_id` dropped here while still borrowed

The error is very confusing because, obviously, the closure and the future are consumed before the end of the main function. And the 'static lifetime makes no real sense.

I originally though the issue was due to automatic lifetimes assigned by rustc to the Future in the .transaction function: give the same lifetime to the conn and to the future, so I tried to explicit them:

// extract of `src/transaction_manager.rs`
    async fn transaction<'conn, 'fut, F, R, E>(conn: &'conn mut Conn, callback: F) -> Result<R, E>
    where
        F: FnOnce(&mut Conn) -> BoxFuture<'fut, Result<R, E>> + Send,
        E: From<Error> + Send,
        R: Send,
    {
        Self::begin_transaction(conn).await?;

With this, the compile error changes, but is as confusing as ever:

error: lifetime may not live long enough
  --> src/main.rs:30:29
   |
30 |         .transaction(|conn| Box::pin(async { users::table.find(*&user_id).first(conn).await }))
   |                       ----- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ returning this value requires that `'1` must outlive `'2`
   |                       |   |
   |                       |   return type of closure is Pin<Box<(dyn futures::Future<Output = Result<User, diesel::result::Error>> + std::marker::Send + '2)>>
   |                       has type `&'1 mut AsyncPgConnection`

At least this time the lifetimes involved make sense: '1 (i.e. the reborrowed conn) must outlive '2 (i.e. the future that uses said conn). But it makes no sense in this case that the future outlives the connection (because this is what the compiler effectively tells us, that it may outlive it).

The full code is below.

Many thanks, this is the last hurdle for me to adopt diesel 2 + diesel-async!

What are you trying to accomplish?

Running a transaction that borrows data from around it.

What is the expected output?

A binary.

What is the actual output?

A lifetime error I cannot wrap my head around.

Are you seeing any additional errors?

Steps to reproduce

main.rs:

use anyhow::Result;
use diesel::{prelude::*, table, Identifiable, Queryable};
use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl};

table! {
    users(id) {
        id -> Integer,
        name -> Text,
    }
}

#[derive(Queryable, Identifiable, Debug, Clone)]
#[diesel(table_name = users)]
struct User {
    id: i32,
    name: String,
}

#[tokio::main]
async fn main() -> Result<()> {
    let mut conn = AsyncPgConnection::establish(&std::env::var("DATABASE_URL")?).await?;

    let user_id = 4242;
    // The *& is intended to make the closure capture a reference since i32 is Copy
    let user: User = conn
        .transaction(|conn| Box::pin(async { users::table.find(*&user_id).first(conn).await }))
        .await?;
    println!("Found user {user:?}");

    Ok(())
}

Cargo.toml:

[package]
name = "diesel-2-issue"
version = "0.1.0"
edition = "2021"

[dependencies]
anyhow = "1"
diesel = { version = "2", features = ["postgres"] }
diesel-async = { path = "../diesel_async", version = "0.1", features = ["postgres"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

Checklist

• [x] I have already looked over the issue tracker for similar possible closed issues.

• [x] This issue can be reproduced on Rust's stable channel. (Your issue will be closed if this is not the case)
• [x] This issue can be reproduced without requiring a third party crate

Hi,

I am sorry about not using your template. It is just question. Do you have any vision on support connections pooling here? Any plans? Any interest to contribution?

Not sure if this is intended, but TransactionManagerStatus is effectively hidden from public access; however, this enum is required if you want to implement the trait TransactionManager, as the trait function TransactionManager::transaction_manager_status_mut returns a TransactionManagerStatus directly. As far as I can tell, it is accessible in diesel, even if gated behind a feature.

Adds a postgresql-native-tls dependency and uses it by default. This brings the postgresql inline with mysql_async which also depends on native_tls.

I am not sure if this is the best way to implement this feature, I can re-work it to be behind a feature flag if required.

edit: I think the error handling could be improved as well, I tried to keep it close to diesel but may need some changes.

Since we disabled mysql_async's caching mechanism and in that way it requires manual close the statement, we need to close all prepared stmts before dropping it.

From my understanding, and correct me if I'm wrong, as long as statements are cacheable, we will never drop them. So we only need to drop non-cacheable stmts. This PR implements this.

Concern

Is it better to close prepare statements right after execution or close before the next execution of non-cacheable statement (the implementation)?

close #26

This PR updates diesel to use the recently released 2.0 stable version, instead of the RC1. It also removes the Cargo.lock file from the repository, so it overrides #16.

Setup

Versions

• Rust: rustc 1.62.1 (e092d0b6b 2022-07-16)
• Diesel: 2.0.00-rc.1
• Diesel_async: main branch @ 3c9e976
• Database: PostgreSQL
• Operating System: MacOS Monterey

Feature Flags

• diesel: ["postgres_backend", "chrono", "uuid", "numeric", "ipnet-address", "serde_json" ], default features inactive
• diesel_async: ["postgres"]
• diesel_migrations: ["postgres"]

Problem Description

Not really a problem per se, but a missing feature. I would like to not depend on diesel_cli to manage migrations, so I was using diesel_migrations to handle this, but the MigrationHarness needed to retrieve the applied migrations requires the asynchronous connection to implement MigrationConnection, which makes sure that the DB has the correct table.

Unfortunately, this is not yet implemented for asynchronous connections.

What are you trying to accomplish?

I'm trying to embed the migrations with my code, and even create the DB if needed, so that on startup, it will check that the DB is at the correct migration.

What is the expected output?

I would expect to be able to do this with an asynchronous connection, the same way as with a synchronous connection.

What is the actual output?

error[E0277]: the trait bound `AsyncPgConnection: MigrationConnection` is not satisfied
   --> backend/src/lib.rs:152:20
    |
152 |     run_migrations(&mut conn).expect("couldn't run migrations");
    |     -------------- ^^^^^^^^^ the trait `MigrationConnection` is not implemented for `AsyncPgConnection`
    |     |
    |     required by a bound introduced by this call
    |
    = help: the trait `MigrationConnection` is implemented for `PgConnection`
    = note: required because of the requirements on the impl of `MigrationHarness<Pg>` for `AsyncPgConnection`
note: required by a bound in `update_database::{closure#0}::run_migrations`
   --> backend/src/lib.rs:123:31
    |
122 |     fn run_migrations(
    |        -------------- required by a bound in this
123 |         connection: &mut impl MigrationHarness<Pg>,
    |                               ^^^^^^^^^^^^^^^^^^^^ required by this bound in `update_database::{closure#0}::run_migrations`

error[E0277]: the trait bound `AsyncPgConnection: LoadConnection` is not satisfied
   --> backend/src/lib.rs:152:20
    |
152 |     run_migrations(&mut conn).expect("couldn't run migrations");
    |     -------------- ^^^^^^^^^ the trait `LoadConnection` is not implemented for `AsyncPgConnection`
    |     |
    |     required by a bound introduced by this call
    |
    = help: the trait `LoadConnection<B>` is implemented for `PgConnection`
    = note: required because of the requirements on the impl of `diesel::query_dsl::LoadQuery<'_, AsyncPgConnection, MigrationVersion<'static>>` for `SelectStatement<FromClause<diesel_migrations::migration_harness::__diesel_schema_migrations::table>, query_builder::select_clause::SelectClause<diesel_migrations::migration_harness::__diesel_schema_migrations::columns::version>, query_builder::distinct_clause::NoDistinctClause, query_builder::where_clause::NoWhereClause, query_builder::order_clause::OrderClause<diesel::expression::operators::Desc<diesel_migrations::migration_harness::__diesel_schema_migrations::columns::version>>>`
    = note: required because of the requirements on the impl of `MigrationHarness<Pg>` for `AsyncPgConnection`
note: required by a bound in `update_database::{closure#0}::run_migrations`
   --> backend/src/lib.rs:123:31
    |
122 |     fn run_migrations(
    |        -------------- required by a bound in this
123 |         connection: &mut impl MigrationHarness<Pg>,
    |                               ^^^^^^^^^^^^^^^^^^^^ required by this bound in `update_database::{closure#0}::run_migrations`

Are you seeing any additional errors?

No additional error is shown.

Steps to reproduce

    /// Embedded DB migrations.
    const MIGRATIONS: EmbeddedMigrations = embed_migrations!("../migrations");

    /// Runs the migrations in the database.
    #[allow(trivial_casts)]
    fn run_migrations(
        connection: &mut impl MigrationHarness<Pg>,
    ) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
        let migrations = (&MIGRATIONS as &dyn MigrationSource<Pg>).migrations()?;
        let last = migrations
            .last()
            .expect("could not find the last migration");

        if let Some(last_applied) = connection.applied_migrations()?.first() {
            if last_applied > &last.name().version() {
                // TODO: downgrade, probably add a confirmation parameter in the executable.
                // But how? We would need to get the migrations for downgrade, but we don't have them.
                panic!("Not possible to downgrade the database");
            }
        }

        let _ = connection.run_pending_migrations(MIGRATIONS)?;
        Ok(())
    }

    let mut conn = match db::Connection::establish(db_cfg.conn_str()).await {
        Err(e) => {
            eprintln!("WARNING: error connecting to {}", db_cfg.conn_str());
            db::create(db_cfg)
                .await
                .expect("could not create the database")
        }
        Ok(c) => c,
    };

Checklist

• [x] I have already looked over the issue tracker for similar possible closed issues.

• [x] This issue can be reproduced on Rust's stable channel. (Your issue will be closed if this is not the case)
• [ ] This issue can be reproduced without requiring a third party crate

Suggestion

I can work on this, but, should I add a feature to add this migrations connection implementation, so that it's not active by default for those who don't want to use migrations?

Also, I see that the diesel_migrations trait expects the connection to run the SQL synchronously, so this might be trickier than expected.