isolate underway migrations

This is something of a hack to work around the fact that SQLx migrations
do not currently support specifying a schema under which the migrations
table will live.

Here we provide a search path throughout our migrations and in the
transaction that will run the migrations to ensure that migrations are
applied to `underway._sqlx_migrations`. Note that this assumes a
`public._sqlx_migrations` exists.

In the future we should be able to use `sqlx.toml` to address this more
robustly. That's expected as part of the `0.9.0` release of SQLx. Please
see: launchbadge/sqlx#3383

Closes #11

README.md

@@ -73,7 +73,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()
@@ -148,7 +148,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()
@@ -207,7 +207,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()

examples/basic/src/main.rs

@@ -20,7 +20,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()

examples/graceful_shutdown/src/main.rs

@@ -51,7 +51,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()

examples/multitask/src/main.rs

@@ -155,7 +155,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Create the task queue.
     let queue = Queue::builder()

examples/rag/src/main.rs

@@ -152,7 +152,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let database_url = &env::var("DATABASE_URL").expect("DATABASE_URL should be set");
     let pool = PgPool::connect(database_url).await?;
 
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     let openai_client = Client::new();
 

examples/scheduled/src/main.rs

@@ -12,7 +12,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(&database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()

examples/step/src/main.rs

@@ -35,7 +35,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Create our job.
     let job = Job::builder()

examples/tracing/src/main.rs

@@ -29,7 +29,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = PgPool::connect(database_url).await?;
 
     // Run migrations.
-    underway::MIGRATOR.run(&pool).await?;
+    underway::run_migrations(&pool).await?;
 
     // Build the job.
     let job = Job::builder()

migrations/20240921151751_0.sql

@@ -1,5 +1,12 @@
 create schema if not exists underway;
 
+-- Make sure we're in the Underway schema for the migration.
+set search_path to underway;
+
+-- Manage Underway migrations within the Underway schema.
+create table if not exists underway._sqlx_migrations
+(like public._sqlx_migrations including all);
+
 create table underway.task_queue (
     name       text not null,
     dlq_name   text,

migrations/20241024174106_1.sql

@@ -1,3 +1,6 @@
+-- Make sure we're in the Underway schema for the migration.
+set search_path to underway;
+
 -- function to notify about task changes
 create or replace function underway.task_change_notify()
 returns trigger as $$

src/lib.rs

@@ -60,7 +60,7 @@
 //!     let pool = PgPool::connect(database_url).await?;
 //!
 //!     // Run migrations.
-//!     underway::MIGRATOR.run(&pool).await?;
+//!     underway::run_migrations(&pool).await?;
 //!
 //!     // Build the job.
 //!     let job = Job::builder()
@@ -135,7 +135,7 @@
 //!     let pool = PgPool::connect(database_url).await?;
 //!
 //!     // Run migrations.
-//!     underway::MIGRATOR.run(&pool).await?;
+//!     underway::run_migrations(&pool).await?;
 //!
 //!     // Build the job.
 //!     let job = Job::builder()
@@ -194,7 +194,7 @@
 //!     let pool = PgPool::connect(database_url).await?;
 //!
 //!     // Run migrations.
-//!     underway::MIGRATOR.run(&pool).await?;
+//!     underway::run_migrations(&pool).await?;
 //!
 //!     // Build the job.
 //!     let job = Job::builder()
@@ -265,7 +265,7 @@
 
 #![warn(clippy::all, nonstandard_style, future_incompatible, missing_docs)]
 
-use sqlx::migrate::Migrator;
+use sqlx::{migrate::Migrator, PgPool};
 
 pub use crate::{
     job::{Job, To},
@@ -281,7 +281,16 @@ mod scheduler;
 pub mod task;
 pub mod worker;
 
-/// A SQLx [`Migrator`] which provides Underway's schema migrations.
+static MIGRATOR: Migrator = sqlx::migrate!();
+
+/// Runs Underway migrations.
+///
+/// A transaction is acquired via the provided pool and migrations are run via
+/// this transaction.
+///
+/// As no direct support for specifying the schema under which the migrations
+/// table will live, we manually specify this via the search path. This ensures
+/// that migrations are isolated to underway._sqlx_migrations.
 ///
 /// These migrations must be applied before queues, tasks, and workers can be
 /// run.
@@ -304,8 +313,21 @@ pub mod worker;
 /// let pool = PgPool::connect(database_url).await?;
 ///
 /// // Run migrations.
-/// underway::MIGRATOR.run(&pool).await?;
+/// underway::run_migrations(&pool).await?;
 /// # Ok::<(), Box<dyn std::error::Error>>(())
 /// # });
 /// # }
-pub static MIGRATOR: Migrator = sqlx::migrate!();
+pub async fn run_migrations(pool: &PgPool) -> Result<(), sqlx::Error> {
+    let mut tx = pool.begin().await?;
+
+    // Manually specify "underway" search path to ensure the right migrations table
+    // is used.
+    sqlx::raw_sql("create schema if not exists underway; set search_path to underway;")
+        .execute(&mut *tx)
+        .await?;
+    MIGRATOR.run(&mut *tx).await?;
+
+    tx.commit().await?;
+
+    Ok(())
+}