diff --git a/AGENTS.md b/AGENTS.md index 751c8c0..3a8c4f9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,14 +66,14 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi ## Code Conventions & Common Patterns -- **Errors via `thiserror` derive** (no anyhow): the public, stringified errors — `FetchError` (`Http`/`Json`/`Pixiv`/`Site`/`NotFound`/`Blocked`/`Disabled`/`Sensitive`/`TooLarge`/`MediaPrep`/`Transient`/`Io`) and `PixivError` — derive `thiserror::Error` with `#[from]` conversions; `Display`/`source()` come from the derive. The internal control-flow enums — `QueueError` (`Retryable { delay_seconds, payload }` / `Permanent`), `SendError` (Retryable/Permanent), `Classification`, `FallbackError` — carry no `Display` and are handled by direct variant matching. New errors should follow the same split: stringified/public errors derive `thiserror`, internal flow enums stay plain. +- **Errors via `thiserror` derive** (no anyhow): the public, stringified errors — `FetchError` (`Http`/`Json`/`Pixiv`/`Site`/`NotFound`/`Blocked`/`Disabled`/`Sensitive`/`TooLarge`/`MediaPrep`/`Transient`/`RateLimited`/`Io`) and `PixivError` — derive `thiserror::Error` with `#[from]` conversions; `Display`/`source()` come from the derive. The internal control-flow enums — `QueueError` (`Retryable { delay_seconds, payload }` / `Permanent`), `SendError` (Retryable/Permanent), `Classification`, `FallbackError` — carry no `Display` and are handled by direct variant matching. New errors should follow the same split: stringified/public errors derive `thiserror`, internal flow enums stay plain. - **Global state via `std::sync::LazyLock` statics**, not DI: `CONFIG`, `CHAT_STORE`, `TASK_QUEUE` in `handlers/statics.rs`; shared reqwest `CLIENT` in `x-media/src/site/mod.rs`. `Bot` is passed/cloned into handlers; queue workers share the process-wide `send::BOT` (`LazyLock`, force-initialized in `main` so a missing token fails at startup). - **Async**: tokio multi-thread runtime (`#[tokio::main]` default). All rusqlite I/O inside `tokio::task::spawn_blocking`. Long loops use `tokio::select!` with `tokio::sync::{watch, Notify}` stop/wake channels. No streams. - **Blocking sync primitives**: `parking_lot::Mutex` for hot caches, `tokio::sync::Mutex` for async-shared state (pixiv token cache), `AtomicBool` for feature gates. - **Site adapter convention**: each site module exports `PATTERN: LazyLock`, `fetch_from_url(url) -> Result` and `cache_key`, plus a unit struct `Site` implementing `site::Site`; `enabled`/`is_retryable`/`media_headers` come from the trait's defaults unless the site overrides them (only pixiv does); 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. 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). +- **Retries**: only `x-media::site::fetch` retries (3 attempts, a doubling backoff widened by a random slice of itself so workers that failed together do not recover together, with a 429's `Retry-After` honored up to `MAX_RETRY_AFTER_SECS` = 60 s, over 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; user-supplied text that does reach a line (display names, callback data, channel handles) goes through `handlers::log_escape`, whose escapes keep a crafted value from splitting or forging a log entry, 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 diff --git a/crates/x-media/src/site/bilibili/interface.rs b/crates/x-media/src/site/bilibili/interface.rs index 647a3d0..a0a4153 100644 --- a/crates/x-media/src/site/bilibili/interface.rs +++ b/crates/x-media/src/site/bilibili/interface.rs @@ -217,7 +217,7 @@ pub async fn fetch(dynamic_id: &str) -> Result { // posts permanent, 429/5xx retried). The local fallback used to // disagree: a bilibili 404 came back Transient here. 412 above is // bilibili's risk control, which does clear on its own. - _ => crate::site::status_error("bilibili", status), + _ => crate::site::status_error("bilibili", &response), }); } let detail: model::Detail = response.json().await.map_err(|e| FetchError::Site { diff --git a/crates/x-media/src/site/bsky/interface.rs b/crates/x-media/src/site/bsky/interface.rs index 87b302d..91ce158 100644 --- a/crates/x-media/src/site/bsky/interface.rs +++ b/crates/x-media/src/site/bsky/interface.rs @@ -132,7 +132,7 @@ fn concat_list(files: &mut [(usize, std::path::PathBuf)]) -> String { /// more, so the second attempt belongs on the request that actually failed. async fn fetch_hls(url: &str, cap: u64) -> Result { match crate::site::download_media_limited(url, cap, crate::site::DOWNLOAD_TOTAL_TIMEOUT).await { - Err(FetchError::Http(_) | FetchError::Transient(_)) => { + Err(FetchError::Http(_) | FetchError::Transient(_) | FetchError::RateLimited { .. }) => { tokio::time::sleep(std::time::Duration::from_secs(1)).await; crate::site::download_media_limited(url, cap, crate::site::DOWNLOAD_TOTAL_TIMEOUT) .await @@ -304,7 +304,7 @@ pub async fn fetch(handle: &str, rkey: &str) -> Result { // 404/410 = gone (permanent); 429/5xx = transient and retried by fetch. let status = response.status(); if !status.is_success() { - return Err(crate::site::status_error("bsky", status)); + return Err(crate::site::status_error("bsky", &response)); } let text = response.text().await?; Post::from_json(&text, rkey.to_string()) diff --git a/crates/x-media/src/site/download.rs b/crates/x-media/src/site/download.rs index e6fec8e..96d62cf 100644 --- a/crates/x-media/src/site/download.rs +++ b/crates/x-media/src/site/download.rs @@ -119,7 +119,7 @@ async fn send_download(request: reqwest::RequestBuilder) -> Result Result { // The local fallback used to disagree with the center: a misskey // 404 came back Transient here and was fetched three more times // for a note that is simply gone. - _ => crate::site::status_error("misskey", status), + _ => crate::site::status_error("misskey", &response), }); } response.json().await.map_err(|e| FetchError::Site { diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index 9c191d3..b399f53 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -278,24 +278,63 @@ pub enum FetchError { /// A transient server-side failure (429 / 5xx); [`fetch`] retries these. #[error("transient: {0}")] Transient(String), + /// The source answered 429 *with* a `Retry-After` and named its own + /// delay: [`fetch`] sleeps at least that long instead of guessing one + /// (capped by [`MAX_RETRY_AFTER_SECS`] — the header is server-supplied + /// and must not park one of the fetch slots). + #[error("{site} rate limited, retry after {retry_after_secs}s")] + RateLimited { + site: &'static str, + retry_after_secs: u64, + }, /// A local I/O failure while streaming a download to disk /// (see [`download_media_to_file`]). #[error("io error: {0}")] Io(std::io::Error), } +/// Cap on a server-supplied `Retry-After`: honored so a retry stops hammering +/// a source that asked for air, bounded so the same untrusted header cannot +/// park a fetch slot for an hour. +pub const MAX_RETRY_AFTER_SECS: u64 = 60; + /// The error class for a non-success HTTP status, shared by the site /// adapters, the media downloads and twitter's auth fallback: 404/410 mean /// the post is gone (permanent), any other client error the source answers /// on sight is a refusal (permanent too — three retries only delay the same -/// answer), and only 408/429/5xx are a bad moment, retried by [`fetch`]. +/// answer), 408/429/5xx are a bad moment retried by [`fetch`], and a 429 +/// that carries `Retry-After` keeps the delay the source asked for (the +/// seconds form only — a HTTP-date value parses to `None` and falls back to +/// the plain transient path). /// `site` only names the adapter in the message (`"media"` for downloads); /// a site whose statuses mean something else (bilibili's 412 risk control, /// misskey's 400 with `NO_SUCH_NOTE`) maps those before falling back here. -pub fn status_error(site: &'static str, status: reqwest::StatusCode) -> FetchError { +pub fn status_error(site: &'static str, response: &reqwest::Response) -> FetchError { + let retry_after = response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.trim().parse::().ok()); + classify_status(site, response.status(), retry_after) +} + +/// [`status_error`]'s table, split out so tests reach it without building an +/// HTTP response — the retry delay only ever shapes the 429 arm. +pub(crate) fn classify_status( + site: &'static str, + status: reqwest::StatusCode, + retry_after: Option, +) -> FetchError { match status.as_u16() { 404 | 410 => FetchError::NotFound, code if status.is_client_error() && !matches!(code, 408 | 429) => FetchError::Blocked, + 429 => match retry_after { + Some(retry_after_secs) => FetchError::RateLimited { + site, + retry_after_secs: retry_after_secs.min(MAX_RETRY_AFTER_SECS), + }, + None => FetchError::Transient(format!("{site} status {status}")), + }, _ => FetchError::Transient(format!("{site} status {status}")), } } @@ -354,9 +393,13 @@ pub trait Site: Send + Sync { fn cache_key(&self, url: &str) -> Option; /// Fetches and normalizes a post. fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched>; - /// Retry policy for fetch errors: transient classes only. + /// Retry policy for fetch errors: transient classes only (a 429's + /// named `Retry-After` included — it is a bad moment, just a louder one). fn is_retryable(&self, err: &FetchError) -> bool { - matches!(err, FetchError::Http(_) | FetchError::Transient(_)) + matches!( + err, + FetchError::Http(_) | FetchError::Transient(_) | FetchError::RateLimited { .. } + ) } /// Extra headers for downloading this site's media (hotlink protection, /// e.g. pixiv's Referer for pximg.net). Matched on the media URL, not @@ -490,7 +533,7 @@ async fn fetch_with_attempts(url: &str, attempts: u32) -> Result } Err(err) => { if site.is_retryable(&err) && attempt + 1 < attempts { - tokio::time::sleep(Duration::from_secs(1 << attempt)).await; + tokio::time::sleep(retry_wait(attempt, rand::random::(), &err)).await; } else { return Err(err); } @@ -500,6 +543,23 @@ async fn fetch_with_attempts(url: &str, attempts: u32) -> Result unreachable!("retry loop always returns") } +/// How long to sleep before retrying `attempt` (0-based) after `err`: the +/// doubling base plus a random slice of it (roll in [0, base) → [base, 2×base)) +/// so workers that failed together do not recover together, floored at the +/// delay a 429's `Retry-After` named — already capped by the classifier at +/// [`MAX_RETRY_AFTER_SECS`], so an untrusted server cannot park a slot. +pub(crate) fn retry_wait(attempt: u32, roll: u64, err: &FetchError) -> Duration { + let base = 1u64 << attempt.min(16); + let mut secs = base + roll % base; + if let FetchError::RateLimited { + retry_after_secs, .. + } = err + { + secs = secs.max(*retry_after_secs); + } + Duration::from_secs(secs) +} + /// Whether fetching `url` requires site-specific headers (pixiv's `Referer` /// for `pximg.net` hotlink protection, see [`Site::media_headers`]). Telegram's /// own fetch of a media URL sends none of them, so a URL that needs them fails @@ -730,34 +790,76 @@ mod tests { // in two local fallbacks — twitter syndication's broken-token 400, for // one, burned three retries per link before saying the same thing. assert!(matches!( - status_error("x", StatusCode::NOT_FOUND), + classify_status("x", StatusCode::NOT_FOUND, None), FetchError::NotFound )); assert!(matches!( - status_error("x", StatusCode::BAD_REQUEST), + classify_status("x", StatusCode::BAD_REQUEST, None), FetchError::Blocked )); assert!(matches!( - status_error("x", StatusCode::PAYLOAD_TOO_LARGE), + classify_status("x", StatusCode::PAYLOAD_TOO_LARGE, None), FetchError::Blocked )); assert!(matches!( - status_error("x", StatusCode::REQUEST_TIMEOUT), + classify_status("x", StatusCode::REQUEST_TIMEOUT, None), FetchError::Transient(_) )); assert!(matches!( - status_error("x", StatusCode::TOO_MANY_REQUESTS), + classify_status("x", StatusCode::TOO_MANY_REQUESTS, None), FetchError::Transient(_) )); assert!(matches!( - status_error("x", StatusCode::INTERNAL_SERVER_ERROR), + classify_status("x", StatusCode::INTERNAL_SERVER_ERROR, None), FetchError::Transient(_) )); // The download path delegates under its own name, same classes. assert!(matches!( - status_error("media", StatusCode::BAD_REQUEST), + classify_status("media", StatusCode::BAD_REQUEST, None), FetchError::Blocked )); + // A 429 that named its delay keeps it — and the cap means the + // (server-supplied) header cannot park a fetch slot for an hour. + assert!(matches!( + classify_status("x", StatusCode::TOO_MANY_REQUESTS, Some(12)), + FetchError::RateLimited { + retry_after_secs: 12, + .. + } + )); + assert!(matches!( + classify_status("x", StatusCode::TOO_MANY_REQUESTS, Some(9999)), + FetchError::RateLimited { + retry_after_secs: crate::site::MAX_RETRY_AFTER_SECS, + .. + } + )); + } + + #[test] + fn retry_wait_jitters_and_respects_a_named_delay() { + // Doubling base plus a random slice: attempt 0 → exactly 1 s (any + // slice of 1 is 0), attempt 2 with roll 3 → 4 + 3 s. + assert_eq!( + retry_wait(0, 0, &FetchError::Transient("x".into())), + Duration::from_secs(1) + ); + assert_eq!( + retry_wait(2, 3, &FetchError::Transient("x".into())), + Duration::from_secs(7) + ); + // A named delay floors the wait: roll 0 would sleep 1 s, the source said 60. + assert_eq!( + retry_wait( + 0, + 0, + &FetchError::RateLimited { + site: "x", + retry_after_secs: 60 + } + ), + Duration::from_secs(60) + ); } #[tokio::test] diff --git a/crates/x-media/src/site/pixiv/api.rs b/crates/x-media/src/site/pixiv/api.rs index fdc2005..0e27e50 100644 --- a/crates/x-media/src/site/pixiv/api.rs +++ b/crates/x-media/src/site/pixiv/api.rs @@ -248,7 +248,9 @@ impl PixivAPI { // made one hiccup permanently fail the whole ugoira post, // while the bot's own upload downloads retry the same // classes. - transient @ (FetchError::Transient(_) | FetchError::Io(_)) => { + transient @ (FetchError::Transient(_) + | FetchError::RateLimited { .. } + | FetchError::Io(_)) => { PixivError::Transient(format!("frame zip download failed: {transient}")) } other => PixivError::Api(format!("frame zip download failed: {other}")), diff --git a/crates/x-media/src/site/pixiv/interface.rs b/crates/x-media/src/site/pixiv/interface.rs index bb3137b..c6be888 100644 --- a/crates/x-media/src/site/pixiv/interface.rs +++ b/crates/x-media/src/site/pixiv/interface.rs @@ -91,7 +91,7 @@ pub fn cache_key(url: &str) -> Option { /// API/auth errors, unparseable bodies and missing auth are not retried. pub fn is_retryable(err: &FetchError) -> bool { match err { - FetchError::Http(_) | FetchError::Transient(_) => true, + FetchError::Http(_) | FetchError::Transient(_) | FetchError::RateLimited { .. } => true, FetchError::Pixiv(e) => pixiv_error_is_retryable(e), _ => false, } @@ -464,6 +464,10 @@ mod tests { // Transient: network errors, explicit transient, pixiv 429/5xx, and a // failed media download (the frame zip's own bad moment). assert!(is_retryable(&FetchError::Transient("429".into()))); + assert!(is_retryable(&FetchError::RateLimited { + site: "pixiv", + retry_after_secs: 30 + })); assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(429)))); assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(500)))); assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(503)))); diff --git a/crates/x-media/src/site/twitter/auth.rs b/crates/x-media/src/site/twitter/auth.rs index c21e02e..91beb24 100644 --- a/crates/x-media/src/site/twitter/auth.rs +++ b/crates/x-media/src/site/twitter/auth.rs @@ -130,7 +130,7 @@ pub async fn fetch(id: &str) -> Result { let status = response.status(); if !status.is_success() { log::warn!("twitter auth fetch {id}: HTTP {status}"); - return Err(crate::site::status_error("twitter auth", status)); + return Err(crate::site::status_error("twitter auth", &response)); } let text = response.text().await?; let json: Value = serde_json::from_str(&text)?; diff --git a/crates/x-media/src/site/twitter/interface.rs b/crates/x-media/src/site/twitter/interface.rs index 1502983..1ae0e5a 100644 --- a/crates/x-media/src/site/twitter/interface.rs +++ b/crates/x-media/src/site/twitter/interface.rs @@ -87,7 +87,7 @@ pub async fn fetch(id: &str) -> Result { // 404/410 = gone (permanent); 429/5xx = transient and retried by fetch. let status = response.status(); if !status.is_success() { - return Err(crate::site::status_error("twitter", status)); + return Err(crate::site::status_error("twitter", &response)); } let text = response.text().await?; // Classify before building the tweet (see [`parse_syndication_body`]), and diff --git a/crates/xmedia-bot/src/handlers/urls.rs b/crates/xmedia-bot/src/handlers/urls.rs index 048351f..1212c98 100644 --- a/crates/xmedia-bot/src/handlers/urls.rs +++ b/crates/xmedia-bot/src/handlers/urls.rs @@ -507,7 +507,7 @@ fn fetch_error_message(err: &x_media::site::FetchError) -> String { FetchError::Disabled { site } => { format!("{} support is disabled on this bot.", site_title(site)) } - FetchError::Transient(_) | FetchError::Http(_) => { + FetchError::RateLimited { .. } | FetchError::Transient(_) | FetchError::Http(_) => { "The source site is unavailable right now (tried 3 times). Try again later.".to_string() } FetchError::MediaPrep(_) => concat!(