WIP feat: filling out axum-multi-tenant example

Author: abonander

Cargo.lock

@@ -12,7 +12,17 @@ authors.workspace = true
 accounts = { path = "accounts" }
 payments = { path = "payments" }
 
+tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
+
 sqlx = { path = "../../..", version = "0.8.3", features = ["runtime-tokio", "postgres"] }
 
+axum = "0.8.1"
+
+clap = { version = "4.5.30", features = ["derive", "env"] }
+color-eyre = "0.6.3"
+dotenvy = "0.15.7"
+tracing-subscriber = "0.3.19"
+
+
 [lints]
 workspace = true

examples/postgres/axum-multi-tenant/Cargo.toml

@@ -3,9 +3,20 @@
 This example project involves three crates, each owning a different schema in one database,
 with their own set of migrations.
 
-* The main crate, an Axum app. 
-  * Owns the `public` schema (tables are referenced unqualified).
+* The main crate, an Axum app.
+    * Owns the `public` schema (tables are referenced unqualified).
 * `accounts`: a subcrate simulating a reusable account-management crate.
-  * Owns schema `accounts`.
+    * Owns schema `accounts`.
 * `payments`: a subcrate simulating a wrapper for a payments API.
-  * Owns schema `payments`.
+    * Owns schema `payments`.
+
+## Note: Schema-Qualified Names
+
+This example uses schema-qualified names everywhere for clarity.
+
+It can be tempting to change the `search_path` of the connection (MySQL, Postgres) to eliminate the need for schema
+prefixes, but this can cause some really confusing issues when names conflict.
+
+This example will generate a `_sqlx_migrations` table in three different schemas, and if `search_path` is set
+to `public,accounts,payments` and the migrator for the main application attempts to reference the table unqualified,
+it would throw an error.

examples/postgres/axum-multi-tenant/README.md

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-sqlx = { workspace = true, features = ["postgres", "time", "uuid"] }
+sqlx = { workspace = true, features = ["postgres", "time", "uuid", "macros", "sqlx-toml"] }
 tokio = { version = "1", features = ["rt", "sync"] }
 
 argon2 = { version = "0.5.3", features = ["password-hash"] }
@@ -13,3 +13,8 @@ password-hash = { version = "0.5", features = ["std"] }
 uuid = "1"
 thiserror = "1"
 rand = "0.8"
+
+time = "0.3.37"
+
+[dev-dependencies]
+sqlx = { workspace = true, features = ["runtime-tokio"] }

examples/postgres/axum-multi-tenant/accounts/Cargo.toml

