DatabaseConnection 与 ConnectOptions

    pub async fn connect<C>(opt: C) -> Result<DatabaseConnection, DbErr>
    where
        C: Into<ConnectOptions>,
    {
        let opt: ConnectOptions = opt.into();

        if url::Url::parse(&opt.url).is_err() {
            return Err(conn_err(format!(
                "The connection string '{}' cannot be parsed.",
                opt.url
            )));
        }

        // ...
        #[cfg(feature = "sqlx-postgres")]
        if DbBackend::Postgres.is_prefix_of(&opt.url) {
            return crate::SqlxPostgresConnector::connect(opt).await;
        }
        // ...

        Err(conn_err(format!(
            "The connection string '{}' has no supporting driver.",
            opt.url
        )))
    }

postgresql://lab:lab@192.168.64.106/lab => Postgres
mysql://lab:s3cret@db.internal/lab => Connection Error: The connection string 'mysql://lab:s3cret@db.internal/lab' has no supporting driver.
lab:s3cret@192.168.64.106/lab => Connection Error: The connection string 'lab:s3cret@192.168.64.106/lab' has no supporting driver.
DatabaseConnection::default() => Connection Error: Disconnected / Conn(Internal("Disconnected"))

#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct DatabaseConnection {
    /// Driver-specific connection or pool. Held in a field so we can attach
    /// orthogonal state (e.g. RBAC) alongside.
    pub inner: DatabaseConnectionType,
    #[cfg(feature = "rbac")]
    pub(crate) rbac: crate::RbacEngineMount,
}

#[derive(Clone)]
pub enum DatabaseConnectionType {
    // ...
    /// PostgreSQL connection pool (`sqlx-postgres`).
    #[cfg(feature = "sqlx-postgres")]
    SqlxPostgresPoolConnection(crate::SqlxPostgresPoolConnection),
    // ...
    /// In-memory mock connection used for testing (`mock`).
    #[cfg(feature = "mock")]
    MockDatabaseConnection(Arc<crate::MockDatabaseConnection>),
    // ...
    Disconnected,
}

    pub async fn connect(options: ConnectOptions) -> Result<DatabaseConnection, DbErr> {
        let record_stmt_in_spans = options.get_record_stmt_in_spans();
        let mut sqlx_opts = options
            .url
            .parse::<PgConnectOptions>()
            .map_err(sqlx_error_to_conn_err)?;
        use sqlx::ConnectOptions;
        if !options.sqlx_logging {
            sqlx_opts = sqlx_opts.disable_statement_logging();
        } else {
            sqlx_opts = sqlx_opts.log_statements(options.sqlx_logging_level);
            if options.sqlx_slow_statements_logging_level != LevelFilter::Off {
                sqlx_opts = sqlx_opts.log_slow_statements(
                    options.sqlx_slow_statements_logging_level,
                    options.sqlx_slow_statements_logging_threshold,
                );
            }
        }

        if let Some(application_name) = &options.application_name {
            sqlx_opts = sqlx_opts.application_name(application_name);
        }

        if let Some(timeout) = options.statement_timeout {
            sqlx_opts = sqlx_opts.options([("statement_timeout", timeout.as_millis().to_string())]);
        }

        if let Some(f) = &options.pg_opts_fn {
            sqlx_opts = f(sqlx_opts);
        }

        let set_search_path_sql = // ... 见下文
        // ...
        let mut pool_options = options.sqlx_pool_options();
        pool_options = crate::ConnectOptions::apply_before_acquire::<sqlx::Postgres>(
            pool_options,
            ping_after_idle,
            pg_before_acquire,
        );

        if let Some(sql) = set_search_path_sql {
            pool_options = pool_options.after_connect(move |conn, _| {
                // ... 每条新连接执行一次 SET search_path
            });
        }
        if let Some(f) = &pg_pool_opts_fn {
            pool_options = f(pool_options);
        }
        let pool = if lazy {
            pool_options.connect_lazy_with(sqlx_opts)
        } else {
            pool_options
                .connect_with(sqlx_opts)
                .await
                .map_err(sqlx_error_to_conn_err)?
        };
        // ... 包成 DatabaseConnection
        if let Some(cb) = after_connect {
            cb(conn.clone()).await?;
        }

        Ok(conn)
    }

