test: drop redundant tests, make the vacuous ones real

Audit of all 205 tests (five read-only passes plus a line-by-line
re-check). Ten test functions were removed or merged and eight
subsumed assertion blocks trimmed; the suite is down to 180 tests with
no loss of mutation coverage, and four tests that were passing for
nothing now fail when the code they name is broken.

Redundant (deleted or merged):
- twitter: `syndication_text_only_has_no_media` (re-asserts its own
  empty fixture), `..._keeps_multibyte_text` (both transforms are
  no-ops for that text), `..._strips_trailing_short_link_without_entities`
  (same branch as `..._media_short_link`, which now also covers the
  real multibyte tweet), `..._regardless_of_index_units` (its
  `display_text_range` rationale outlived the function it described).
- pixiv: `test_fetch` (a bare `is_ok()` on the illustration
  `download_media_pixiv_original_with_referer` already asserts and
  downloads, and the only network touch in a plain `cargo test`),
  `startup_validation_only_disables_on_a_definitive_failure` (four rows
  that are a subset of the retry-policy table; the `validate()` branch
  it was named for is not asserted at all).
- bilibili: `from_item_legacy_draw_shape_still_parses` (its fixture is
  the same legacy `draw` shape `from_item_maps_draw_images_and_topic`
  builds, with a subset of its assertions).
- site/mod.rs: two `Ok(None)` cases merged into one test.
- urls.rs: `cache_hit_success_keeps_the_cache_entry` (the `/test` test
  asserts the same two things under stricter settings), plus a
  `assert_ne!` loop that re-states the mapping assertions above it.
- send/mod.rs: `media_group_success_and_forward_ok` (the forward half is
  covered by `post_send_forwards_immediately_when_configured`; the
  `is_ok()` half cannot see the returned file ids), and two boundary
  rows implied by the constant they sit next to.
- commands.rs: the parse tail that `every_documented_invocation_parses`
  already covers per README form, and three `debug_report` rows the
  escaping test pins with stronger input.

Passing for nothing (now real):
- `truncate_caption_does_not_split_an_html_entity` — the cut lands
  inside the entity, so `!contains("&amp")` never fired; it now asserts
  the exact output in both directions and fails when the guard in
  `truncate_caption` is deleted (verified).
- `pipeline_resizes_oversized_jpeg` — magic bytes and a non-empty buffer
  pass for a copy-through; it now decodes the output's headers and
  fails when the JPEG branch skips the resize (verified).
- `live_validate_with_bogus_token_fails` — expected `PixivError::Api`,
  which the status check before the body read made unreachable; a bogus
  token is a 4xx. Confirmed against the live endpoint: the old
  assertion fails with `got Err(Status(400))`, the new one passes.
- bsky `live_fetch_with_photos` — its URL is a text-only post and it had
  a byte-identical twin, so no live test pinned media; it now points at a
  labelled post with photos and asserts media + the label (live-verified).

Also fixed, found by turning the runtime-sweep test into a real one:
the 30 s lease-expiry sweep recovered crashed rows but never woke a
worker, so a recovered task waited for the next unrelated enqueue (every
worker is parked on `notify` when no row is pending). `recover_update`
now reports its count, `recover_expired` wakes a worker when it changed
something, and `runtime_sweep_recovers_expired_lease` drives the spawned
loop with a paused clock instead of calling the recovery by hand — it
fails on both the missing wake-up and a sweep that recovers nothing.

Verified: `cargo fmt --check`, `cargo clippy --workspace --all-targets
--locked -- -D warnings`, `cargo test --workspace --locked` (180 passed,
14 ignored) and `cargo test -p x-media -- --ignored live` (13 passed).

crates/xmedia-bot/src/queue.rs

@@ -92,13 +92,30 @@ struct QueueWorker {
 }
 
 /// Resets rows left `in_progress` with an expired lock TTL back to `pending`
