Compare commits
7 Commits
790002181a
...
67212a52b3
| Author | SHA1 | Date | |
|---|---|---|---|
| 67212a52b3 | |||
| 20e352bf85 | |||
| 91d83f714c | |||
| 356f4354dc | |||
| 283d8b5f4d | |||
| 920e26e213 | |||
| d044304969 |
12
migrate.sh
12
migrate.sh
@ -1,4 +1,16 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# 数据库迁移脚本(手动 / CI 备用)。
|
||||
#
|
||||
# 注意:服务器二进制现在会在启动时自动执行迁移(见 src/db/migrate.rs),
|
||||
# 正常情况下无需手动运行本脚本。保留它是为了:
|
||||
# - 运维 escape hatch:二进制因迁移失败起不来时,手动救库
|
||||
# - CI/CD 中“先迁移再滚动发布”的工作流
|
||||
#
|
||||
# 本脚本与服务器内置运行器读取相同的 migrations/*.sql 文件,且这些 SQL
|
||||
# 都是幂等的(IF NOT EXISTS / IF EXISTS / ON CONFLICT DO NOTHING),
|
||||
# 因此两者混用安全。
|
||||
#
|
||||
set -euo pipefail
|
||||
|
||||
# 从 .env 加载环境变量
|
||||
|
||||
@ -7,7 +7,7 @@
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
use axum::{
|
||||
extract::{ConnectInfo, Multipart},
|
||||
extract::{ConnectInfo, Extension, Multipart},
|
||||
http::{HeaderMap, StatusCode},
|
||||
response::Json,
|
||||
};
|
||||
@ -70,13 +70,19 @@ fn validate_raw_image(data: &[u8], mime_type: &str) -> bool {
|
||||
///
|
||||
/// 流程:限流 → 解析 session → 校验 admin → 读取 multipart → 校验类型/大小 →
|
||||
/// 转码(如适用)→ 按日期落盘 → 返回相对 URL。
|
||||
///
|
||||
/// `ConnectInfo` 以可选扩展注入:`dioxus::server::serve()` 接管了 listener,
|
||||
/// 无法调用 `into_make_service_with_connect_info::<SocketAddr>()`,所以这里
|
||||
/// 与 `serve_image` 保持一致的优雅降级——扩展缺失时退回 `"unknown"` 限流桶。
|
||||
/// 生产环境应在反向代理后部署并配置 `TRUSTED_PROXY_COUNT`,让限流拿到真实 IP。
|
||||
pub async fn upload_image(
|
||||
ConnectInfo(addr): ConnectInfo<SocketAddr>,
|
||||
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
|
||||
headers: HeaderMap,
|
||||
mut multipart: Multipart,
|
||||
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
// 0. Rate limit check
|
||||
let ip = crate::api::rate_limit::get_client_ip_with_peer(&headers, Some(addr));
|
||||
let peer = connect_info.map(|Extension(ConnectInfo(addr))| addr);
|
||||
let ip = crate::api::rate_limit::get_client_ip_with_peer(&headers, peer);
|
||||
if let Err(msg) = crate::api::rate_limit::check_upload_limit(&ip) {
|
||||
return Err((
|
||||
StatusCode::TOO_MANY_REQUESTS,
|
||||
|
||||
317
src/db/migrate.rs
Normal file
317
src/db/migrate.rs
Normal file
@ -0,0 +1,317 @@
|
||||
//! 数据库迁移运行器。
|
||||
//!
|
||||
//! 在服务器启动时(`dioxus::server::serve()` 之前)自动执行迁移。
|
||||
//! 设计要点:
|
||||
//! - 迁移 SQL 通过 `include_str!` 内联进二进制,部署只需单个二进制。
|
||||
//! - `schema_migrations` 表记录已应用版本,避免重复执行。
|
||||
//! - 每个迁移在独立事务里执行,失败自动回滚,版本行不会写入。
|
||||
//! - 咨询锁(`pg_advisory_lock`)保证多实例启动时只有一个进程执行迁移。
|
||||
//!
|
||||
//! 仅在 `feature = "server"` 时编译。
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
/// 咨询锁的固定 key。Postgres 咨询锁是数据库级唯一的;
|
||||
/// 这里用一个项目专属的大整数,避免与同库其它应用冲突。
|
||||
/// 该值无语义,仅要求唯一性。
|
||||
const ADVISORY_LOCK_KEY: i64 = 0x5947_4752_4153_494C;
|
||||
|
||||
/// 所有迁移的 (version, sql) 列表,按 version 升序排列。
|
||||
///
|
||||
/// 新增迁移时:
|
||||
/// 1. 在 `migrations/` 下创建 `NNN_描述.sql`
|
||||
/// 2. 在本数组末尾追加一行 `("NNN", include_str!("../../migrations/NNN_描述.sql"))`
|
||||
///
|
||||
/// version 用字符串而非整数,与文件名前缀直接对应,且未来可支持日期戳/语义版本。
|
||||
const MIGRATIONS: &[(&str, &str)] = &[
|
||||
("001", include_str!("../../migrations/001_init.sql")),
|
||||
("002", include_str!("../../migrations/002_posts.sql")),
|
||||
("003", include_str!("../../migrations/003_indexes.sql")),
|
||||
("004", include_str!("../../migrations/004_search_trgm.sql")),
|
||||
("005", include_str!("../../migrations/005_comments.sql")),
|
||||
("006", include_str!("../../migrations/006_add_toc_html.sql")),
|
||||
("007", include_str!("../../migrations/007_settings.sql")),
|
||||
("008", include_str!("../../migrations/008_comments_cascade.sql")),
|
||||
(
|
||||
"009",
|
||||
include_str!("../../migrations/009_cleanup_duplicate_indexes.sql"),
|
||||
),
|
||||
(
|
||||
"010",
|
||||
include_str!("../../migrations/010_post_word_counts.sql"),
|
||||
),
|
||||
("011", include_str!("../../migrations/011_perf_indexes.sql")),
|
||||
(
|
||||
"012",
|
||||
include_str!("../../migrations/012_session_generation.sql"),
|
||||
),
|
||||
(
|
||||
"013",
|
||||
include_str!("../../migrations/013_comment_content_hash_index.sql"),
|
||||
),
|
||||
(
|
||||
"014",
|
||||
include_str!("../../migrations/014_drop_ineffective_trgm_index.sql"),
|
||||
),
|
||||
// 新增迁移在此追加,同时在 migrations/ 下创建对应 .sql 文件。
|
||||
];
|
||||
|
||||
/// 迁移执行错误。
|
||||
#[derive(Debug)]
|
||||
pub enum MigrateError {
|
||||
/// 无法从连接池获取连接。
|
||||
Pool(deadpool_postgres::PoolError),
|
||||
/// 执行查询(建表、查版本、咨询锁等)失败。
|
||||
Query(tokio_postgres::Error),
|
||||
/// 某个具体迁移执行失败(包含版本号便于定位)。
|
||||
Apply {
|
||||
version: String,
|
||||
source: tokio_postgres::Error,
|
||||
},
|
||||
}
|
||||
|
||||
impl From<deadpool_postgres::PoolError> for MigrateError {
|
||||
fn from(e: deadpool_postgres::PoolError) -> Self {
|
||||
MigrateError::Pool(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<tokio_postgres::Error> for MigrateError {
|
||||
fn from(e: tokio_postgres::Error) -> Self {
|
||||
MigrateError::Query(e)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for MigrateError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
MigrateError::Pool(e) => write!(f, "database pool error: {}", e),
|
||||
MigrateError::Query(e) => write!(f, "database query error: {}", e),
|
||||
MigrateError::Apply { version, source } => {
|
||||
write!(f, "migration {} failed: {}", version, source)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for MigrateError {}
|
||||
|
||||
/// 执行所有未应用的迁移。
|
||||
///
|
||||
/// 在 `main.rs` 启动时调用一次。流程:
|
||||
/// 1. 获取一个独占连接(咨询锁是 session 级,需在同一连接上 lock/unlock)
|
||||
/// 2. 抢咨询锁(多实例启动时串行化)
|
||||
/// 3. 确保 `schema_migrations` 表存在
|
||||
/// 4. 查询已应用版本集合
|
||||
/// 5. 按序应用未应用的迁移(每个一个事务)
|
||||
/// 6. 释放咨询锁
|
||||
///
|
||||
/// 失败时返回错误;调用方(`main.rs`)应让进程退出,避免启动半残服务。
|
||||
pub async fn run() -> Result<(), MigrateError> {
|
||||
use crate::db::pool::get_conn;
|
||||
|
||||
tracing::info!("running database migrations");
|
||||
|
||||
// 咨询锁是 session 级的,必须在同一连接上 lock/unlock。
|
||||
// 因此整个迁移流程独占一个池连接。
|
||||
let mut conn = get_conn().await?;
|
||||
|
||||
// 抢咨询锁:多实例滚动发布时只有一个进程能进入迁移循环,
|
||||
// 其余实例在此等待;锁释放后它们查版本表发现已全部应用,直接返回。
|
||||
conn.execute("SELECT pg_advisory_lock($1)", &[&ADVISORY_LOCK_KEY])
|
||||
.await?;
|
||||
|
||||
// 在已持锁连接上执行迁移主体逻辑。
|
||||
// 锁释放策略:
|
||||
// - 正常返回 / 返回 Err:下面的显式 pg_advisory_unlock 释放锁。
|
||||
// - panic:panic 会跳过显式 unlock。此时 `conn` 在 unwind 中被 drop,
|
||||
// 但 deadpool 会把连接归还池中复用(不关闭 Postgres 会话),
|
||||
// 所以 session 级咨询锁不会立即释放。安全性依赖调用方(main.rs 的
|
||||
// `.expect()`)在 panic 时终止进程——进程退出会关闭所有池连接,
|
||||
// Postgres 检测到会话断开后释放 session 级咨询锁。
|
||||
let result = run_inner(&mut conn).await;
|
||||
|
||||
// 无论成功失败都尝试显式释放锁;释放失败不应掩盖原始错误,仅记录告警。
|
||||
if let Err(unlock_err) = conn
|
||||
.execute("SELECT pg_advisory_unlock($1)", &[&ADVISORY_LOCK_KEY])
|
||||
.await
|
||||
{
|
||||
tracing::warn!("failed to release migration advisory lock: {}", unlock_err);
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// 在已持有咨询锁的连接上执行迁移主体逻辑。
|
||||
async fn run_inner(conn: &mut deadpool_postgres::Object) -> Result<(), MigrateError> {
|
||||
// 确保版本表存在(独立语句,不在事务里,否则建表失败无法记录)。
|
||||
ensure_versions_table(conn).await?;
|
||||
|
||||
// 查询已应用的版本集合。
|
||||
let applied = applied_versions(conn).await?;
|
||||
|
||||
// 按序应用未应用的迁移。
|
||||
let mut applied_count = 0usize;
|
||||
for (version, sql) in MIGRATIONS {
|
||||
if applied.contains(*version) {
|
||||
continue;
|
||||
}
|
||||
tracing::info!("applying migration {}", version);
|
||||
apply_one(conn, version, sql).await?;
|
||||
applied_count += 1;
|
||||
}
|
||||
|
||||
if applied_count == 0 {
|
||||
tracing::info!("database is up to date, 0 migrations applied");
|
||||
} else {
|
||||
tracing::info!("successfully applied {} migration(s)", applied_count);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 创建 `schema_migrations` 表(若不存在)。
|
||||
async fn ensure_versions_table(conn: &deadpool_postgres::Object) -> Result<(), MigrateError> {
|
||||
conn.batch_execute(
|
||||
"CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
)",
|
||||
)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 查询已应用的版本集合。
|
||||
async fn applied_versions(
|
||||
conn: &deadpool_postgres::Object,
|
||||
) -> Result<HashSet<String>, MigrateError> {
|
||||
let rows = conn.query("SELECT version FROM schema_migrations", &[]).await?;
|
||||
let mut set = HashSet::with_capacity(rows.len());
|
||||
for row in rows {
|
||||
set.insert(row.get::<_, String>(0));
|
||||
}
|
||||
Ok(set)
|
||||
}
|
||||
|
||||
/// 在一个事务内应用单个迁移:执行 SQL + 写入版本行,失败则回滚。
|
||||
async fn apply_one(
|
||||
conn: &mut deadpool_postgres::Object,
|
||||
version: &str,
|
||||
sql: &str,
|
||||
) -> Result<(), MigrateError> {
|
||||
let tx = conn.transaction().await.map_err(MigrateError::Query)?;
|
||||
|
||||
// batch_execute 执行整段 SQL(可含多条语句)。
|
||||
if let Err(e) = tx.batch_execute(sql).await {
|
||||
// 显式回滚以尽早释放事务(而非等 Transaction drop 的隐式回滚);
|
||||
// 回滚本身的错误丢弃,因为已有更具信息量的 apply 错误要上报。
|
||||
let _ = tx.rollback().await;
|
||||
return Err(MigrateError::Apply {
|
||||
version: version.to_string(),
|
||||
source: e,
|
||||
});
|
||||
}
|
||||
|
||||
// 记录版本行。显式构造 Apply 错误(不能用 ?,否则会被 blanket From 映射成 Query)。
|
||||
if let Err(e) = tx
|
||||
.execute(
|
||||
"INSERT INTO schema_migrations (version) VALUES ($1)",
|
||||
&[&version],
|
||||
)
|
||||
.await
|
||||
{
|
||||
let _ = tx.rollback().await;
|
||||
return Err(MigrateError::Apply {
|
||||
version: version.to_string(),
|
||||
source: e,
|
||||
});
|
||||
}
|
||||
|
||||
tx.commit().await.map_err(|e| MigrateError::Apply {
|
||||
version: version.to_string(),
|
||||
source: e,
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(all(test, feature = "server"))]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn migrations_are_sorted_ascending() {
|
||||
let mut sorted = MIGRATIONS.iter().map(|(v, _)| *v).collect::<Vec<_>>();
|
||||
sorted.sort_unstable();
|
||||
let original: Vec<&str> = MIGRATIONS.iter().map(|(v, _)| *v).collect();
|
||||
assert_eq!(original, sorted, "MIGRATIONS must be in ascending version order");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migrations_have_unique_versions() {
|
||||
let mut versions: Vec<&str> = MIGRATIONS.iter().map(|(v, _)| *v).collect();
|
||||
let total = versions.len();
|
||||
versions.sort_unstable();
|
||||
versions.dedup();
|
||||
assert_eq!(versions.len(), total, "MIGRATIONS has duplicate version strings");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migrations_non_empty() {
|
||||
assert!(!MIGRATIONS.is_empty(), "MIGRATIONS must not be empty");
|
||||
}
|
||||
|
||||
/// 防止"新建了 .sql 但忘记在 MIGRATIONS 加行"的脚枪。
|
||||
/// 扫描 migrations/ 目录,断言每个 .sql 文件都在 MIGRATIONS 里有对应版本行。
|
||||
/// 仅在 server feature + test 下运行(WASM 无文件系统)。
|
||||
#[test]
|
||||
fn migrations_match_files_on_disk() {
|
||||
use std::collections::HashSet;
|
||||
use std::fs;
|
||||
|
||||
// CARGO_MANIFEST_DIR 指向 crate 根目录(yggdrasil/)。
|
||||
let manifest_dir = env!("CARGO_MANIFEST_DIR");
|
||||
let migrations_dir = std::path::Path::new(manifest_dir).join("migrations");
|
||||
|
||||
let mut files_on_disk: HashSet<String> = HashSet::new();
|
||||
for entry in fs::read_dir(&migrations_dir)
|
||||
.unwrap_or_else(|e| panic!("failed to read {}: {}", migrations_dir.display(), e))
|
||||
{
|
||||
let entry = entry.unwrap();
|
||||
let path = entry.path();
|
||||
if path.extension().and_then(|e| e.to_str()) == Some("sql") {
|
||||
let filename = path
|
||||
.file_name()
|
||||
.and_then(|n| n.to_str())
|
||||
.unwrap_or_else(|| panic!("non-utf8 filename: {}", path.display()));
|
||||
// 文件名形如 "001_init.sql",取前 3 位数字作为 version。
|
||||
let version = filename
|
||||
.split('_')
|
||||
.next()
|
||||
.unwrap_or_else(|| panic!("filename has no '_' separator: {}", filename));
|
||||
files_on_disk.insert(version.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
let versions_in_array: HashSet<String> =
|
||||
MIGRATIONS.iter().map(|(v, _)| v.to_string()).collect();
|
||||
|
||||
// 磁盘上有但数组里没有 → 忘记加行(会静默不执行该迁移)。
|
||||
let missing_in_array: Vec<&String> =
|
||||
files_on_disk.difference(&versions_in_array).collect();
|
||||
assert!(
|
||||
missing_in_array.is_empty(),
|
||||
"migrations/*.sql files not registered in MIGRATIONS: {:?}. \
|
||||
Add a row for each in src/db/migrate.rs.",
|
||||
missing_in_array
|
||||
);
|
||||
|
||||
// 数组里有但磁盘上没有 → include_str! 本就会编译失败,这里只是双保险。
|
||||
let missing_on_disk: Vec<&String> =
|
||||
versions_in_array.difference(&files_on_disk).collect();
|
||||
assert!(
|
||||
missing_on_disk.is_empty(),
|
||||
"MIGRATIONS rows without a corresponding .sql file: {:?}",
|
||||
missing_on_disk
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -16,6 +16,10 @@ pub mod pool;
|
||||
#[cfg(feature = "server")]
|
||||
pub mod retry;
|
||||
|
||||
/// 数据库迁移运行器,仅在启用 server feature 时编译。
|
||||
#[cfg(feature = "server")]
|
||||
pub mod migrate;
|
||||
|
||||
/// 占位连接池实现,仅在不启用 server feature 时编译。
|
||||
///
|
||||
/// `DummyPool` 是一个最小 stub:它提供与真实连接池相同的公开接口形状
|
||||
|
||||
29
src/main.rs
29
src/main.rs
@ -206,6 +206,25 @@ fn main() {
|
||||
std::process::exit(1);
|
||||
}
|
||||
|
||||
// 启动前执行数据库迁移。阻塞:完成前不监听端口。
|
||||
// 失败直接 panic(expect),避免启动一个 schema 不一致的半残服务。
|
||||
// 多实例滚动发布时由咨询锁串行化,详见 src/db/migrate.rs。
|
||||
//
|
||||
// main() 是同步函数,这里用一个独立的多线程 runtime 驱动迁移的异步逻辑,
|
||||
// 完成后再交给 dioxus::server::serve() 启动它自己的 runtime。
|
||||
// 两个 runtime 不重叠,避免与 Dioxus 内部 runtime 产生交互。
|
||||
let migrate_rt = tokio::runtime::Builder::new_multi_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("failed to build migration runtime");
|
||||
migrate_rt.block_on(async {
|
||||
db::migrate::run()
|
||||
.await
|
||||
.expect("Database migration failed on startup");
|
||||
});
|
||||
// 迁移 runtime 用完即弃,显式 drop 以在 serve() 前释放其线程资源。
|
||||
drop(migrate_rt);
|
||||
|
||||
// 启动 Dioxus 服务端,返回构建好的 Axum Router
|
||||
dioxus::server::serve(|| async move {
|
||||
use dioxus::server::{axum, DioxusRouterExt, ServeConfig};
|
||||
@ -307,10 +326,12 @@ fn main() {
|
||||
));
|
||||
|
||||
// 静态资源路由:图片文件服务。
|
||||
// 注意:axum 0.8 没有 ConnectInfoLayer,且 dioxus::server::serve 不会把
|
||||
// ConnectInfo 扩展传播到手动 merge 的路由,所以 serve_image 使用
|
||||
// Option<Extension<ConnectInfo<SocketAddr>>> 优雅降级。生产环境应在反向代理后
|
||||
// 部署并配置 TRUSTED_PROXY_COUNT,使限流能拿到真实客户端 IP。
|
||||
// 注意:`dioxus::server::serve()` 接管了 listener 与 `into_make_service`
|
||||
// 调用,没有机会换成 `into_make_service_with_connect_info::<SocketAddr>()`,
|
||||
// 所以手动 merge 进来的路由(含 static_routes)拿不到 `ConnectInfo` 扩展。
|
||||
// serve_image / upload_image 因此都用 `Option<Extension<ConnectInfo<SocketAddr>>>`
|
||||
// 优雅降级。生产环境应在反向代理后部署并配置 TRUSTED_PROXY_COUNT,
|
||||
// 使限流能拿到真实客户端 IP。
|
||||
let static_routes = axum::Router::new()
|
||||
.route(
|
||||
"/uploads/{*path}",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user