迁移

[migrate]
table-name = "tx._sqlx_migrations"

MIGRATOR.table_name = "tx._sqlx_migrations"
  1 "widgets" ReversibleUp no_tx=false checksum 48 字节
  1 "widgets" ReversibleDown no_tx=false checksum 48 字节
  2 "widget color" ReversibleUp no_tx=false checksum 48 字节
  2 "widget color" ReversibleDown no_tx=false checksum 48 字节
  3 "widgets name idx" ReversibleUp no_tx=true checksum 48 字节
  3 "widgets name idx" ReversibleDown no_tx=true checksum 48 字节

static MIGRATOR: sqlx::migrate::Migrator = {
    ::sqlx::migrate::Migrator {
        migrations: ::std::borrow::Cow::Borrowed(const {
            &[
                ::sqlx::migrate::Migration {
                    version: 1i64,
                    description: ::std::borrow::Cow::Borrowed("widgets"),
                    migration_type: ::sqlx::migrate::MigrationType::ReversibleUp,
                    sql: ::sqlx::SqlStr::from_static(
                        "create table tx.widgets (id int8 primary key, name text not null);\n",
                    ),
                    no_tx: false,
                    checksum: ::std::borrow::Cow::Borrowed(
                        &[
                            35u8, 247u8, 221u8, 183u8, 173u8, 171u8, 232u8, 204u8, 216u8,
                            // ...
                        ],
                    ),
                },
                // ...
            ]
        }),
        create_schemas: ::std::borrow::Cow::Borrowed(&[]),
        table_name: ::std::borrow::Cow::Borrowed("tx._sqlx_migrations"),
        ..::sqlx::migrate::Migrator::DEFAULT
    }
};

        // lock the database for exclusive access by the migrator
        if self.locking {
            conn.lock().await?;
        }

        for schema_name in self.create_schemas.iter() {
            conn.create_schema_if_not_exists(schema_name).await?;
        }

        // creates [_migrations] table only if needed
        // eventually this will likely migrate previous versions of the table
        conn.ensure_migrations_table(&self.table_name).await?;

        let version = conn.dirty_version(&self.table_name).await?;
        if let Some(version) = version {
            return Err(MigrateError::Dirty(version));
        }

        let applied_migrations = conn.list_applied_migrations(&self.table_name).await?;
        validate_applied_migrations(&applied_migrations, self)?;
        // ...
        for migration in self.iter() {
            if target.is_some_and(|target| target < migration.version) {
                // Target version reached
                break;
            }

            if migration.migration_type.is_down_migration() {
                continue;
            }

            match applied_migrations.get(&migration.version) {
                Some(applied_migration) => {
                    if migration.checksum != applied_migration.checksum {
                        return Err(MigrateError::VersionMismatch(migration.version));
                    }
                }
                None => {
                    if skip {
                        conn.skip(&self.table_name, migration).await?;
                    } else {
                        conn.apply(&self.table_name, migration).await?;
                    }
                }
            }
        }

        // unlock the migrator to allow other migrators to run
        // but do nothing as we already migrated
        if self.locking {
            conn.unlock().await?;
        }

current_database = "lab", crc32 = 0x61d6b1c4, lock key = 1685338728340073208
--- 第一次 MIGRATOR.run（锁被别人拿着）---
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_1, "SELECT current_database()"), Describe(S, sqlx_s_1), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 3), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_2, "SELECT pg_advisory_lock($1)"), Describe(S, sqlx_s_2), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, RowDescription, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_2), Execute, Close(P, unnamed), Sync
pg_locks 等待者: classid=392398500 objid=3840617208 objsubid=1 granted=false => (classid<<32)|objid = 1685338728340073208
释放我们手里的锁
  [conn 1] S->C BindComplete, DataRow(列长 0), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Query("\nCREATE TABLE IF NOT EXISTS tx._sqlx_migrations (\n    version BIGINT PRIMARY KEY,\n    description TEXT NOT NULL,\n    installed_on TIMESTAMPTZ NOT NULL DEFAULT now(),\n    success BOOLEAN NOT NULL,\n    checksum BYTEA NOT NULL,\n    execution_time BIGINT NOT NULL\n);\n                ")
  [conn 1] S->C CommandComplete("CREATE TABLE"), ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_3, "SELECT version FROM tx._sqlx_migrations WHERE success = false ORDER BY version LIMIT 1"), Describe(S, sqlx_s_3), Sync
  ...
  [conn 1] C->S Parse(sqlx_s_4, "SELECT version, checksum FROM tx._sqlx_migrations ORDER BY version"), Describe(S, sqlx_s_4), Sync
  ...
  [conn 1] C->S Query("BEGIN")
  [conn 1] S->C CommandComplete("BEGIN"), ReadyForQuery(T)
  [conn 1] C->S Query("create table tx.widgets (id int8 primary key, name text not null);\n")
  [conn 1] S->C CommandComplete("CREATE TABLE"), ReadyForQuery(T)
  [conn 1] C->S Parse(sqlx_s_5, "\n    INSERT INTO tx._sqlx_migrations ( version, description, success, checksum, execution_time )\n    VALUES ( $1, $2, TRUE, $3, -1 )\n                "), Describe(S, sqlx_s_5), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, NoData, ReadyForQuery(T)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_5), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(T)
  [conn 1] C->S Query("COMMIT")
  [conn 1] S->C CommandComplete("COMMIT"), ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_6, "\n    UPDATE tx._sqlx_migrations\n    SET execution_time = $1\n    WHERE version = $2\n                "), Describe(S, sqlx_s_6), Sync
  [conn 1] S->C ParseComplete, ParameterDescription, NoData, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_6), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("UPDATE 1"), CloseComplete, ReadyForQuery(I)
  ... (2 号迁移同上)
  [conn 1] C->S Query("-- no-transaction\ncreate index concurrently widgets_name_idx on tx.widgets (name);\n")
  [conn 1] S->C CommandComplete("CREATE INDEX"), ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_5), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("INSERT 0 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_6), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, CommandComplete("UPDATE 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Bind(portal unnamed, stmt sqlx_s_1), Execute, Close(P, unnamed), Sync
  [conn 1] S->C BindComplete, DataRow(列长 3), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 1] C->S Parse(sqlx_s_7, "SELECT pg_advisory_unlock($1)"), Describe(S, sqlx_s_7), Sync
  ...
  [conn 1] C->S Sync
  [conn 1] S->C ReadyForQuery(I)
