diff --git a/AGENTS.md b/AGENTS.md index a278b06..ef3b087 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,7 +18,7 @@ Telegram update → Dispatcher (polling or axum webhook) → dptree branches └─ callback_query → "forward" (copy to channel) / "template|" (apply caption template) ``` -Message flow: `message_handler` extracts URLs (from `url`/`text_link` entities, text + caption, deduped) → `x_media::site::fetch(url)` → `Fetched` → builds a `Task` → `send::send_media_sequence` (media groups ≤ 10, caption on first item) or `send::send_animation`. On Telegram URL-fetch failure or size error (`send_batch_via_upload`): download via `x_media::site::download_media` to a temp file (≤ 10 MiB), sniff magic bytes (`sniff_ext`), upload via multipart; oversized items fall back to `fallback_url`. On failure: `enqueue_retry` persists resume-state `Task` into the SQLite queue → workers lease (120 s lock TTL) → retry with exponential backoff (≤ 30 s, `MAX_RETRIES = 2`) → dead-letter → `notify_failure`. Success → `post_send_actions`: edit-before-forward prompt with inline buttons, or `copy_messages` to the bound forward channel. +Message flow: `message_handler` extracts URLs (from `url`/`text_link` entities, text + caption, deduped) → `x_media::site::fetch(url)` → `Fetched` → builds a `Task` → `send::send_media_sequence` (media groups ≤ 10, caption on first item) or `send::send_animation`. On Telegram URL-fetch failure or size error (`send_batch_via_upload`): download via `x_media::site::download_media` to a temp file (≤ 10 MiB), sniff magic bytes (`sniff_ext`), upload via multipart; oversized items fall back to `fallback_url`. On failure: `enqueue_retry` persists resume-state `Task` into the SQLite queue (it reports whether the row was really written, and only then does the user get the "retrying in Ns" notice — an enqueue that fails says so instead) → workers lease (120 s lock TTL) → retry with exponential backoff (≤ 30 s for the bot's own delays, `MAX_RETRIES = 2`) → dead-letter → `notify_failure`. Success → `post_send_actions`: edit-before-forward prompt with inline buttons, or `copy_messages` to the bound forward channel. Debug command: `/debug ` runs the same `x_media::site::fetch` and replies with `debug_report` (`handlers/commands.rs`) — site id, normalized cache key, source URL, title/author/tags, sensitive flag, caption and the media list — nothing is sent, cached or forwarded; the report is capped at 4000 chars and sent with HTML parse mode: raw fields are escaped, and the caption is wrapped in a `
` so it renders exactly like the sent media caption (escaped text and links included). The caption it shows is `preview_caption`'s: the chat's per-site format override plus the long-post quoting, i.e. exactly what the send paths produce — showing the raw built-in caption made `/set_format` look like a no-op, and the `/set_format` success reply points users at `/debug` to preview. @@ -41,7 +41,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr | `crates/xmedia-bot/src/main.rs` | Entry point: env/log init, command registration (`register_commands` — `setMyCommands` plus the profile description texts), shared `send::BOT` force-init, queue worker start, site login validation (`site::validate_all`), 300 s edit-expiry sweep (expired prompts are rewritten in place to `EDIT_PROMPT_EXPIRED_TEXT` with an empty keyboard — an edit, never a new message, so a background timer cannot wake a chat), dptree handler tree, webhook vs polling dispatch | | `crates/xmedia-bot/src/config.rs` | Manual env parsing into `Config` | | `crates/xmedia-bot/src/db.rs` | `DbPool`: one shared SQLite connection pool (`POOL_SIZE = 4`, WAL, busy_timeout) for all three tables over `$DATA_DIR/task_queue.db` (default `data/`) — the three stores share it; `open_store` creates file + schema, `with_conn` runs all rusqlite I/O in `spawn_blocking` | -| `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`), `statics.rs` (global statics) | +| `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 | @@ -73,7 +73,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi - **Site adapter convention**: each site module exports `PATTERN: LazyLock`, `enabled() -> bool`, `fetch_from_url(url) -> Result`, plus `cache_key`/`is_retryable`/`media_headers`, and a unit struct `Site` implementing `site::Site`; the central dispatcher (`site/mod.rs`) only iterates the `SITES` registry. Adding a site = new `site//{mod.rs,interface.rs,model.rs}` + one `Box::new(...)` entry in `SITES` — the bot crate never lists sites (SetFormat whitelist, cache-key site lookup and startup validation all derive from the registry). Async trait methods return `SiteFuture` (a boxed `Pin>`) because `async fn` in traits is not dyn-compatible. - **Serde**: per-site `model.rs` are pure `Deserialize` DTOs mirroring API JSON; site structs in `interface.rs` have private fields, a `caption()` builder, and `impl From for Fetched`. Persisted payloads use internally-tagged enums (`#[serde(tag = "kind")]` / `type`). - **Naming**: module-per-concern, snake_case files, `CamelCase` types, `snake_case` fns. `//!` module docs and `///` docs on non-obvious logic (syndication token, ugoira encoding, `display_text_range`). -- **Retries**: only `x-media::site::fetch` retries (3 attempts, `1 << attempt` backoff, HTTP errors only); `site::fetch_once` is the same code path with a single attempt, used by inline queries whose answer window is shorter than the backoff. Queue retries are explicit `QueueError::Retryable` with computed delay (`retry_delay_seconds`). +- **Retries**: only `x-media::site::fetch` retries (3 attempts, `1 << attempt` backoff, HTTP errors only); `site::fetch_once` is the same code path with a single attempt, used by inline queries whose answer window is shorter than the backoff. A status a site answers with is classified by what a *retry* can change: 404/410 are `NotFound` and 401/403 are `Blocked` (permanent, reported at once), 429/5xx are `Transient` and retried. Queue retries are explicit `QueueError::Retryable` with computed delay (`retry_delay_seconds`), scaled per attempt by `scaled_retry_delay` — which only ever scales **up**, so a delay the server asked for (Telegram `retry_after`) is never shortened. `send::classify_request_error` is the send-side counterpart: `RetryAfter` and `Network` are retryable, and so is a 5xx — teloxide sleeps 10 s on a server error and then parses the body, so by then the HTTP status is gone and the condition is recognised by shape instead (a JSON server-error description, or an `InvalidJson` whose raw body is not JSON, i.e. a proxy/error page). - Logging via `log` macros (`pretty_env_logger`, level from `RUST_LOG`). `main.rs` initializes the **timed** builder with a default filter of `info,hyper_util=warn,reqwest=warn` when `RUST_LOG` is unset: the plain `init` had no timestamps and fell back to `error`, so a deployment that forgot the variable logged nothing at all, and at `debug` the HTTP client's own lines outnumbered the bot's two to one. An explicit `RUST_LOG` overrides the default wholesale. Level convention: `info` = lifecycle + per-post business results (`sent`/`forwarded`/`copied`, with `chat=` and the total `ms`), admin/operator actions and anomalies (fallback, retry enqueue, dead-letter is `error`); `debug` = per-request detail (URL extraction, `fetching`/`fetched` with the fetch duration, batch sends, queue processing with the row's `chat=`/`key=` and per-attempt `ms`, photo processing, inline queries); `trace` = user data (the full URL, the message text, the inline query). At `debug` and above links are printed via the normalized cache key (`handlers::log_key`, e.g. `[key=twitter:123...]`), so a `debug` log can be shared without echoing what users pasted, and degradations that leave the user served (a failed cache read/write, a failed chat action) are `warn`, not `error`. The only queue/sweep aggregate is the 300 s sweep's queue line, and it speaks only when the queue is non-empty. ## Important Files @@ -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); `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`; the senders. `upload.rs`: download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`). `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`; `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). `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 | | `crates/x-media/src/site/mod.rs` | Dispatcher, `Fetched`/`FetchError`, shared `CLIENT`, `download_media` (adds `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/x-media/src/site/bilibili/interface.rs b/crates/x-media/src/site/bilibili/interface.rs index ea74bd9..efa88ab 100644 --- a/crates/x-media/src/site/bilibili/interface.rs +++ b/crates/x-media/src/site/bilibili/interface.rs @@ -214,6 +214,9 @@ pub async fn fetch(dynamic_id: &str) -> Result { if !status.is_success() { return Err(match status.as_u16() { 412 => risk_control("412"), + // A refusal or an auth demand is not a bad moment (412 above is + // bilibili's risk control, which does clear on its own). + 401 | 403 => FetchError::Blocked, _ => FetchError::Transient(format!("bilibili status {status}")), }); } diff --git a/crates/x-media/src/site/bsky/interface.rs b/crates/x-media/src/site/bsky/interface.rs index 08ff764..5fda8fc 100644 --- a/crates/x-media/src/site/bsky/interface.rs +++ b/crates/x-media/src/site/bsky/interface.rs @@ -239,6 +239,9 @@ pub async fn fetch(handle: &str, rkey: &str) -> Result { if !status.is_success() { return match status.as_u16() { 404 | 410 => Err(FetchError::NotFound), + // A refusal or an auth demand is not a bad moment: retrying it + // three times only delays an error the user has to see. + 401 | 403 => Err(FetchError::Blocked), _ => Err(FetchError::Transient(format!("bsky status {status}"))), }; } diff --git a/crates/x-media/src/site/misskey/interface.rs b/crates/x-media/src/site/misskey/interface.rs index 55f5a31..ecf9567 100644 --- a/crates/x-media/src/site/misskey/interface.rs +++ b/crates/x-media/src/site/misskey/interface.rs @@ -77,6 +77,8 @@ pub async fn fetch(note_id: &str) -> Result { if !status.is_success() { return Err(match status.as_u16() { 400 => not_found_or_invalid(response).await, + // A refusal or an auth demand is not a bad moment. + 401 | 403 => FetchError::Blocked, _ => FetchError::Transient(format!("misskey status {status}")), }); } diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index 4ab39f4..1bba6f1 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -525,15 +525,30 @@ pub async fn media_size(url: &str) -> Result, FetchError> { Ok(response.content_length()) } +/// Maps a media download's HTTP status onto the same classes the site +/// adapters use, so callers can tell "try again" from "this URL is dead": +/// 4xx is a property of the media (gone, refused by the host), while 429/5xx +/// is a property of the moment. A transport error never reaches this — it +/// fails in `send()` and stays [`FetchError::Http`]. +fn download_status_error(status: reqwest::StatusCode) -> FetchError { + match status.as_u16() { + 401 | 403 => FetchError::Blocked, + 404 | 410 => FetchError::NotFound, + _ => FetchError::Transient(format!("media status {status}")), + } +} + /// Downloads a media file with a hard size cap: the body is streamed and the /// download aborts with [`FetchError::TooLarge`] the moment the cap is /// 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? - .error_for_status()?; + 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())); + }; if let Some(len) = response.content_length() && len > max_bytes { @@ -566,10 +581,12 @@ 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? - .error_for_status()?; + 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())); + }; if let Some(len) = response.content_length() && len > max_bytes { diff --git a/crates/x-media/src/site/twitter/auth.rs b/crates/x-media/src/site/twitter/auth.rs index bdca3e0..8cbe293 100644 --- a/crates/x-media/src/site/twitter/auth.rs +++ b/crates/x-media/src/site/twitter/auth.rs @@ -132,6 +132,9 @@ pub async fn fetch(id: &str) -> Result { log::warn!("twitter auth fetch {id}: HTTP {status}"); return match status.as_u16() { 404 | 410 => Err(FetchError::NotFound), + // A stale/refused `auth_token` is not a bad moment: retrying it + // three times only delays the report. + 401 | 403 => Err(FetchError::Blocked), _ => Err(FetchError::Transient(format!( "twitter auth status {status}" ))), diff --git a/crates/x-media/src/site/twitter/interface.rs b/crates/x-media/src/site/twitter/interface.rs index 7d633ec..67e19b1 100644 --- a/crates/x-media/src/site/twitter/interface.rs +++ b/crates/x-media/src/site/twitter/interface.rs @@ -105,6 +105,9 @@ pub async fn fetch(id: &str) -> Result { if !status.is_success() { return match status.as_u16() { 404 | 410 => Err(FetchError::NotFound), + // A refusal or an auth demand is not a bad moment: retrying it + // three times only delays an error the user has to see. + 401 | 403 => Err(FetchError::Blocked), _ => Err(FetchError::Transient(format!("twitter status {status}"))), }; } @@ -146,7 +149,18 @@ fn parse_syndication_body(text: &str) -> Result { return Err(FetchError::NotFound); } if body.get("id_str").is_none() { - return Err(FetchError::Sensitive); + // Syndication answers an empty `{}` for withheld (NSFW / + // age-restricted) tweets: the documented case, kept as `Sensitive` + // because it is what triggers the logged-in auth fallback. + if body.as_object().is_some_and(|object| object.is_empty()) { + return Err(FetchError::Sensitive); + } + // Any other shape is not a tweet: an interstitial, a truncated body, + // a change on their side. Reporting that as withheld content told the + // user to set TWITTER_AUTH_TOKEN for something auth cannot fix. + return Err(FetchError::Transient( + "unexpected syndication body".to_string(), + )); } Ok(body) } @@ -762,6 +776,17 @@ mod tests { )); } + #[test] + fn syndication_unexpected_shape_is_transient_not_withheld() { + // A 200 that is not a tweet at all (an interstitial, a truncated + // body) must not be reported as withheld content: that message tells + // the user to set TWITTER_AUTH_TOKEN, which cannot fix it. + match parse_syndication_body("{\"foo\":1}") { + Err(FetchError::Transient(_)) => {} + other => panic!("expected Transient, got {other:?}"), + } + } + #[test] fn syndication_tweet_body_passes() { let raw = fixture(serde_json::json!([])); diff --git a/crates/xmedia-bot/src/handlers/callback.rs b/crates/xmedia-bot/src/handlers/callback.rs index 7f69d9a..aa0882b 100644 --- a/crates/xmedia-bot/src/handlers/callback.rs +++ b/crates/xmedia-bot/src/handlers/callback.rs @@ -117,9 +117,24 @@ async fn handle_callback( delay_seconds, task, }) => { - log::info!("forward queued for retry in {delay_seconds:.1}s"); - send::enqueue_retry(ctx.task_queue, *task, delay_seconds).await; - ("Forward queued for retry.".to_string(), false) + // The queued row owns the forward from here (it carries + // the message ids itself), so the prompt is settled + // either way: leaving it live let a second Confirm copy + // the same messages to the channel twice, and let Skip + // answer "nothing was forwarded" while the row still + // delivered it. + let queued = + send::enqueue_retry(ctx.task_queue, &task, delay_seconds).await; + if queued { + log::info!("forward queued for retry in {delay_seconds:.1}s"); + ("Forward queued for retry.".to_string(), true) + } else { + log::error!("forward retry could not be queued"); + ( + "Forward failed and the retry could not be queued.".to_string(), + true, + ) + } } Err(send::SendError::Permanent { message, .. }) => { log::error!("forward failed permanently: {message}"); @@ -314,7 +329,7 @@ mod tests { } #[tokio::test] - async fn retryable_forward_is_queued_and_keeps_the_prompt() { + async fn retryable_forward_is_queued_and_settles_the_prompt() { use teloxide::types::Seconds; let sender = MockSender::scripted(vec![Outcome::CopyErr], || { RequestError::RetryAfter(Seconds::from_seconds(7)) @@ -325,23 +340,45 @@ mod tests { handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; + // The queued row carries the message ids itself, so it owns the + // forward from here and the prompt is closed with it. Keeping it live + // (the old behaviour) let a second Confirm copy the same messages to + // the channel twice, and let Skip answer "nothing was forwarded" while + // the row still delivered it. assert_eq!( sender.calls(), - vec!["copy_messages", "answer_callback_query"] + vec!["copy_messages", "delete_message", "answer_callback_query"] ); assert_eq!( sender.answers(), vec![Some("Forward queued for retry.".to_string())] ); assert_eq!(stores.queued_tasks().await, 1); - // The prompt is not settled: the queued retry still needs the record. assert!( - ctx.chat_store + !ctx.chat_store .get(1) .await .edit_message - .contains_key(&PROMPT_ID) + .contains_key(&PROMPT_ID), + "the record must be dropped so the prompt cannot be used again" ); + + // A second tap finds no record: it cannot enqueue a duplicate copy. + handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; + assert_eq!( + sender.calls(), + vec![ + "copy_messages", + "delete_message", + "answer_callback_query", + "answer_callback_query" + ] + ); + assert_eq!( + sender.answers().last().map(|a| a.as_deref()), + Some(Some("Expired")) + ); + assert_eq!(stores.queued_tasks().await, 1, "no second forward row"); } #[tokio::test] diff --git a/crates/xmedia-bot/src/handlers/urls.rs b/crates/xmedia-bot/src/handlers/urls.rs index cd216dc..73ec930 100644 --- a/crates/xmedia-bot/src/handlers/urls.rs +++ b/crates/xmedia-bot/src/handlers/urls.rs @@ -210,19 +210,23 @@ async fn dispatch_send( "send for [key={}] chat={chat_id} failed after {ms}ms, queued for retry in {delay_seconds:.1}s", log_key(url) ); - send::enqueue_retry(ctx.task_queue, *task, delay_seconds).await; // Name the post and the wait: "queued for retry" alone left the - // user guessing which link it was and how long the wait is. - let _ = reply( - ctx.sender, - chat_id, - reply_to, + // user guessing which link it was and how long the wait is. The + // promise is made only when the retry was really persisted — an + // enqueue that failed (DB write) would leave the user waiting for + // a retry nothing can deliver. + let promised = if send::enqueue_retry(ctx.task_queue, &task, delay_seconds).await { format!( "Send failed for {} — retrying in {delay_seconds:.0}s.", log_key(url) - ), - ) - .await; + ) + } else { + format!( + "Send failed for {} and the retry could not be queued — please send the link again.", + log_key(url) + ) + }; + let _ = reply(ctx.sender, chat_id, reply_to, promised).await; } Err(send::SendError::Permanent { message: err_message, diff --git a/crates/xmedia-bot/src/queue.rs b/crates/xmedia-bot/src/queue.rs index fe44ff6..174f2f5 100644 --- a/crates/xmedia-bot/src/queue.rs +++ b/crates/xmedia-bot/src/queue.rs @@ -81,10 +81,11 @@ fn recover_update(conn: &rusqlite::Connection) -> rusqlite::Result<()> { /// Base delay × 2^attempts (attempts = retries already done), capped at 300s. /// Applied at the queue layer so the attempt count actually reaches the -/// backoff computation; Telegram `RetryAfter` delays get the same treatment -/// (conservatively larger wait, no API change needed). +/// backoff computation. The cap only ever scales *up*: a delay the server +/// asked for (Telegram `RetryAfter`) must not be shortened, retrying earlier +/// than allowed just re-triggers the flood control it came from. fn scaled_retry_delay(base: f64, attempts: i32) -> f64 { - (base * 2f64.powi(attempts)).min(300.0) + (base * 2f64.powi(attempts)).min(300.0).max(base) } impl PersistentTaskQueue { @@ -519,6 +520,10 @@ mod tests { assert_eq!(scaled_retry_delay(1.5, 1), 3.0); assert_eq!(scaled_retry_delay(1.0, 10), 300.0, "capped at 300s"); assert_eq!(scaled_retry_delay(300.0, 0), 300.0); + // A server-asked delay above the cap is honoured, not truncated: a + // 1800s flood-control wait used to become 300s and earn another 429. + assert_eq!(scaled_retry_delay(1800.0, 0), 1800.0); + assert_eq!(scaled_retry_delay(1800.0, 1), 1800.0); } async fn new_queue() -> (PersistentTaskQueue, tempfile::TempDir) { diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index 40563d3..0ecd352 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -271,7 +271,7 @@ pub fn retry_delay_seconds(attempts: u32) -> f64 { /// 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; 6] = [ + const MARKERS: [&str; 7] = [ "webpage_media_empty", "media_empty", "empty_web_media", @@ -280,6 +280,11 @@ pub fn is_media_fetch_failure(e: &ApiError) -> bool { // 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)) @@ -320,10 +325,31 @@ pub fn classify_request_error(e: &RequestError) -> Classification { 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 { @@ -332,6 +358,21 @@ pub fn classify_request_error(e: &RequestError) -> Classification { } } +/// 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 { @@ -918,6 +959,15 @@ mod tests { } } + #[test] + fn is_media_fetch_failure_matches_the_single_media_url_description() { + // `sendPhoto`/`sendAnimation`-style URL sends answer with this one + // instead of the `webpage_*` markers; without it the URL send failed + // permanently instead of going through the reupload fallback. + let api = ApiError::Unknown("Bad Request: failed to get HTTP URL content".into()); + assert!(is_media_fetch_failure(&api)); + } + #[test] fn is_size_error_matches_known_errors() { // 413 upload cap. @@ -979,6 +1029,39 @@ mod tests { classify_request_error(&e), Classification::MediaFetchFailure )); + // A server-error description (teloxide drops the HTTP status, so a + // JSON 5xx arrives as an unknown description) -> Retryable. Without + // this a Telegram 502 dead-lettered the post. + let e = RequestError::Api(ApiError::Unknown("Internal Server Error".into())); + assert!(matches!( + classify_request_error(&e), + Classification::Retryable { .. } + )); + // An HTML/proxy error page in place of the API's JSON -> Retryable. + let e = RequestError::InvalidJson { + source: std::sync::Arc::new( + serde_json::from_str::("502").unwrap_err(), + ), + raw: "502 Bad Gateway".into(), + }; + assert!(matches!( + classify_request_error(&e), + Classification::Retryable { .. } + )); + // A JSON body of the wrong shape is a type mismatch, not a transport + // problem: still permanent. + // (The `source` is only ever rendered, so an unrelated parse error + // stands in for the shape mismatch; `raw` is what the classifier reads.) + let e = RequestError::InvalidJson { + source: std::sync::Arc::new( + serde_json::from_str::("x").unwrap_err(), + ), + raw: "{\"ok\":true,\"result\":true}".into(), + }; + assert!(matches!( + classify_request_error(&e), + Classification::Permanent { .. } + )); // MigrateToChatId -> Permanent let e = RequestError::MigrateToChatId(ChatId(123)); assert!(matches!( diff --git a/crates/xmedia-bot/src/send/post_send.rs b/crates/xmedia-bot/src/send/post_send.rs index d6237de..416cff8 100644 --- a/crates/xmedia-bot/src/send/post_send.rs +++ b/crates/xmedia-bot/src/send/post_send.rs @@ -277,7 +277,20 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message }) .await; } - Err(e) => log::error!("failed to send edit prompt: {e}"), + Err(e) => { + log::error!("failed to send edit prompt: {e}"); + // Nothing is forwarded until the prompt is confirmed, so a + // prompt that never arrived means this post is never forwarded. + // Tell the chat instead of letting it wait for a prompt that + // will not come. + notify_failure( + ctx.sender, + notify_chat_id, + notify_message_id, + "Could not open the edit-before-forward prompt — nothing was forwarded.", + ) + .await; + } } return; } @@ -301,7 +314,17 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message delay_seconds, task, }) => { - enqueue_retry(ctx.task_queue, *task, delay_seconds).await; + // The forward is already committed from the user's side; if it + // cannot be queued, say so rather than going quiet. + if !enqueue_retry(ctx.task_queue, &task, delay_seconds).await { + notify_failure( + ctx.sender, + notify_chat_id, + notify_message_id, + &failure_text(Some(&task), "retry could not be queued"), + ) + .await; + } } Err(SendError::Permanent { message, .. }) => { notify_failure( @@ -316,16 +339,24 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message } } -/// Enqueues a task for a later attempt (retry / forward resume). When the -/// enqueue itself fails the task can never be sent again, so its keep-alive -/// temp media is released instead of leaking until process exit. -pub(crate) async fn enqueue_retry(queue: &PersistentTaskQueue, task: Task, delay_seconds: f64) { - let payload = serde_json::to_value(&task).expect("task serializes"); +/// Enqueues a task for a later attempt (retry / forward resume). Returns +/// whether the retry is actually persisted: when the enqueue itself fails the +/// task can never run again, so its keep-alive temp media is released instead +/// of leaking until process exit — and the caller must not tell the user a +/// retry is coming (nothing would ever deliver it). +pub(crate) async fn enqueue_retry( + queue: &PersistentTaskQueue, + task: &Task, + delay_seconds: f64, +) -> bool { + let payload = serde_json::to_value(task).expect("task serializes"); let run_after = now_f64() + delay_seconds; if let Err(e) = queue.enqueue(payload, run_after).await { log::error!("failed to enqueue retry: {e}"); - release_keep_alive(&task); + release_keep_alive(task); + return false; } + true } /// Queue entry point: parses the stored task and dispatches. diff --git a/crates/xmedia-bot/src/send/upload.rs b/crates/xmedia-bot/src/send/upload.rs index f3680e4..3151d55 100644 --- a/crates/xmedia-bot/src/send/upload.rs +++ b/crates/xmedia-bot/src/send/upload.rs @@ -71,19 +71,7 @@ async fn download_to_temp( }; let bytes = match x_media::site::download_media_limited(media_url, limit).await { Ok(bytes) => bytes, - Err(FetchError::Http(_)) => { - return Err(FallbackError::Retryable { - delay_seconds: retry_delay_seconds(0), - }); - } - Err(FetchError::TooLarge) => { - return Err(FallbackError::MediaTooLarge); - } - Err(e) => { - return Err(FallbackError::Permanent { - message: format!("download failed: {e}"), - }); - } + Err(e) => return Err(classify_download_error(e)), }; let ext = sniff_ext(&bytes); let mut file = tempfile::Builder::new() @@ -93,14 +81,37 @@ async fn download_to_temp( message: format!("temp file failed: {e}"), })?; use std::io::Write; - file.as_file_mut() - .write_all(&bytes) - .map_err(|e| FallbackError::Permanent { - message: format!("temp file write failed: {e}"), - })?; + // A write failure is resource exhaustion far more often than a broken temp + // dir (ENOSPC / EDQUOT), and that clears on its own — worth an attempt + // instead of dropping the post on the first try. Creating the file (above) + // stays permanent: a temp dir that cannot be created at all is a + // deployment fault that should fail loudly and immediately. `Retryable` + // carries no message, so the cause is logged here. + file.as_file_mut().write_all(&bytes).map_err(|e| { + log::error!("temp file write failed: {e}"); + FallbackError::Retryable { + delay_seconds: retry_delay_seconds(0), + } + })?; Ok((file, bytes)) } +/// Which failure class a media download belongs to. Transport errors and +/// server-side hiccups (429/5xx, see `download_media_limited`) are worth +/// another attempt; a 4xx means the media itself is gone or refused, and a +/// retry could only ask the same URL again. +fn classify_download_error(err: FetchError) -> FallbackError { + match err { + FetchError::Http(_) | FetchError::Transient(_) => FallbackError::Retryable { + delay_seconds: retry_delay_seconds(0), + }, + FetchError::TooLarge => FallbackError::MediaTooLarge, + e => FallbackError::Permanent { + message: format!("download failed: {e}"), + }, + } +} + /// Builds the media group item from an uploaded file. fn media_from_file( item: &MediaItemPayload, @@ -343,3 +354,31 @@ pub(super) async fn send_batch_via_upload( Err(e) => Err(classify_to_send_error(&e, task, "upload failed")), } } + +#[cfg(test)] +mod download_class_tests { + use super::*; + + #[test] + fn download_errors_split_by_whether_a_retry_can_help() { + // Transport failure and a server-side hiccup: try again. + assert!(matches!( + classify_download_error(FetchError::Transient("media status 503".into())), + FallbackError::Retryable { .. } + )); + // The media is gone / the host refuses us: a retry repeats the 4xx. + assert!(matches!( + classify_download_error(FetchError::NotFound), + FallbackError::Permanent { .. } + )); + assert!(matches!( + classify_download_error(FetchError::Blocked), + FallbackError::Permanent { .. } + )); + // Over the cap: degrade to the smaller URL, never retry. + assert!(matches!( + classify_download_error(FetchError::TooLarge), + FallbackError::MediaTooLarge + )); + } +}