From 3fb8421c3a81eab540b10e98ceea1a0c53e39075 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Wed, 16 Sep 2026 21:23:53 +0800 Subject: [PATCH] perf: stop the sweep stealing worker wakeups, retry-free inline fetch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 18 ++++++++++++++++-- crates/xmedia-bot/src/handlers/inline.rs | 4 +++- crates/xmedia-bot/src/queue.rs | 18 +++++++++++++----- crates/xmedia-bot/src/send.rs | 18 ++++++++++++------ 4 files changed, 44 insertions(+), 14 deletions(-) diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index c6d3e0f..a54c766 100644 --- a/crates/x-media/src/site/mod.rs +++ b/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, 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, 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, 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, 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); diff --git a/crates/xmedia-bot/src/handlers/inline.rs b/crates/xmedia-bot/src/handlers/inline.rs index aa0e041..fd5bdc0 100644 --- a/crates/xmedia-bot/src/handlers/inline.rs +++ b/crates/xmedia-bot/src/handlers/inline.rs @@ -121,7 +121,9 @@ async fn answer_inline_query(bot: Bot, query: InlineQuery) -> Result { let mut results: Vec = Vec::new(); // Inline results have the same 1024-char caption limit as regular diff --git a/crates/xmedia-bot/src/queue.rs b/crates/xmedia-bot/src/queue.rs index abf7e3f..a002f8b 100644 --- a/crates/xmedia-bot/src/queue.rs +++ b/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, + /// 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, + /// Wakes the lease-expiry sweep; `stop` is the only producer. + sweep_notify: Arc, stop: Arc, worker: Mutex>>, 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; diff --git a/crates/xmedia-bot/src/send.rs b/crates/xmedia-bot/src/send.rs index 375b8c9..d166c85 100644 --- a/crates/xmedia-bot/src/send.rs +++ b/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(items: Vec) -> Vec> { - 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(items: Vec) -> Vec> { + let mut items = items.into_iter(); + let mut batches = Vec::new(); + loop { + let batch: Vec = 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