fix: propagate startup repair database errors

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 {

crates/xmedia-bot/src/handlers/repair.rs

@@ -21,6 +21,7 @@ struct Refetched {
     caption: String,
     items: Vec<MediaItemPayload>,
     cache_data: Option<CachedPost>,
+    keep_alive: Option<std::sync::Arc<tempfile::TempDir>>,
 }
 
 /// 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,
+/// 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
+/// This runs at startup before queue workers exist, so any SQLite error is
-/// that cannot come back. Returns how many rows were rewritten.
+/// returned to the caller and prevents workers from starting on unrepaired
-///
+/// state.
-/// Startup only, before the queue workers start: no worker can lease a row while
+pub(crate) async fn repair_lost_local_media(
-/// this writes, which is what lets it replace payloads without the lease-token
+    ctx: &AppContext<'_>,
-/// guard every worker write-back carries.
+) -> Result<usize, rusqlite::Error> {
-pub(crate) async fn repair_lost_local_media(ctx: &AppContext<'_>) -> usize {
+    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::<Task>(&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 {
+                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} (its local media did not survive the restart)",
+                            "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 {

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)");
     }

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<Vec<(String, String)>> {
         self.pool
-            .with_conn_or(
+            .with_conn(|conn| {
-                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::<rusqlite::Result<Vec<(String, String)>>>()
-                },
+            })
-            )
             .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<bool> {
         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;
+            .await?;
-        match result {
+        if !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
-            }
         }
+        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<f64>>(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()
         );
     }