mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-28 00:02:12 +00:00
refactor(send): move the Bot API error policy into send/error.rs
classify_request_error, the marker tables it matches on, Classification and SendError (with its fallback conversion) are one policy — which failures are retried, which are permanent, which the reupload fallback owns — and were interleaved with the payload types and the senders. They move whole into send/error.rs and are re-exported, so every existing send::… path is unchanged.
This commit is contained in:
@@ -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/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/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/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/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 |
|
| `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/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 <url>` send-only, `/debug <url>` 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/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 <url>` send-only, `/debug <url>` 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/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/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` |
|
| `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` |
|
||||||
|
|||||||
@@ -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<Task> },
|
||||||
|
Permanent { message: String, task: Box<Task> },
|
||||||
|
}
|
||||||
|
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"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -4,9 +4,10 @@
|
|||||||
//! blocked by hotlink protection, so the bot downloads the file itself and
|
//! blocked by hotlink protection, so the bot downloads the file itself and
|
||||||
//! uploads it via multipart), [`post_send`] covers everything around a send
|
//! uploads it via multipart), [`post_send`] covers everything around a send
|
||||||
//! (cache write, keep-alive media, settlement, post-send actions, queue
|
//! (cache write, keep-alive media, settlement, post-send actions, queue
|
||||||
//! entry points). This module keeps the payload types, the error
|
//! entry points) and [`error`] holds the Bot API error policy. This module
|
||||||
//! classification and the senders themselves.
|
//! keeps the payload types and the senders themselves.
|
||||||
|
|
||||||
|
mod error;
|
||||||
mod input_media;
|
mod input_media;
|
||||||
mod post_send;
|
mod post_send;
|
||||||
mod upload;
|
mod upload;
|
||||||
@@ -14,15 +15,19 @@ mod upload;
|
|||||||
use crate::ctx::AppContext;
|
use crate::ctx::AppContext;
|
||||||
use crate::handlers::log_key;
|
use crate::handlers::log_key;
|
||||||
use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
|
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 input_media::{build_media_group, input_file_for, item_url};
|
||||||
use post_send::{cache_animation_send, cache_sent_task};
|
use post_send::{cache_animation_send, cache_sent_task};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::borrow::Cow;
|
use std::borrow::Cow;
|
||||||
use std::sync::LazyLock;
|
use std::sync::LazyLock;
|
||||||
|
use teloxide::RequestError;
|
||||||
use teloxide::prelude::*;
|
use teloxide::prelude::*;
|
||||||
use teloxide::types::{ChatId, InputMedia, MessageId};
|
use teloxide::types::{ChatId, InputMedia, MessageId};
|
||||||
use teloxide::{ApiError, RequestError};
|
use upload::{PreparedItem, prepare_upload_item, send_batch_via_upload};
|
||||||
use upload::{FallbackError, PreparedItem, prepare_upload_item, send_batch_via_upload};
|
|
||||||
|
|
||||||
// The crate-facing API of this module lives in its submodules; re-export the
|
// The crate-facing API of this module lives in its submodules; re-export the
|
||||||
// parts other modules use so call sites stay `send::x`.
|
// 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)
|
(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<Task> },
|
|
||||||
Permanent { message: String, task: Box<Task> },
|
|
||||||
}
|
|
||||||
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<i64>) -> Task {
|
fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec<i64>) -> Task {
|
||||||
let mut updated = task.clone();
|
let mut updated = task.clone();
|
||||||
match &mut updated {
|
match &mut updated {
|
||||||
@@ -836,6 +692,7 @@ mod tests {
|
|||||||
use crate::ctx::test_support::{TestStores, cached_photo, photo_item};
|
use crate::ctx::test_support::{TestStores, cached_photo, photo_item};
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
use teloxide::ApiError;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn oversized_photo_boundary() {
|
fn oversized_photo_boundary() {
|
||||||
|
|||||||
Reference in New Issue
Block a user