diff --git a/AGENTS.md b/AGENTS.md index 5f1e030..b01b031 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -46,7 +46,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr | `crates/xmedia-bot/src/link_cache.rs` | `LinkCache`: SQLite-backed cache (`link_cache` table) of successfully sent posts — raw caption fields + the source media URLs + Telegram `file_id`s; repeat links re-send locally (no fetch/upload), TTL + prune; a permanent send failure *degrades* the entry instead of dropping it (the file ids go, the URLs stay, so the next request re-sends from those without a fetch), and a degraded entry that fails again is removed | | `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 `lease_token` fence: `lease_next` stamps a random token and every write-back (heartbeat, `delete`, `reschedule`, `mark_done`) is guarded by it, so a lease that expired and was re-leased cannot be written by its former holder — a lost lease stops the attempt instead; 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), `runnable_rows`/`replace_payload` (the startup repair's read/rewrite path: it runs before the workers exist, which is why it needs no lease token), `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; the sweep does notify the workers after it actually recovered a row, since a recovered task is due immediately while every worker may be parked on `notify` with no pending row to sleep on), `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, and the module also carries the fixtures those tests share — the canonical cached post (`cached_photo`), the edit-before-forward prompt (`seed_prompt` with its `PROMPT_ID`/`FORWARDED_ID`) and a scripted API error (`api_error`) — so no two test modules keep their own copies | -| `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/send/` | `send/mod.rs`: `Task`/`MediaItemPayload` payloads, `send_media_sequence`/`send_animation`/`forward_messages`; `send/error.rs`: the Bot API error policy (`SendError`/`Classification`, `classify_request_error`, the media-fetch/size markers); `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/{mod.rs,test_support.rs}` | `MediaSender` trait: the user-flow surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`answer_callback_query`/`edit_message_text`/`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`. `test_support.rs` (cfg(test)-only) holds the scripted `MockSender` and `fake_api` (the stand-in API the real-`Bot` tests drive) | | `crates/xmedia-bot/src/rate_limit.rs` | Two token buckets paced before sends reach the API so batch forwards don't trip flood control: one per chat (`CAPACITY = 20`, ~20 msg/min refill) and one bot-wide (`acquire_global`, 30/s — Telegram's per-bot ceiling, invisible to any per-chat bucket and only binding when a batch fans out over many chats). `prune_idle` drops the per-chat buckets that refilled while unheld | @@ -82,7 +82,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi |---|---| | `crates/xmedia-bot/src/main.rs` | Startup sequence, webhook vs polling, graceful shutdown (SIGINT via teloxide ctrlc / SIGTERM via `stop_token` for docker, → sweep stop → admin msg → queue stop) | | `crates/xmedia-bot/src/handlers/` | `statics.rs` = `CHAT_STORE`/`TASK_QUEUE`/`CONFIG` singletons (open `$DATA_DIR/task_queue.db`, default `data/` **relative to CWD**, dir auto-created); `mod.rs` also holds `apply_caption_edit`, the one place a caption edit is applied and its failure classified: a short retryable delay is retried once, anything else is reported to the user instead of being swallowed (`callback.rs`'s template button answers its toast with the failure and leaves the record alone); `commands.rs` = command dispatch (incl. `/test ` send-only, `/debug ` parse-only, the read-only `/settings` every chat member can read — unlike the admin-only `/bot_dict` raw dump — and template removal; `/start`/`/help` carry the guidance teloxide's `descriptions()` cannot render, and `/set_format` rejects unknown `{…}` placeholders, resetting with `-`); `urls.rs` = URL extraction + the per-URL pipeline (`url_media` takes a `PostSend` mode: chat settings vs `/test`'s suppressed actions); `inline.rs` = debounced inline queries (hotlink-protected and local media skipped); `callback.rs` = edit-before-forward buttons (dptree entry + testable `handle_callback` core, incl. `skip`) | -| `crates/xmedia-bot/src/send/` | `mod.rs`: constants `MAX_MEDIA_GROUP = 10`; `classify_request_error` (5xx/non-JSON bodies retry, see the Retries bullet) and the media-fetch markers that route a URL send into the reupload fallback — including `failed to get HTTP url content`, the description single-media URL sends answer with; the senders. `upload.rs`: download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`), with a download's class from `classify_download_error` (transport/429/5xx retry; 4xx is permanent — the media itself is gone or refused — and a temp-file *write* failure retries, being resource exhaustion far more often than a broken temp dir). Item preparation is bounded **process-wide** (`PREP_SLOTS` in `upload.rs`: URL workers and queue workers can each be inside a batch, so a per-batch bound is not a memory bound), and the check that routes an oversized item to `fallback_url` is the download's own declared-Content-Length abort (`FetchError::TooLarge` → `MediaTooLarge`) — there is no separate size probe, which used to cost a second request per item. `post_send.rs`: settlement (`settle_task`), cache write, post-send actions (dead-letter text via `failure_text`: post key + cause, since the raw error alone does not say which link died), queue handlers. `input_media.rs`: payload → `InputMedia` | +| `crates/xmedia-bot/src/send/` | `mod.rs`: constants `MAX_MEDIA_GROUP = 10` and the senders; `error.rs`: `classify_request_error` (5xx/non-JSON bodies retry, see the Retries bullet) and the media-fetch markers that route a URL send into the reupload fallback — including `failed to get HTTP url content`, the description single-media URL sends answer with. `upload.rs`: download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`), with a download's class from `classify_download_error` (transport/429/5xx retry; 4xx is permanent — the media itself is gone or refused — and a temp-file *write* failure retries, being resource exhaustion far more often than a broken temp dir). Item preparation is bounded **process-wide** (`PREP_SLOTS` in `upload.rs`: URL workers and queue workers can each be inside a batch, so a per-batch bound is not a memory bound), and the check that routes an oversized item to `fallback_url` is the download's own declared-Content-Length abort (`FetchError::TooLarge` → `MediaTooLarge`) — there is no separate size probe, which used to cost a second request per item. `post_send.rs`: settlement (`settle_task`), cache write, post-send actions (dead-letter text via `failure_text`: post key + cause, since the raw error alone does not say which link died), queue handlers. `input_media.rs`: payload → `InputMedia` | | `crates/xmedia-bot/src/photo.rs` | Pure-Rust photo processing (no ffmpeg): `png` (image-png) decode/encode + `zune-jpeg` decode + `fast_image_resize` Lanczos3 downscale + `jpeg-encoder`. Photos over Telegram's limits (width + height > 10000 px → `PHOTO_INVALID_DIMENSIONS`; bytes > 10 MiB) are decoded, downscaled keeping the format, PNG bit depth > 24 (RGBA 32-bit / 16-bit per channel) reduced to 24-bit RGB with alpha flattened white (≤24-bit untouched, never upconverted), and transcoded to JPEG only if still over the cap; memory budget guarded, otherwise the item's smaller fallback URL. Two budgets, not one: `MAX_PHOTO_DOWNLOAD_BYTES` (32 MiB) caps the *download* in the send fallback — the whole body is buffered, once per prep slot — while `MAX_DECODE_BYTES` (512 MiB) stays the pre-allocation guard that decides whether a decoded photo can be processed at all; over either one the item degrades to its smaller URL | | `crates/x-media/src/site/mod.rs` | Dispatcher, `Fetched`/`FetchError`, shared `CLIENT`, `download_media_limited`/`download_media_to_file` (add `Referer: https://www.pixiv.net/` for `pximg.net` hotlink protection), `needs_media_headers` (the same per-site rule, asked by the inline path to skip what Telegram cannot fetch) | | `crates/x-media/src/site/pixiv/api.rs` | OAuth token exchange (hardcoded app client id/secret), access-token cache, ugoira zip→MP4 via ffmpeg in `spawn_blocking` | diff --git a/crates/xmedia-bot/src/send/error.rs b/crates/xmedia-bot/src/send/error.rs new file mode 100644 index 0000000..e56a08d --- /dev/null +++ b/crates/xmedia-bot/src/send/error.rs @@ -0,0 +1,161 @@ +//! The Telegram error policy: which failures the send paths retry, which are +//! permanent, and which the download-and-reupload fallback owns. A status a +//! *site* answers with is classified in `x_media::site`; this is the Bot API's +//! side of the same question. + +use super::upload::FallbackError; +use super::{Task, retry_delay_seconds}; +use teloxide::{ApiError, RequestError}; + +/// Telegram's servers failed to fetch a media URL (hotlink protection etc.): +/// these errors are handled by the download-and-reupload fallback, NOT by a +/// queue retry (resending the URL cannot succeed). +pub fn is_media_fetch_failure(e: &ApiError) -> bool { + const MARKERS: [&str; 7] = [ + "webpage_media_empty", + "media_empty", + "empty_web_media", + "webpage_curl_failed", + "timeout", + // Oversized photos (width + height > 10000 px) are rejected on URL + // sends too; route them to the download-and-resize fallback. + "photo_invalid_dimensions", + // Telegram refused to fetch the URL it was handed. Single-media URL + // sends answer with this one (the media-group verbs use the + // `webpage_*`/`media_empty` markers above), and it is exactly the + // case the download-and-reupload fallback exists for. + "failed to get http url content", + ]; + let description = e.to_string().to_lowercase(); + MARKERS.iter().any(|marker| description.contains(marker)) +} + +/// Telegram reported the media file as too large (HTTP 413 on multipart +/// upload, or a "too large" message for URL-fetched media). These errors are +/// handled by the size-check fallback (use a smaller media URL), NOT by a +/// queue retry. +pub fn is_size_error(e: &ApiError) -> bool { + if matches!(e, ApiError::RequestEntityTooLarge) { + return true; + } + let description = e.to_string().to_lowercase(); + ["too large", "too big"] + .iter() + .any(|marker| description.contains(marker)) +} + +/// Task-free classification of a Telegram request error. The callers attach +/// the (updated) task when building a [`SendError`]. +pub enum Classification { + Retryable { + delay_seconds: f64, + }, + Permanent { + message: String, + }, + /// Handled by the download fallback, not a queue retry. + MediaFetchFailure, +} + +pub fn classify_request_error(e: &RequestError) -> Classification { + match e { + RequestError::RetryAfter(seconds) => Classification::Retryable { + delay_seconds: seconds.seconds() as f64, + }, + RequestError::Network(_) => Classification::Retryable { + delay_seconds: retry_delay_seconds(0), + }, + // A 5xx from the API — or from a proxy in front of it — is transient. + // teloxide only sleeps 10s on a server error and then parses whatever + // body came back, so by the time we see the error the HTTP status is + // gone: a JSON 5xx body arrives as an unknown description, an HTML + // error page as `InvalidJson`. Both used to be Permanent, which + // dead-lettered a post over a Telegram-side blip. + RequestError::Api(api) if is_server_error_text(&api.to_string()) => { + Classification::Retryable { + delay_seconds: retry_delay_seconds(0), + } + } + RequestError::Api(api) if is_media_fetch_failure(api) => Classification::MediaFetchFailure, + RequestError::Api(api) => Classification::Permanent { + message: api.to_string(), + }, + // An unparsable body can only come from something that is not the Bot + // API (which always answers JSON): a 5xx/error page from an + // intermediary, cut off mid-response. A JSON body that merely does not + // match the expected type cannot be fixed by retrying, so that case + // stays permanent. + RequestError::InvalidJson { raw, .. } if !raw.trim_start().starts_with('{') => { + Classification::Retryable { + delay_seconds: retry_delay_seconds(0), + } + } + RequestError::MigrateToChatId(_) + | RequestError::InvalidJson { .. } + | RequestError::Io(_) => Classification::Permanent { + message: e.to_string(), + }, + } +} + +/// Descriptions a 5xx carries when its body *is* JSON (teloxide keeps only the +/// description text, never the status code). Matched like the media-fetch +/// markers below; anything unmatched stays permanent, so a new permanent API +/// error is not retried just because it is unfamiliar. +fn is_server_error_text(description: &str) -> bool { + const MARKERS: [&str; 4] = [ + "server error", + "bad gateway", + "gateway timeout", + "service unavailable", + ]; + let description = description.to_lowercase(); + MARKERS.iter().any(|marker| description.contains(marker)) +} + +/// Task boxed to keep the error size within `result_large_err` limits. +#[derive(Debug)] +pub enum SendError { + Retryable { delay_seconds: f64, task: Box }, + Permanent { message: String, task: Box }, +} +pub(crate) fn classify_to_send_error( + e: &RequestError, + task: Task, + fetch_failure_label: &str, +) -> SendError { + match classify_request_error(e) { + Classification::Retryable { delay_seconds } => SendError::Retryable { + delay_seconds, + task: Box::new(task), + }, + Classification::Permanent { message } => SendError::Permanent { + message, + task: Box::new(task), + }, + Classification::MediaFetchFailure => SendError::Permanent { + message: fetch_failure_label.into(), + task: Box::new(task), + }, + } +} + +impl SendError { + /// Attaches the (updated) task to a task-free [`FallbackError`] from the + /// download/upload pipeline. [`FallbackError::MediaTooLarge`] never + /// escapes the pipeline (it is handled by falling back to the smaller + /// URL), so it is unreachable here. + pub(super) fn from_fallback(f: FallbackError, task: Task) -> SendError { + match f { + FallbackError::Retryable { delay_seconds } => SendError::Retryable { + delay_seconds, + task: Box::new(task), + }, + FallbackError::Permanent { message } => SendError::Permanent { + message, + task: Box::new(task), + }, + FallbackError::MediaTooLarge => unreachable!("handled inside the upload fallback"), + } + } +} diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index 0fa2830..0c33585 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -4,9 +4,10 @@ //! blocked by hotlink protection, so the bot downloads the file itself and //! uploads it via multipart), [`post_send`] covers everything around a send //! (cache write, keep-alive media, settlement, post-send actions, queue -//! entry points). This module keeps the payload types, the error -//! classification and the senders themselves. +//! entry points) and [`error`] holds the Bot API error policy. This module +//! keeps the payload types and the senders themselves. +mod error; mod input_media; mod post_send; mod upload; @@ -14,15 +15,19 @@ mod upload; use crate::ctx::AppContext; use crate::handlers::log_key; use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost}; +pub(crate) use error::classify_to_send_error; +pub use error::{ + Classification, SendError, classify_request_error, is_media_fetch_failure, is_size_error, +}; use input_media::{build_media_group, input_file_for, item_url}; use post_send::{cache_animation_send, cache_sent_task}; use serde::{Deserialize, Serialize}; use std::borrow::Cow; use std::sync::LazyLock; +use teloxide::RequestError; use teloxide::prelude::*; use teloxide::types::{ChatId, InputMedia, MessageId}; -use teloxide::{ApiError, RequestError}; -use upload::{FallbackError, PreparedItem, prepare_upload_item, send_batch_via_upload}; +use upload::{PreparedItem, prepare_upload_item, send_batch_via_upload}; // The crate-facing API of this module lives in its submodules; re-export the // parts other modules use so call sites stay `send::x`. @@ -363,155 +368,6 @@ pub fn retry_delay_seconds(attempts: u32) -> f64 { (2f64.powi(attempts as i32) + jitter).min(30.0) } -/// Telegram's servers failed to fetch a media URL (hotlink protection etc.): -/// these errors are handled by the download-and-reupload fallback, NOT by a -/// queue retry (resending the URL cannot succeed). -pub fn is_media_fetch_failure(e: &ApiError) -> bool { - const MARKERS: [&str; 7] = [ - "webpage_media_empty", - "media_empty", - "empty_web_media", - "webpage_curl_failed", - "timeout", - // Oversized photos (width + height > 10000 px) are rejected on URL - // sends too; route them to the download-and-resize fallback. - "photo_invalid_dimensions", - // Telegram refused to fetch the URL it was handed. Single-media URL - // sends answer with this one (the media-group verbs use the - // `webpage_*`/`media_empty` markers above), and it is exactly the - // case the download-and-reupload fallback exists for. - "failed to get http url content", - ]; - let description = e.to_string().to_lowercase(); - MARKERS.iter().any(|marker| description.contains(marker)) -} - -/// Telegram reported the media file as too large (HTTP 413 on multipart -/// upload, or a "too large" message for URL-fetched media). These errors are -/// handled by the size-check fallback (use a smaller media URL), NOT by a -/// queue retry. -pub fn is_size_error(e: &ApiError) -> bool { - if matches!(e, ApiError::RequestEntityTooLarge) { - return true; - } - let description = e.to_string().to_lowercase(); - ["too large", "too big"] - .iter() - .any(|marker| description.contains(marker)) -} - -/// Task-free classification of a Telegram request error. The callers attach -/// the (updated) task when building a [`SendError`]. -pub enum Classification { - Retryable { - delay_seconds: f64, - }, - Permanent { - message: String, - }, - /// Handled by the download fallback, not a queue retry. - MediaFetchFailure, -} - -pub fn classify_request_error(e: &RequestError) -> Classification { - match e { - RequestError::RetryAfter(seconds) => Classification::Retryable { - delay_seconds: seconds.seconds() as f64, - }, - RequestError::Network(_) => Classification::Retryable { - delay_seconds: retry_delay_seconds(0), - }, - // A 5xx from the API — or from a proxy in front of it — is transient. - // teloxide only sleeps 10s on a server error and then parses whatever - // body came back, so by the time we see the error the HTTP status is - // gone: a JSON 5xx body arrives as an unknown description, an HTML - // error page as `InvalidJson`. Both used to be Permanent, which - // dead-lettered a post over a Telegram-side blip. - RequestError::Api(api) if is_server_error_text(&api.to_string()) => { - Classification::Retryable { - delay_seconds: retry_delay_seconds(0), - } - } - RequestError::Api(api) if is_media_fetch_failure(api) => Classification::MediaFetchFailure, - RequestError::Api(api) => Classification::Permanent { - message: api.to_string(), - }, - // An unparsable body can only come from something that is not the Bot - // API (which always answers JSON): a 5xx/error page from an - // intermediary, cut off mid-response. A JSON body that merely does not - // match the expected type cannot be fixed by retrying, so that case - // stays permanent. - RequestError::InvalidJson { raw, .. } if !raw.trim_start().starts_with('{') => { - Classification::Retryable { - delay_seconds: retry_delay_seconds(0), - } - } - RequestError::MigrateToChatId(_) - | RequestError::InvalidJson { .. } - | RequestError::Io(_) => Classification::Permanent { - message: e.to_string(), - }, - } -} - -/// Descriptions a 5xx carries when its body *is* JSON (teloxide keeps only the -/// description text, never the status code). Matched like the media-fetch -/// markers below; anything unmatched stays permanent, so a new permanent API -/// error is not retried just because it is unfamiliar. -fn is_server_error_text(description: &str) -> bool { - const MARKERS: [&str; 4] = [ - "server error", - "bad gateway", - "gateway timeout", - "service unavailable", - ]; - let description = description.to_lowercase(); - MARKERS.iter().any(|marker| description.contains(marker)) -} - -/// Task boxed to keep the error size within `result_large_err` limits. -#[derive(Debug)] -pub enum SendError { - Retryable { delay_seconds: f64, task: Box }, - Permanent { message: String, task: Box }, -} -fn classify_to_send_error(e: &RequestError, task: Task, fetch_failure_label: &str) -> SendError { - match classify_request_error(e) { - Classification::Retryable { delay_seconds } => SendError::Retryable { - delay_seconds, - task: Box::new(task), - }, - Classification::Permanent { message } => SendError::Permanent { - message, - task: Box::new(task), - }, - Classification::MediaFetchFailure => SendError::Permanent { - message: fetch_failure_label.into(), - task: Box::new(task), - }, - } -} - -impl SendError { - /// Attaches the (updated) task to a task-free [`FallbackError`] from the - /// download/upload pipeline. [`FallbackError::MediaTooLarge`] never - /// escapes the pipeline (it is handled by falling back to the smaller - /// URL), so it is unreachable here. - fn from_fallback(f: FallbackError, task: Task) -> SendError { - match f { - FallbackError::Retryable { delay_seconds } => SendError::Retryable { - delay_seconds, - task: Box::new(task), - }, - FallbackError::Permanent { message } => SendError::Permanent { - message, - task: Box::new(task), - }, - FallbackError::MediaTooLarge => unreachable!("handled inside the upload fallback"), - } - } -} - fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec) -> Task { let mut updated = task.clone(); match &mut updated { @@ -836,6 +692,7 @@ mod tests { use crate::ctx::test_support::{TestStores, cached_photo, photo_item}; use std::collections::HashMap; use std::time::Duration; + use teloxide::ApiError; #[test] fn oversized_photo_boundary() {