connect_timeout=-      acquire_timeout=-      => PoolOptions { max_connections: 1, min_connections: 0, connect_timeout: 30s, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }
connect_timeout=1000ms acquire_timeout=-      => PoolOptions { max_connections: 1, min_connections: 0, connect_timeout: 1s, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }
connect_timeout=-      acquire_timeout=1000ms => PoolOptions { max_connections: 1, min_connections: 0, connect_timeout: 1s, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }
connect_timeout=1500ms acquire_timeout=500ms  => PoolOptions { max_connections: 1, min_connections: 0, connect_timeout: 500ms, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }
connect_timeout=500ms  acquire_timeout=1500ms => PoolOptions { max_connections: 1, min_connections: 0, connect_timeout: 1.5s, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }
ConnectOptions::new 默认 => PoolOptions { max_connections: 10, min_connections: 0, connect_timeout: 30s, max_lifetime: Some(1800s), idle_timeout: Some(600s), test_before_acquire: true }

        if let Some(connect_timeout) = self.connect_timeout {
            opt = opt.acquire_timeout(connect_timeout);
        }
        if let Some(idle_timeout) = self.idle_timeout {
            opt = opt.idle_timeout(idle_timeout);
        }
        if let Some(acquire_timeout) = self.acquire_timeout {
            opt = opt.acquire_timeout(acquire_timeout);
        }

for (c, a) in CASES {
  let started = Instant::now();
  let res = Database::connect(opts("postgres://lab:lab@127.0.0.1:1/lab", c, a)).await;
  let err = res.err().expect("must fail");
  println!("{} 用时 {:.2}s  {err}", label(c, a), started.elapsed().as_secs_f64());
}

--- 2. Database::connect 到没人监听的端口 127.0.0.1:1 ---
connect_timeout=1000ms acquire_timeout=-      用时 1.00s  Connection Error: pool timed out while waiting for an open connection
connect_timeout=-      acquire_timeout=1000ms 用时 1.00s  Connection Error: pool timed out while waiting for an open connection
connect_timeout=1500ms acquire_timeout=500ms  用时 0.50s  Connection Error: pool timed out while waiting for an open connection
connect_timeout=500ms  acquire_timeout=1500ms 用时 1.50s  Connection Error: pool timed out while waiting for an open connection
--- 3. Database::connect 到只收不答的端口 ---
connect_timeout=1000ms acquire_timeout=-      用时 1.01s  Conn(SqlxError(PoolTimedOut))
connect_timeout=-      acquire_timeout=1000ms 用时 1.00s  Conn(SqlxError(PoolTimedOut))
connect_timeout=1500ms acquire_timeout=500ms  用时 0.50s  Conn(SqlxError(PoolTimedOut))
connect_timeout=500ms  acquire_timeout=1500ms 用时 1.50s  Conn(SqlxError(PoolTimedOut))

=== 默认 ===
-- 第 1 条（上一条结束后又等了 200+0ms）
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_1, "select pg_backend_pid() as pid"), Describe(S, sqlx_s_1), Sync
...
-- 第 2 条（上一条结束后又等了 200+50ms）
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
=== test_before_acquire(false) ===
-- 第 1 条（上一条结束后又等了 200+0ms）
  [conn 2] C->S Parse(sqlx_s_1, "select pg_backend_pid() as pid"), Describe(S, sqlx_s_1), Sync
...
-- 第 3 条（上一条结束后又等了 200+500ms）
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
=== test_before_acquire_if_idle_for(300ms) ===
...
-- 第 2 条（上一条结束后又等了 200+50ms）
  [conn 3] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
-- 第 3 条（上一条结束后又等了 200+500ms）
  [conn 3] C->S Sync
  [conn 3] S->C ReadyForQuery(I)
  [conn 3] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
=== before_acquire：age 超过 400ms 就丢 ===
-- 第 1 条（上一条结束后又等了 200+0ms）
  before_acquire: age=204.712459ms idle_for=204.709875ms -> true
  [conn 4] C->S Parse(sqlx_s_1, "select pg_backend_pid() as pid"), Describe(S, sqlx_s_1), Sync
...
   pg_backend_pid = 1932
-- 第 2 条（上一条结束后又等了 200+50ms）
  before_acquire: age=463.819959ms idle_for=256.242875ms -> false
  [conn 4] C->S Terminate
  [conn 5] C->S StartupMessage(v3.0)
