perf: degrade a link-cache entry instead of dropping it on a failed send

`link_cache` exists so a repeat link costs nothing: no source request, no
download, no upload. It was written only on a *successful* send, and a send
that failed permanently deleted the entry — so the user's immediate retry, the
one case where they are most likely to try again, re-fetched everything:
site requests, a download, and for a ugoira or a bsky video a full ffmpeg
encode. Invalidation is right about the cause (the cached Telegram file id is
what went stale) and wrong about the cure (the media and its URLs are usually
fine).

Cached media now carries the source URL it was sent from, and a permanent
failure *degrades* the entry: the file ids are cleared, the URLs and the
caption fields stay, and the next request sends from those URLs — Telegram
fetches the media (or the upload fallback does) with no source round trip.
That is the same media a fresh fetch would have produced (site CDN URLs are
stable per post), and it is bounded: an entry that is already degraded, or one
from before this field existed, is removed instead, so a dead post still ends
up re-fetched and reported rather than retried forever.

Verified: a cached send that fails permanently leaves the entry with its URL
and no file id, a second failure drops it, and a degraded entry sends the
media with no fetch at all (the mock records no reply, which is what the
fetch-error path would have produced). 124 bot tests + 91 x-media tests pass,
including a direct test of the two payload shapes.

`cargo fmt --check`, `cargo clippy --workspace --all-targets --locked -- -D
warnings` and `cargo test --workspace --locked` clean.

AGENTS.md

