From 4e156ac683b1e93532b315a45c657669fa01fb79 Mon Sep 17 00:00:00 2001 From: Noah Date: Mon, 21 Sep 2026 02:48:33 +0000 Subject: [PATCH] =?UTF-8?q?fix(db):=20=E5=90=AF=E5=8A=A8=E6=97=B6=E5=AF=B9?= =?UTF-8?q?=E9=BD=90=E8=A2=AB=E6=94=B9=E5=86=99=E8=BF=87=E7=9A=84=E8=BF=81?= =?UTF-8?q?=E7=A7=BB=E6=A0=A1=E9=AA=8C=E5=92=8C=EF=BC=8C1.2.4=20=E7=9A=84?= =?UTF-8?q?=E8=80=81=E5=BA=93=E4=B8=8D=E5=86=8D=E6=97=A0=E6=B3=95=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=20(#1703)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(db): 启动时对齐被改写过的迁移校验和，老库不再无法启动 1.2.5 就地修改了已经发布的迁移 2（#1684 把 SET tags = null 改成 SET tags = '[]'）。 sqlx 会用文件内容的 SHA-384 校验每条已应用的迁移，于是 1.1.8 ~ 1.2.4 期间建立的 数据库升级后一律以 VersionMismatch 整体失败：进程退出码 1，容器反复重启。 在跑迁移之前加一步对齐：只有当 _sqlx_migrations 里记的校验和恰好等于登记在册的 历史摘要时，才在同一个事务里补跑新旧内容的差异（幂等 UPDATE）并改写校验和。 其余任何失配仍按原样报错，不会把用户自己改过的迁移悄悄放行。 * test(db): 钉死已发布迁移的校验和，防止再次就地改写 这次故障的根因是改了一条已经发布过的迁移。加一条断言把迁移 1..5 的 SHA-384 写死： 再有人就地改动已发布的迁移，CI 当场失败，而不是等用户升级时容器起不来。 --- .../server/infrastructure/connection_pool.rs | 294 +++++++++++++++++- 1 file changed, 286 insertions(+), 8 deletions(-) diff --git a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs index 7764f21..6261582 100644 --- a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs +++ b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs @@ -1,13 +1,45 @@ use crate::server::errors::{AppError, AppResult}; use error_stack::ResultExt; +use sqlx::migrate::Migrator; use sqlx::sqlite::SqlitePoolOptions; use sqlx::{Pool, Sqlite}; +use std::fmt::Write as _; use std::path::Path; -use tracing::info; +use tracing::{info, warn}; /// SQLite连接池类型别名 pub type ConnectionPool = Pool; +/// 一条曾经被就地改写过的迁移。 +/// +/// `legacy_checksum` 是历史版本里该文件内容的 SHA-384（sqlx 记在 `_sqlx_migrations` +/// 里的那一列），`repair` 是把旧内容的执行效果补成新内容所需的语句。 +struct SupersededMigration { + version: i64, + legacy_checksum: &'static str, + /// 必须幂等：老库可能已经处于新内容期望的状态，也可能重复启动多次。 + repair: &'static [&'static str], +} + +/// 迁移文件是只读历史：sqlx 会用文件内容的 SHA-384 校验每条已应用的迁移，一旦 +/// 就地改写，所有老库都会在启动时以 +/// `migration N was previously applied but has been modified` 整体失败 —— 服务起不来， +/// 容器无限重启（#1701 / #1702）。修 SQL 的正确做法始终是新增一个迁移文件。 +/// +/// 这张表只给**已经发生过**的改写兜底，不是继续改写的许可。登记一条的前提是： +/// 对已经跑过旧内容的库来说，新旧内容的差异能用下面这组幂等语句补齐。 +const SUPERSEDED_MIGRATIONS: &[SupersededMigration] = &[ + // 迁移 2 原本写 `UPDATE uploadstreamers SET tags = null`，而该列是 JSON NOT NULL， + // 从 Python 版升级且 tags 为空串的实例会卡死在这一步；v1.2.5 把它改成了 + // `SET tags = '[]'`（#1684）。改动本身是对的，但让 v1.2.4 及更早版本建立的库 + // 全部校验失配。两版内容的唯一差异就是这一条 UPDATE。 + SupersededMigration { + version: 2, + legacy_checksum: "fcc6436a889297e5c28f2a0f12196e5eee5975ba352032c8201dfa61fb1b9c3fd01ddc41ddfa71bd7423999521b895eb", + repair: &["UPDATE uploadstreamers SET tags = '[]' WHERE tags = '' OR tags = 'null'"], + }, +]; + /// 连接管理器 /// 负责管理SQLite数据库连接池的创建和配置 pub struct ConnectionManager; @@ -48,21 +80,267 @@ impl ConnectionManager { ))?; // 运行数据库迁移，确保数据库结构是最新的 + let migrator = sqlx::migrate!(); + Self::reconcile_superseded_migrations(&pool, &migrator).await?; + info!("migrations enabled, running..."); - sqlx::migrate!() - .run(&pool) - .await - .change_context(AppError::Custom( - "error while running database migrations".to_string(), - ))?; + migrator.run(&pool).await.change_context(AppError::Custom( + "error while running database migrations".to_string(), + ))?; Ok(pool) } + + /// 把历史上被改写过的迁移记录对齐到当前文件内容，让老库还能继续升级。 + /// + /// 只在记录里的校验和**恰好等于**登记过的历史摘要时才动手；其余任何失配都原样 + /// 留给 sqlx 报错，免得把用户自己改过的迁移悄悄放行。补丁语句与校验和改写在同一 + /// 个事务里，中途崩溃不会留下「补丁没跑但校验和已对齐」的中间态。 + async fn reconcile_superseded_migrations( + pool: &ConnectionPool, + migrator: &Migrator, + ) -> AppResult<()> { + let bookkeeping_exists: Option = sqlx::query_scalar( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = '_sqlx_migrations'", + ) + .fetch_optional(pool) + .await + .change_context(AppError::Custom( + "error while inspecting database migration state".to_string(), + ))?; + // 全新安装：连记账表都还没有，没有任何历史需要对齐。 + if bookkeeping_exists.is_none() { + return Ok(()); + } + + for entry in SUPERSEDED_MIGRATIONS { + let Some(current) = migrator.iter().find(|m| m.version == entry.version) else { + continue; + }; + let applied: Option<(Vec, bool)> = + sqlx::query_as("SELECT checksum, success FROM _sqlx_migrations WHERE version = ?") + .bind(entry.version) + .fetch_optional(pool) + .await + .change_context(AppError::Custom( + "error while inspecting database migration state".to_string(), + ))?; + // 没跑过（含跑失败留下的记录）的迁移由 sqlx 正常应用，不需要对齐。 + let Some((checksum, true)) = applied else { + continue; + }; + if checksum == current.checksum.as_ref() { + continue; + } + if hex_lower(&checksum) != entry.legacy_checksum { + warn!( + version = entry.version, + "已应用的迁移既不是当前内容也不是已知的历史内容，不做自动对齐" + ); + continue; + } + + let mut tx = pool.begin().await.change_context(AppError::Custom( + "error while reconciling database migration state".to_string(), + ))?; + for statement in entry.repair { + sqlx::query(statement) + .execute(&mut *tx) + .await + .change_context(AppError::Custom( + "error while reconciling database migration state".to_string(), + ))?; + } + sqlx::query("UPDATE _sqlx_migrations SET checksum = ? WHERE version = ?") + .bind(current.checksum.to_vec()) + .bind(entry.version) + .execute(&mut *tx) + .await + .change_context(AppError::Custom( + "error while reconciling database migration state".to_string(), + ))?; + tx.commit().await.change_context(AppError::Custom( + "error while reconciling database migration state".to_string(), + ))?; + + info!( + version = entry.version, + "该迁移在新版本中被修正，已补跑差异并对齐校验和" + ); + } + + Ok(()) + } +} + +fn hex_lower(bytes: &[u8]) -> String { + bytes + .iter() + .fold(String::with_capacity(bytes.len() * 2), |mut out, byte| { + let _ = write!(out, "{byte:02x}"); + out + }) } #[cfg(test)] mod tests { - use super::ConnectionManager; + use super::{ConnectionManager, ConnectionPool, SUPERSEDED_MIGRATIONS, hex_lower}; + + fn decode_hex(hex: &str) -> Vec { + (0..hex.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap()) + .collect() + } + + /// 把一个全新库回退成 v1.2.4 那一代的样子：迁移 4/5 尚未应用，迁移 2 记的是 + /// 改写前的校验和。 + async fn rewind_to_v1_2_4(pool: &ConnectionPool, migration_2_checksum: &[u8]) { + sqlx::query("DELETE FROM _sqlx_migrations WHERE version IN (4, 5)") + .execute(pool) + .await + .unwrap(); + sqlx::query("ALTER TABLE uploadstreamers DROP COLUMN tid_v2") + .execute(pool) + .await + .unwrap(); + sqlx::query("UPDATE _sqlx_migrations SET checksum = ? WHERE version = 2") + .bind(migration_2_checksum.to_vec()) + .execute(pool) + .await + .unwrap(); + } + + /// 已发布迁移的内容摘要，逐条钉死。 + /// + /// 这个断言挂掉说明有人改动了一条**已经发布过**的迁移 —— 所有老库都会在升级时 + /// 以校验和失配启动失败（#1701 就是这么来的）。正确做法是把改动挪进一个新的迁移 + /// 文件；确实只能就地改的话，把旧摘要连同幂等补丁登记进 `SUPERSEDED_MIGRATIONS`， + /// 再更新这里。 + #[test] + fn shipped_migration_checksums_are_frozen() { + const FROZEN: &[(i64, &str)] = &[ + ( + 1, + "f0dbce08e61eb0722d99ea5415742b6786a00904aa87bf1d9c21eceabced7484e340c4b51185546cb0578106bbb0c865", + ), + ( + 2, + "aef99c14523e93ceed5602d79b6c40c61891c794ea7f5fe8583ec1858315f57afeebfcaf9167a108ba7589745d7e34bf", + ), + ( + 3, + "bda5d2a2757539c071a1887ffba044a2660dd03200680d9f8537546680556c987047422c074bd438ecb7ef25a73be51b", + ), + ( + 4, + "8c93f71ec95e78418443f93395375eaeb9e0f16cf666fec555aa3165e73faa75e9a336794e292556d8d717a979085aaa", + ), + ( + 5, + "f6f913ecbbdf1f4d437bf2ce4778b8d0317353056c4b240795fc9347b4eeac7f71e719d4c799ac2df72b06802ee064c7", + ), + ]; + let embedded = sqlx::migrate!(); + let actual: Vec<(i64, String)> = embedded + .iter() + .map(|m| (m.version, hex_lower(&m.checksum))) + .collect(); + let expected: Vec<(i64, String)> = + FROZEN.iter().map(|(v, c)| (*v, (*c).to_string())).collect(); + assert_eq!(actual, expected); + } + + /// #1701：v1.2.5 就地改写了迁移 2，所有 v1.2.4 及更早版本建立的库都会以 + /// `migration 2 was previously applied but has been modified` 启动失败。 + /// 升级必须照常完成：待应用的迁移要跑完，既有数据不能丢。 + #[tokio::test] + async fn upgrade_from_pre_1_2_5_database_reconciles_rewritten_migration() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("data.sqlite3"); + let pool = ConnectionManager::new_pool(db.to_str().unwrap()) + .await + .unwrap(); + rewind_to_v1_2_4(&pool, &decode_hex(SUPERSEDED_MIGRATIONS[0].legacy_checksum)).await; + sqlx::query( + "INSERT INTO uploadstreamers (id, template_name, tags) VALUES \ + (1, 'from-python', ''), (2, 'normal', '[\"直播录像\"]')", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO livestreamers (url, remark, override) VALUES \ + ('https://live.bilibili.com/1', 'placeholder', '{\"file_size\":null}')", + ) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + + let pool = ConnectionManager::new_pool(db.to_str().unwrap()) + .await + .expect("v1.2.4 建立的库必须能升级上来"); + + let migrations: Vec<(i64, Vec)> = + sqlx::query_as("SELECT version, checksum FROM _sqlx_migrations ORDER BY version") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + migrations.iter().map(|m| m.0).collect::>(), + vec![1, 2, 3, 4, 5], + "待应用的迁移必须补齐" + ); + let embedded = sqlx::migrate!(); + let expected = embedded.iter().find(|m| m.version == 2).unwrap(); + assert_eq!( + hex_lower(&migrations[1].1), + hex_lower(&expected.checksum), + "迁移 2 的校验和必须对齐到当前文件内容" + ); + + let tags: Vec<(i64, String)> = + sqlx::query_as("SELECT id, tags FROM uploadstreamers ORDER BY id") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(tags[0].1, "[]", "空 tags 必须被补成空数组，而不是 NULL"); + assert_eq!(tags[1].1, "[\"直播录像\"]", "正常数据原样保留"); + + // 迁移 4/5 确实跑过：占位 null 被清掉，tid_v2 列已存在。 + let override_json: String = sqlx::query_scalar("SELECT override FROM livestreamers") + .fetch_one(&pool) + .await + .unwrap(); + assert!(!override_json.contains("file_size"), "{override_json}"); + sqlx::query("SELECT tid_v2 FROM uploadstreamers") + .fetch_all(&pool) + .await + .expect("迁移 5 必须已应用"); + } + + /// 自愈只针对登记在册的历史内容。用户自己改过的迁移仍然要报错，不能被悄悄放行。 + #[tokio::test] + async fn unknown_migration_checksum_is_not_silently_reconciled() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("data.sqlite3"); + let pool = ConnectionManager::new_pool(db.to_str().unwrap()) + .await + .unwrap(); + sqlx::query("UPDATE _sqlx_migrations SET checksum = ? WHERE version = 2") + .bind(vec![0u8; 48]) + .execute(&pool) + .await + .unwrap(); + pool.close().await; + + assert!( + ConnectionManager::new_pool(db.to_str().unwrap()) + .await + .is_err() + ); + } /// 旧版本落库的覆写里 `file_size: null` 只是占位，迁移后必须消失（跟随全局）， /// 其它显式设置的字段与真正的数值原样保留。