From 33b1f04f0ef17b2951c015d6fefbbd39f61d601a Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Thu, 24 Sep 2026 00:48:28 +0800 Subject: [PATCH] fix(send): release one keep-alive reference per settled task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit release_keep_alive retained every KEEP_ALIVE entry whose path matched the settling task, deleting the shared TempDir out from under a concurrent duplicate of the same post (a shared fetch pushes one Arc per pipeline): the duplicate's queued retry then dead-lettered on local media that no longer existed. Release now removes exactly one matching entry, which requires settle to run once per task — handle_task settled on Permanent right before the queue invoked dead_letter_notify, which settles the same payload again, so the redundant settle is dropped. A regression test pins one settle to one entry removed. --- crates/xmedia-bot/src/send/mod.rs | 35 +++++++++++++++++++++++++ crates/xmedia-bot/src/send/post_send.rs | 29 +++++++++++++------- 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index 860dea5..4c63160 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -1499,6 +1499,41 @@ mod tests { ); } + #[tokio::test] + async fn one_settle_drops_only_one_shared_keep_alive_holder() { + // Two pipelines of the same post (shared fetch) push the same temp + // dir once each. One task settling must drop only its own reference — + // clearing every holder would delete the file out from under the + // other task's queued retry, which would then dead-letter on a local + // media that no longer exists. + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("ugoira.mp4"); + std::fs::write(&file, b"not-a-real-mp4").unwrap(); + let task = sequence_task(file.to_str().unwrap()); + let dir_path = dir.path().to_path_buf(); + let dir = std::sync::Arc::new(dir); + { + let mut alive = KEEP_ALIVE.lock(); + alive.push(std::sync::Arc::clone(&dir)); + alive.push(std::sync::Arc::clone(&dir)); + } + + // `Settled::Sent` never talks to the API, so no outcome is scripted. + let sender = MockSender::scripted(vec![], media_fetch_error); + let stores = TestStores::new(); + let ctx = stores.ctx(&sender); + settle_task(&ctx, &task, Settled::Sent).await; + + let holders = KEEP_ALIVE + .lock() + .iter() + .filter(|d| d.path() == dir_path) + .count(); + // Drop this test's entries so the registry does not outlive it. + KEEP_ALIVE.lock().retain(|d| d.path() != dir_path); + assert_eq!(holders, 1, "one settle must drop exactly one shared holder"); + } + #[tokio::test] async fn post_send_opens_and_records_the_edit_prompt() { let sender = MockSender::scripted(vec![Outcome::MessageOk], media_fetch_error); diff --git a/crates/xmedia-bot/src/send/post_send.rs b/crates/xmedia-bot/src/send/post_send.rs index 56cf4c2..8f82706 100644 --- a/crates/xmedia-bot/src/send/post_send.rs +++ b/crates/xmedia-bot/src/send/post_send.rs @@ -122,24 +122,29 @@ async fn invalidate_cache(ctx: &AppContext<'_>, task: &Task) { /// that drop, so without this the local file would be gone by the time the /// retry sends it. `Arc` because one fetch can serve several tasks (a /// concurrent duplicate of the same link shares it): each holder keeps the -/// directory alive until its own task settles. Entries are removed when the +/// directory alive until its own task settles. Its entry is removed when that /// task settles (see [`release_keep_alive`]). pub(crate) 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. +/// Drops the keep-alive reference this task's pipeline pushed (one entry, +/// matched by path prefix). Called once a task settles — sent or permanently +/// failed — so retry-only temp files do not leak; retryable tasks keep theirs. +/// Exactly one entry goes per call: a shared fetch pushes one per pipeline, so +/// clearing every holder would delete the directory out from under a +/// concurrent duplicate's queued retry. pub(crate) 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)) - }); + if let Some(index) = alive + .iter() + .position(|dir| paths.iter().any(|p| p.starts_with(dir.path()))) + { + alive.remove(index); + } } /// The edit-before-forward prompt's text. It names both controls and the TTL, @@ -426,7 +431,10 @@ pub(crate) async fn handle_task( }); } Err(SendError::Permanent { message, task }) => { - settle_task(ctx, &task, Settled::Failed).await; + // The queue dead-letters this payload into + // `dead_letter_notify`, which settles the task — settling + // here as well would release a shared keep-alive + // directory twice. return Err(QueueError::Permanent { message, payload: serde_json::to_value(task).expect("task serializes"), @@ -454,7 +462,8 @@ pub(crate) async fn handle_task( payload: serde_json::to_value(task).expect("task serializes"), }), Err(SendError::Permanent { message, task }) => { - settle_task(ctx, &task, Settled::Failed).await; + // Settled by `dead_letter_notify`, which the queue invokes for + // this payload. Err(QueueError::Permanent { message, payload: serde_json::to_value(task).expect("task serializes"),