@@ -0,0 +1,30 @@
+-- We try to ensure every table has `created_at` and `updated_at` columns, which can help immensely with debugging
+-- and auditing.
+--
+-- While `created_at` can just be `default now()`, setting `updated_at` on update requires a trigger which
+-- is a lot of boilerplate. These two functions save us from writing that every time as instead we can just do
+--
+-- select accounts.trigger_updated_at('<table name>');
+--
+-- after a `CREATE TABLE`.
+create or replace function accounts.set_updated_at()
+    returns trigger as
+$$
+begin
+    NEW.updated_at = now();
+return NEW;
+end;
+$$ language plpgsql;
+
+create or replace function accounts.trigger_updated_at(tablename regclass)
+    returns void as
+$$
+begin
+execute format('CREATE TRIGGER set_updated_at
+        BEFORE UPDATE
+        ON %s
+        FOR EACH ROW
+        WHEN (OLD is distinct from NEW)
+    EXECUTE FUNCTION accounts.set_updated_at();', tablename);
+end;
+$$ language plpgsql;

examples/postgres/axum-multi-tenant/accounts/migrations/01_setup.sql

@@ -1,8 +1,10 @@
 create table accounts.account
 (
-    account_id uuid primary key default gen_random_uuid(),
-    email text unique not null,
-    password_hash text not null,
-    created_at timestamptz not null default now(),
-    updated_at timestamptz
+    account_id    uuid primary key     default gen_random_uuid(),
+    email         text unique not null,
+    password_hash text        not null,
+    created_at    timestamptz not null default now(),
+    updated_at    timestamptz
 );
+
+select accounts.trigger_updated_at('accounts.account');

examples/postgres/axum-multi-tenant/accounts/migrations/02_account.sql

@@ -1,6 +1,6 @@
 [migrate]
 create-schemas = ["accounts"]
-migrations-table = "accounts._sqlx_migrations"
+table-name = "accounts._sqlx_migrations"
 
 [macros.table-overrides.'accounts.account']
 'account_id' = "crate::AccountId"

examples/postgres/axum-multi-tenant/accounts/sqlx.toml

@@ -1,11 +1,9 @@
 use argon2::{password_hash, Argon2, PasswordHasher, PasswordVerifier};
-use std::error::Error;
 use std::sync::Arc;
 
 use password_hash::PasswordHashString;
 
-use sqlx::{PgConnection, PgTransaction};
-use sqlx::types::Text;
+use sqlx::{PgConnection, PgPool, PgTransaction};
 
 use uuid::Uuid;
 
@@ -16,6 +14,37 @@ use tokio::sync::Semaphore;
 pub struct AccountId(pub Uuid);
 
 pub struct AccountsManager {
+    /// Controls how many blocking tasks are allowed to run concurrently for Argon2 hashing.
+    ///
+    /// ### Motivation
+    /// Tokio blocking tasks are generally not designed for CPU-bound work.
+    ///
+    /// If no threads are idle, Tokio will automatically spawn new ones to handle
+    /// new blocking tasks up to a very high limit--512 by default.
+    ///
+    /// This is because blocking tasks are expected to spend their time *blocked*, e.g. on
+    /// blocking I/O, and thus not consume CPU resources or require a lot of context switching.
+    ///
+    /// This strategy is not the most efficient way to use threads for CPU-bound work, which
+    /// should schedule work to a fixed number of threads to minimize context switching
+    /// and memory usage (each new thread needs significant space allocated for its stack).
+    ///
+    /// We can work around this by using a purpose-designed thread-pool, like Rayon,
+    /// but we still have the problem that those APIs usually are not designed to support `async`,
+    /// so we end up needing blocking tasks anyway, or implementing our own work queue using
+    /// channels. Rayon also does not shut down idle worker threads.
+    ///
+    /// `block_in_place` is not a silver bullet, either, as it simply uses `spawn_blocking`
+    /// internally to take over from the current thread while it is executing blocking work.
+    /// This also prevents futures from being polled concurrently in the current task.
+    ///
+    /// We can lower the limit for blocking threads when creating the runtime, but this risks
+    /// starving other blocking tasks that are being created by the application or the Tokio
+    /// runtime itself
+    /// (which are used for `tokio::fs`, stdio, resolving of hostnames by `ToSocketAddrs`, etc.).
+    ///
+    /// Instead, we can just use a Semaphore to limit how many blocking tasks are spawned at once,
+    /// emulating the behavior of a thread pool like Rayon without needing any additional crates.
     hashing_semaphore: Arc<Semaphore>,
 }
 
@@ -57,7 +86,7 @@ pub enum GeneralError {
     PasswordHash(
         #[source]
         #[from]
-        argon2::password_hash::Error,
+        password_hash::Error,
     ),
     #[error("task panicked")]
     Task(
@@ -68,12 +97,9 @@ pub enum GeneralError {
 }
 
 impl AccountsManager {
-    pub async fn new(
-        conn: &mut PgConnection,
-        max_hashing_threads: usize,
-    ) -> Result<Self, GeneralError> {
+    pub async fn setup(pool: &PgPool, max_hashing_threads: usize) -> Result<Self, GeneralError> {
         sqlx::migrate!()
-            .run(conn)
+            .run(pool)
             .await
             .map_err(sqlx::Error::from)?;
 
@@ -147,8 +173,8 @@ impl AccountsManager {
         let hash = self.hash_password(password).await?;
 
         // Thanks to `sqlx.toml`, `account_id` maps to `AccountId`
-        // language=PostgreSQL
         sqlx::query_scalar!(
+            // language=PostgreSQL
             "insert into accounts.account(email, password_hash) \
              values ($1, $2) \
              returning account_id",
@@ -158,7 +184,9 @@ impl AccountsManager {
         .fetch_one(&mut **txn)
         .await
         .map_err(|e| {
-            if e.as_database_error().and_then(|dbe| dbe.constraint()) == Some("account_account_id_key") {
+            if e.as_database_error().and_then(|dbe| dbe.constraint())
+                == Some("account_account_id_key")
+            {
                 CreateError::EmailInUse
             } else {
                 GeneralError::from(e).into()
@@ -193,7 +221,8 @@ impl AccountsManager {
             return Err(AuthenticateError::UnknownEmail);
         };
 
-        self.verify_password(password, account.password_hash.into_inner()).await?;
+        self.verify_password(password, account.password_hash.into_inner())
+            .await?;
 
         Ok(account.account_id)
     }

examples/postgres/axum-multi-tenant/accounts/src/lib.rs

@@ -4,4 +4,14 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-sqlx = { workspace = true, features = ["postgres", "time", "uuid"] }
+accounts = { path = "../accounts" }
+
+sqlx = { workspace = true, features = ["postgres", "time", "uuid", "rust_decimal", "sqlx-toml"] }
+
+rust_decimal = "1.36.0"
+
+time = "0.3.37"
+uuid = "1.12.1"
+
+[dev-dependencies]
+sqlx = { workspace = true, features = ["runtime-tokio"] }

examples/postgres/axum-multi-tenant/payments/Cargo.toml

@@ -0,0 +1,30 @@
+-- We try to ensure every table has `created_at` and `updated_at` columns, which can help immensely with debugging
+-- and auditing.
+--
+-- While `created_at` can just be `default now()`, setting `updated_at` on update requires a trigger which
+-- is a lot of boilerplate. These two functions save us from writing that every time as instead we can just do
+--
+-- select payments.trigger_updated_at('<table name>');
+--
+-- after a `CREATE TABLE`.
+create or replace function payments.set_updated_at()
+    returns trigger as
+$$
+begin
+    NEW.updated_at = now();
+return NEW;
+end;
+$$ language plpgsql;
+
+create or replace function payments.trigger_updated_at(tablename regclass)
+    returns void as
+$$
+begin
+execute format('CREATE TRIGGER set_updated_at
+        BEFORE UPDATE
+        ON %s
+        FOR EACH ROW
+        WHEN (OLD is distinct from NEW)
+    EXECUTE FUNCTION payments.set_updated_at();', tablename);
+end;
+$$ language plpgsql;