Fix running multiple migration process in parallel

crates/cli/src/commands/config.rs

@@ -12,10 +12,9 @@ use clap::Parser;
 use figment::Figment;
 use mas_config::{ConfigurationSection, RootConfig, SyncConfig};
 use mas_data_model::{Clock as _, SystemClock};
-use mas_storage_pg::MIGRATOR;
 use rand::SeedableRng;
 use tokio::io::AsyncWriteExt;
-use tracing::{Instrument, info, info_span};
+use tracing::{info, info_span};
 
 use crate::util::database_connection_from_config;
 
@@ -129,9 +128,7 @@ impl Options {
                 // Grab a connection to the database
                 let mut conn = database_connection_from_config(&config.database).await?;
 
-                MIGRATOR
-                    .run(&mut conn)
-                    .instrument(info_span!("db.migrate"))
+                mas_storage_pg::migrate(&mut conn)
                     .await
                     .context("could not run migrations")?;
 

crates/cli/src/commands/database.rs

@@ -10,8 +10,7 @@ use anyhow::Context;
 use clap::Parser;
 use figment::Figment;
 use mas_config::{ConfigurationSectionExt, DatabaseConfig};
-use mas_storage_pg::MIGRATOR;
-use tracing::{Instrument, info_span};
+use tracing::info_span;
 
 use crate::util::database_connection_from_config;
 
@@ -35,9 +34,7 @@ impl Options {
         let mut conn = database_connection_from_config(&config).await?;
 
         // Run pending migrations
-        MIGRATOR
-            .run(&mut conn)
-            .instrument(info_span!("db.migrate"))
+        mas_storage_pg::migrate(&mut conn)
             .await
             .context("could not run migrations")?;
 

crates/cli/src/commands/server.rs

@@ -4,7 +4,7 @@
 // SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 // Please see LICENSE files in the repository root for full details.
 
-use std::{collections::BTreeSet, process::ExitCode, sync::Arc, time::Duration};
+use std::{process::ExitCode, sync::Arc, time::Duration};
 
 use anyhow::Context;
 use clap::Parser;
@@ -18,9 +18,8 @@ use mas_data_model::SystemClock;
 use mas_handlers::{ActivityTracker, CookieManager, Limiter, MetadataCache};
 use mas_listener::server::Server;
 use mas_router::UrlBuilder;
-use mas_storage_pg::{MIGRATOR, PgRepositoryFactory};
-use sqlx::migrate::Migrate;
-use tracing::{Instrument, info, info_span, warn};
+use mas_storage_pg::PgRepositoryFactory;
+use tracing::{info, info_span, warn};
 
 use crate::{
     app_state::AppState,
@@ -73,24 +72,20 @@ impl Options {
         let pool = database_pool_from_config(&config.database).await?;
 
         if self.no_migrate {
-            // Check that we applied all the migrations
             let mut conn = pool.acquire().await?;
-            let applied = conn.list_applied_migrations().await?;
-            let applied: BTreeSet<_> = applied.into_iter().map(|m| m.version).collect();
-            let has_missing_migrations = MIGRATOR.iter().any(|m| !applied.contains(&m.version));
-            if has_missing_migrations {
+            let pending_migrations = mas_storage_pg::pending_migrations(&mut conn).await?;
+            if !pending_migrations.is_empty() {
                 // Refuse to start if there are pending migrations
                 return Err(anyhow::anyhow!(
-                    "The server is running with `--no-migrate` but there are pending. Please run them first with `mas-cli database migrate`, or omit the `--no-migrate` flag to apply them automatically on startup."
+                    "The server is running with `--no-migrate` but there are pending migrations. Please run them first with `mas-cli database migrate`, or omit the `--no-migrate` flag to apply them automatically on startup."
                 ));
             }
         } else {
             info!("Running pending database migrations");
-            MIGRATOR
-                .run(&pool)
-                .instrument(info_span!("db.migrate"))
+            let mut conn = pool.acquire().await?;
+            mas_storage_pg::migrate(&mut conn)
                 .await
-                .context("could not run database migrations")?;
+                .context("could not run migrations")?;
         }
 
         let encrypter = config.secrets.encrypter().await?;

crates/cli/src/commands/syn2mas.rs

@@ -14,13 +14,12 @@ use mas_config::{
     UpstreamOAuth2Config,
 };
 use mas_data_model::SystemClock;
-use mas_storage_pg::MIGRATOR;
 use rand::thread_rng;
 use sqlx::{Connection, Either, PgConnection, postgres::PgConnectOptions, types::Uuid};
 use syn2mas::{
     LockedMasDatabase, MasWriter, Progress, ProgressStage, SynapseReader, synapse_config,
 };
-use tracing::{Instrument, error, info, info_span};
+use tracing::{Instrument, error, info};
 
 use crate::util::{DatabaseConnectOptions, database_connection_from_config_with_options};
 
@@ -122,9 +121,7 @@ impl Options {
         )
         .await?;
 
-        MIGRATOR
-            .run(&mut mas_connection)
-            .instrument(info_span!("db.migrate"))
+        mas_storage_pg::migrate(&mut mas_connection)
             .await
             .context("could not run migrations")?;
 

crates/storage-pg/Cargo.toml

@@ -19,6 +19,7 @@ workspace = true
 [dependencies]
 async-trait.workspace = true
 chrono.workspace = true
+crc.workspace = true
 futures-util.workspace = true
 opentelemetry-semantic-conventions.workspace = true
 opentelemetry.workspace = true
@@ -31,6 +32,7 @@ sha2.workspace = true
 sqlx.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
+tokio.workspace = true
 ulid.workspace = true
 url.workspace = true
 uuid.workspace = true