fix(send): keep local media (ugoira/bsky remux MP4) alive across queue retries

A task whose media is a locally produced file (pixiv ugoira MP4, bsky HLS
remux MP4) references a path inside a tempfile TempDir owned by Fetched.
The retry ran after that TempDir was dropped, so the file was already gone
and the retry always failed permanently ("local media file missing") — and
the upload fallback even tried to GET the local path as a URL.

Keep the temp dirs in a process-wide registry (Fetched::take_keep_alive ->
send::KEEP_ALIVE) that is only released when the task settles (sent or
permanently failed); the upload fallback now uploads local files directly
instead of attempting to download them.

Restart-mid-queue still loses the files (documented behavior in
input_file_for) — only the in-process retry path is fixed here.

crates/x-media/src/site/mod.rs

@@ -95,6 +95,14 @@ impl Fetched {
             )
         })
     }
+
+    /// Hands over the temp dir keeping locally produced media (ugoira MP4,
+    /// bsky remux MP4) alive. The bot keeps it while its task may still be
+    /// retried by the queue, which runs after this [`Fetched`] is dropped and
+    /// its temp files would otherwise be gone. `None` when no such dir exists.
+    pub fn take_keep_alive(&mut self) -> Option<tempfile::TempDir> {
+        self._keep_alive.take()
+    }
 }
 
 /// Renders a user-supplied caption format from raw (already-escaped) field

crates/xmedia-bot/src/handlers.rs

@@ -516,6 +516,8 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) {
         Ok(message_ids) => {
             log::info!("sent {} message(s) for {url}", message_ids.len());
             send::post_send_actions(&bot, task, message_ids).await;
+            // The task settled: drop any keep-alive temp media.
+            send::release_keep_alive(task);
         }
         Err(send::SendError::Retryable {
             delay_seconds,
@@ -530,6 +532,7 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) {
             task,
         }) => {
             send::invalidate_cache(&task).await;
+            send::release_keep_alive(&task);
             log::error!("send for {url} failed permanently: {err_message}");
             let _ = reply(bot, message.clone(), format!("Send failed: {err_message}")).await;
         }
@@ -667,7 +670,7 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
             )
             .await;
         }
-        Ok(Some(fetched)) => {
+        Ok(Some(mut fetched)) => {
             if fetched.media.is_empty() {
                 let _ = reply(
                     bot,
@@ -712,6 +715,13 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
                 items,
                 cache_data,
             );
+            // Hand the keep-alive temp dir (ugoira / bsky remux MP4) to the
+            // retry registry: a queued retry runs after this function returns
+            // and the fetch's own TempDir is dropped, so without this the
+            // local file would be gone by the time the retry sends it.
+            if let Some(dir) = fetched.take_keep_alive() {
+                send::KEEP_ALIVE.lock().push(dir);
+            }
             dispatch_send(bot, message, &task, url).await;
         }
     }

crates/xmedia-bot/src/send.rs

@@ -147,6 +147,39 @@ impl Task {
     fn is_cached_send(&self) -> bool {
         self.cache_data().is_some_and(|c| !c.media.is_empty())
     }
+
+    /// All media payloads of this task (sequence batches flattened plus the
+    /// lone animation).
+    fn media_items(&self) -> Vec<&MediaItemPayload> {
+        match self {
+            Task::SendMediaSequence { media_batches, .. } => {
+                media_batches.iter().flatten().collect()
+            }
+            Task::SendAnimation { animation, .. } => std::slice::from_ref(animation).iter().collect(),
+            Task::ForwardMessages { .. } => Vec::new(),
+        }
+    }
+
+    /// Local file paths referenced by this task's media (ugoira / bsky remux
+    /// MP4 and the like); empty for URL or Telegram file-id sends.
+    fn local_media_paths(&self) -> Vec<std::path::PathBuf> {
+        let mut out = Vec::new();
+        for item in self.media_items() {
+            let is_file_id = match item {
+                MediaItemPayload::Photo { file_id, .. }
+                | MediaItemPayload::Video { file_id, .. }
+                | MediaItemPayload::Animation { file_id, .. } => *file_id,
+            };
+            if is_file_id {
+                continue;
+            }
+            let media = item_url(item);
+            if !media.starts_with("http://") && !media.starts_with("https://") {
+                out.push(std::path::PathBuf::from(media));
+            }
+        }
+        out
+    }
 }
 
 /// Telegram file id of the message's media, matched to the payload kind.
@@ -227,6 +260,30 @@ pub async fn invalidate_cache(task: &Task) {
     }
 }
 
