perf: stop spending API calls the link pipeline cannot use

Two calls per link went out that could not affect anything:

- `run_with_chat_action` awaited the opening `send_chat_action` to completion
  before the pipeline was polled at all, and again inside the loop on every
  `ACTION_REFRESH`. Telegram round trips are hundreds of ms: the first delay
  came out of the user's wait for every link, and each refresh suspended the
  fetch (an ugoira encode or HLS remux runs for seconds) by the same amount.
- `handle_message` enqueued *every* URL a private chat posted, including
  links no site adapter claims. Those cost a queue slot, a worker wake-up,
  a `Message` clone and (through the action above) one Telegram call, only
  for `url_media_inner` to conclude there was nothing to send. The group
  branch has always made the `cache_key(url).is_some()` test before it acts;
  the private branch now makes it before it enqueues.

The in-flight action is held (`Option<BoxFuture>` — the sender surface is
already type-erased, so it is `Unpin`) and polled as its own `select!`
branch: still polled *before* the pipeline, so the indicator is on screen
before the first send, but a slow Telegram response can no longer delay the
pipeline, and none of the branch bodies ever awaits one. One action is in
flight at a time; a refresh while one is unanswered is skipped rather than
dropping the request mid-flight. Note that `select!` evaluates every branch's
future expression eagerly, so the `None` case is an `async` block whose
`unwrap` only runs when the branch is polled (the eager form panicked).

Behavior pinned by the existing tests, unchanged: the opening action precedes
the first send, a 12s pipeline still sees exactly three actions
(`a_long_pipeline_keeps_the_chat_action_alive`), and an unsupported URL
reaching `url_media` still gets the one indicator before the pipeline settles
— in production it no longer reaches `url_media` at all.

`cargo fmt --check`, `cargo clippy --workspace --all-targets --locked -- -D
warnings` and `cargo test --workspace --locked` clean.

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

@@ -240,11 +240,18 @@ pub(crate) async fn handle_message(
         return respond(());
     }
     if is_private {
-        let urls = extract_urls(&message);
+        // Only links a site adapter claims: an unsupported URL never gets a
+        // media message, so enqueuing it would spend a queue slot, a worker
+        // wake-up and (through `run_with_chat_action`) a Telegram call on
+        // nothing. Same test the group branch below makes for its hint.
+        let urls: Vec<String> = extract_urls(&message)
+            .into_iter()
+            .filter(|url| x_media::site::cache_key(url).is_some())
+            .collect();
         if !urls.is_empty() {
             // Debug only, and echo the normalized keys instead of the raw URLs.
             let keys: Vec<String> = urls.iter().map(|u| log_key(u)).collect();
-            log::debug!("extracted {} URL(s): {keys:?}", urls.len());
+            log::debug!("queuing {} supported URL(s): {keys:?}", urls.len());
         }
         for url in urls {
             // Clone out of the lock: the parking_lot guard is !Send and must

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

@@ -10,6 +10,7 @@ use crate::state::ChatData;
 use std::collections::HashSet;
 use std::future::Future;
 use std::sync::LazyLock;
+use teloxide::RequestError;
 use teloxide::types::{ChatAction, ChatId, Message, MessageEntityKind, MessageId};
 use x_media::media::Media;
 
@@ -367,33 +368,50 @@ pub(crate) async fn url_media(
 /// expires an action after ~5s, while a fetch (ugoira encode, HLS remux) plus a
 /// download-and-reupload fallback routinely takes longer. The pipeline updates
 /// `hint` when it knows what it is sending.
+///
+/// No action is ever awaited *ahead* of the pipeline: doing that held the loop
+/// — and with it the fetch the user is waiting for — for a Telegram round trip,
+/// once before the pipeline was polled at all and again every
+/// [`ACTION_REFRESH`]. The in-flight send is held and polled *beside* the
+/// pipeline instead: the opening indicator still goes out before the pipeline's
+/// own first call (that is what it is for), but a slow API can no longer delay
+/// anything but the next indicator.
 async fn run_with_chat_action<F: Future<Output = ()>>(
     sender: &dyn MediaSender,
     chat_id: i64,
     hint: &parking_lot::Mutex<ActionHint>,
     pipeline: F,
 ) {
-    // The guard is released before the await: a parking_lot guard held across
-    // it makes the future !Send, and the URL workers spawn these.
-    let action = hint.lock().action();
-    if let Err(e) = sender.send_chat_action(ChatId(chat_id), action).await {
+    let warn = |e: RequestError| {
         // Cosmetic indicator: a failure degrades the experience, it does not
         // break the send (a group where the bot cannot send actions).
         log::warn!("send_chat_action failed for chat {chat_id}: {e}");
-    }
+    };
+    // The guard is released before the await: a parking_lot guard held across
+    // it makes the future !Send, and the URL workers spawn these.
+    let mut action = Some(sender.send_chat_action(ChatId(chat_id), hint.lock().action()));
     tokio::pin!(pipeline);
     loop {
         tokio::select! {
-            // `biased` polls the pipeline first, so a finished pipeline returns
-            // without ever arming the refresh timer (no stray actions).
+            // `biased` fixes the order below: the indicator is polled ahead of
+            // the pipeline, and a finished pipeline returns without arming the
+            // refresh timer (no stray actions).
             biased;
+            // `select!` evaluates every branch's future expression eagerly, so
+            // the `None` case is an inert block: the guard is what keeps it
+            // from being polled (and from unwrapping a `None`).
+            result = async { action.as_mut().unwrap().await }, if action.is_some() => {
+                action = None;
+                if let Err(e) = result {
+                    warn(e);
+                }
+            }
             () = &mut pipeline => return,
             () = tokio::time::sleep(ACTION_REFRESH) => {
-                let action = hint.lock().action();
-                if let Err(e) = sender.send_chat_action(ChatId(chat_id), action).await {
-                    // Cosmetic indicator: a failure degrades the experience, it does not
-        // break the send (a group where the bot cannot send actions).
-        log::warn!("send_chat_action failed for chat {chat_id}: {e}");
+                // One action in flight at a time: re-arming while the previous
+                // send is still unanswered would drop it mid-request.
+                if action.is_none() {
+                    action = Some(sender.send_chat_action(ChatId(chat_id), hint.lock().action()));
                 }
             }
         }