postgres: fail when the search path no longer resolves to the migrations table

sqlx-core/src/migrate/error.rs

@@ -43,6 +43,37 @@ pub enum MigrateError {
     #[error("database driver does not support creation of schemas at migrate time: {0}")]
     CreateSchemasNotSupported(String),
 
+    /// The migrations table exists, but not where an unqualified reference to it resolves.
+    ///
+    /// Currently only returned by the PostgreSQL driver, where an unqualified name is resolved
+    /// through `search_path`.
+    #[error(
+        "cannot migrate: `{table_name}` does not exist in the default schema `{default_schema}`, \
+         but does exist at: {}.\n\n\
+         This suggests that the search path for the current user or database has changed since \
+         the migrations table was created. This may happen explicitly, or implicitly if a schema \
+         is created with the same name as the user \
+         (https://www.postgresql.org/docs/current/ddl-schemas.html#DDL-SCHEMAS-PATH).\n\n\
+         Since SQLx cannot know which of the existing `{table_name}` tables is correct for the \
+         current context, migration cannot continue.\n\n\
+         To resolve this ambiguity, either change the default search path for this database \
+         (using `ALTER DATABASE`), or create `{default_schema}.{table_name}` with the correct \
+         set of migrations.",
+        .other_schemas
+            .iter()
+            .map(|schema| format!("`{schema}.{table_name}`"))
+            .collect::<Vec<_>>()
+            .join(", ")
+    )]
+    AmbiguousMigrationsTable {
+        /// The unqualified name of the migrations table.
+        table_name: String,
+        /// The schema an unqualified reference to `table_name` currently resolves to.
+        default_schema: String,
+        /// The schemas in the search path that do contain `table_name`, in priority order.
+        other_schemas: Vec<String>,
+    },
+
     #[error("database driver does not support skipping migrations")]
     SkipNotSupported(),
 }

sqlx-postgres/src/migrate.rs

@@ -17,6 +17,12 @@ use crate::query_as::query_as;
 use crate::query_scalar::query_scalar;
 use crate::{PgConnectOptions, PgConnection, Postgres};
 
