From a54029c19be17037e77dfcf6e48b8739f23118a7 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Thu, 24 Sep 2026 21:53:04 +0800 Subject: [PATCH] fix: propagate startup repair database errors --- crates/xmedia-bot/src/ctx.rs | 6 +++ crates/xmedia-bot/src/handlers/repair.rs | 68 +++++++++++++++--------- crates/xmedia-bot/src/main.rs | 8 ++- crates/xmedia-bot/src/queue.rs | 46 ++++++---------- 4 files changed, 73 insertions(+), 55 deletions(-) diff --git a/crates/xmedia-bot/src/ctx.rs b/crates/xmedia-bot/src/ctx.rs index 712902a..b5c680a 100644 --- a/crates/xmedia-bot/src/ctx.rs +++ b/crates/xmedia-bot/src/ctx.rs @@ -186,6 +186,12 @@ pub(crate) mod test_support { &self.task_queue } + /// Path to the shared test database, for tests that need to corrupt or + /// inspect schema through a separate connection. + pub(crate) fn db_path(&self) -> &str { + self.pool.path() + } + /// Rows persisted in the task queue: what "queued for retry" looks like /// from the outside. pub(crate) async fn queued_tasks(&self) -> i64 { diff --git a/crates/xmedia-bot/src/handlers/repair.rs b/crates/xmedia-bot/src/handlers/repair.rs index 19ff75e..b2ffb53 100644 --- a/crates/xmedia-bot/src/handlers/repair.rs +++ b/crates/xmedia-bot/src/handlers/repair.rs @@ -21,6 +21,7 @@ struct Refetched { caption: String, items: Vec, cache_data: Option, + keep_alive: Option>, } /// Whether a queued task should have its post re-fetched, because it still @@ -117,28 +118,24 @@ async fn refetch( if items.is_empty() { return Ok(None); } - // The re-fetch may produce a fresh local file (ugoira / bsky remux): hand it - // to the same keep-alive registry the first fetch uses. - if let Some(dir) = fetched.keep_alive() { - send::KEEP_ALIVE.lock().push(dir); - } Ok(Some(Refetched { caption, items, cache_data, + keep_alive: fetched.keep_alive(), })) } -/// Re-fetches every queued task whose local media did not survive the restart, -/// so the user's link is still delivered instead of dead-lettering on a file -/// that cannot come back. Returns how many rows were rewritten. -/// -/// Startup only, before the queue workers start: no worker can lease a row while -/// this writes, which is what lets it replace payloads without the lease-token -/// guard every worker write-back carries. -pub(crate) async fn repair_lost_local_media(ctx: &AppContext<'_>) -> usize { +/// Re-fetches every queued task whose local media did not survive the restart. +/// This runs at startup before queue workers exist, so any SQLite error is +/// returned to the caller and prevents workers from starting on unrepaired +/// state. +pub(crate) async fn repair_lost_local_media( + ctx: &AppContext<'_>, +) -> Result { + let rows = ctx.task_queue.runnable_rows().await?; let mut repaired = 0; - for (id, payload) in ctx.task_queue.runnable_rows().await { + for (id, payload) in rows { let Ok(task) = serde_json::from_str::(&payload) else { continue; }; @@ -155,17 +152,26 @@ pub(crate) async fn repair_lost_local_media(ctx: &AppContext<'_>) -> usize { continue; }; let updated = serde_json::to_value(&updated).expect("task serializes"); - if ctx.task_queue.replace_payload(&id, &updated).await { - repaired += 1; - log::info!( - "startup repair: re-fetched [key={}] for chat={chat_id} (its local media did not survive the restart)", - log_key(&url) - ); + match ctx.task_queue.replace_payload(&id, &updated).await { + Ok(true) => { + if let Some(dir) = fresh.keep_alive { + send::KEEP_ALIVE.lock().push(dir); + } + repaired += 1; + log::info!( + "startup repair: re-fetched [key={}] for chat={chat_id}", + log_key(&url) + ); + } + Ok(false) => { + log::warn!("startup repair: queue row {id} disappeared before rewrite") + } + Err(e) => { + log::error!("startup repair: queue row {id} rewrite failed: {e}"); + return Err(e); + } } } - // The post is gone or withheld now: the retry could not have - // delivered anything either, so say why instead of letting it - // dead-letter on a missing file. Ok(None) | Err(_) => { let (notify_chat_id, notify_message_id) = task.notify_target(); log::warn!( @@ -185,7 +191,7 @@ pub(crate) async fn repair_lost_local_media(ctx: &AppContext<'_>) -> usize { } } } - repaired + Ok(repaired) } #[cfg(test)] @@ -259,6 +265,7 @@ mod tests { caption: "fresh caption".into(), items: vec![photo_item("https://cdn/fresh.jpg", true, false)], cache_data: None, + keep_alive: None, }; match apply_refresh(&task, &fresh).expect("a repairable task") { Task::SendMediaSequence { @@ -303,6 +310,17 @@ mod tests { } } + #[tokio::test] + async fn a_queue_scan_error_is_reported_to_the_startup_caller() { + let stores = TestStores::new(); + let sender = MockSender::scripted(vec![], permanent_error); + let ctx = stores.ctx(&sender); + let raw = rusqlite::Connection::open(stores.db_path()).unwrap(); + raw.execute_batch("DROP TABLE tasks").unwrap(); + + assert!(repair_lost_local_media(&ctx).await.is_err()); + } + /// The whole repair against a real post: a queued row whose media is a local /// file the restart took away is re-fetched from its `source_url` and /// rewritten in place, so the retry can still deliver it. @@ -323,7 +341,7 @@ mod tests { .await .unwrap(); - assert_eq!(repair_lost_local_media(&ctx).await, 1); + assert_eq!(repair_lost_local_media(&ctx).await, Ok(1)); let updated: Task = serde_json::from_value(stores.queued_payload().await).unwrap(); match updated { diff --git a/crates/xmedia-bot/src/main.rs b/crates/xmedia-bot/src/main.rs index 25d9e44..c8e27b9 100644 --- a/crates/xmedia-bot/src/main.rs +++ b/crates/xmedia-bot/src/main.rs @@ -157,7 +157,13 @@ async fn main() { // succeed after a restart — the registry that kept those files alive is in // memory — so those rows are re-fetched from their post instead of // dead-lettering the user's link. - let repaired = handlers::repair_lost_local_media(&CONTEXT).await; + let repaired = match handlers::repair_lost_local_media(&CONTEXT).await { + Ok(repaired) => repaired, + Err(e) => { + log::error!("startup repair failed: {e}; refusing to start queue workers"); + return; + } + }; if repaired > 0 { log::info!("startup repair: re-fetched {repaired} queued task(s)"); } diff --git a/crates/xmedia-bot/src/queue.rs b/crates/xmedia-bot/src/queue.rs index 32ee9ca..2ca5536 100644 --- a/crates/xmedia-bot/src/queue.rs +++ b/crates/xmedia-bot/src/queue.rs @@ -247,20 +247,15 @@ impl PersistentTaskQueue { /// `(id, payload)` of every row that can still run (`pending`, /// `in_progress`). The startup repair reads these before the workers start: /// with no worker running, no row can be leased while it writes. - pub async fn runnable_rows(&self) -> Vec<(String, String)> { + pub async fn runnable_rows(&self) -> rusqlite::Result> { self.pool - .with_conn_or( - log::Level::Error, - "queue row scan failed", - Vec::new(), - |conn| { - let mut stmt = conn.prepare( - "SELECT id, payload FROM tasks WHERE status IN ('pending', 'in_progress') ORDER BY run_after", - )?; - let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?; - rows.collect::>>() - }, - ) + .with_conn(|conn| { + let mut stmt = conn.prepare( + "SELECT id, payload FROM tasks WHERE status IN ('pending', 'in_progress') ORDER BY run_after", + )?; + let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?; + rows.collect::>>() + }) .await } @@ -269,7 +264,7 @@ impl PersistentTaskQueue { /// already spent do not carry over. Startup repair only — a worker's /// write-back is lease-token guarded instead (`replace_payload` cannot race /// one: it runs before any worker does). - pub async fn replace_payload(&self, id: &str, payload: &Value) -> bool { + pub async fn replace_payload(&self, id: &str, payload: &Value) -> rusqlite::Result { let logged_id = id.to_string(); let (id, payload) = (id.to_string(), payload.to_string()); let result = self @@ -282,18 +277,11 @@ impl PersistentTaskQueue { )?; Ok(affected == 1) }) - .await; - match result { - Ok(true) => true, - Ok(false) => { - log::warn!("queue: row {logged_id} vanished before its payload could be replaced"); - false - } - Err(e) => { - log::error!("queue payload replace failed for {logged_id}: {e}"); - false - } + .await?; + if !result { + log::warn!("queue: row {logged_id} vanished before its payload could be replaced"); } + Ok(result) } pub async fn pending_backlog(&self) -> Option<(i64, f64)> { @@ -308,7 +296,6 @@ impl PersistentTaskQueue { [], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option>(1)?)), )?; - // `MIN` over zero rows is NULL, so the count is what decides. Ok(match oldest { Some(oldest) if count > 0 => Some((count, oldest)), _ => None, @@ -1025,7 +1012,7 @@ mod tests { .enqueue(serde_json::json!({"s": 1}), now_f64()) .await .unwrap(); - let rows = queue.runnable_rows().await; + let rows = queue.runnable_rows().await.unwrap(); assert_eq!(rows.len(), 1); let (id, payload) = rows[0].clone(); assert_eq!(payload, "{\"s\":1}"); @@ -1047,19 +1034,20 @@ mod tests { queue .replace_payload(&id, &serde_json::json!({"s": 2})) .await + .unwrap() ); - let rows = queue.runnable_rows().await; + let rows = queue.runnable_rows().await.unwrap(); assert_eq!(rows[0].1, "{\"s\":2}"); assert_eq!( queue.pending_backlog().await.map(|(n, _)| n), Some(1), "a repaired row is pending work again" ); - // A row that is gone (or done) is not rewritten. assert!( !queue .replace_payload("task_missing", &serde_json::json!({})) .await + .unwrap() ); }