mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-23 23:32:05 +00:00
fix(retry): stop losing posts to transient failures and broken promises
P0 of the retry audit. The main finding: a Telegram 5xx was classified
Permanent, so one Telegram-side blip dead-lettered the post.
- `classify_request_error`: a server error is retryable again. teloxide sleeps
10s on a 5xx and then parses the body, so the HTTP status is gone by the
time the error arrives; it is recognised by shape instead — a JSON
server-error description, or an `InvalidJson` whose raw body is not JSON
(a proxy/error page). A JSON body of the wrong shape stays permanent, since
retrying a type mismatch cannot help. Reproduced end to end: with the old
classification a fake 502 (HTML body) logged "failed permanently" and
dead-lettered; now it logs "queued for retry" and the retry delivers.
- The same class of mistake elsewhere: `is_media_fetch_failure` was missing
`failed to get HTTP url content`, the description single-media URL sends
answer with, so hotlink-rejected media failed permanently instead of going
through the reupload fallback.
- `enqueue_retry` now reports whether the row was written, and the callers
only promise a retry when it was — a failed enqueue (DB write) used to tell
the user "retrying in Ns" and then deliver nothing, ever.
- A forward that fails retryably now settles the prompt instead of leaving it
live: the queued row carries the message ids itself, and a live prompt let
a second Confirm copy the same messages to the channel twice and let Skip
answer "nothing was forwarded" while the row still delivered.
- A prompt that could not be sent no longer swallows the gated forward
silently: the chat is told, since nothing would ever forward.
- `scaled_retry_delay` only scales up, so a server-asked `retry_after` above
the 300s cap is honoured instead of retried early (which earned another 429
and then dead-lettered the post).
- Download classification: a 4xx media download is permanent (the media is
gone or refused) while transport errors and 429/5xx retry — previously every
download error counted as retryable and burned the whole budget. A temp-file
*write* failure retries too (resource exhaustion clears; a temp dir that
cannot be created stays permanent).
- Site status mapping: 401/403 are `Blocked` (permanent) rather than
`Transient`, so a refusal is reported at once instead of after three
wasted attempts; and a twitter 200 that is not a tweet is no longer
reported as withheld content (the empty `{}` withheld shape keeps
`Sensitive`, which is what triggers the auth fallback).
This commit is contained in:
@@ -18,7 +18,7 @@ Telegram update → Dispatcher (polling or axum webhook) → dptree branches
|
||||
└─ callback_query → "forward" (copy to channel) / "template|<name>" (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 <url>` 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 `<blockquote>` 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 <url>` (send-only) / `/debug <url>` (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 <url>` (send-only) / `/debug <url>` (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<HashMap>` 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<Regex>`, `enabled() -> bool`, `fetch_from_url(url) -> Result<Fetched, FetchError>`, plus `cache_key`/`is_retryable`/`media_headers`, and a unit struct `<Name>Site` implementing `site::Site`; the central dispatcher (`site/mod.rs`) only iterates the `SITES` registry. Adding a site = new `site/<name>/{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<Box<dyn Future + Send>>`) 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<SiteStruct> 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 <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`; 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` |
|
||||
|
||||
@@ -214,6 +214,9 @@ pub async fn fetch(dynamic_id: &str) -> Result<model::Item, FetchError> {
|
||||
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}")),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -239,6 +239,9 @@ pub async fn fetch(handle: &str, rkey: &str) -> Result<Post, FetchError> {
|
||||
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}"))),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -77,6 +77,8 @@ pub async fn fetch(note_id: &str) -> Result<model::Note, FetchError> {
|
||||
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}")),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -525,15 +525,30 @@ pub async fn media_size(url: &str) -> Result<Option<u64>, 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<bytes::Bytes, FetchError> {
|
||||
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<u64, FetchError> {
|
||||
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
|
||||
{
|
||||
|
||||
@@ -132,6 +132,9 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
|
||||
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}"
|
||||
))),
|
||||
|
||||
@@ -105,6 +105,9 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
|
||||
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<serde_json::Value, FetchError> {
|
||||
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!([]));
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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::<serde_json::Value>("<html>502</html>").unwrap_err(),
|
||||
),
|
||||
raw: "<html>502 Bad Gateway</html>".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::<serde_json::Value>("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!(
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user