+/// The default name of the migrations table, as an unqualified identifier.
+///
+/// If the user has picked a different name, or qualified this one, they have taken
+/// responsibility for how it resolves and [`check_migrations_table_resolution()`] is skipped.
+const DEFAULT_MIGRATIONS_TABLE: &str = "_sqlx_migrations";
+
 fn parse_for_maintenance(url: &str) -> Result<(PgConnectOptions, String), Error> {
     let mut options = PgConnectOptions::from_str(url)?;
 
@@ -124,6 +130,10 @@ impl Migrate for PgConnection {
         table_name: &'e str,
     ) -> BoxFuture<'e, Result<(), MigrateError>> {
         Box::pin(async move {
+            if table_name == DEFAULT_MIGRATIONS_TABLE {
+                check_migrations_table_resolution(self, table_name).await?;
+            }
+
             // language=SQL
             self.execute(AssertSqlSafe(format!(
                 r#"
@@ -312,6 +322,63 @@ CREATE TABLE IF NOT EXISTS {table_name} (
     }
 }
 
+/// Check that an unqualified reference to `table_name` still resolves where it was created.
+///
+/// An unqualified name is resolved through `search_path`, which contains `"$user"` by default.
+/// A schema is ignored while it does not exist, so a migration that creates a schema named after
+/// the connecting role makes that schema shadow `public` for every later session. The migrations
+/// table then resolves to the new, empty schema, and every migration is applied a second time.
+///
+/// Returns [`MigrateError::AmbiguousMigrationsTable`] if `table_name` exists somewhere in the
+/// search path but not in the schema an unqualified reference resolves to first, since SQLx
+/// cannot tell which of those tables describes the current context.
+async fn check_migrations_table_resolution(
+    conn: &mut PgConnection,
+    table_name: &str,
+) -> Result<(), MigrateError> {
+    // `current_schemas()` returns the effective search path in priority order, with entries that
+    // name no existing schema omitted, which is exactly what object resolution searches.
+    // language=SQL
+    let search_path: Vec<String> = query_scalar("SELECT current_schemas(false)::text[]")
+        .fetch_one(&mut *conn)
+        .await?;
+
+    // With an empty search path, nothing can shadow anything, and the `CREATE TABLE` that
+    // follows reports the missing target schema itself.
+    let Some(default_schema) = search_path.first() else {
+        return Ok(());
+    };
+
+    // language=SQL
+    let found_in: Vec<String> =
+        query_scalar("SELECT schemaname::text FROM pg_catalog.pg_tables WHERE tablename = $1")
+            .bind(table_name)
+            .fetch_all(&mut *conn)
+            .await?;
+
+    // Keep search path order so that the first entry is the one an unqualified name resolves to.
+    let mut found_in_search_path = search_path
+        .iter()
+        .filter(|schema| found_in.contains(schema));
+
+    match found_in_search_path.next() {
+        // The table does not exist yet, so it is about to be created in `default_schema` and
+        // will resolve there. A later move is caught on the next run.
+        None => Ok(()),
+        // The table already resolves to `default_schema`.
+        Some(schema) if schema == default_schema => Ok(()),
+        // The table exists, but an unqualified reference no longer points at it.
+        Some(schema) => Err(MigrateError::AmbiguousMigrationsTable {
+            table_name: table_name.to_owned(),
+            default_schema: default_schema.clone(),
+            other_schemas: std::iter::once(schema)
+                .chain(found_in_search_path)
+                .cloned()
+                .collect(),
+        }),
+    }
+}
+
 async fn execute_migration(
     conn: &mut PgConnection,
     table_name: &str,

tests/postgres/migrate.rs

@@ -1,8 +1,9 @@
-use sqlx::migrate::Migrator;
+use sqlx::migrate::{MigrateError, Migration, MigrationType, Migrator};
 use sqlx::pool::PoolConnection;
-use sqlx::postgres::{PgConnection, Postgres};
+use sqlx::postgres::{PgConnection, PgPool, Postgres};
 use sqlx::Executor;
 use sqlx::Row;
+use sqlx::{AssertSqlSafe, ConnectOptions, Connection, SqlSafeStr};
 use std::path::Path;
 
 #[sqlx::test(migrations = false)]
@@ -130,6 +131,111 @@ async fn no_tx(mut conn: PoolConnection<Postgres>) -> anyhow::Result<()> {
     Ok(())
 }
 
+/// A migration that creates a schema named after the connecting role makes that schema shadow
+/// `public` in the default search path of every session that starts afterwards. An unqualified
+/// reference to `_sqlx_migrations` then resolves to the new, empty schema.
+///
+/// The migrator must report the ambiguity instead of creating a second migrations table and
+/// applying every migration a second time.
+#[sqlx::test(migrations = false)]
+async fn migrations_table_shadowed_by_role_schema(pool: PgPool) -> anyhow::Result<()> {
+    let connect_options = pool.connect_options();
+
+    let role: String = sqlx::query_scalar("SELECT current_user")
+        .fetch_one(&pool)
+        .await?;
+
+    let migrator = Migrator::with_migrations(vec![
+        Migration::new(
+            1,
+            "create role schema".into(),
+            MigrationType::Simple,
+            AssertSqlSafe(format!(r#"CREATE SCHEMA "{role}""#)).into_sql_str(),
+            false,
+        ),
+        Migration::new(
+            2,
+            "add table".into(),
+            MigrationType::Simple,
+            AssertSqlSafe(format!(
+                r#"CREATE TABLE "{role}".migrations_shadowed_test (id INT PRIMARY KEY)"#
+            ))
+            .into_sql_str(),
+            false,
+        ),
+    ]);
+
+    // The schema only shadows `public` for sessions that start after it exists,
+    // so each run needs its own connection.
+    let mut conn = connect_options.connect().await?;
+    migrator.run(&mut conn).await?;
+    conn.close().await?;
+
+    let mut conn = connect_options.connect().await?;
+    let error = migrator
+        .run(&mut conn)
+        .await
+        .expect_err("second run did not detect the shadowed migrations table");
+    conn.close().await?;
+
+    assert!(
+        matches!(error, MigrateError::AmbiguousMigrationsTable { .. }),
+        "unexpected error: {error:?}"
+    );
+
+    // The second run must not have left an empty migrations table behind in the new schema.
+    let tracking_tables: Vec<String> = sqlx::query_scalar(
+        "SELECT schemaname::text FROM pg_catalog.pg_tables \
+         WHERE tablename = '_sqlx_migrations' ORDER BY schemaname",
+    )
+    .fetch_all(&pool)
+    .await?;
+
+    assert_eq!(tracking_tables, ["public"]);
+
+    Ok(())
+}
+
+/// An explicitly chosen table name is the user's responsibility, so the shadowing check is
+/// skipped for it and migrations keep running against the name as written.
+#[sqlx::test(migrations = false)]
+async fn qualified_migrations_table_ignores_role_schema(pool: PgPool) -> anyhow::Result<()> {
+    let connect_options = pool.connect_options();
+
+    let role: String = sqlx::query_scalar("SELECT current_user")
+        .fetch_one(&pool)
+        .await?;
+
+    let mut migrator = Migrator::with_migrations(vec![Migration::new(
+        1,
+        "create role schema".into(),
+        MigrationType::Simple,
+        AssertSqlSafe(format!(r#"CREATE SCHEMA "{role}""#)).into_sql_str(),
+        false,
+    )]);
+    migrator.dangerous_set_table_name("public._sqlx_migrations");
+
+    let mut conn = connect_options.connect().await?;
+    migrator.run(&mut conn).await?;
+    conn.close().await?;
+
+    // A qualified name cannot move, so the second run is a no-op.
+    let mut conn = connect_options.connect().await?;
+    migrator.run(&mut conn).await?;
+    conn.close().await?;
+
+    let tracking_tables: Vec<String> = sqlx::query_scalar(
+        "SELECT schemaname::text FROM pg_catalog.pg_tables \
+         WHERE tablename = '_sqlx_migrations' ORDER BY schemaname",
+    )
+    .fetch_all(&pool)
+    .await?;
+
+    assert_eq!(tracking_tables, ["public"]);
+
+    Ok(())
+}
+
 /// Ensure that we have a clean initial state.
 async fn clean_up(conn: &mut PgConnection) -> anyhow::Result<()> {
     conn.execute("DROP DATABASE IF EXISTS test_db").await.ok();