perf: stop the sweep stealing worker wakeups, retry-free inline fetch

- queue: the lease-expiry sweep waited on the workers' `Notify`. `notify_one`
  stores a permit, so a sweep wakeup could consume the one meant for a worker,
  which then blocked on `notified()` (it only waits when the table looked
  empty, i.e. indefinitely) with a due row sitting there. The sweep now has
  its own notify, woken only by stop.
- x-media: split fetch's retry loop into `fetch` (3 attempts, unchanged) and
  `fetch_once` (1 attempt); inline queries use the latter — the 800ms debounce
  plus 1s/2s backoffs were outlasting the answer window of the query.
- send: chunk_media_items now moves items out of the input Vec instead of
  requiring `T: Clone` and copying every payload.

fmt/clippy clean, 55 + 69 tests pass.

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

@@ -375,10 +375,24 @@ pub async fn validate_all() -> Vec<(&'static str, String)> {
 /// are returned immediately; retrying them only wastes attempts against the
 /// source site.
 pub async fn fetch(url: &str) -> Result<Option<Fetched>, FetchError> {
+    fetch_with_attempts(url, MAX_FETCH_ATTEMPTS).await
+}
+
+/// [`fetch`] without the retry backoff (one attempt). For callers with a
+/// short deadline: an inline query's answer window is measured in seconds, so
+/// the 1s + 2s retry sleeps would outlast the query the answer belongs to.
+pub async fn fetch_once(url: &str) -> Result<Option<Fetched>, FetchError> {
+    fetch_with_attempts(url, 1).await
+}
+
+/// Total attempts of the retried [`fetch`] (3: the initial try plus two).
+const MAX_FETCH_ATTEMPTS: u32 = 3;
+
+async fn fetch_with_attempts(url: &str, attempts: u32) -> Result<Option<Fetched>, FetchError> {
     let Some(site) = find_site(url) else {
         return Ok(None);
     };
-    for attempt in 0..3u32 {
+    for attempt in 0..attempts.max(1) {
         match site.fetch_from_url(url).await {
             Ok(fetched) => {
                 // Per-request detail: debug only, keyed by the post id.
@@ -391,7 +405,7 @@ pub async fn fetch(url: &str) -> Result<Option<Fetched>, FetchError> {
                 return Ok(Some(fetched));
             }
             Err(err) => {
-                if site.is_retryable(&err) && attempt < 2 {
+                if site.is_retryable(&err) && attempt + 1 < attempts {
                     tokio::time::sleep(Duration::from_secs(1 << attempt)).await;
                 } else {
                     return Err(err);

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

@@ -121,7 +121,9 @@ async fn answer_inline_query(bot: Bot, query: InlineQuery) -> Result<bool, Reque
         query.query,
         log_key(&query.query)
     );
-    match x_media::site::fetch(&query.query).await {
+    // No retries: the debounce plus a 1s/2s backoff would outlast the inline
+    // query the answer belongs to.
+    match x_media::site::fetch_once(&query.query).await {
         Ok(Some(fetched)) => {
             let mut results: Vec<InlineQueryResult> = Vec::new();
             // Inline results have the same 1024-char caption limit as regular

crates/xmedia-bot/src/queue.rs

@@ -41,7 +41,13 @@ type DeadLetter = dyn Fn(Value, String) -> BoxFuture<'static, ()> + Send + Sync;
 
 pub struct PersistentTaskQueue {
     pool: std::sync::Arc<crate::db::DbPool>,
+    /// Wakes the workers when a row becomes leasable. `notify_one` stores a
+    /// permit, so nothing else may share it: a waiter that is not a worker
+    /// (the sweep) can consume the permit and leave the due row pending until
+    /// the next enqueue.
     notify: Arc<Notify>,
+    /// Wakes the lease-expiry sweep; `stop` is the only producer.
+    sweep_notify: Arc<Notify>,
     stop: Arc<AtomicBool>,
     worker: Mutex<Vec<JoinHandle<()>>>,
     counter: AtomicU64,
@@ -88,6 +94,7 @@ impl PersistentTaskQueue {
         Self {
             pool,
             notify: Arc::new(Notify::new()),
+            sweep_notify: Arc::new(Notify::new()),
             stop: Arc::new(AtomicBool::new(false)),
             worker: Mutex::new(Vec::new()),
             counter: AtomicU64::new(0),
@@ -119,12 +126,12 @@ impl PersistentTaskQueue {
             handles.push(tokio::spawn(worker.run_loop_supervised()));
         }
         // Periodic lease-expiry sweep: recovers rows a crashed/panicked
-        // worker left `in_progress` (the lock TTL bounds the wait). Woken by
-        // the same notify as the workers, so enqueue and stop interrupt the
-        // sleep; the first interval tick fires immediately (harmless extra
-        // recovery at startup).
+        // worker left `in_progress` (the lock TTL bounds the wait). Its own
+        // notify (not the workers'): sharing that one let this task consume a
+        // `notify_one` permit meant for a worker, which then slept through a
+        // due row until some later event. Only `stop` wakes it.
         let sweep_pool = std::sync::Arc::clone(&self.pool);
-        let sweep_notify = Arc::clone(&self.notify);
+        let sweep_notify = Arc::clone(&self.sweep_notify);
         let sweep_stop = Arc::clone(&self.stop);
         handles.push(tokio::spawn(async move {
             let mut interval = tokio::time::interval(Duration::from_secs(30));
@@ -150,6 +157,7 @@ impl PersistentTaskQueue {
     pub async fn stop(&self) {
         self.stop.store(true, Ordering::Relaxed);
         self.notify.notify_waiters();
+        self.sweep_notify.notify_waiters();
         let handles = std::mem::take(&mut *self.worker.lock());
         for handle in handles {
             let _ = handle.await;

crates/xmedia-bot/src/send.rs

@@ -295,12 +295,18 @@ pub fn release_keep_alive(task: &Task) {
 
 pub const MAX_MEDIA_GROUP: usize = 9;
 
-/// Splits media into batches of at most [`MAX_MEDIA_GROUP`] items.
-pub fn chunk_media_items<T: Clone>(items: Vec<T>) -> Vec<Vec<T>> {
-    items
-        .chunks(MAX_MEDIA_GROUP)
-        .map(|chunk| chunk.to_vec())
-        .collect()
+/// Splits media into batches of at most [`MAX_MEDIA_GROUP`] items, moving the
+/// items out (no per-item clone).
+pub fn chunk_media_items<T>(items: Vec<T>) -> Vec<Vec<T>> {
+    let mut items = items.into_iter();
+    let mut batches = Vec::new();
+    loop {
+        let batch: Vec<T> = items.by_ref().take(MAX_MEDIA_GROUP).collect();
+        if batch.is_empty() {
+            return batches;
+        }
+        batches.push(batch);
+    }
 }
 
 /// Orders media for a Telegram media group: when photos and videos are