saddle-db 0.2.0

Saddle managed asynchronous database access and transactions
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use std::{
    env,
    future::{Future, pending},
    pin::Pin,
    sync::Arc,
    task::{Context, Poll, Wake, Waker},
};

use sqlx::{MySqlPool, mysql::MySqlPoolOptions};

struct NoopWake;

impl Wake for NoopWake {
    fn wake(self: Arc<Self>) {}
}

fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
    let waker = Waker::from(Arc::new(NoopWake));
    let mut context = Context::from_waker(&waker);
    future.poll(&mut context)
}

async fn pool(url: &str, min_connections: u32, max_connections: u32) -> MySqlPool {
    MySqlPoolOptions::new()
        .min_connections(min_connections)
        .max_connections(max_connections)
        .test_before_acquire(false)
        .idle_timeout(None)
        .max_lifetime(None)
        .connect(url)
        .await
        .unwrap()
}

#[tokio::test(flavor = "current_thread")]
async fn logical_capacity_cannot_use_try_acquire_without_preopened_connections() {
    let Ok(url) = env::var("SADDLE_C6_STOP_DATABASE_URL") else {
        eprintln!("skipping C6 STOP evidence: SADDLE_C6_STOP_DATABASE_URL is not set");
        return;
    };
    let pool = pool(&url, 0, 2).await;

    assert_eq!(pool.options().get_max_connections(), 2);
    assert_eq!(
        pool.size(),
        1,
        "sqlx connect opens one connection when min_connections is zero"
    );

    let mut first = pool.try_acquire().expect("the startup connection is idle");
    assert!(
        pool.try_acquire().is_none(),
        "try_acquire correctly refuses to create or wait for the second physical connection"
    );
    first.return_to_pool().await;
    drop(first);
    pool.close().await;
}

#[tokio::test(flavor = "current_thread")]
async fn cancellation_returns_pool_connection_only_through_hidden_spawn() {
    let Ok(url) = env::var("SADDLE_C6_STOP_DATABASE_URL") else {
        eprintln!("skipping C6 STOP evidence: SADDLE_C6_STOP_DATABASE_URL is not set");
        return;
    };
    let pool = pool(&url, 1, 1).await;
    assert_eq!(pool.size(), 1);
    assert_eq!(pool.num_idle(), 1);

    let execution_pool = pool.clone();
    let mut execution = Box::pin(async move {
        let mut connection = execution_pool
            .try_acquire()
            .expect("preopened fixed connection");
        let _ = sqlx::query("SELECT SLEEP(60)")
            .fetch_optional(&mut *connection)
            .await
            .unwrap();
        pending::<()>().await;
        connection.return_to_pool().await;
    });

    assert!(matches!(poll_once(execution.as_mut()), Poll::Pending));
    assert_eq!(pool.num_idle(), 0);

    // This models Runtime dropping a pending DB operation. sqlx's
    // PoolConnection::drop cannot return the slot synchronously and calls its
    // own runtime spawn. On a current-thread runtime that task cannot run until
    // this test yields.
    drop(execution);
    assert_eq!(
        pool.num_idle(),
        0,
        "the physical slot is not reconciled when the operation Drop returns"
    );

    for _ in 0..200 {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        if pool.num_idle() == 1 {
            break;
        }
    }
    assert_eq!(
        pool.num_idle(),
        1,
        "sqlx's hidden return_to_pool task eventually restores the slot"
    );
    pool.close().await;
}