--- 第一次 run 返回 ---

tx._sqlx_migrations:
  (1, "widgets", true, "23f7ddb7adabe8cc", 48, 7641125)
  (2, "widget color", true, "8dada9d67fce8b19", 48, 1685000)
  (3, "widgets name idx", true, "d321ecb11b0fce22", 48, 3436042)
  列 ("version", "bigint", "NO")
  列 ("description", "text", "NO")
  列 ("installed_on", "timestamp with time zone", "NO")
  列 ("success", "boolean", "NO")
  列 ("checksum", "bytea", "NO")
  列 ("execution_time", "bigint", "NO")
运行后 (public, tx) = (None, Some("tx._sqlx_migrations"))
MIGRATOR 里 1 号的 checksum 前 8 字节 = Some("23f7ddb7adabe8cc")

    fn lock(&mut self) -> BoxFuture<'_, Result<(), MigrateError>> {
        Box::pin(async move {
            let database_name = current_database(self).await?;
            let lock_id = generate_lock_id(&database_name);

            // create an application lock over the database
            // this function will not return until the lock is acquired
            // ...
            let _ = query("SELECT pg_advisory_lock($1)")
                .bind(lock_id)
                .execute(self)
                .await?;

            Ok(())
        })
    }

// inspired from rails: https://github.com/rails/rails/blob/6e49cc77ab3d16c06e12f93158eaf3e507d4120e/activerecord/lib/active_record/migration.rb#L1308
fn generate_lock_id(database_name: &str) -> i64 {
    const CRC_IEEE: crc::Crc<u32> = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC);
    // 0x3d32ad9e chosen by fair dice roll
    0x3d32ad9e * (CRC_IEEE.checksum(database_name.as_bytes()) as i64)
}

--- 1. 改一个已应用的迁移 ---
运行期 Migrator 的 table_name = "_sqlx_migrations"
Display: migration 1 was previously applied but has been modified
Debug:   VersionMismatch(1)
出错之后池里那条连接 (pid 361) 持有的 advisory 锁: [(361, true)]

            // execute migration queries
            if migration.no_tx {
                execute_migration(self, table_name, migration).await?;
            } else {
                // Use a single transaction for the actual migration script and the essential bookkeeping so we never
                // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
                // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
                // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
                // and update it once the actual transaction completed.
                let mut tx = self.begin().await?;
                execute_migration(&mut tx, table_name, migration).await?;
                tx.commit().await?;
            }

--- 3. CREATE INDEX CONCURRENTLY 没写 -- no-transaction ---
Display: while executing migration 2: error returned from database: CREATE INDEX CONCURRENTLY cannot run inside a transaction block at line 3497
tx._sqlx_migrations_err: [(1, true)]

        // opt-out of migration transaction
        let no_tx = sql.starts_with("-- no-transaction");

--- 4. 行首不是 -- no-transaction ---
  1 no_tx = true
  2 no_tx = false
  3 no_tx = false
  4 no_tx = false

no_tx = true
Display: while executing migration 1: error returned from database: CREATE INDEX CONCURRENTLY cannot run inside a transaction block at line 3497
tx.multi 上的索引: []

Found `sqlx.toml` in current directory; reading...
error: SQLx found config file at "sqlx.toml" but the `sqlx-toml` feature was not enabled

$ sqlx migrate info
1/installed widgets
2/installed widget color
3/installed widgets name idx
$ sqlx migrate revert
Applied 3/revert widgets name idx (3.78075ms)
$ sqlx migrate info
1/installed widgets
2/installed widget color
3/pending widgets name idx
$ sqlx migrate revert --target-version 0
Applied 2/revert widget color (2.421542ms)
Applied 1/revert widgets (2.514375ms)
$ sqlx migrate info
1/pending widgets
2/pending widget color
3/pending widgets name idx

let mut ws = ResolveConfig::new();
ws.ignore_chars([' ', '\t', '\r', '\n']);
let b = ResolveWith(dir.clone(), ws).resolve().await.map_err(|e| anyhow::anyhow!(e))?;

--- 2. ignored-chars 和 checksum ---
原文件     默认 0fa7f87ac75a0c6e  忽略空白 5457a13cba8fe4a9
重新排版后 默认 660fe9b15a7205dd  忽略空白 5457a13cba8fe4a9

let mut m = Migrator::new(dir.as_path()).await?; // 1_never.sql, 2_conc.sql
m.dangerous_set_table_name("tx._sqlx_migrations_skip");
m.skip(&pool, Some(1)).await?;
m.run(&pool).await?;

--- 5. Migrator::skip（#3846）---
tx._sqlx_migrations_skip: [(1, "never", -1), (2, "conc", 1153625)]
(tx.never, tx.conc) = (None, Some("tx.conc"))

6
加了 4_extra.sql 之后:
6
touch 1_widgets.up.sql 之后:
7