From bd43a12dee16caffa7cbb3af2ba300ad7e53a8f7 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Thu, 13 Aug 2026 22:05:07 +0800 Subject: [PATCH] fix(send): keep local media (ugoira/bsky remux MP4) alive across queue retries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 8 +++ crates/xmedia-bot/src/handlers.rs | 12 ++++- crates/xmedia-bot/src/send.rs | 83 +++++++++++++++++++++++++++++-- 3 files changed, 98 insertions(+), 5 deletions(-) diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index b8bed21..340b1bb 100644 --- a/crates/x-media/src/site/mod.rs +++ b/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 { + self._keep_alive.take() + } } /// Renders a user-supplied caption format from raw (already-escaped) field diff --git a/crates/xmedia-bot/src/handlers.rs b/crates/xmedia-bot/src/handlers.rs index 4af07ba..19f8114 100644 --- a/crates/xmedia-bot/src/handlers.rs +++ b/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; } } diff --git a/crates/xmedia-bot/src/send.rs b/crates/xmedia-bot/src/send.rs index acfe088..f162eb3 100644 --- a/crates/xmedia-bot/src/send.rs +++ b/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 { + 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>> = + 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 { - message, - payload: serde_json::to_value(task).expect("task serializes"), - }), + Err(SendError::Permanent { message, task }) => { + release_keep_alive(&task); + Err(QueueError::Permanent { + message, + payload: serde_json::to_value(task).expect("task serializes"), + }) + } }, } }