...
   pg_backend_pid = 1933

        let set_search_path_sql = options.schema_search_path.as_ref().map(|schema| {
            let mut string = "SET search_path = ".to_owned();
            if schema.starts_with('"') {
                write!(&mut string, "{schema}").expect("Infallible");
            } else {
                for (i, schema) in schema.split(',').enumerate() {
                    if i > 0 {
                        write!(&mut string, ",").expect("Infallible");
                    }
                    if schema.starts_with('"') {
                        write!(&mut string, "{schema}").expect("Infallible");
                    } else {
                        write!(&mut string, "\"{schema}\"").expect("Infallible");
                    }
                }
            }
            string
        });

## set_schema_search_path("so_conn1")
 INFO sqlx::query: summary="SET search_path = \"so_conn1\"" db.statement="" rows_affected=0 rows_returned=0 elapsed=543.334µs elapsed_secs=0.000543334
search_path = so_conn1
## set_schema_search_path("so_conn1,public")
 INFO sqlx::query: summary="SET search_path = \"so_conn1\",\"public\"" db.statement="" rows_affected=0 rows_returned=0 elapsed=300.167µs elapsed_secs=0.000300167
search_path = so_conn1, public
## set_schema_search_path("so_conn1, public")
 INFO sqlx::query: summary="SET search_path = \"so_conn1\",\" …" db.statement="\n\nSET search_path = \"so_conn1\",\" public\"\n" rows_affected=0 rows_returned=0 elapsed=245.708µs elapsed_secs=0.000245708
search_path = so_conn1, " public"
## set_schema_search_path("\"so_conn1\"")
 INFO sqlx::query: summary="SET search_path = \"so_conn1\"" db.statement="" rows_affected=0 rows_returned=0 elapsed=317.5µs elapsed_secs=0.0003175
search_path = so_conn1
## set_schema_search_path("my\"schema")
 INFO sqlx::query: summary="SET search_path = \"my\"schema\"" db.statement="" rows_affected=0 rows_returned=0 elapsed=1.061042ms elapsed_secs=0.001061042
ERROR sqlx_core::pool::inner: error returned from after_connect error=error returned from database: syntax error at or near "schema" at line 1244
 INFO sqlx::query: summary="SET search_path = \"my\"schema\"" db.statement="" rows_affected=0 rows_returned=0 elapsed=559.583µs elapsed_secs=0.000559583
ERROR sqlx_core::pool::inner: error returned from after_connect error=error returned from database: syntax error at or near "schema" at line 1244
...
connect 失败，用时 1.00s：Connection Error: pool timed out while waiting for an open connection
--- 2. 以引号开头的值原样拼进 SQL ---
 INFO sqlx::query: summary="SET search_path = \"so_conn1\"; …" db.statement="\n\nSET search_path = \"so_conn1\"; SET application_name = 'injected'\n" rows_affected=0 rows_returned=0 elapsed=324.667µs elapsed_secs=0.000324667
search_path = so_conn1
application_name = injected

PgConnectOptions.options = Some("-c search_path=so_conn1 -c statement_timeout=300")
PgConnectOptions.application_name = Some("conn1-app")
statement_timeout = 300ms
search_path = so_conn1
application_name = conn1-app
 INFO sqlx::query: summary="select pg_sleep(1)" db.statement="" rows_affected=0 rows_returned=0 elapsed=307.602208ms elapsed_secs=0.307602208
pg_sleep(1)：Execution Error: error returned from database: canceling statement due to statement timeout at line 3394

--- 4. ConnectOptions::after_connect 被调用几次 ---
池里 3 条连接，after_connect 调用了 1 次
--- 5. map_sqlx_postgres_pool_opts 里再设 after_connect ---
 INFO sqlx::query: summary="select current_setting('search_path') as v" db.statement="" rows_affected=1 rows_returned=1 elapsed=1.140042ms elapsed_secs=0.001140042
search_path = "$user", public

        if ping_after_idle.is_none() && user_cb.is_none() {
            return opt;
        }
        if ping_after_idle.is_some() {
            opt = opt.test_before_acquire(false);
        }
        opt.before_acquire(move |conn, meta| {
            let user_cb = user_cb.clone();
            Box::pin(async move {
                if let Some(threshold) = ping_after_idle {
                    // ...
                    if meta.idle_for >= threshold {
                        conn.ping().await?;
                    }
                }
                match user_cb {
                    Some(user_cb) => user_cb(conn, meta).await,
                    None => Ok(true),
                }
            })
        })

