diff --git a/AGENTS.md b/AGENTS.md index ef3b087..758747a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,7 +44,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr | `crates/xmedia-bot/src/handlers/` | Handler modules: `mod.rs` (message entry point, `reply`, `log_key`, the group-only `GROUP_LINK_HINT` for a supported link posted outside a private chat), `commands.rs` (teloxide `BotCommands` enum + command executor, incl. `/test ` (send-only) / `/debug ` (parse-only) and the admin-only `/bot_dict` state dump; `/set_format` rejects unknown `{…}` placeholders and resets with `-`), `urls.rs` (URL extraction + bounded job channel (256) drained by `URL_WORKERS = 8` workers (`start_url_workers`) — backpressure instead of unbounded spawns; teloxide's per-chat workers are sequential — batch-forwards need concurrency), `inline.rs`/`callback.rs` (inline queries / edit-before-forward buttons, incl. `skip`; a forward that fails retryably is both queued *and* settles the prompt — the queued row carries the message ids itself, and a prompt left live let a second Confirm copy the same messages twice and let Skip answer "nothing was forwarded" while the row still delivered), `statics.rs` (global statics) | | `crates/xmedia-bot/src/state.rs` | `ChatStore`: parking_lot `Mutex` cache + SQLite write-through (`chat_state` table) | | `crates/xmedia-bot/src/link_cache.rs` | `LinkCache`: SQLite-backed cache (`link_cache` table) of successfully sent posts — raw caption fields + Telegram `file_id`s; repeat links re-send locally (no fetch/upload), TTL + prune, invalidated on permanent send failure | -| `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, `notify_one` worker wakeup plus a separate `Notify` for the 30 s lease-expiry sweep (a shared one let the sweep steal the workers' wakeup permit), `busy_timeout` on all connections | +| `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, a finished row's `DELETE`/reschedule retried and a failed delete falling back to a `done` tombstone (the lease query and the sweep only look at `pending`/`in_progress`, so a task that already ran cannot be resurrected and re-run), `notify_one` worker wakeup plus a separate `Notify` for the 30 s lease-expiry sweep (a shared one let the sweep steal the workers' wakeup permit), `busy_timeout` on all connections | | `crates/xmedia-bot/src/ctx.rs` | `AppContext`: the injected collaborators (`sender` + `ChatStore`/`PersistentTaskQueue`/`LinkCache`/`Config`), `from_statics` for production and the `CONTEXT` static the worker closures hold. `test_support::TestStores` backs handler tests with a tempdir store set | | `crates/xmedia-bot/src/send/` | `send/mod.rs`: `Task`/`MediaItemPayload` payloads, `SendError`/`Classification`, `send_media_sequence`/`send_animation`/`forward_messages`; `send/input_media.rs`: payload → `InputFile`/`InputMedia` + `build_media_group` (caption on the first item only); `send/upload.rs`: the download-and-reupload fallback (`prepare_upload_item`/`send_batch_via_upload`, photo downscale handoff); `send/post_send.rs`: link-cache write, `KEEP_ALIVE` registry, `settle_task`, `post_send_actions`, `handle_task`/`dead_letter_notify` | | `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the user-flow surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`answer_callback_query`/`edit_message_caption`/`delete_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a recording `MockSender` in tests. Admin/setup APIs (`get_chat`, `set_my_commands`, …) stay on the concrete `Bot` | diff --git a/crates/x-media/src/site/bsky/interface.rs b/crates/x-media/src/site/bsky/interface.rs index 5fda8fc..8134efe 100644 --- a/crates/x-media/src/site/bsky/interface.rs +++ b/crates/x-media/src/site/bsky/interface.rs @@ -55,6 +55,10 @@ pub async fn fetch_from_url(url: &str) -> Result { // working on: the media URL is derived from what the user pasted, and // `warn` is a level operators share. let key = cache_key(url).unwrap_or_else(|| "?".into()); + // A failed remux is remembered: if it leaves the post with no media at + // all, returning `Ok` would read as "this post has no media" and skip the + // retry that a transient segment-download failure deserves. + let mut remux_failure: Option = None; for item in fetched.media { let is_hls = matches!(&item, Media::Video { url, .. } if url.contains("playlist") || url.ends_with(".m3u8")); @@ -76,10 +80,23 @@ pub async fn fetch_from_url(url: &str) -> Result { }); fetched._keep_alive = Some(keep_alive); } + // No ffmpeg: a deployment gap, not a bad moment — retrying it + // would only waste the fetch budget, so the post degrades (and an + // all-video post reports the media type as unsupported). Ok(None) => log::warn!("bsky video remux unavailable for [key={key}]"), - Err(e) => log::warn!("bsky video remux failed for [key={key}]: {e}"), + Err(e) => { + log::warn!("bsky video remux failed for [key={key}]: {e}"); + remux_failure = Some(e); + } } } + if media.is_empty() + && let Some(reason) = remux_failure + { + return Err(FetchError::Transient(format!( + "bsky video remux failed: {reason}" + ))); + } fetched.media = media; Ok(fetched) } diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index 1bba6f1..2608f64 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -269,15 +269,23 @@ pub enum FetchError { Io(std::io::Error), } -/// Shared HTTP client (browser User-Agent) for twitter/bsky fetches and -/// [`download_media`]. -pub(crate) static CLIENT: LazyLock = LazyLock::new(|| { +/// How long a download may make no progress: the response head, and then each +/// individual chunk, must arrive within this window. Deliberately *not* a +/// total timeout — see [`MEDIA_CLIENT`]. +const DOWNLOAD_IDLE_TIMEOUT: Duration = Duration::from_secs(30); + +/// Builds a client with the shared configuration (browser User-Agent, the +/// Bot API's proxy, per-runtime pools under test). `total_timeout` is what +/// differs between the two clients below. +fn build_client(total_timeout: Option) -> reqwest::Client { let mut builder = reqwest::Client::builder() .user_agent("Mozilla/5.0") + .connect_timeout(Duration::from_secs(10)); + if let Some(total) = total_timeout { // reqwest has no total timeout by default; a stalled connection // would otherwise pin a fetch/handler forever. - .timeout(Duration::from_secs(30)) - .connect_timeout(Duration::from_secs(10)); + builder = builder.timeout(total); + } // Route site fetches through the same proxy the Bot API uses, so a // network that needs TELOXIDE_PROXY (e.g. behind the GFW) does not // leave site fetches dead while the bot itself works. @@ -295,7 +303,53 @@ pub(crate) static CLIENT: LazyLock = LazyLock::new(|| { #[cfg(test)] let builder = builder.pool_max_idle_per_host(0); builder.build().expect("failed to build HTTP client") -}); +} + +/// Shared HTTP client (browser User-Agent) for the site fetches — metadata +/// requests, where 30s is generous. +pub(crate) static CLIENT: LazyLock = + LazyLock::new(|| build_client(Some(Duration::from_secs(30)))); + +/// Client for media *downloads*, with no total timeout: a 10 MiB fallback +/// download, or an ugoira frame zip that may be hundreds of MB, legitimately +/// takes minutes on a slow link — a 30s total cap made those posts impossible +/// to deliver at all (the size cap said 512 MiB, the clock said 30s). What a +/// stalled connection cannot do is hang a worker: the head and every chunk are +/// bounded by [`DOWNLOAD_IDLE_TIMEOUT`] instead (see [`next_chunk`]). +static MEDIA_CLIENT: LazyLock = LazyLock::new(|| build_client(None)); + +/// The error a download reports when it stops making progress. +fn download_stalled() -> FetchError { + FetchError::Transient(format!( + "download stalled for {}s", + DOWNLOAD_IDLE_TIMEOUT.as_secs() + )) +} + +/// Sends a media-download request: the response head must arrive within the +/// idle window, and a non-2xx status is classified by [`download_status_error`]. +async fn send_download(request: reqwest::RequestBuilder) -> Result { + let response = match tokio::time::timeout(DOWNLOAD_IDLE_TIMEOUT, request.send()).await { + Ok(Ok(response)) => response, + Ok(Err(e)) => return Err(e.into()), + Err(_) => return Err(download_stalled()), + }; + if response.status().is_success() { + Ok(response) + } else { + Err(download_status_error(response.status())) + } +} + +/// One body chunk, or `None` at the end. A body that stops delivering is a +/// transient download error rather than a hang. +async fn next_chunk(response: &mut reqwest::Response) -> Result, FetchError> { + match tokio::time::timeout(DOWNLOAD_IDLE_TIMEOUT, response.chunk()).await { + Ok(Ok(chunk)) => Ok(chunk), + Ok(Err(e)) => Err(e.into()), + Err(_) => Err(download_stalled()), + } +} /// Whether a usable `ffmpeg` binary is on PATH (probed once). Shared by the /// pixiv ugoira encoder and the bsky HLS remuxer. @@ -543,12 +597,7 @@ fn download_status_error(status: reqwest::StatusCode) -> FetchError { /// crossed (or when a declared Content-Length already exceeds it). Keeps the /// bot from buffering arbitrarily large bodies into memory. pub async fn download_media_limited(url: &str, max_bytes: u64) -> Result { - let response = apply_media_headers(CLIENT.get(url), url).send().await?; - let response = if response.status().is_success() { - response - } else { - return Err(download_status_error(response.status())); - }; + let response = send_download(apply_media_headers(MEDIA_CLIENT.get(url), url)).await?; if let Some(len) = response.content_length() && len > max_bytes { @@ -556,7 +605,7 @@ pub async fn download_media_limited(url: &str, max_bytes: u64) -> Result max_bytes { return Err(FetchError::TooLarge); @@ -581,12 +630,7 @@ pub async fn download_media_to_file( out: &mut std::fs::File, ) -> Result { use std::io::Write; - let response = apply_media_headers(CLIENT.get(url), url).send().await?; - let response = if response.status().is_success() { - response - } else { - return Err(download_status_error(response.status())); - }; + let response = send_download(apply_media_headers(MEDIA_CLIENT.get(url), url)).await?; if let Some(len) = response.content_length() && len > max_bytes { @@ -594,7 +638,7 @@ pub async fn download_media_to_file( } let mut response = response; let mut total: u64 = 0; - while let Some(chunk) = response.chunk().await? { + while let Some(chunk) = next_chunk(&mut response).await? { total += chunk.len() as u64; if total > max_bytes { return Err(FetchError::TooLarge); diff --git a/crates/x-media/src/site/pixiv/api.rs b/crates/x-media/src/site/pixiv/api.rs index 8bc49f0..0414714 100644 --- a/crates/x-media/src/site/pixiv/api.rs +++ b/crates/x-media/src/site/pixiv/api.rs @@ -76,6 +76,13 @@ impl PixivAPI { .header("User-Agent", AUTH_USER_AGENT) .send() .await?; + // Check the status *before* reading the body: a 429/5xx from the + // token endpoint is worth retrying (the class comes from + // `is_retryable`), while parsing a maintenance page as JSON turned it + // into a permanent `Api`/`Json` error with no retry at all. + if !response.status().is_success() { + return Err(PixivError::Status(response.status().as_u16())); + } let json: serde_json::Value = serde_json::from_str(&response.text().await?)?; let access_token = json .get("access_token") @@ -134,8 +141,11 @@ impl PixivAPI { let mut illustration = Illustration::from_model(&model); if matches!(&model.r#type, TypeModel::Ugoira) { // Real ugoira support: download the frame zip and encode an MP4. - // Without ffmpeg (or on encode failure) the post stays - // unsupported (empty media, like Python). + // Without ffmpeg the post stays unsupported (empty media, like + // Python) — but a *failed* download/encode is reported instead: + // a ugoira post has no static image to fall back to, so + // swallowing it would present a transient zip-download error as + // "this post has no media", with the retries skipped. match self.ugoira_video(illust_id).await { Ok(Some((mp4_path, _keep_alive))) => { illustration.media.push(Media::Video { @@ -146,7 +156,10 @@ impl PixivAPI { illustration._keep_alive = Some(_keep_alive); } Ok(None) => {} - Err(e) => log::error!("ugoira encode failed for {illust_id}: {e}"), + Err(e) => { + log::error!("ugoira encode failed for {illust_id}: {e}"); + return Err(FetchError::Pixiv(e)); + } } } Ok(illustration) diff --git a/crates/x-media/src/site/pixiv/interface.rs b/crates/x-media/src/site/pixiv/interface.rs index 978f434..1d926ab 100644 --- a/crates/x-media/src/site/pixiv/interface.rs +++ b/crates/x-media/src/site/pixiv/interface.rs @@ -50,8 +50,14 @@ impl Site for PixivSite { match super::api::validate().await { Ok(()) => Ok(()), Err(e) => { - // Keep the old behavior: a failed login disables pixiv - // for the rest of this process. + // A rejected credential disables pixiv for the rest of + // this process (it will not fix itself). A bad *moment* — + // a 5xx or a network error while the container comes up — + // must not: disabling on any error turned every later + // pixiv link into "pixiv support is disabled". + if pixiv_error_is_retryable(&e) { + return Err(format!("{e} (transient — pixiv stays enabled)")); + } super::api::disable(); Err(format!("{e}")) } @@ -84,18 +90,24 @@ pub fn cache_key(url: &str) -> Option { pub fn is_retryable(err: &FetchError) -> bool { match err { FetchError::Http(_) | FetchError::Transient(_) => true, - FetchError::Pixiv(e) => match e { - PixivError::Http(_) => true, - PixivError::Status(code) if *code == 429 || *code >= 500 => true, - PixivError::Status(_) - | PixivError::Api(_) - | PixivError::Json(_) - | PixivError::NoAuth => false, - }, + FetchError::Pixiv(e) => pixiv_error_is_retryable(e), _ => false, } } +/// The pixiv-specific half of the retry policy, shared with startup +/// validation: a bad moment (429/5xx, a network error) is retryable, a +/// rejected credential is not. +fn pixiv_error_is_retryable(err: &PixivError) -> bool { + match err { + PixivError::Http(_) => true, + PixivError::Status(code) if *code == 429 || *code >= 500 => true, + PixivError::Status(_) | PixivError::Api(_) | PixivError::Json(_) | PixivError::NoAuth => { + false + } + } +} + /// pximg.net is hotlink-protected: downloads must carry the pixiv Referer. /// The match is on the media host, not the site PATTERN — pixiv's PATTERN /// only matches `pixiv.net/artworks/...`, never `i.pximg.net`. @@ -410,6 +422,16 @@ mod tests { } } + #[test] + fn startup_validation_only_disables_on_a_definitive_failure() { + // A bad moment: the site must stay enabled for later links. + assert!(pixiv_error_is_retryable(&PixivError::Status(503))); + assert!(pixiv_error_is_retryable(&PixivError::Status(429))); + // A rejected credential is what `disable()` is for. + assert!(!pixiv_error_is_retryable(&PixivError::Status(403))); + assert!(!pixiv_error_is_retryable(&PixivError::NoAuth)); + } + #[test] fn is_retryable_classifies_transient_and_permanent() { // Transient: network errors, explicit transient, pixiv 429/5xx. diff --git a/crates/xmedia-bot/src/queue.rs b/crates/xmedia-bot/src/queue.rs index 174f2f5..aa2ba0f 100644 --- a/crates/xmedia-bot/src/queue.rs +++ b/crates/xmedia-bot/src/queue.rs @@ -17,6 +17,17 @@ use tokio::sync::Notify; use tokio::task::JoinHandle; pub const MAX_RETRIES: u32 = 2; + +/// Attempts for a *terminal* row write (delete / reschedule). These are not +/// like a task retry: failing them leaves the row in `in_progress`, where the +/// expiry sweep can re-run a task that already ran, so a contended DB gets a +/// few quick chances before the caller falls back to a terminal state. +const TERMINAL_WRITE_ATTEMPTS: u32 = 3; + +/// 100ms, 200ms, … between terminal write attempts. +fn terminal_write_backoff(attempt: u32) -> Duration { + Duration::from_millis(100 * (1u64 << attempt.min(4))) +} pub const LOCK_TTL_SECONDS: f64 = 120.0; /// Number of concurrent worker loops. Tasks are independent (retries and @@ -238,6 +249,22 @@ impl PersistentTaskQueue { } } +/// Last-resort terminal state for a row whose `DELETE` would not go through: +/// `done` is invisible to `lease_next` (`status='pending'`), to the expiry +/// sweep (`status='in_progress'`) and to the backlog line, so a task that +/// already ran cannot be leased and run again. +async fn mark_done(pool: &std::sync::Arc, id: &str) -> rusqlite::Result<()> { + let id = id.to_string(); + pool.with_conn(move |conn| { + conn.execute( + "UPDATE tasks SET status='done', locked_until=0 WHERE id = ?1", + params![id], + )?; + Ok(()) + }) + .await +} + /// Which task a lease/retry/dead-letter line is about: the chat from the /// stored payload, plus the post's normalized cache key when the payload /// carries one (`ForwardMessages` has no source URL). Without these a queue @@ -474,36 +501,88 @@ impl QueueWorker { } } + /// Deletes a finished row. A failure here is not cosmetic: the row would + /// stay `in_progress` with a live lease, the next sweep would flip it back + /// to `pending`, and the *completed* task would run again — a second album, + /// a second edit prompt, a second channel copy. So the delete is retried + /// (a busy/contended DB is the usual cause and clears), and if the DB still + /// refuses, the row is marked `done` — a status neither the lease query + /// (`pending`) nor the sweep (`in_progress`) looks at — so a task that + /// already ran can never be re-leased. Both writes failing is logged at + /// error level with the row id, since that is the one case where a + /// duplicate send stays possible. async fn delete_row(&self, id: &str) { + for attempt in 0..TERMINAL_WRITE_ATTEMPTS { + match self.try_delete_row(id).await { + Ok(()) => return, + Err(e) => { + log::error!("queue delete failed (attempt {}): {e}", attempt + 1); + tokio::time::sleep(terminal_write_backoff(attempt)).await; + } + } + } + match mark_done(&self.pool, id).await { + Ok(()) => log::warn!("queue: row {id} marked done instead of deleted"), + Err(e) => log::error!( + "queue: row {id} could not be deleted or marked done ({e}); \ + the expiry sweep may run this finished task again" + ), + } + } + + async fn try_delete_row(&self, id: &str) -> rusqlite::Result<()> { let id = id.to_string(); - let result = self - .pool + self.pool .with_conn(move |conn| { conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])?; Ok(()) }) - .await; - if let Err(e) = result { - log::error!("queue delete failed: {e}"); - } + .await } + /// Writes back a retryable attempt's state. A failure is retried: the row + /// would otherwise stay `in_progress`, and the expiry sweep would re-run + /// the attempt from its *previous* payload — re-sending batches the last + /// attempt had already delivered. Unlike [`Self::delete_row`] there is no + /// safe terminal fallback here (marking it done would drop the retry + /// without telling anyone), so a persistent failure is logged loudly and + /// the sweep's re-run — at-least-once, the documented trade — is named. async fn reschedule(&self, id: &str, payload: Value, delay_seconds: f64, attempts: i32) { let id = id.to_string(); let payload = payload.to_string(); - let result = self.pool.with_conn(move |conn| { - conn.execute( - "UPDATE tasks SET payload=?1, run_after=?2, attempts=?3, status='pending', locked_until=0 WHERE id=?4", - params![payload, now_f64() + delay_seconds, attempts, id], - )?; - Ok(()) - }) - .await; - if let Err(e) = result { - log::error!("queue reschedule failed: {e}"); + let run_after = now_f64() + delay_seconds; + let mut last_error = None; + for attempt in 0..TERMINAL_WRITE_ATTEMPTS { + let id = id.clone(); + let payload = payload.clone(); + let result = self + .pool + .with_conn(move |conn| { + conn.execute( + "UPDATE tasks SET payload=?1, run_after=?2, attempts=?3, status='pending', locked_until=0 WHERE id=?4", + params![payload, run_after, attempts, id], + )?; + Ok(()) + }) + .await; + match result { + Ok(()) => { + // Same permit semantics as enqueue: never lose the wakeup. + self.notify.notify_one(); + return; + } + Err(e) => { + log::error!("queue reschedule failed (attempt {}): {e}", attempt + 1); + last_error = Some(e.to_string()); + tokio::time::sleep(terminal_write_backoff(attempt)).await; + } + } } - // Same permit semantics as enqueue: never lose the wakeup. - self.notify.notify_one(); + log::error!( + "queue: row {id} could not be rescheduled ({}); the expiry sweep will \ + re-run this attempt from its previous state", + last_error.unwrap_or_default() + ); } } @@ -578,6 +657,47 @@ mod tests { assert_eq!(row_fields(&serde_json::Value::Null), ""); } + /// The row a leaked deletion would resurrect: `done` is invisible to the + /// lease query, so a task that already ran cannot be run again. + #[tokio::test] + async fn done_rows_are_never_leased() { + let (queue, _dir) = new_queue().await; + let runs = Arc::new(AtomicUsize::new(0)); + queue + .enqueue(serde_json::json!({"chat_id": 1}), now_f64()) + .await + .unwrap(); + let id: String = queue + .pool + .with_conn(|conn| conn.query_row("SELECT id FROM tasks", [], |r| r.get(0))) + .await + .unwrap(); + mark_done(&queue.pool, &id).await.unwrap(); + + assert_eq!( + queue.pending_backlog().await, + None, + "a done row is not pending work" + ); + let runs_worker = runs.clone(); + queue + .start( + move |_payload| { + runs_worker.fetch_add(1, AtomicOrdering::SeqCst); + async { Ok(()) } + }, + |_payload, _message| async {}, + ) + .await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + runs.load(AtomicOrdering::SeqCst), + 0, + "the finished row must not run again" + ); + queue.stop().await; + } + #[tokio::test] async fn pending_backlog_counts_only_unleased_rows() { let (queue, _dir) = new_queue().await;