@@ -43,7 +43,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr
 | `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 and then applies the `PRAGMA user_version` migration chain (`MIGRATIONS` + `migrate` — append-only; `schema_init` is the version-0 baseline and must not gain columns an existing database would never receive — `db.rs`'s tests pin a pre-migration database upgrading intact, the shipped migration text frozen (appending is the only allowed change) and a fresh database landing at the latest version), `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; one *shared* in-flight fetch per cache key (`fetch_shared`: a second chat, a batch forward or a retry asking for the same post meanwhile waits for the first caller's result, the entry is dropped the moment the fetch settles so nothing is ever answered from an old fetch, and a waiter whose sharer was cancelled fetches for itself); plus the startup repair `repair_lost_local_media`, whose decision (`needs_refetch`) and rewrite (`apply_refresh`) are pure and tested while the fetch itself is a live test), `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); the 300 s sweep's `prune_expired` evicts any chat with no live edit-before-forward prompt, so the cache (and the per-chat lock map) stays bounded to active prompts — durable settings reload from the DB on next use |
-| `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/link_cache.rs` | `LinkCache`: SQLite-backed cache (`link_cache` table) of successfully sent posts — raw caption fields + the source media URLs + Telegram `file_id`s; repeat links re-send locally (no fetch/upload), TTL + prune; a permanent send failure *degrades* the entry instead of dropping it (the file ids go, the URLs stay, so the next request re-sends from those without a fetch), and a degraded entry that fails again is removed |
 | `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, a `lease_token` fence: `lease_next` stamps a random token and every write-back (heartbeat, `delete`, `reschedule`, `mark_done`) is guarded by it, so a lease that expired and was re-leased cannot be written by its former holder — a lost lease stops the attempt instead; a finished row's `DELETE`/reschedule retried and a failed delete falling back to a `done` tombstone (the lease query and the sweep only look at `pending`/`in_progress`, so a task that already ran cannot be resurrected and re-run), `runnable_rows`/`replace_payload` (the startup repair's read/rewrite path: it runs before the workers exist, which is why it needs no lease token), `notify_one` worker wakeup plus a separate `Notify` for the 30 s lease-expiry sweep (a shared one let the sweep steal the workers' wakeup permit; the sweep does notify the workers after it actually recovered a row, since a recovered task is due immediately while every worker may be parked on `notify` with no pending row to sleep on), `busy_timeout` on all connections |
 | `crates/xmedia-bot/src/ctx.rs` | `AppContext`: the injected collaborators (`sender` + `ChatStore`/`PersistentTaskQueue`/`LinkCache`/`Config`), `from_statics` for production and the `CONTEXT` static the worker closures hold. `test_support::TestStores` backs handler tests with a tempdir store set, and the module also carries the fixtures those tests share — the canonical cached post (`cached_photo`), the edit-before-forward prompt (`seed_prompt` with its `PROMPT_ID`/`FORWARDED_ID`) and a scripted API error (`api_error`) — so no two test modules keep their own copies |
 | `crates/xmedia-bot/src/send/` | `send/mod.rs`: `Task`/`MediaItemPayload` payloads, `SendError`/`Classification`, `send_media_sequence`/`send_animation`/`forward_messages`; `send/input_media.rs`: payload → `InputFile`/`InputMedia` + `build_media_group` (caption on the first item only); `send/upload.rs`: the download-and-reupload fallback (`prepare_upload_item`/`send_batch_via_upload`, photo downscale handoff); `send/post_send.rs`: link-cache write, `KEEP_ALIVE` registry, `settle_task`, `post_send_actions`, `handle_task`/`dead_letter_notify` |

crates/xmedia-bot/src/ctx.rs

@@ -80,6 +80,7 @@ pub(crate) mod test_support {
             media: vec![CachedMedia {
                 kind: CachedMediaKind::Photo,
                 file_id: "AgAC-file-id".into(),
+                url: "https://pbs.twimg.com/media/photo.jpg".into(),
             }],
         }
     }

crates/xmedia-bot/src/handlers/urls.rs

@@ -3,7 +3,7 @@
 
 use super::{log_key, reply};
 use crate::ctx::{AppContext, CONTEXT};
-use crate::link_cache::{CachedMediaKind, CachedPost};
+use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
 use crate::media_sender::MediaSender;
 use crate::send::{self, MediaItemPayload, Task};
 use crate::state::ChatData;
@@ -361,6 +361,47 @@ async fn dispatch_send(
     }
 }
 
+/// A cached entry's item as a send payload: the Telegram file id when the entry
+/// still has one, otherwise the source URL.
+///
+/// The URL case is a *degraded* entry — `send::post_send`'s `invalidate_cache`
+/// drops the file ids of an entry whose cached send failed permanently, keeping
+/// the URLs, because a stale file id says nothing about the media. Sending from
+/// the URL costs Telegram a fetch (or the upload fallback a download) and saves
+/// the whole source round trip, including a ugoira encode or an HLS remux.
+fn cached_media_payload(media: &CachedMedia, sensitive: bool) -> MediaItemPayload {
+    let has_file_id = !media.file_id.is_empty();
+    let source = if has_file_id {
+        media.file_id.clone()
+    } else {
+        media.url.clone()
+    };
+    // A degraded item carries no smaller variant: a fresh fetch's would, but
+    // the item is what the source itself sent, so an oversize is handled by the
+    // upload fallback rather than by a URL that was never recorded.
+    let fallback_url = None;
+    match media.kind {
+        CachedMediaKind::Photo => MediaItemPayload::Photo {
+            media: source,
+            has_spoiler: sensitive,
+            fallback_url,
+            file_id: has_file_id,
+        },
+        CachedMediaKind::Video => MediaItemPayload::Video {
+            media: source,
+            has_spoiler: sensitive,
+            thumbnail: None,
+            fallback_url,
+            file_id: has_file_id,
+        },
+        CachedMediaKind::Animation => MediaItemPayload::Animation {
+            media: source,
+            has_spoiler: sensitive,
+            file_id: has_file_id,
+        },
+    }
+}
+
 /// Whether a send also runs the chat's post-send actions. `/test` sends with
 /// them suppressed so a test can never forward to the channel or open the
 /// edit-before-forward prompt; a normal link uses whatever the chat is
@@ -636,26 +677,7 @@ async fn url_media_inner(
         let items: Vec<MediaItemPayload> = cached
             .media
             .iter()
-            .map(|m| match m.kind {
-                CachedMediaKind::Photo => MediaItemPayload::Photo {
-                    media: m.file_id.clone(),
-                    has_spoiler: cached.sensitive,
-                    fallback_url: None,
-                    file_id: true,
-                },
-                CachedMediaKind::Video => MediaItemPayload::Video {
-                    media: m.file_id.clone(),
-                    has_spoiler: cached.sensitive,
-                    thumbnail: None,
-                    fallback_url: None,
-                    file_id: true,
-                },
-                CachedMediaKind::Animation => MediaItemPayload::Animation {
-                    media: m.file_id.clone(),
-                    has_spoiler: cached.sensitive,
-                    file_id: true,
-                },
-            })
+            .map(|m| cached_media_payload(m, cached.sensitive))
             .collect();
         // The indicator switches to "sending photo/video" once the kinds are
         // known; `items` is moved into the task below.
@@ -989,7 +1011,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn cache_hit_sends_file_ids_and_invalidates_on_permanent_failure() {
+    async fn cache_hit_sends_file_ids_and_degrades_on_permanent_failure() {
         let stores = TestStores::new();
         let sender = MockSender::scripted(
             vec![Outcome::GroupErr, Outcome::MessageErr],
@@ -1006,14 +1028,83 @@ mod tests {
             sender.calls(),
             vec!["send_chat_action", "send_media_group", "send_message"]
         );
-        // The stale cache entry was invalidated so the next request re-fetches.
-        assert!(
-            stores
-                .link_cache()
-                .get("twitter:1", Duration::from_secs(3600))
-                .await
-                .is_none()
-        );
+        // The entry is degraded, not dropped: the next request re-sends from
+        // the source URL without a fetch.
+        let entry = stores
+            .link_cache()
+            .get("twitter:1", Duration::from_secs(3600))
+            .await
+            .expect("a stale file id must not cost the whole entry");
+        assert!(entry.media[0].file_id.is_empty());
+        assert_eq!(entry.media[0].url, "https://pbs.twimg.com/media/photo.jpg");
+    }
+
+    /// A degraded entry (see the test above) sends the source URL: no fetch,
+    /// no download of the post's data, and no upload of our own — Telegram
+    /// fetches the media it is pointed at. The absence of a `send_message`
+    /// (which the fetch-error path would emit) is what proves no fetch ran.
+    #[tokio::test]
+    async fn a_degraded_entry_sends_by_url_without_fetching() {
+        let stores = TestStores::new();
+        let sender = MockSender::scripted(vec![Outcome::GroupOk], permanent_error);
+        let ctx = stores.ctx(&sender);
+        let mut cached = cached_photo();
+        cached.media[0].file_id.clear();
+        stores.link_cache().put("twitter:1", &cached).await;
+
+        url_media(&ctx, 1, 2, "https://x.com/u/status/1", PostSend::FromChat).await;
+
+        assert_eq!(sender.calls(), vec!["send_chat_action", "send_media_group"]);
+        // Still cached, still degraded: a degraded entry keeps serving.
+        let entry = stores
+            .link_cache()
+            .get("twitter:1", Duration::from_secs(3600))
+            .await
+            .expect("the entry must stay");
+        assert!(entry.media[0].file_id.is_empty());
+    }
+
+    /// The two payload shapes a cached item can take, asserted directly: the
+    /// file id when there is one, the source URL when the entry was degraded.
+    #[test]
+    fn cached_media_payload_prefers_the_file_id_over_the_url() {
+        let with_id = CachedMedia {
+            kind: CachedMediaKind::Photo,
+            file_id: "AgAC".into(),
+            url: "https://p/1.jpg".into(),
+        };
+        match cached_media_payload(&with_id, true) {
+            MediaItemPayload::Photo {
+                media,
+                has_spoiler,
+                file_id,
+                ..
+            } => {
+                assert_eq!(media, "AgAC");
+                assert!(file_id, "a cached send must go by file id");
+                assert!(has_spoiler);
+            }
+            _ => panic!("expected a photo payload"),
+        }
+
+        let degraded = CachedMedia {
+            kind: CachedMediaKind::Video,
+            file_id: String::new(),
+            url: "https://v/1.mp4".into(),
+        };
+        match cached_media_payload(&degraded, false) {
+            MediaItemPayload::Video {
+                media,
+                file_id,
+                fallback_url,
+                ..
+            } => {
+                assert_eq!(media, "https://v/1.mp4");
+                assert!(!file_id, "a degraded send must go by URL");
+                assert!(fallback_url.is_none());
+            }
+            _ => panic!("expected a video payload"),
+        }
     }
 
     /// The caption-quote threshold matches the post's text inside the caption,

crates/xmedia-bot/src/link_cache.rs

@@ -26,6 +26,12 @@ pub enum CachedMediaKind {
 pub struct CachedMedia {
     pub kind: CachedMediaKind,
     pub file_id: String,
+    /// The media URL the send used, kept so an entry whose file ids stopped
+    /// working can still be re-sent without touching the source site (see the
+    /// bot's `invalidate_cache`). Empty for entries written before this field
+    /// existed — those can only be dropped and re-fetched.
+    #[serde(default)]
+    pub url: String,
 }
 
 /// Everything needed to re-send a post without touching the source site:
@@ -237,6 +243,8 @@ mod tests {
         let got = got.unwrap();
         assert_eq!(got.url, "https://x.com/u/status/1");
         assert_eq!(got.media[0].file_id, "AgAC-file-id");
+        // The source URL rides along: it is what a degraded entry falls back to.
+        assert_eq!(got.media[0].url, "https://pbs.twimg.com/media/photo.jpg");
     }
 
     #[tokio::test]

crates/xmedia-bot/src/send/mod.rs

@@ -258,6 +258,9 @@ fn collect_file_ids(messages: &[Message], batch: &[MediaItemPayload], out: &mut
             out.push(CachedMedia {
                 kind: kind_of_item(item),
                 file_id,
+                // A fresh send's item is the source URL (file ids only appear
+                // in a *cached* send, and `cache_sent_task` skips those).
+                url: item_url(item).to_string(),
             });
         }
     }
@@ -684,7 +687,7 @@ pub async fn send_animation(ctx: &AppContext<'_>, task: &Task) -> Result<Vec<i64
     {
         Ok(message) => {
             let id = message.id.0 as i64;
-            cache_animation_send(ctx, task, &message).await;
+            cache_animation_send(ctx, task, &message, media_url).await;
             Ok(vec![id])
         }
         Err(RequestError::Api(api)) if is_media_fetch_failure(&api) || is_size_error(&api) => {
@@ -719,7 +722,7 @@ pub async fn send_animation(ctx: &AppContext<'_>, task: &Task) -> Result<Vec<i64
                     {
                         Ok(message) => {
                             let id = message.id.0 as i64;
-                            cache_animation_send(ctx, task, &message).await;
+                            cache_animation_send(ctx, task, &message, media_url).await;
                             Ok(vec![id])
                         }
                         Err(e) => Err(classify_to_send_error(
@@ -1663,7 +1666,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn settled_failed_drops_the_cache_entry() {
+    async fn settled_failed_degrades_the_cache_entry_then_drops_it() {
         let sender = MockSender::scripted(vec![], media_fetch_error);
         let stores = TestStores::new();
         let ctx = stores.ctx(&sender);
@@ -1672,13 +1675,27 @@ mod tests {
 
         settle_task(&ctx, &task, Settled::Failed).await;
 
+        // The file id is what failed, not the media: the entry survives with
+        // its source URLs, so the next request re-sends without a fetch.
+        let entry = stores
+            .link_cache()
+            .get("twitter:1", Duration::from_secs(3600))
+            .await
+            .expect("a failed cached send must not drop the entry outright");
+        assert_eq!(entry.media.len(), 1);
+        assert!(entry.media[0].file_id.is_empty(), "the stale id must go");
+        assert_eq!(entry.media[0].url, "https://pbs.twimg.com/media/photo.jpg");
+        assert_eq!(entry.caption, "cap", "the text is still good");
+
+        // A second failure — this time the URLs did not work either — drops it.
+        settle_task(&ctx, &task, Settled::Failed).await;
         assert!(
             stores
                 .link_cache()
                 .get("twitter:1", Duration::from_secs(3600))
                 .await
                 .is_none(),
-            "a permanently failed cached send must drop the entry"
+            "a degraded entry that fails again must be dropped"
         );
     }
 }

crates/xmedia-bot/src/send/post_send.rs

@@ -7,7 +7,7 @@ use super::{SendError, Task, forward_messages, send_animation, send_media_sequen
 use crate::ctx::AppContext;
 use crate::db::{now_f64, unix_now};
 use crate::handlers::log_key;
-use crate::link_cache::{CachedMedia, CachedMediaKind, LinkCache};
+use crate::link_cache::{CachedMedia, CachedMediaKind};
 use crate::media_sender::MediaSender;
 use crate::queue::{PersistentTaskQueue, QueueError};
 use crate::state::EditMessage;
@@ -33,7 +33,12 @@ pub(super) async fn cache_sent_task(ctx: &AppContext<'_>, task: &Task, media: Ve
 }
 
 /// Persists a lone animation send under the post's cache key.
-pub(super) async fn cache_animation_send(ctx: &AppContext<'_>, task: &Task, message: &Message) {
+pub(super) async fn cache_animation_send(
+    ctx: &AppContext<'_>,
+    task: &Task,
+    message: &Message,
+    source_url: &str,
+) {
     if let Some(file_id) = message.animation().map(|a| a.file.id.to_string()) {
         cache_sent_task(
             ctx,
@@ -41,6 +46,7 @@ pub(super) async fn cache_animation_send(ctx: &AppContext<'_>, task: &Task, mess
             vec![CachedMedia {
                 kind: CachedMediaKind::Animation,
                 file_id,
+                url: source_url.to_string(),
             }],
         )
         .await;
@@ -57,25 +63,52 @@ pub(crate) enum Settled {
 /// Every path that ends a task's life — sent, permanently failed, or
 /// dead-lettered after the last retry — funnels through here, so the cleanup a
 /// settled task owes cannot be forgotten by a new path: release the keep-alive
-/// temp media (retryable tasks keep it, they will be resent) and drop the
-/// link-cache entry that a failed send's stale file ids would keep poisoning.
+/// temp media (retryable tasks keep it, they will be resent) and deal with the
+/// link-cache entry a failed send's stale file ids would keep poisoning
+/// (degraded to its source URLs, dropped once those fail too).
 pub(crate) async fn settle_task(ctx: &AppContext<'_>, task: &Task, outcome: Settled) {
     if matches!(outcome, Settled::Failed) {
-        invalidate_cache(ctx.link_cache, task).await;
+        invalidate_cache(ctx, task).await;
     }
     release_keep_alive(task);
 }
 
-/// A cached Telegram file id failed permanently (stale/expired); drop the
-/// cache entry so the next request re-fetches instead of repeating it.
-async fn invalidate_cache(cache: &LinkCache, task: &Task) {
-    if task.is_cached_send()
-        && let Some(url) = task.source_url()
-        && let Some(key) = x_media::site::cache_key(url)
-    {
-        log::debug!("removing stale link cache entry for [key={}]", log_key(url));
-        cache.remove(&key).await;
+/// A cached Telegram file id failed permanently (stale/expired). The media
+/// itself is usually fine, so the entry is *degraded* rather than dropped: its
+/// file ids go away and the source URLs stay, and the next request re-sends the
+/// post from those — no source request, no ugoira encode, no HLS remux — with
+/// the media fetched by Telegram (or by the upload fallback). An entry that is
+/// already degraded, or whose older rows carry no URLs, is removed instead: its
+/// URLs did not work either, and the next request should fetch the post again
+/// and report what the source says.
+async fn invalidate_cache(ctx: &AppContext<'_>, task: &Task) {
+    if !task.is_cached_send() {
+        return;
     }
+    let Some(url) = task.source_url() else {
+        return;
+    };
+    let Some(key) = x_media::site::cache_key(url) else {
+        return;
+    };
+    let Some(mut entry) = ctx.link_cache.get(&key, ctx.config.link_cache_ttl).await else {
+        return;
+    };
+    let degradable = entry.media.iter().all(|m| !m.url.is_empty())
+        && entry.media.iter().any(|m| !m.file_id.is_empty());
+    if !degradable {
+        log::debug!("removing stale link cache entry for [key={}]", log_key(url));
+        ctx.link_cache.remove(&key).await;
+        return;
+    }
+    log::debug!(
+        "degrading stale link cache entry to its source URLs for [key={}]",
+        log_key(url)
+    );
+    for media in &mut entry.media {
+        media.file_id.clear();
+    }
+    ctx.link_cache.put(&key, &entry).await;
 }
 
 /// Locally produced media files (ugoira MP4, bsky remux MP4) whose temp dirs