Struct SqlxBroker 

pub struct SqlxBroker<DB: Database, D = BuiltIn<DB>> { /* private fields */ }

The inbox broker: task queues in the service’s own tables, served through a sqlx pool.

from_url and from_connect_options record the pool’s settings and do no I/O, so the service builds the broker in its app’s builder; connect builds the pool, and shutdown closes it. new takes a pool the service shares with code of its own, and the broker never closes that one. connect takes one connection to check the database, and picks the built-in dialect by it where the service passed none. A message in work holds one of the pool’s connections until it settles (in the lease form outside transactional mode only while it settles), and a publish takes another, so the pool is sized for both. Subscriptions read tables through InboxQueue descriptors; publishing writes them through Repository policies or through the routes this builder records.

Examples

use ruststream::OutgoingMessage;
use ruststream::prelude::*;
use ruststream_sqlx::{Inbox, InboxQueue, Publish, SqlxBroker};
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use sqlx::{PgConnection, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The options read the `PG*` environment variables; the broker builds the pool when the
    // service starts and closes it when the service stops.
    let broker = SqlxBroker::from_connect_options(PgConnectOptions::new(), PgPoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        broker.route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

Testing

A test hands TestApp::start the app main runs, unchanged, and addresses the broker by its production type (feature testing). The broker connects in process to the database the test’s pool reaches, so the service’s SQL runs as in production, and the harness waits for every publish: a row counts for the queue it lands in, its table and group, whatever name a subscription reads it by. SQLite in memory needs no server. After tokio::time::pause() the queue reads “now” from the test’s clock, which tb.advance(d) moves past a delayed retry or a lease; a table on DatabaseClock reads the database’s time instead. The crate registers SqlxBroker<Postgres>, SqlxBroker<MySql>, SqlxBroker<Sqlite> and SqlxBroker<Any> with the harness, and a broker on a dialect of the service’s own is registered by the service (with_dialect).

use std::error::Error;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::testing::TestApp;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{Connection, Sqlite, SqliteConnection, SqlitePool};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "thumbnail_jobs")]
pub struct MakeThumbnail {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Sqlite> for MakeThumbnail {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO thumbnail_jobs (payload) VALUES (?)")
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Serialize, Deserialize, Outgoing)]
pub struct Image {
    path: String,
}

#[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
async fn thumbnail(image: &Image) -> HandlerOutcome {
    tracing::info!(path = %image.path, "making a thumbnail");
    HandlerOutcome::ack()
}

pub fn app(pool: SqlitePool) -> RustStream {
    RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<MakeThumbnail>("thumbnails"),
        |b| {
            b.include(thumbnail);
        },
    )
}

const SCHEMA: &str = "CREATE TABLE thumbnail_jobs (id INTEGER PRIMARY KEY, \
    attempt INTEGER NOT NULL DEFAULT 1, locked_until TEXT, payload BLOB NOT NULL)";

// The body of a `#[tokio::test]`.
pub async fn a_thumbnail_is_made() -> Result<(), Box<dyn Error>> {
    // A database in memory that every connection naming it shares, alive while one of them is
    // open: the test holds that one beside the pool.
    let options: SqliteConnectOptions =
        "sqlite:file:thumbnails?mode=memory&cache=shared".parse()?;
    let _keeper = SqliteConnection::connect_with(&options).await?;
    let pool = SqlitePoolOptions::new()
        .min_connections(1)
        .idle_timeout(None)
        .connect_with(options)
        .await?;
    sqlx::raw_sql(SCHEMA).execute(&pool).await?;

    let tb = TestApp::start(app(pool)).await?;
    // Through the route and the service's own `Publish`, as in production.
    tb.broker::<SqlxBroker<Sqlite>>()
        .message(&Image { path: "cat.png".to_owned() })
        .to("thumbnails")
        .publish()
        .await?;
    tb.broker::<SqlxBroker<Sqlite>>()
        .subscriber("thumbnails")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    Ok(())
}

Implementations

impl<DB: BuiltInDialect> SqlxBroker<DB, BuiltIn<DB>>

pub fn from_url(url: impl Into<String>, pool: PoolOptions<DB>) -> Self

A broker that builds its pool from url and pool when it connects, with the dialect built into the crate for the database.

Synchronous and free of I/O, so the service builds it in its app’s builder, before the runtime starts. connect builds the pool, takes one connection to check the database, and picks the dialect by it; shutdown closes the pool once the broker’s own work is done. The driver reads the URL at connect: a URL it cannot read stops the service’s start with SqlxBrokerError::Url. A pool the service also queries outside its handlers is the service’s own: new takes it.

Examples

use std::env;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::postgres::PgPoolOptions;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let url = env::var("DATABASE_URL").unwrap_or_else(|_| "postgres://localhost/mail".into());
    // Nothing connects here: the broker builds the pool when the service starts.
    let broker = SqlxBroker::from_url(url, PgPoolOptions::new().max_connections(8));
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

pub fn from_connect_options( options: <<DB as Database>::Connection as Connection>::Options, pool: PoolOptions<DB>, ) -> Self

