fix(send): release one keep-alive reference per settled task

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

@@ -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);

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<parking_lot::Mutex<Vec<std::sync::Arc<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.
+/// 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"),