+/// Locally produced media files (ugoira MP4, bsky remux MP4) whose temp dirs
+/// must stay alive while their task may be retried by the queue. The fetch
+/// pipeline hands ownership here via [`x_media::site::Fetched::take_keep_alive`]
+/// before the [`Fetched`] is dropped; a queued retry runs after that drop, so
+/// without this the local file would be gone by the time the retry sends it.
+/// Entries are removed when the task settles (see [`release_keep_alive`]).
+pub static KEEP_ALIVE: LazyLock<parking_lot::Mutex<Vec<tempfile::TempDir>>> =
+    LazyLock::new(|| parking_lot::Mutex::new(Vec::new()));
+
+/// Drops the keep-alive temp dirs holding media referenced by `task` (matched
+/// by path prefix). Called once a task settles — sent or permanently failed —
+/// so retry-only temp files do not leak; retryable tasks keep them alive.
+pub fn release_keep_alive(task: &Task) {
+    let paths = task.local_media_paths();
+    if paths.is_empty() {
+        return;
+    }
+    let mut alive = KEEP_ALIVE.lock();
+    alive.retain(|dir| {
+        let dir_path = dir.path();
+        !paths.iter().any(|p| p.starts_with(dir_path))
+    });
+}
+
 pub const MAX_MEDIA_GROUP: usize = 9;
 
 /// Splits media into batches of at most [`MAX_MEDIA_GROUP`] items.
@@ -595,6 +652,18 @@ async fn send_batch_via_upload(
     let mut items = Vec::new();
     for (i, item) in batch.iter().enumerate() {
         let item_caption = if i == 0 { caption } else { None };
+        // Locally produced files (ugoira / bsky remux MP4): nothing to
+        // download or shrink — upload the file directly. The send is a
+        // multipart upload, so the only remaining failure is an upload-cap
+        // error, which is permanent (a video cannot be re-encoded here).
+        let media_url = item_url(item);
+        if !media_url.starts_with("http://") && !media_url.starts_with("https://") {
+            let path = std::path::PathBuf::from(media_url);
+            let media = media_from_file(item, path, item_caption, item.thumbnail_url())
+                .map_err(|message| FallbackError::Permanent { message })?;
+            items.push(media);
+            continue;
+        }
         // Size check before downloading/uploading: over the cap, use the
         // smaller URL instead of the file. Photos are exempt — they are
         // downloaded and processed (downscale / PNG→JPEG) before uploading.
@@ -1199,6 +1268,8 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
                 }
                 Err(SendError::Permanent { message, task }) => {
                     invalidate_cache(&task).await;
+                    // The task settles here: drop any keep-alive temp media.
+                    release_keep_alive(&task);
                     return Err(QueueError::Permanent {
                         message,
                         payload: serde_json::to_value(task).expect("task serializes"),
@@ -1208,6 +1279,7 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
             if !resumed {
                 post_send_actions(&bot, &task, message_ids).await;
             }
+            release_keep_alive(&task);
             Ok(())
         }
         Task::ForwardMessages { .. } => match forward_messages(&bot, &task).await {
@@ -1219,10 +1291,13 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
                 delay_seconds,
                 payload: serde_json::to_value(task).expect("task serializes"),
             }),
-            Err(SendError::Permanent { message, task }) => Err(QueueError::Permanent {
+            Err(SendError::Permanent { message, task }) => {
-                message,
+                release_keep_alive(&task);
-                payload: serde_json::to_value(task).expect("task serializes"),
+                Err(QueueError::Permanent {
-            }),
+                    message,
+                    payload: serde_json::to_value(task).expect("task serializes"),
+                })
+            }
         },
     }
 }