A broker that builds its pool from the driver’s connect options and pool when it connects, with the dialect built into the crate for the database.

The options a service sets in code or reads from the environment (PgConnectOptions::new() reads the PG* variables); otherwise as from_url: synchronous, free of I/O, the pool built at connect and closed at shutdown.

Examples

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The options read the `PG*` environment variables; the application name is the
    // service's own.
    let options = PgConnectOptions::new().application_name("mailer");
    let broker = SqlxBroker::from_connect_options(options, PgPoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

pub fn new(pool: Pool<DB>) -> Self

A broker on the service’s own pool, with the dialect built into the crate for its database.

For a pool the service shares with the rest of its code: an HTTP endpoint that writes rows, a migration run before the app. The broker never closes it; the service does, once the app has stopped. Synchronous and free of I/O: the database type comes from the pool, and nothing connects before connect, which picks the dialect from the connection it checks. An AnyPool reaches a database named only then, and a database whose dialect’s feature is off stops connect with SqlxBrokerError::Backend. sqlx builds a pool only inside a Tokio runtime, so a service that owns one builds it in its own main; a broker that builds its pool itself is from_url.

Examples

use std::error::Error;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPool, PgPoolOptions};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(send);
    })
}

// The service shares the pool with code of its own, so it builds the pool in its `main`,
// inside the runtime, and closes it after the app.
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> {
    // A lazy pool: nothing connects before the service starts. The options read the
    // `PG*` environment variables.
    let pool = PgPoolOptions::new().connect_lazy_with(PgConnectOptions::new());
    let app = app(pool.clone());
    app.run().await?;
    pool.close().await;
    Ok(())
}

impl<DB: QueueDatabase, D: Dialect + 'static> SqlxBroker<DB, D>

pub fn with_dialect(pool: Pool<DB>, dialect: D) -> Self

A broker on pool whose statements dialect, a dialect of the service’s own, builds: a driver this crate has no dialect for, or a built-in dialect the service wraps to write a statement its own way.

The dialect’s type is the broker’s second type parameter, so what a subscription asks of it is checked where the subscription mounts: a table in the row lock form needs the dialect to implement RowLock, a lease table Lease, and a table that names an isolation level or a SQLite mode Opens for it. The dialect runs while subscriptions start; messages travel through the statements it built.

The crate registers the brokers of its built-in dialects with the test harness. A test of a broker on a dialect of the service’s own registers its type once, ruststream::register_testable_broker!(SqlxBroker<Postgres, Audited>);, and addresses it by that type, tb.broker::<SqlxBroker<Postgres, Audited>>().

Examples

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

/// Postgres with statements of the service's own.
#[derive(Debug)]
pub struct Audited;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    // A `SqlxBroker<Postgres, Audited>`: `send` mounts only because `Audited` implements
    // `RowLock`, the form of `email_jobs`.
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        .with_broker(SqlxBroker::with_dialect(pool, Audited), |b| {
            b.include(send);
        })
}

pub fn route<Row>(self, name: impl Into<Cow<'static, str>>) -> Self

where Row: Publish<DB> + Events<DB> + PayloadRow, Row::Form: FormOn<D>, D: Opens<Row::Opening>,

Leads publishes to name into the table of Row, through its Publish, and opens a subscription by that name (#[subscriber("name")]) on the same table.

A name ending in * leads every name that starts with what precedes it; an exact name wins over a prefix, and a longer prefix over a shorter one. Routing a name twice leads it to the last row type. The broker’s default publisher is this route table: a reply or an Out slot without a policy of its own publishes through it.

A route asks of the broker’s dialect what an InboxQueue of the row asks: the table’s form, and the isolation level or SQLite mode the struct declares. A by-name subscription names no row type, so the route is where its table is checked.

Examples

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
pub struct Report {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for Report {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO report_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize)]
struct Daily {
    day: u32,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "reports.weekly")]
struct Weekly {
    week: u32,
}

// Read by name from `report_jobs`, answered with a row of `report_jobs` in group `reports.weekly`.
#[subscriber("reports.daily", reply)]
async fn roll_up(daily: &Daily) -> Weekly {
    Weekly { week: daily.day / 7 }
}

pub fn app(pool: PgPool) -> RustStream {
    // `reports.daily` and `reports.weekly` both lead into `report_jobs`.
    let broker = SqlxBroker::new(pool).route::<Report>("reports.*");
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(roll_up);
    })
}

pub const fn poll_interval(self, interval: Duration) -> Self

How long a subscription waits between claims that found its queue empty; one second unless set. A subscription’s own poll_interval overrides it. A publish through this broker wakes the subscriptions of its table and group before then, and so does a notification with listen_notify (waking a subscription).

Examples

use std::time::Duration;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    // Every queue on this broker checks twice a second when it runs dry.
    let broker = SqlxBroker::new(pool).poll_interval(Duration::from_millis(500));
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

pub const fn lease(self, lease: Duration) -> Self