-/// so they can be leased again (crash/panic recovery).
-fn recover_update(conn: &rusqlite::Connection) -> rusqlite::Result<()> {
+/// so they can be leased again (crash/panic recovery). Returns how many rows
+/// came back, which is what decides whether a worker needs waking.
+fn recover_update(conn: &rusqlite::Connection) -> rusqlite::Result<usize> {
     conn.execute(
         "UPDATE tasks SET status='pending', locked_until=0 WHERE status='in_progress' AND locked_until < ?1",
         params![now_f64()],
-    )?;
-    Ok(())
+    )
+}
+
+/// Runs [`recover_update`] and wakes a worker when something actually came
+/// back. A recovered row is due immediately, but every worker may be parked on
+/// `notify` — with no pending row there is no `earliest_run_after` to sleep on
+/// — so without this the recovered task waits for the next unrelated enqueue.
+/// Same permit semantics as `enqueue`: `notify_one` stores a permit when no
+/// worker is registered.
+async fn recover_expired(pool: &std::sync::Arc<crate::db::DbPool>, notify: &Notify) {
+    match pool.with_conn(move |conn| recover_update(conn)).await {
+        Ok(recovered) if recovered > 0 => {
+            log::warn!("queue: recovered {recovered} row(s) from an expired lease");
+            notify.notify_one();
+        }
+        Ok(_) => {}
+        Err(e) => log::error!("queue recovery failed: {e}"),
+    }
 }
 
 /// Base delay × 2^attempts (attempts = retries already done), capped at 300s.
@@ -136,7 +153,7 @@ impl PersistentTaskQueue {
         let handler: Arc<Handler> = Arc::new(move |payload| Box::pin(handler(payload)));
         let dead_letter: Arc<DeadLetter> =
             Arc::new(move |payload, message| Box::pin(dead_letter(payload, message)));
-        self.recover_stale().await;
+        recover_expired(&self.pool, &self.notify).await;
         let mut handles = Vec::with_capacity(QUEUE_WORKERS + 1);
         for _ in 0..QUEUE_WORKERS {
             let worker = QueueWorker {
@@ -156,6 +173,7 @@ impl PersistentTaskQueue {
         let sweep_pool = std::sync::Arc::clone(&self.pool);
         let sweep_notify = Arc::clone(&self.sweep_notify);
         let sweep_stop = Arc::clone(&self.stop);
+        let sweep_workers = Arc::clone(&self.notify);
         handles.push(tokio::spawn(async move {
             let mut interval = tokio::time::interval(Duration::from_secs(30));
             loop {
@@ -168,10 +186,7 @@ impl PersistentTaskQueue {
                 if sweep_stop.load(Ordering::Relaxed) {
                     break;
                 }
-                let result = sweep_pool.with_conn(move |conn| recover_update(conn)).await;
-                if let Err(e) = result {
-                    log::error!("queue sweep failed: {e}");
-                }
+                recover_expired(&sweep_pool, &sweep_workers).await;
             }
         }));
         *self.worker.lock() = handles;
@@ -247,17 +262,6 @@ impl PersistentTaskQueue {
             }
         }
     }
-
-    async fn recover_stale(&self) {
-        self.recover_sweep().await;
-    }
-
-    async fn recover_sweep(&self) {
-        let result = self.pool.with_conn(move |conn| recover_update(conn)).await;
-        if let Err(e) = result {
-            log::error!("queue recovery failed: {e}");
-        }
-    }
 }
 
 /// Last-resort terminal state for a row whose `DELETE` would not go through:
@@ -960,7 +964,12 @@ mod tests {
         queue.stop().await;
     }
 
-    #[tokio::test]
+    /// A row that goes stale *after* startup is picked up by the periodic
+    /// sweep — the spawned task, its 30 s interval included — and the worker
+    /// that sweeps it gets woken. The paused clock is what makes this the real
+    /// test: this used to call the recovery by hand, which proved the SQL but
+    /// left the wiring free to be deleted.
+    #[tokio::test(start_paused = true)]
     async fn runtime_sweep_recovers_expired_lease() {
         let (queue, _dir) = new_queue().await;
         let calls = Arc::new(AtomicUsize::new(0));
@@ -975,8 +984,12 @@ mod tests {
                 |_payload, _message| async {},
             )
             .await;
-        // Insert a stale leased row AFTER startup: without a runtime sweep it
-        // would stay `in_progress` forever (only start() used to recover).
+        // Let every worker park and the sweep consume its immediate first tick,
+        // so only a later tick can see the row. The clock is paused: yields do
+        // not advance it, and the sleep below does.
+        for _ in 0..16 {
+            tokio::task::yield_now().await;
+        }
         {
             let conn = rusqlite::Connection::open(queue.pool.path()).unwrap();
             conn.execute(
@@ -986,12 +999,18 @@ mod tests {
             )
             .unwrap();
         }
-        queue.recover_sweep().await;
-        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(
+            calls.load(AtomicOrdering::SeqCst),
+            0,
+            "the row is stale but no sweep has run since it appeared"
+        );
+
+        tokio::time::sleep(Duration::from_secs(31)).await;
+
         assert_eq!(
             calls.load(AtomicOrdering::SeqCst),
             1,
-            "expired lease must be recovered and processed exactly once"
+            "the periodic sweep must recover the row and wake a worker"
         );
         queue.stop().await;
     }