=== 默认 ===
-- 第 1 条
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_1, "select pg_backend_pid() as pid"), Describe(S, sqlx_s_1), Sync
...
=== test_before_acquire(false) ===
-- 第 1 条
  [conn 2] C->S Parse(sqlx_s_1, "select pg_backend_pid() as pid"), Describe(S, sqlx_s_1), Sync
...
=== test_before_acquire_if_idle_for(300ms) ===
-- 第 2 条（上一条结束后又等了 200+50ms）
  [conn 3] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
-- 第 3 条（上一条结束后又等了 200+500ms）
  [conn 3] C->S Sync
  [conn 3] S->C ReadyForQuery(I)
  [conn 3] C->S Bind(portal unnamed, stmt sqlx_s_1, ), Execute, Close(P, unnamed), Sync
...
=== before_acquire：age 超过 400ms 就丢 ===
-- 第 2 条（上一条结束后又等了 200+50ms）
  before_acquire: age=460.341875ms idle_for=254.167ms -> false
  [conn 4] C->S Terminate
  [conn 5] C->S StartupMessage(v3.0)
...
   pg_backend_pid = 1923

let q = Query::select().expr(Expr::val(7).add(1)).to_owned();
db.execute(&q).await?;
let stmt = Statement::from_sql_and_values(DbBackend::Postgres, "select $1::int + 1", [7.into()]);
db.execute_raw(stmt).await?;
let stmt = Statement::from_string(DbBackend::Postgres, "select 1; select 2");
db.execute_raw(stmt).await.err().expect("two statements");
db.execute_unprepared("select 1; select 2").await?;

## execute(&SelectStatement)
  [conn 1] C->S Parse(sqlx_s_1, "SELECT $1 + $2"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23/23), RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1, $1=0x00000007 $2=0x00000001), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
 INFO sqlx::query: summary="SELECT $1 + $2" db.statement="" rows_affected=1 rows_returned=1 elapsed=3.024875ms elapsed_secs=0.003024875
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
rows_affected = 1
## execute_raw(Statement::from_sql_and_values)
  [conn 1] C->S Parse(sqlx_s_2, "select $1::int + 1"), Describe(S, sqlx_s_2), Sync
  [conn 1] S->C ParseComplete, ParameterDescription(OID 23), RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2, $1=0x00000007), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 4), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
 INFO sqlx::query: summary="select $1::int + 1" db.statement="" rows_affected=1 rows_returned=1 elapsed=1.835625ms elapsed_secs=0.001835625
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
rows_affected = 1
## execute_raw(Statement::from_string(两条语句))
  [conn 1] C->S Parse(sqlx_s_3, "select 1; select 2"), Describe(S, sqlx_s_3), Sync
  [conn 1] S->C ErrorResponse, ReadyForQuery(I)
 INFO sqlx::query: summary="select 1; select 2" db.statement="" rows_affected=0 rows_returned=0 elapsed=1.351542ms elapsed_secs=0.001351542
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
Execution Error: error returned from database: cannot insert multiple commands into a prepared statement at line 1478
## execute_unprepared(两条语句)
  [conn 1] C->S Query("select 1; select 2")
  [conn 1] S->C RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), RowDescription, DataRow(列长 1), CommandComplete("SELECT 1"), ReadyForQuery(I)
 INFO sqlx::query: summary="select 1; select 2" db.statement="" rows_affected=2 rows_returned=2 elapsed=1.112833ms elapsed_secs=0.001112833
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
rows_affected = 2

let pool = db.get_postgres_connection_pool();
let n: i64 = sqlx::query_scalar("select 40::int8 + 2").fetch_one(pool).await?;
let from_pool: DatabaseConnection = pool.clone().into();

 INFO sqlx::query: summary="select 40::int8 + 2" db.statement="" rows_affected=1 rows_returned=1 elapsed=2.680792ms elapsed_secs=0.002680792
sqlx::query_scalar(..).fetch_one(pool) = 42，池里 1 条
From<PgPool>：Postgres

--- 4. 池满（max_connections=1，唯一一条被占住）时执行一条语句 ---
connect_timeout=1000ms acquire_timeout=-      用时 1.00s  Failed to acquire connection from pool: Connection pool timed out
connect_timeout=-      acquire_timeout=1000ms 用时 1.00s  Failed to acquire connection from pool: Connection pool timed out
connect_timeout=1500ms acquire_timeout=500ms  用时 0.50s  Failed to acquire connection from pool: Connection pool timed out
connect_timeout=500ms  acquire_timeout=1500ms 用时 1.50s  Failed to acquire connection from pool: Connection pool timed out