How long a claim leases a row of a table in the lease form (one with a #[field(locked_until)] field); thirty seconds unless set. A subscription’s own lease overrides it, and a table in another form ignores it.

A lease is whole seconds, at least one: a shorter one is rounded up. The subscription extends the lease of every delivery in work each half lease, so a handler may run longer than its lease; the lease is how long a row stays out of the queue after its process crashed. A delivery dropped unsettled releases its row at once. A lease runs out under a running handler only when its extensions fail or stop (the database out of reach, the subscription closed, the broker shut down): the row then goes to the next claim, and the late settlement fails with SqlxBrokerError::LeaseLost.

Examples

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "video_jobs")]
pub struct Transcode {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Video {
    path: String,
}

#[subscriber(InboxQueue::<Transcode>::new("videos"))]
async fn transcode(video: &Video) -> HandlerOutcome {
    tracing::info!(path = %video.path, "transcoding");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    // A video whose process crashed goes back to the queue once its ten-second lease runs out.
    let broker = SqlxBroker::new(pool).lease(Duration::from_secs(10));
    RustStream::new(AppInfo::new("transcoder", "0.1.0")).with_broker(broker, |b| {
        b.include(transcode);
    })
}

pub const fn connection_limit(self, limit: NonZeroU32) -> Self

The most connections the service opens on the database: the pool’s size, and the connections of the dedicated threads of every subscription mounted with on_threads. No limit unless set.

Each such subscription counts its threads times their connections when it opens. One that would take the sum past the limit refuses to open, with SqlxBrokerError::ConnectionLimit naming the numbers, and the service does not start. Set it to what the database lets this service open.

Examples

use ruststream::nonzero;
use ruststream_sqlx::prelude::*;
use sqlx::{PgPool, Postgres};

#[subscriber(InboxQueue::<ScanJob>::new("scans"))]
async fn scan(image: &Image, Ctx(pool): Ctx<keys::Pool<Postgres>>) -> HandlerOutcome {
    let noted = sqlx::query("INSERT INTO scanned (path) VALUES ($1)")
        .bind(&image.path)
        .execute(&pool)
        .await;
    if noted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    // The database lets this service open 20 connections: the pool's and 4 threads' own.
    let broker = SqlxBroker::new(pool).connection_limit(nonzero!(20u32));
    RustStream::new(AppInfo::new("scanner", "0.1.0")).with_broker(broker, |b| {
        b.include(scan.on_threads(InboxThreads::new(nonzero!(4))));
    })
}

pub fn listen_notify(self) -> Self

where DB: Notifies,

Wakes the broker’s subscriptions on Postgres notifications: a row another process announced with pg_notify is claimed at once, not on the next poll.

The broker holds one connection of the pool for its life, from connect to shutdown, and listens there. Each subscription listens on the channel of its table, qualified with its schema, as it opens; a notification whose payload names a group wakes that group’s subscription, and one with an empty payload wakes every subscription of the table. Each publish of the broker sends SELECT pg_notify(table, group) on the connection that wrote the row, right after the write: one more statement per publish. A publish whose notification fails is still a publish; the row waits for the poll interval and the failure is logged.

The poll interval stays: a notification sent while the listening connection was lost is gone, so after a reconnect every subscription claims once, and a row written by the service’s own SQL or a handler’s Tx wakes nobody unless the service sends the notification itself. It is opt-in because of its cost: the pool connection it holds, the statement each publish adds, and the lock, global to the server, that Postgres takes at the commit of each transaction that notifies, so those commits run one after another, which limits many concurrent writers.

A table whose qualified name is longer than Postgres’s 63-byte channel names stops its subscription at startup with SqlxBrokerError::Declaration.

Examples

use std::time::Duration;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    // A row another service writes and announces with
    // `SELECT pg_notify('email_jobs', 'emails')` is sent at once; the poll each minute
    // catches a row whose notification was lost.
    let broker = SqlxBroker::new(pool)
        .poll_interval(Duration::from_secs(60))
        .listen_notify();
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

Trait Implementations

impl<DB: QueueDatabase, D: Dialect + 'static> Broker for SqlxBroker<DB, D>

type Error = SqlxBrokerError

The error type returned by broker-level operations.

type Connected = ConnectedSqlxBroker<DB, D>

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl<DB: Database, D: Dialect> Debug for SqlxBroker<DB, D>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Postgres, D>

Available on crate feature postgres only.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<MySql, D>

Available on crate feature mysql only.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Sqlite, D>

Available on crate feature sqlite only.

SQLite runs inside the service, on a file or in memory: the description names the protocol and no host, so it holds neither the database’s path nor any other part of its URL, and SQLite takes no credentials.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Any, D>

Available on crate feature any only.

An AnyPool is described by the scheme of its URL, as the broker of that database describes itself: the host and port of a server, or SQLite with no host.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<DB: QueueDatabase, D: Dialect + 'static> InProcess for SqlxBroker<DB, D>

async fn connect_in_process( self, ) -> Result<ConnectedSqlxBroker<DB, D>, SqlxBrokerError>

Connects to the database the test’s pool reaches, as connect does, with every database call of the connection kept off a paused clock.