feat(admin): crawler observability dashboard + reliability hardening (0.55.0)

Admin-only crawler dashboard backed by an SSE live-status stream,
coordinated browser restart, runtime PHPSESSID refresh, dead-letter
requeue, and a batch of reliability fixes. Closes everything from
the two-pass audit (10 commits' worth) and bumps 0.52.0 -> 0.55.0.

Backend:

- New /admin/crawler/* surface (cookie-auth, RequireAdmin) split
  into status / control / dead_jobs / backlog modules. SSE stream
  composes in-memory status with DB-derived queue counts, memoizes
  the counts for 1s and debounces watch pokes for 250ms (~10x QPS
  reduction per subscriber). One-shot GET /admin/crawler shares the
  same compose path.
- POST /admin/crawler/run gated by manual_pass_lock try_lock_owned
  (409 Conflict on overlapping click); browser restart goes through
  the coordinated_restart gate (drain + relaunch + auto-clear of the
  sticky session_expired flag on Ok).
- Runtime PHPSESSID refresh via SessionController (allow-list
  validation, never logged, audit row carries SHA-256 fingerprint).
  Storage layer is repo::crawler::runtime_session_{load,persist}.
- Dead-letter requeue with four scopes (all/manga/chapter/job);
  scope=all requires confirm:true; DISTINCT ON dedup keeps the
  partial unique index from rejecting requeues for chapters with
  multiple dead rows. SQL is four &'static str constants per scope.
- StatusHandle + ChapterGuard / CoverGuard RAII model survives
  panics; last-writer-wins on cover so concurrent dispatches don't
  clobber each other's slot. Pure functions (should_stop /
  should_mark_clean_exit / should_abort_pass) with named regression
  tests.
- Reliability bundle: per-lease heartbeat, jitter on retries,
  per-job timeout, circuit breaker on consecutive failures, BrowserManager
  coordinated restart gate, request fingerprint changes.
- Streaming page download: Storage::put_stream trait method,
  LocalStorage impl atomic via temp + fsync + UUID-suffixed rename.
  Pages stream through with peak memory ~one HTTP chunk + 64-byte
  sniff prefix instead of one full image per dispatch.
- New partial indexes (migration 0022): mangas_missing_cover_idx
  and crawler_jobs_dead_idx, both ordered by updated_at DESC to
  match the dashboard's LIMIT/OFFSET reads.
- Security hardening: admin_csrf_guard (Origin/Referer allowlist
  on /admin/* mutations, opt-in via ADMIN_ALLOWED_ORIGINS),
  admin_no_store_guard (Cache-Control: no-store on admin
  responses), audit rows carry per-scope target_id.

Frontend:

- /admin/crawler page decomposed into lib/components/crawler/
  (11 components: ProgressBar, SearchBar, CrawlerHero,
  CrawlerControls, ActiveChaptersCard, ActiveJobsTable,
  MissingCoversTable, DeadJobsTable, RestartConfirmModal,
  RequeueAllConfirmModal, SessionModal). Page is 532 LOC of
  orchestration; each component 22-148 LOC.
- EventSource lifecycle wired to visibilitychange / pagehide /
  pageshow (BFCache); after 5 consecutive errors probes the status
  endpoint so a 401 routes through the global on401Hook instead of
  infinite silent reconnects.
- Backlog $effect refetches debounced 500ms with per-loader
  AbortControllers; refresh after a control action only runs when
  the SSE stream is dead.
- Inline requeue button on /admin/mangas patches the affected row's
  sync_state locally (no full chapter-list refetch); proper
  aria-label. Requeue-all gets its own confirm modal; both confirm
  modals autofocus Cancel.
- SvelteKit reverse proxy bypasses its 5-minute AbortController
  for Accept: text/event-stream; pure shouldBypassProxyTimeout
  helper covered by unit tests.

Config / docs:

- New env vars (.env.example): ADMIN_ALLOWED_ORIGINS,
  CRAWLER_JOB_TIMEOUT_SECS, CRAWLER_METADATA_MAX_CONSECUTIVE_FAILURES,
  CRAWLER_BROWSER_RESTART_THRESHOLD.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/src/crawler/browser_manager.rs

@@ -13,7 +13,7 @@
 //! until [`BrowserManager::shutdown`].
 
 use std::ops::Deref;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -71,12 +71,42 @@ impl ActiveTracker {
     }
 }
 
+/// Lifecycle gate for a coordinated browser restart. `acquire()` parks
+/// while not [`RestartPhase::Healthy`] so no new navigation starts mid-
+/// restart; long-lived lease holders (the metadata pass) cooperate by
+/// checking [`BrowserManager::is_restart_pending`] at safe boundaries.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum RestartPhase {
+    /// Normal operation — acquires proceed.
+    Healthy,
+    /// Restart requested; new acquires park, waiting for in-flight leases
+    /// to drain.
+    Draining,
+    /// Chromium is being closed + relaunched.
+    Restarting,
+}
+
+const PHASE_HEALTHY: u8 = 0;
+const PHASE_DRAINING: u8 = 1;
+const PHASE_RESTARTING: u8 = 2;
+
 pub struct BrowserManager {
     inner: Mutex<Inner>,
     active: Arc<ActiveTracker>,
     launch_opts: LaunchOptions,
     idle_timeout: Duration,
     on_launch: OnLaunch,
+    /// Coarse lifecycle phase (one of the `PHASE_*` constants).
+    phase: AtomicU8,
+    /// Woken when the phase returns to `Healthy` so parked acquires resume.
+    resume: Notify,
+    /// Serialises coordinated restarts so concurrent requests collapse into
+    /// a single relaunch.
+    restart_lock: Mutex<()>,
+    /// Result of the most recent relaunch, so a caller that coalesced into
+    /// an in-progress restart reports that restart's real outcome instead
+    /// of a blind success.
+    last_restart_ok: AtomicBool,
 }
 
 struct Inner {
@@ -99,28 +129,72 @@ impl BrowserManager {
             launch_opts,
             idle_timeout,
             on_launch,
+            phase: AtomicU8::new(PHASE_HEALTHY),
+            resume: Notify::new(),
+            restart_lock: Mutex::new(()),
+            last_restart_ok: AtomicBool::new(true),
         })
     }
 
+    /// Current restart phase.
+    pub fn phase(&self) -> RestartPhase {
+        match self.phase.load(Ordering::Acquire) {
+            PHASE_DRAINING => RestartPhase::Draining,
+            PHASE_RESTARTING => RestartPhase::Restarting,
+            _ => RestartPhase::Healthy,
+        }
+    }
+
+    fn set_phase(&self, phase: RestartPhase) {
+        let v = match phase {
+            RestartPhase::Healthy => PHASE_HEALTHY,
+            RestartPhase::Draining => PHASE_DRAINING,
+            RestartPhase::Restarting => PHASE_RESTARTING,
+        };
+        self.phase.store(v, Ordering::Release);
+    }
+
+    /// Whether a coordinated restart is in progress. Long-lived lease
+    /// holders poll this at safe boundaries and yield their lease so the
+    /// drain can complete promptly.
+    pub fn is_restart_pending(&self) -> bool {
+        self.phase() != RestartPhase::Healthy
+    }
+
+    /// Launch Chromium into `guard`, running the `on_launch` hook before
+    /// publishing the handle so a probe failure doesn't leave a half-
+    /// initialised browser behind.
+    async fn launch_into(&self, guard: &mut Inner) -> anyhow::Result<()> {
+        let handle = browser::launch(self.launch_opts.clone())
+            .await
+            .context("BrowserManager: launch chromium")?;
+        let shared = handle.shared();
+        if let Err(e) = (self.on_launch)(Arc::clone(&shared)).await {
+            let _ = handle.close().await;
+            return Err(e.context("BrowserManager: on_launch hook failed"));
+        }
+        guard.handle = Some(handle);
+        guard.shared = Some(shared);
+        Ok(())
+    }
+
     /// Acquire a shared browser lease. The first acquire after a teardown
     /// launches a fresh Chromium (and runs `on_launch`); subsequent acquires
     /// while a process is alive just bump the counter and clone the `Arc`.
     pub async fn acquire(&self) -> anyhow::Result<BrowserLease> {
+        // Park while a coordinated restart is draining/relaunching so no new
+        // navigation starts against a browser that's about to be torn down.
+        // The short sleep fallback guarantees liveness even if a `resume`
+        // notification is missed (classic Notify lost-wakeup).
+        while self.phase() != RestartPhase::Healthy {
+            tokio::select! {
+                _ = self.resume.notified() => {}
+                _ = tokio::time::sleep(Duration::from_millis(100)) => {}
+            }
+        }
         let mut guard = self.inner.lock().await;
         if guard.handle.is_none() {
-            let handle = browser::launch(self.launch_opts.clone())
-                .await
-                .context("BrowserManager: launch chromium")?;
-            let shared = handle.shared();
-            // Run the on-launch hook before publishing the handle so a session
-            // probe failure doesn't leave a half-initialized browser behind.
-            if let Err(e) = (self.on_launch)(Arc::clone(&shared)).await {
-                // Close the just-launched browser since we won't be using it.
-                let _ = handle.close().await;
-                return Err(e.context("BrowserManager: on_launch hook failed"));
-            }
-            guard.handle = Some(handle);
-            guard.shared = Some(shared);
+            self.launch_into(&mut guard).await?;
         }
         let browser = guard
             .shared
@@ -134,6 +208,51 @@ impl BrowserManager {
         })
     }
 
+    /// Coordinated restart: block new acquires, wait for in-flight leases
+    /// to drain (up to `drain_deadline`, then force), close + relaunch
+    /// Chromium (re-running `on_launch` → re-inject session + probe), then
+    /// resume parked acquirers. Concurrent calls collapse into one
+    /// relaunch. The phase is always returned to `Healthy` — even if the
+    /// relaunch errors — so a failed restart never permanently wedges
+    /// acquisition (the next acquire retries the launch lazily).
+    pub async fn coordinated_restart(&self, drain_deadline: Duration) -> anyhow::Result<()> {
+        // Dedup: if a restart is already running, wait for it and report
+        // that restart's real outcome (not a blind success).
+        let _restart_guard = match self.restart_lock.try_lock() {
+            Ok(g) => g,
+            Err(_) => {
+                let _ = self.restart_lock.lock().await;
+                return if self.last_restart_ok.load(Ordering::Acquire) {
+                    Ok(())
+                } else {
+                    Err(anyhow::anyhow!("a concurrent coordinated browser restart failed"))
+                };
+            }
+        };
+
+        self.set_phase(RestartPhase::Draining);
+        await_drain(&self.active, drain_deadline).await;
+
+        self.set_phase(RestartPhase::Restarting);
+        let relaunch = {
+            let mut guard = self.inner.lock().await;
+            guard.shared = None;
+            if let Some(handle) = guard.handle.take() {
+                let _ = handle.close().await;
+            }
+            self.launch_into(&mut guard).await
+        };
+
+        self.last_restart_ok.store(relaunch.is_ok(), Ordering::Release);
+        self.set_phase(RestartPhase::Healthy);
+        self.resume.notify_waiters();
+        match &relaunch {
+            Ok(()) => tracing::info!("BrowserManager: coordinated restart complete"),
+            Err(e) => tracing::error!(error = ?e, "BrowserManager: coordinated restart relaunch failed"),
+        }
+        relaunch.context("coordinated_restart: relaunch")
+    }
+
     /// Forcefully close the cached browser regardless of active count.
     /// Used on daemon shutdown. After this returns the next acquire will
     /// re-launch from scratch.
@@ -176,6 +295,29 @@ impl BrowserManager {
     }
 }
 
+/// Wait for the active-lease count to reach zero, up to `deadline`. Wakes
+/// on the tracker's idle signal and re-checks on a short poll so a missed
+/// signal can't strand the drain. Returns when drained or when the
+/// deadline elapses (the caller then force-restarts). Extracted as a free
+/// fn so the timing logic is unit-testable without launching Chromium.
+async fn await_drain(active: &Arc<ActiveTracker>, deadline: Duration) {
+    let start = tokio::time::Instant::now();
+    while active.current() > 0 {
+        let Some(remaining) = deadline.checked_sub(start.elapsed()) else {
+            tracing::warn!(
+                active = active.current(),
+                "coordinated_restart: drain deadline exceeded — forcing relaunch"
+            );
+            return;
+        };
+        let nap = remaining.min(Duration::from_millis(250));
+        tokio::select! {
+            _ = active.idle_signal().notified() => {}
+            _ = tokio::time::sleep(nap) => {}
+        }
+    }
+}
+
 /// Background reaper. Returns immediately when `idle_timeout == 0`.
 /// Otherwise spawns a task that:
 ///   1. Waits on `idle_signal` (woken when active hits zero).
@@ -270,6 +412,63 @@ mod tests {
         mgr.invalidate().await;
     }
 
+    #[tokio::test]
+    async fn await_drain_returns_immediately_when_already_idle() {
+        let active = ActiveTracker::new();
+        let start = tokio::time::Instant::now();
+        await_drain(&active, Duration::from_secs(5)).await;
+        assert!(start.elapsed() < Duration::from_millis(200), "no wait when idle");
+    }
+
+    #[tokio::test]
+    async fn await_drain_completes_when_lease_released() {
+        let active = ActiveTracker::new();
+        active.acquire();
+        let bg = {
+            let a = Arc::clone(&active);
+            tokio::spawn(async move {
+                tokio::time::sleep(Duration::from_millis(100)).await;
+                a.release();
+            })
+        };
+        // Generous deadline; should return shortly after the release, not
+        // at the deadline.
+        let start = tokio::time::Instant::now();
+        await_drain(&active, Duration::from_secs(5)).await;
+        assert!(start.elapsed() < Duration::from_secs(2), "drained on release");
+        assert_eq!(active.current(), 0);
+        bg.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn await_drain_force_returns_after_deadline_when_stuck() {
+        let active = ActiveTracker::new();
+        active.acquire(); // never released
+        let start = tokio::time::Instant::now();
+        await_drain(&active, Duration::from_millis(300)).await;
+        let elapsed = start.elapsed();
+        assert!(elapsed >= Duration::from_millis(250), "waited ~deadline: {elapsed:?}");
+        assert!(elapsed < Duration::from_secs(2), "but not forever: {elapsed:?}");
+        assert_eq!(active.current(), 1, "still held — caller force-restarts");
+    }
+
+    #[test]
+    fn phase_transitions_reflect_is_restart_pending() {
+        let mgr = BrowserManager::new(
+            crate::crawler::browser::LaunchOptions::default(),
+            Duration::ZERO,
+            noop_on_launch(),
+        );
+        assert_eq!(mgr.phase(), RestartPhase::Healthy);
+        assert!(!mgr.is_restart_pending());
+        mgr.set_phase(RestartPhase::Draining);
+        assert!(mgr.is_restart_pending());
+        mgr.set_phase(RestartPhase::Restarting);
+        assert!(mgr.is_restart_pending());
+        mgr.set_phase(RestartPhase::Healthy);
+        assert!(!mgr.is_restart_pending());
+    }
+
     #[tokio::test]
     async fn active_tracker_signals_idle_only_on_zero_transition() {
         let tracker = ActiveTracker::new();

backend/src/crawler/content.rs

@@ -18,9 +18,9 @@ use uuid::Uuid;
 
 use crate::crawler::detect::PageError;
 use crate::crawler::rate_limit::HostRateLimiters;
-use crate::crawler::safety::{fetch_bytes_capped, looks_like_image, DownloadAllowlist};
+use crate::crawler::safety::{fetch_stream, looks_like_image, DownloadAllowlist};
 use crate::crawler::session::{self, ChapterProbe};
-use crate::storage::Storage;
+use crate::storage::{Storage, StorageError};
 
 /// Parse the chapter page DOM and return the page images in `pageN`
 /// order. Filters out the loader `<img class="loading">` and any
@@ -186,11 +186,17 @@ where
     }
 }
 
-/// Fetch all images for one chapter and persist them atomically. On
-/// any error after the first storage put, the DB transaction rolls
-/// back so the chapter stays at `page_count = 0` and is retried on the
-/// next run. Bytes already written to storage become orphans; a future
-/// reaper sweeps them.
+/// Fetch one chapter's images and persist them. Each image is
+/// streamed straight to storage via `Storage::put_stream` after a
+/// short prefix is peeked off the body for content-type sniffing —
+/// peak memory per concurrent dispatch is one HTTP chunk plus the
+/// sniff prefix, not a full multi-MB image. The per-image size cap
+/// (`CRAWLER_MAX_IMAGE_BYTES`) is enforced inside the stream so a
+/// server that omits Content-Length still can't exhaust memory. The
+/// page rows + `page_count` are then written in one short transaction.
+/// On any failure the chapter stays at `page_count = 0` (no partial
+/// rows) and the blobs already written are deleted best-effort by
+/// [`cleanup_orphans`], so a retry starts clean.
 #[allow(clippy::too_many_arguments)]
 pub async fn sync_chapter_content(
     browser: &chromiumoxide::Browser,
@@ -205,6 +211,10 @@ pub async fn sync_chapter_content(
     allowlist: &DownloadAllowlist,
     max_image_bytes: usize,
     tor: Option<&crate::crawler::tor::TorController>,
+    // Optional live-status sink for the realtime page counter. The daemon
+    // dispatcher passes the shared handle (the chapter has already been
+    // registered via `begin_chapter`); the CLI / admin resync pass `None`.
+    progress: Option<&crate::crawler::status::StatusHandle>,
 ) -> anyhow::Result<SyncOutcome> {
     // Skip if already fetched, unless caller explicitly forces.
     if !force_refetch {
@@ -260,56 +270,189 @@ pub async fn sync_chapter_content(
     // Resolve image URLs against the chapter URL (they may be relative).
     let base = reqwest::Url::parse(source_url).context("parse chapter URL")?;
 
-    // Fetch every image bytes-first into memory before writing
-    // anything. Lets us bail the whole chapter cleanly if any image
-    // fails — DB stays at page_count=0, no partial rows persisted.
-    let mut fetched: Vec<(i32, Vec<u8>, &'static str)> = Vec::with_capacity(images.len());
+    // Stream each image straight to storage as it's fetched, capping peak
+    // memory at a single image rather than the whole chapter. Track the
+    // keys written so they can be rolled back if a later page (or the
+    // final DB commit) fails — preserving the all-or-nothing guarantee
+    // without holding a DB transaction open across the network puts
+    // (which matters once `Storage` is backed by S3).
+    let total = images.len();
+    // Publish the now-known page total so the dashboard shows "0/N".
+    if let Some(p) = progress {
+        p.set_chapter_pages(chapter_id, 0, Some(total));
+    }
+    let mut written_keys: Vec<String> = Vec::with_capacity(total);
+    let mut stored: Vec<StoredPage> = Vec::with_capacity(total);
     for img in &images {
-        let url = base.join(&img.url).with_context(|| {
-            format!("join image URL {} onto {source_url}", img.url)
-        })?;
-        rate.wait_for(url.as_str()).await?;
-        let bytes = fetch_bytes_capped(
+        match download_and_store_page(
+            storage,
             http,
-            url.as_str(),
-            Some(source_url),
+            rate,
+            &base,
+            source_url,
+            manga_id,
+            chapter_id,
+            img,
             allowlist,
             max_image_bytes,
         )
-        .await?
-        .to_vec();
-        // Reject any non-image response: the only valid output of an
-        // image URL is an image. `infer` returns None on truncated
-        // bytes too, which also wants to be a failure not a silent
-        // `.bin` extension.
-        if !looks_like_image(&bytes) {
-            anyhow::bail!(
-                "image URL {url} returned non-image bytes \
-                 (first 16: {:?}); refusing to store as binary blob",
-                &bytes.get(..16.min(bytes.len()))
-            );
+        .await
+        {
+            Ok(page) => {
+                written_keys.push(page.storage_key.clone());
+                stored.push(page);
+                // Live page counter: push the climbing count to subscribers.
+                if let Some(p) = progress {
+                    p.set_chapter_pages(chapter_id, stored.len(), Some(total));
+                }
+            }
+            Err(e) => {
+                cleanup_orphans(storage, &written_keys).await;
+                return Err(e);
+            }
         }
-        let ext = infer::get(&bytes)
-            .map(|k| k.extension())
-            .expect("looks_like_image asserted infer succeeded");
-        fetched.push((img.page_number, bytes, ext));
     }
 
-    // Atomic write: storage puts + page row inserts + page_count
-    // update, all in one transaction. If anything fails, rollback +
-    // the chapter is retried next run. Storage orphans the bytes; a
-    // reaper sweeps them later.
-    let mut tx = db.begin().await.context("open chapter sync tx")?;
-    for (page_number, bytes, ext) in &fetched {
-        let key = format!(
-            "mangas/{manga_id}/chapters/{chapter_id}/pages/{:04}.{ext}",
-            page_number
+    // Short transaction: page rows + page_count only, no network I/O. On
+    // failure, roll back the stored bytes so the chapter stays at
+    // page_count=0 and is retried cleanly next run.
+    if let Err(e) = persist_pages(db, chapter_id, &stored).await {
+        cleanup_orphans(storage, &written_keys).await;
+        return Err(e);
+    }
+
+    Ok(SyncOutcome::Fetched { pages: stored.len() })
+}
+
+/// A page image that has been written to storage and is awaiting its DB
+/// row. Carries everything `persist_pages` needs.
+pub(crate) struct StoredPage {
+    page_number: i32,
+    storage_key: String,
+    content_type: String,
+}
+
+/// Bytes accumulated for content-type sniffing. `infer` only needs the
+/// first few bytes for image formats (the longest signature in our
+/// allow-list is AVIF at 12 bytes), but we read up to this many so a
+/// fragmented TCP frame still produces a confident sniff and the
+/// "first 16 bytes" diagnostic in the error path is useful.
+const SNIFF_PREFIX_BYTES: usize = 64;
+
+/// Download a single page image, validate it's really an image, and
+/// stream it to storage. Returns the storage key + content type. Does
+/// not touch the DB — persistence is batched into one short transaction
+/// afterward.
+///
+/// Streaming path: we peek the first [`SNIFF_PREFIX_BYTES`] from the
+/// HTTP body to determine the file extension (and thus the storage
+/// key), then re-emit those bytes followed by the rest of the response
+/// stream via `Storage::put_stream`. Peak memory per concurrent
+/// dispatch is one HTTP chunk (~16 KiB) plus the sniff prefix, not a
+/// full multi-MB image. The per-image cap is enforced as bytes flow.
+#[allow(clippy::too_many_arguments)]
+async fn download_and_store_page(
+    storage: &dyn Storage,
+    http: &reqwest::Client,
+    rate: &HostRateLimiters,
+    base: &reqwest::Url,
+    source_url: &str,
+    manga_id: Uuid,
+    chapter_id: Uuid,
+    img: &ChapterImage,
+    allowlist: &DownloadAllowlist,
+    max_image_bytes: usize,
+) -> anyhow::Result<StoredPage> {
+    use futures_util::StreamExt as _;
+    let url = base
+        .join(&img.url)
+        .with_context(|| format!("join image URL {} onto {source_url}", img.url))?;
+    rate.wait_for(url.as_str()).await?;
+    let resp = fetch_stream(http, url.as_str(), Some(source_url), allowlist).await?;
+    let mut body = resp.bytes_stream();
+
+    // Drain chunks until we have enough bytes to sniff confidently
+    // (or the body is shorter than the prefix). Enforces the per-image
+    // cap on the prefix accumulation too.
+    let mut prefix = bytes::BytesMut::new();
+    while prefix.len() < SNIFF_PREFIX_BYTES {
+        let Some(chunk) = body.next().await else { break };
+        let chunk = chunk
+            .with_context(|| format!("stream chunk for {url}"))?;
+        if prefix.len().saturating_add(chunk.len()) > max_image_bytes {
+            anyhow::bail!(
+                "image {url} exceeds {max_image_bytes}-byte cap (received >{}+{})",
+                prefix.len(),
+                chunk.len()
+            );
+        }
+        prefix.extend_from_slice(&chunk);
+    }
+    let prefix = prefix.freeze();
+
+    // Reject any non-image response: the only valid output of an image
+    // URL is an image. `infer` returns None on truncated bytes too,
+    // which is also a failure not a silent `.bin` extension.
+    if !looks_like_image(&prefix) {
+        anyhow::bail!(
+            "image URL {url} returned non-image bytes \
+             (first 16: {:?}); refusing to store as binary blob",
+            &prefix.get(..16.min(prefix.len()))
         );
-        storage
-            .put(&key, bytes)
-            .await
-            .with_context(|| format!("put {key}"))?;
-        // (chapter_id, page_number) is unique — re-runs idempotent.
+    }
+    let ext = infer::get(&prefix)
+        .map(|k| k.extension())
+        .expect("looks_like_image asserted infer succeeded");
+    let key = format!(
+        "mangas/{manga_id}/chapters/{chapter_id}/pages/{:04}.{ext}",
+        img.page_number
+    );
+
+    // Build a single stream of (prefix + remaining body) and pipe it
+    // straight to storage. The cap is enforced via a running total in
+    // the stream adapter so a server that omits Content-Length still
+    // can't exhaust memory.
+    let prefix_stream = futures_util::stream::once(async move {
+        Ok::<bytes::Bytes, StorageError>(prefix)
+    });
+    let prefix_len = SNIFF_PREFIX_BYTES.min(max_image_bytes);
+    let mut remaining = max_image_bytes.saturating_sub(prefix_len);
+    let url_for_err = url.clone();
+    let rest_stream = body.map(move |frame| match frame {
+        Ok(chunk) => {
+            if chunk.len() > remaining {
+                return Err(StorageError::Io(std::io::Error::other(format!(
+                    "image {url_for_err} exceeds {max_image_bytes}-byte cap"
+                ))));
+            }
+            remaining -= chunk.len();
+            Ok(chunk)
+        }
+        Err(e) => Err(StorageError::Io(std::io::Error::other(format!(
+            "stream chunk for {url_for_err}: {e}"
+        )))),
+    });
+    let combined = prefix_stream.chain(rest_stream);
+    storage
+        .put_stream(&key, Box::pin(combined))
+        .await
+        .with_context(|| format!("put_stream {key}"))?;
+    Ok(StoredPage {
+        page_number: img.page_number,
+        storage_key: key,
+        content_type: format!("image/{ext}"),
+    })
+}
+
+/// Persist the page rows + chapter `page_count` in one short transaction.
+/// `(chapter_id, page_number)` is unique so re-runs are idempotent.
+pub(crate) async fn persist_pages(
+    db: &PgPool,
+    chapter_id: Uuid,
+    stored: &[StoredPage],
+) -> anyhow::Result<()> {
+    let mut tx = db.begin().await.context("open chapter sync tx")?;
+    for page in stored {
         sqlx::query(
             "INSERT INTO pages (chapter_id, page_number, storage_key, content_type)
              VALUES ($1, $2, $3, $4)
@@ -318,22 +461,36 @@ pub async fn sync_chapter_content(
                  content_type = EXCLUDED.content_type",
         )
         .bind(chapter_id)
-        .bind(page_number)
-        .bind(&key)
-        .bind(format!("image/{ext}"))
+        .bind(page.page_number)
+        .bind(&page.storage_key)
+        .bind(&page.content_type)
         .execute(&mut *tx)
         .await
-        .with_context(|| format!("insert page row {page_number}"))?;
+        .with_context(|| format!("insert page row {}", page.page_number))?;
     }
     sqlx::query("UPDATE chapters SET page_count = $1 WHERE id = $2")
-        .bind(fetched.len() as i32)
+        .bind(stored.len() as i32)
         .bind(chapter_id)
         .execute(&mut *tx)
         .await
         .context("update page_count")?;
     tx.commit().await.context("commit chapter sync")?;
+    Ok(())
+}
 
-    Ok(SyncOutcome::Fetched { pages: fetched.len() })
+/// Best-effort delete of partially-written page blobs after a chapter sync
+/// fails, so a retry doesn't accumulate orphans. Errors are logged, not
+/// raised — a leftover blob is harmless and a future reaper can sweep it.
+pub(crate) async fn cleanup_orphans(storage: &dyn Storage, keys: &[String]) {
+    for key in keys {
+        if let Err(e) = storage.delete(key).await {
+            tracing::warn!(
+                %key,
+                error = ?e,
+                "failed to delete orphaned page blob after chapter sync failure"
+            );
+        }
+    }
 }
 
 // Suppress unused-import warning for `session::registrable_domain`
@@ -347,6 +504,90 @@ fn _keep_session_in_scope() {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::storage::LocalStorage;
+
+    #[tokio::test]
+    async fn cleanup_orphans_deletes_written_keys() {
+        let dir = tempfile::tempdir().unwrap();
+        let storage = LocalStorage::new(dir.path());
+        let keys = vec![
+            "mangas/m/chapters/c/pages/0001.jpg".to_string(),
+            "mangas/m/chapters/c/pages/0002.jpg".to_string(),
+        ];
+        for k in &keys {
+            storage.put(k, b"\xff\xd8\xff\xe0 jpeg-ish").await.unwrap();
+            assert!(storage.exists(k).await.unwrap());
+        }
+        cleanup_orphans(&storage, &keys).await;
+        for k in &keys {
+            assert!(!storage.exists(k).await.unwrap(), "{k} should be deleted");
+        }
+    }
+
+    #[tokio::test]
+    async fn cleanup_orphans_tolerates_missing_keys() {
+        // A key that was never written (e.g. the put itself failed) must
+        // not make cleanup error — it's best-effort.
+        let dir = tempfile::tempdir().unwrap();
+        let storage = LocalStorage::new(dir.path());
+        cleanup_orphans(&storage, &["never/written.jpg".to_string()]).await;
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn persist_pages_inserts_rows_and_sets_page_count(pool: PgPool) {
+        let manga_id = Uuid::new_v4();
+        let chapter_id = Uuid::new_v4();
+        sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')")
+            .bind(manga_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
+            .bind(chapter_id)
+            .bind(manga_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+
+        let stored = vec![
+            StoredPage {
+                page_number: 1,
+                storage_key: "k/0001.jpg".into(),
+                content_type: "image/jpeg".into(),
+            },
+            StoredPage {
+                page_number: 2,
+                storage_key: "k/0002.jpg".into(),
+                content_type: "image/jpeg".into(),
+            },
+        ];
+        persist_pages(&pool, chapter_id, &stored).await.unwrap();
+
+        let page_count: i32 =
+            sqlx::query_scalar("SELECT page_count FROM chapters WHERE id = $1")
+                .bind(chapter_id)
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(page_count, 2);
+        let rows: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
+                .bind(chapter_id)
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(rows, 2);
+
+        // Idempotent re-run (force refetch path): same rows, page_count stable.
+        persist_pages(&pool, chapter_id, &stored).await.unwrap();
+        let rows2: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
+                .bind(chapter_id)
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(rows2, 2, "re-run is idempotent via ON CONFLICT");
+    }
 
     #[test]
     fn parse_chapter_pages_skips_loader_and_sorts_by_id() {

backend/src/crawler/daemon.rs

@@ -48,6 +48,7 @@ use tokio_util::sync::CancellationToken;
 use crate::crawler::content::SyncOutcome;
 use crate::crawler::jobs::{self, JobPayload, Lease, KIND_SYNC_CHAPTER_CONTENT};
 use crate::crawler::pipeline;
+use crate::crawler::status::{Phase, StatusHandle};
 
 /// Fixed `pg_try_advisory_lock` key. ASCII "MANGALRD" interpreted as a
 /// big-endian i64. Hardcoded so every replica agrees on the lock identity
@@ -56,6 +57,15 @@ pub const CRON_LOCK_KEY: i64 = 0x4D414E47414C5244;
 
 const STATE_KEY_LAST_TICK: &str = "last_metadata_tick_at";
 
+/// Lease window handed to `jobs::lease`. Kept short, but continuously
+/// extended by the per-job heartbeat (see [`WorkerContext::process_lease`])
+/// so a long-but-healthy job never lapses and gets stolen.
+const LEASE_DURATION: Duration = Duration::from_secs(60);
+
+/// How often the heartbeat renews the lease while a job runs. A third of
+/// the lease window leaves two missed-beat's slack before expiry.
+const LEASE_HEARTBEAT: Duration = Duration::from_secs(20);
+
 #[async_trait]
 pub trait MetadataPass: Send + Sync {
     async fn run(&self) -> anyhow::Result<pipeline::MetadataStats>;
@@ -77,6 +87,13 @@ pub struct DaemonConfig {
     pub tz: Tz,
     pub retention_days: u32,
     pub session_expired: Arc<AtomicBool>,
+    /// Live status surface updated by the cron + workers.
+    pub status: StatusHandle,
+    /// Hard upper bound on a single job's dispatch. A job that exceeds it
+    /// is acked failed (exponential backoff) rather than wedging a worker
+    /// forever. Must exceed [`LEASE_HEARTBEAT`] and the realistic
+    /// single-job runtime.
+    pub job_timeout: Duration,
     /// Tasks that should run alongside the cron + workers and be cancelled
     /// on shutdown. Used to hand the daemon ownership of the browser
     /// manager's idle reaper.
@@ -123,6 +140,8 @@ pub fn spawn(pool: PgPool, cancel: CancellationToken, cfg: DaemonConfig) -> Daem
         tz,
         retention_days,
         session_expired,
+        status,
+        job_timeout,
         extra_tasks,
     } = cfg;
 
@@ -134,6 +153,7 @@ pub fn spawn(pool: PgPool, cancel: CancellationToken, cfg: DaemonConfig) -> Daem
             tz,
             retention_days,
             metadata,
+            status: status.clone(),
         };
         join.spawn(async move { ctx.run().await });
     } else {
@@ -146,6 +166,8 @@ pub fn spawn(pool: PgPool, cancel: CancellationToken, cfg: DaemonConfig) -> Daem
             cancel: cancel.clone(),
             dispatcher: Arc::clone(&dispatcher),
             session_expired: Arc::clone(&session_expired),
+            status: status.clone(),
+            job_timeout,
             id: worker_id,
         };
         join.spawn(async move { ctx.run().await });
@@ -169,6 +191,7 @@ struct CronContext {
     tz: Tz,
     retention_days: u32,
     metadata: Arc<dyn MetadataPass>,
+    status: StatusHandle,
 }
 
 impl CronContext {
@@ -196,6 +219,11 @@ impl CronContext {
             // (NTP step, suspend/resume) don't strand us on a stale instant.
             let next = next_fire(Utc::now(), self.daily_at, self.tz);
             let wait = (next - Utc::now()).to_std().unwrap_or(Duration::ZERO);
+            self.status
+                .set_phase(Phase::Idle {
+                    next_fire: Some(next),
+                })
+                .await;
             tracing::info!(
                 next_fire_utc = %next.to_rfc3339(),
                 wait_seconds = wait.as_secs(),
@@ -243,9 +271,13 @@ impl CronContext {
         let metadata = &self.metadata;
         let pool = &self.pool;
         let retention_days = self.retention_days;
+        let status = &self.status;
         let body = async move {
             match metadata.run().await {
-                Ok(stats) => tracing::info!(?stats, "cron: metadata pass done"),
+                Ok(stats) => {
+                    status.record_pass(&stats, Utc::now()).await;
+                    tracing::info!(?stats, "cron: metadata pass done");
+                }
                 Err(e) => tracing::error!(?e, "cron: metadata pass failed"),
             }
             match pipeline::enqueue_bookmarked_pending(pool).await {
@@ -283,6 +315,8 @@ struct WorkerContext {
     cancel: CancellationToken,
     dispatcher: Arc<dyn ChapterDispatcher>,
     session_expired: Arc<AtomicBool>,
+    status: StatusHandle,
+    job_timeout: Duration,
     id: usize,
 }
 
@@ -303,7 +337,7 @@ impl WorkerContext {
                 &self.pool,
                 Some(KIND_SYNC_CHAPTER_CONTENT),
                 1,
-                Duration::from_secs(60),
+                LEASE_DURATION,
             )
             .await
             {
@@ -341,9 +375,59 @@ impl WorkerContext {
             }
         }
 
-        let outcome = AssertUnwindSafe(self.dispatcher.dispatch(lease.payload.clone()))
-            .catch_unwind()
-            .await;
+        // Heartbeat: keep the lease fresh while the (potentially long)
+        // dispatch runs, so a slow-but-healthy job is never re-leased and
+        // never inflates `attempts` toward `max_attempts`. Stops itself
+        // once the job is no longer ours (renew returns false).
+        let heartbeat = {
+            let hb_pool = self.pool.clone();
+            let hb_id = lease.id;
+            tokio::spawn(async move {
+                loop {
+                    tokio::time::sleep(LEASE_HEARTBEAT).await;
+                    match jobs::renew(&hb_pool, hb_id, LEASE_DURATION).await {
+                        Ok(true) => {}
+                        Ok(false) => break,
+                        Err(e) => {
+                            tracing::warn!(lease_id = %hb_id, ?e, "heartbeat renew failed");
+                        }
+                    }
+                }
+            })
+        };
+
+        // The "currently crawling" chapter (with its live page count) is
+        // registered by the dispatcher itself (RealChapterDispatcher) so it
+        // carries the manga/chapter identity + page progress and is removed
+        // via an RAII guard on every exit path.
+
+        // Outer timeout: a dispatch that exceeds `job_timeout` is acked
+        // failed (exponential backoff) rather than wedging the worker.
+        let dispatch = AssertUnwindSafe(self.dispatcher.dispatch(lease.payload.clone()))
+            .catch_unwind();
+        let outcome = tokio::time::timeout(self.job_timeout, dispatch).await;
+        heartbeat.abort();
+
+        let outcome = match outcome {
+            Ok(o) => o,
+            Err(_elapsed) => {
+                tracing::warn!(
+                    worker = self.id,
+                    lease_id = %lease.id,
+                    timeout_secs = self.job_timeout.as_secs(),
+                    "worker: dispatch timed out — ack failed"
+                );
+                let _ = jobs::ack_failed(
+                    &self.pool,
+                    lease.id,
+                    "dispatch timed out",
+                    lease.attempts,
+                    lease.max_attempts,
+                )
+                .await;
+                return;
+            }
+        };
         match outcome {
             Ok(Ok(SyncOutcome::Fetched { .. } | SyncOutcome::Skipped)) => {
                 let _ = jobs::ack_done(&self.pool, lease.id).await;
@@ -355,6 +439,8 @@ impl WorkerContext {
                     "session expired — workers will idle until restart"
                 );
                 self.session_expired.store(true, Ordering::Release);
+                // Push the session-expired flip to live status subscribers.
+                self.status.poke();
                 let _ = jobs::release(&self.pool, lease.id).await;
             }
             Ok(Err(e)) => {

backend/src/crawler/jobs.rs

@@ -66,16 +66,33 @@ pub struct Lease {
     pub max_attempts: i32,
 }
 
-/// Exponential backoff for `ack_failed` retries. `attempts` is the
-/// post-increment value reported by `lease()` (so the first failure has
-/// `attempts == 1` and waits 60s, the second 120s, etc.). Capped at 1h to
-/// avoid runaway long sleeps that would outlive the daemon process.
-fn backoff_for(attempts: i32) -> Duration {
+/// Deterministic exponential backoff base for `ack_failed` retries.
+/// `attempts` is the post-increment value reported by `lease()` (so the
+/// first failure has `attempts == 1` and waits 60s, the second 120s,
+/// etc.). Capped at 1h to avoid runaway long sleeps that would outlive
+/// the daemon process. Jitter is applied separately by [`apply_jitter`].
+fn backoff_base(attempts: i32) -> Duration {
     let shift = attempts.saturating_sub(1).clamp(0, 20) as u32;
     let secs = 60u64.saturating_mul(1u64 << shift);
     Duration::from_secs(secs.min(3600))
 }
 
+/// Apply ±20% jitter to a backoff duration. `jitter` is a fraction in
+/// `[0.0, 1.0)` (e.g. `rand::random::<f64>()`), mapped to a multiplier in
+/// `[0.8, 1.2)`. Pure so the bounds stay unit-testable. Spreading retries
+/// avoids a thundering herd when a source outage fails many jobs at once.
+fn apply_jitter(base: Duration, jitter: f64) -> Duration {
+    let frac = jitter.clamp(0.0, 1.0);
+    let mult = 0.8 + 0.4 * frac; // [0.8, 1.2)
+    Duration::from_secs((base.as_secs_f64() * mult).round() as u64)
+}
+
+/// Jittered exponential backoff for `ack_failed`. Wraps [`backoff_base`]
+/// with a random ±20% spread.
+fn backoff_for(attempts: i32) -> Duration {
+    apply_jitter(backoff_base(attempts), rand::random::<f64>())
+}
+
 /// Insert a new pending job. For `SyncChapterContent` payloads the
 /// partial unique index `crawler_jobs_chapter_content_dedup_idx` blocks
 /// a second `(pending|running)` insert per chapter_id, returning
@@ -159,6 +176,35 @@ pub async fn lease(
     Ok(leases)
 }
 
+/// Extend the lease on a still-owned `running` job. Returns `true` if the
+/// row was updated (we still hold the lease), `false` if the job is no
+/// longer `running` (re-leased after a missed heartbeat, or already
+/// acked) — the caller's heartbeat loop should stop. The `state =
+/// 'running'` guard mirrors [`ack_done`]'s rationale.
+///
+/// This is the heartbeat primitive: a worker renews periodically while a
+/// long-but-healthy job runs so `leased_until` never lapses, which would
+/// otherwise let another worker steal the in-flight job and spuriously
+/// inflate `attempts` toward `max_attempts`.
+pub async fn renew(
+    pool: &PgPool,
+    lease_id: Uuid,
+    lease_duration: Duration,
+) -> sqlx::Result<bool> {
+    let lease_ms: i64 = lease_duration.as_millis().min(i64::MAX as u128) as i64;
+    let res = sqlx::query(
+        "UPDATE crawler_jobs \
+            SET leased_until = now() + ($2::bigint || ' milliseconds')::interval, \
+                updated_at = now() \
+          WHERE id = $1 AND state = 'running'",
+    )
+    .bind(lease_id)
+    .bind(lease_ms)
+    .execute(pool)
+    .await?;
+    Ok(res.rows_affected() > 0)
+}
+
 /// Mark a leased job as successfully completed. The `state = 'running'`
 /// predicate guards against a late ack from a worker whose lease expired
 /// and was already re-leased by another worker: without it, the late ack
@@ -278,19 +324,48 @@ mod tests {
     use super::*;
 
     #[test]
-    fn backoff_grows_exponentially_and_caps_at_one_hour() {
+    fn backoff_base_grows_exponentially_and_caps_at_one_hour() {
         // attempts == 1 → 60s, doubling each step.
-        assert_eq!(backoff_for(1), Duration::from_secs(60));
-        assert_eq!(backoff_for(2), Duration::from_secs(120));
-        assert_eq!(backoff_for(3), Duration::from_secs(240));
-        assert_eq!(backoff_for(4), Duration::from_secs(480));
-        assert_eq!(backoff_for(5), Duration::from_secs(960));
-        assert_eq!(backoff_for(6), Duration::from_secs(1920));
+        assert_eq!(backoff_base(1), Duration::from_secs(60));
+        assert_eq!(backoff_base(2), Duration::from_secs(120));
+        assert_eq!(backoff_base(3), Duration::from_secs(240));
+        assert_eq!(backoff_base(4), Duration::from_secs(480));
+        assert_eq!(backoff_base(5), Duration::from_secs(960));
+        assert_eq!(backoff_base(6), Duration::from_secs(1920));
         // 7th: 60 * 64 = 3840 → capped to 3600.
-        assert_eq!(backoff_for(7), Duration::from_secs(3600));
-        assert_eq!(backoff_for(20), Duration::from_secs(3600));
+        assert_eq!(backoff_base(7), Duration::from_secs(3600));
+        assert_eq!(backoff_base(20), Duration::from_secs(3600));
         // Garbage / zero / negatives stay sane.
-        assert_eq!(backoff_for(0), Duration::from_secs(60));
-        assert_eq!(backoff_for(-5), Duration::from_secs(60));
+        assert_eq!(backoff_base(0), Duration::from_secs(60));
+        assert_eq!(backoff_base(-5), Duration::from_secs(60));
+    }
+
+    #[test]
+    fn apply_jitter_stays_within_plus_minus_twenty_percent() {
+        let base = Duration::from_secs(100);
+        // Lower bound (jitter = 0.0) → 0.8x.
+        assert_eq!(apply_jitter(base, 0.0), Duration::from_secs(80));
+        // Midpoint (jitter = 0.5) → 1.0x.
+        assert_eq!(apply_jitter(base, 0.5), Duration::from_secs(100));
+        // Upper end (jitter → 1.0) → ~1.2x.
+        assert_eq!(apply_jitter(base, 1.0), Duration::from_secs(120));
+        // Out-of-range inputs are clamped, never panic.
+        assert_eq!(apply_jitter(base, -3.0), Duration::from_secs(80));
+        assert_eq!(apply_jitter(base, 9.0), Duration::from_secs(120));
+    }
+
+    #[test]
+    fn backoff_for_random_jitter_stays_in_band() {
+        // The production wrapper draws its own randomness; assert the
+        // result for a mid-range attempt always lands within the jitter
+        // band of the base, across many draws.
+        let base = backoff_base(3).as_secs_f64(); // 240s
+        for _ in 0..1000 {
+            let v = backoff_for(3).as_secs_f64();
+            assert!(
+                v >= base * 0.8 - 1.0 && v <= base * 1.2 + 1.0,
+                "jittered backoff {v} outside band of base {base}"
+            );
+        }
     }
 }

backend/src/crawler/mod.rs

@@ -26,6 +26,8 @@ pub mod rate_limit;
 pub mod resync;
 pub mod safety;
 pub mod session;
+pub mod session_control;
 pub mod source;
+pub mod status;
 pub mod tor;
 pub mod url_utils;

backend/src/crawler/pipeline.rs

@@ -65,6 +65,17 @@ pub(crate) fn should_mark_clean_exit(
     walked_to_completion || hit_stop_condition
 }
 
+/// Circuit-breaker: abort the walk once `consecutive` `fetch_manga`
+/// failures reach `threshold`. A `threshold` of 0 disables the breaker
+/// (unbounded — the legacy behaviour). When it fires the caller must NOT
+/// mark a clean exit, so the next tick does a recovery sweep over the
+/// catalog tail the aborted pass never reached.
+///
+/// Pure so the rule is unit-testable without the walker.
+pub(crate) fn should_abort_pass(consecutive: u32, threshold: u32) -> bool {
+    threshold > 0 && consecutive >= threshold
+}
+
 /// Runs the discover → fetch → upsert → cover → chapter-list-diff pipeline
 /// for the target source. Pure metadata; chapter content is enqueued as
 /// separate `SyncChapterContent` jobs by the caller after this returns.
@@ -103,6 +114,8 @@ pub async fn run_metadata_pass(
     skip_chapters: bool,
     allowlist: &DownloadAllowlist,
     max_image_bytes: usize,
+    max_consecutive_failures: u32,
+    status: Option<&crate::crawler::status::StatusHandle>,
     tor: Option<&crate::crawler::tor::TorController>,
 ) -> anyhow::Result<MetadataStats> {
     let lease = browser_manager
@@ -110,6 +123,9 @@ pub async fn run_metadata_pass(
         .await
         .context("acquire browser lease for metadata pass")?;
     let browser_ref: &chromiumoxide::Browser = &lease;
+    if let Some(s) = status {
+        s.set_phase(crate::crawler::status::Phase::WalkingList).await;
+    }
 
     let source = {
         let s = TargetSource::new(start_url.to_string());
@@ -165,6 +181,11 @@ pub async fn run_metadata_pass(
     let mut walked_to_completion = false;
     let mut hit_limit = false;
     let mut hit_stop_condition = false;
+    // Circuit-breaker state: consecutive fetch_manga failures. A sustained
+    // run abort (source outage) leaves the pass un-clean → recovery sweep
+    // next tick.
+    let mut consecutive_failures = 0u32;
+    let mut hit_failure_breaker = false;
 
     'outer: loop {
         let batch = match walker.next_batch(&ctx).await? {
@@ -175,6 +196,17 @@ pub async fn run_metadata_pass(
             }
         };
         for r in batch {
+            // Cooperative checkpoint: if a coordinated browser restart is
+            // pending, yield our (long-lived) lease so the drain can
+            // proceed instead of stalling for the rest of the walk. The
+            // pass exits un-clean, so the next tick recovery-sweeps the
+            // tail we didn't reach.
+            if browser_manager.is_restart_pending() {
+                tracing::info!(
+                    "metadata pass: browser restart pending — yielding (recovery sweep next tick)"
+                );
+                break 'outer;
+            }
             if max_refs.map(|m| stats.discovered >= m).unwrap_or(false) {
                 hit_limit = true;
                 tracing::info!(cap = ?max_refs, "max_results reached; halting walk");
@@ -198,13 +230,24 @@ pub async fn run_metadata_pass(
                 continue;
             }
             stats.discovered += 1;
+            if let Some(s) = status {
+                s.set_phase(crate::crawler::status::Phase::FetchingMetadata {
+                    index: stats.discovered,
+                    total: max_refs,
+                    title: r.title.clone(),
+                })
+                .await;
+            }
             tracing::info!(
                 idx = stats.discovered,
                 key = %r.source_manga_key,
                 "fetching metadata"
             );
             let manga = match source.fetch_manga(&ctx, &r).await {
-                Ok(m) => m,
+                Ok(m) => {
+                    consecutive_failures = 0;
+                    m
+                }
                 Err(e) => {
                     tracing::warn!(
                         key = %r.source_manga_key,
@@ -213,6 +256,17 @@ pub async fn run_metadata_pass(
                         "fetch_manga failed"
                     );
                     stats.mangas_failed += 1;
+                    consecutive_failures += 1;
+                    if should_abort_pass(consecutive_failures, max_consecutive_failures) {
+                        hit_failure_breaker = true;
+                        tracing::error!(
+                            consecutive_failures,
+                            threshold = max_consecutive_failures,
+                            "metadata pass: too many consecutive fetch_manga failures; \
+                             aborting (recovery sweep on next tick)"
+                        );
+                        break 'outer;
+                    }
                     continue;
                 }
             };
@@ -295,7 +349,16 @@ pub async fn run_metadata_pass(
                 || matches!(upsert.status, repo::crawler::UpsertStatus::Updated);
             if needs_cover {
                 if let Some(cover_url) = manga.cover_url.as_deref() {
-                    match download_and_store_cover(
+                    // RAII: the guard clears `current_cover` on every
+                    // exit path (success, panic, future early-return).
+                    // Mirrors the chapter-side ChapterGuard.
+                    let _cover_guard = status.map(|s| {
+                        s.begin_cover(crate::crawler::status::CoverTarget {
+                            manga_id: upsert.manga_id,
+                            manga_title: manga.title.clone(),
+                        })
+                    });
+                    let cover_result = download_and_store_cover(
                         db,
                         storage,
                         http,
@@ -306,8 +369,8 @@ pub async fn run_metadata_pass(
                         allowlist,
                         max_image_bytes,
                     )
-                    .await
-                    {
+                    .await;
+                    match cover_result {
                         Ok(()) => stats.covers_fetched += 1,
                         Err(e) => tracing::warn!(
                             manga_id = %upsert.manga_id,
@@ -390,6 +453,7 @@ pub async fn run_metadata_pass(
         walked_to_completion,
         hit_limit,
         hit_stop_condition,
+        hit_failure_breaker,
         exited_cleanly,
         "metadata pass complete"
     );
@@ -560,6 +624,7 @@ pub async fn backfill_missing_covers(
     max_mangas: usize,
     allowlist: &DownloadAllowlist,
     max_image_bytes: usize,
+    status: Option<&crate::crawler::status::StatusHandle>,
     tor: Option<&crate::crawler::tor::TorController>,
 ) -> anyhow::Result<CoverBackfillStats> {
     let mut stats = CoverBackfillStats::default();
@@ -582,8 +647,13 @@ pub async fn backfill_missing_covers(
     let browser_ref: &chromiumoxide::Browser = &lease;
     let ctx = FetchContext { browser: browser_ref, rate, tor };
 
-    for entry in entries {
+    let total = entries.len();
+    for (index, entry) in entries.into_iter().enumerate() {
         stats.considered += 1;
+        if let Some(s) = status {
+            s.set_phase(crate::crawler::status::Phase::CoverBackfill { index, total })
+                .await;
+        }
         // Metadata-only TargetSource: skip chapter-list parsing so a
         // missing-cover refetch doesn't soft-drop chapters on a partial
         // render. Cover URL alone is what we need.
@@ -593,8 +663,8 @@ pub async fn backfill_missing_covers(
             title: String::new(),
             url: entry.source_url.clone(),
         };
-        let cover_url = match source.fetch_manga(&ctx, &r).await {
-            Ok(manga) => manga.cover_url,
+        let manga = match source.fetch_manga(&ctx, &r).await {
+            Ok(manga) => manga,
             Err(e) => {
                 tracing::warn!(
                     manga_id = %entry.manga_id,
@@ -606,7 +676,7 @@ pub async fn backfill_missing_covers(
                 continue;
             }
         };
-        let Some(cover_url) = cover_url else {
+        let Some(cover_url) = manga.cover_url.clone() else {
             tracing::warn!(
                 manga_id = %entry.manga_id,
                 url = %entry.source_url,
@@ -615,7 +685,16 @@ pub async fn backfill_missing_covers(
             stats.failed += 1;
             continue;
         };
-        match download_and_store_cover(
+        // RAII guard: clears the live current_cover on every exit path,
+        // including a panic inside download_and_store_cover. Mirrors the
+        // chapter-side ChapterGuard.
+        let _cover_guard = status.map(|s| {
+            s.begin_cover(crate::crawler::status::CoverTarget {
+                manga_id: entry.manga_id,
+                manga_title: manga.title.clone(),
+            })
+        });
+        let cover_result = download_and_store_cover(
             db,
             storage,
             http,
@@ -626,8 +705,8 @@ pub async fn backfill_missing_covers(
             allowlist,
             max_image_bytes,
         )
-        .await
-        {
+        .await;
+        match cover_result {
             Ok(()) => stats.fetched += 1,
             Err(e) => {
                 tracing::warn!(
@@ -756,6 +835,18 @@ mod tests {
         assert!(!should_stop(false, UpsertStatus::New, None));
     }
 
+    #[test]
+    fn abort_pass_fires_at_threshold_and_respects_disable() {
+        // Disabled (0) never fires, no matter how many failures.
+        assert!(!should_abort_pass(0, 0));
+        assert!(!should_abort_pass(100, 0));
+        // Below threshold: keep going.
+        assert!(!should_abort_pass(9, 10));
+        // At/above threshold: abort.
+        assert!(should_abort_pass(10, 10));
+        assert!(should_abort_pass(11, 10));
+    }
+
     #[test]
     fn clean_exit_when_walked_to_completion() {
         // End-of-walk reached the catalog tail — the recovery flag may

backend/src/crawler/resync.rs

@@ -235,7 +235,7 @@ impl ResyncService for RealResyncService {
         let row = repo::chapter::dispatch_target(&self.db, chapter_id)
             .await
             .context("look up chapter_sources for resync")?;
-        let Some((manga_id, source_url)) = row else {
+        let Some((manga_id, source_url, _title, _number)) = row else {
             return Err(ResyncError::NoChapterSource.into());
         };
 
@@ -257,6 +257,8 @@ impl ResyncService for RealResyncService {
             &self.download_allowlist,
             self.max_image_bytes,
             self.tor.as_deref(),
+            // Admin resync isn't a daemon worker slot — no live status.
+            None,
         )
         .await;
         drop(lease);

backend/src/crawler/safety.rs

@@ -241,6 +241,31 @@ pub async fn fetch_bytes_capped(
         .with_context(|| format!("download body for {url}"))
 }
 
+/// Send `req` and return the response body as a stream after the
+/// safety check + 2xx status check. Caller owns chunking, capping, and
+/// piping to storage. Used by `download_and_store_page` so peak memory
+/// stays at one chunk per concurrent dispatch instead of one full
+/// image.
+pub async fn fetch_stream(
+    http: &reqwest::Client,
+    url: &str,
+    referer: Option<&str>,
+    allow: &DownloadAllowlist,
+) -> anyhow::Result<reqwest::Response> {
+    is_safe_url(url, allow).with_context(|| format!("reject unsafe URL {url}"))?;
+    let mut req = http.get(url);
+    if let Some(r) = referer {
+        req = req.header(reqwest::header::REFERER, r);
+    }
+    let resp = req
+        .send()
+        .await
+        .with_context(|| format!("GET {url}"))?
+        .error_for_status()
+        .with_context(|| format!("non-2xx for {url}"))?;
+    Ok(resp)
+}
+
 /// True when `bytes` sniffs as one of the *renderable* image formats
 /// the `/files/*key` endpoint can serve with a correct Content-Type:
 /// JPEG, PNG, WebP, GIF, AVIF. Matches the upload pipeline's

backend/src/crawler/session_control.rs

@@ -0,0 +1,184 @@
+//! Runtime-updatable crawler session (PHPSESSID).
+//!
+//! At startup the session comes from `CRAWLER_PHPSESSID`, but it expires
+//! and previously needed a container restart to refresh. This controller
+//! lets an admin push a fresh cookie at runtime: it rewrites the reqwest
+//! cookie jar (CDN image fetches), updates the in-memory value the browser
+//! `on_launch` hook reads, persists it to `crawler_state` (so it survives
+//! a restart), and clears the sticky `session_expired` flag. A subsequent
+//! coordinated browser restart re-runs `on_launch`, re-injecting the new
+//! cookie into Chromium and re-probing.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+use anyhow::Context;
+use sqlx::PgPool;
+use tokio::sync::RwLock;
+
+use crate::repo;
+
+pub struct SessionController {
+    /// Current PHPSESSID — what `on_launch` injects into a fresh browser.
+    phpsessid: RwLock<Option<String>>,
+    /// The same `Arc<Jar>` handed to the reqwest client; updating it here
+    /// updates the client's cookies (the jar is internally mutable).
+    cookie_jar: Arc<reqwest::cookie::Jar>,
+    cookie_domain: Option<String>,
+    start_url: Option<String>,
+    db: PgPool,
+    session_expired: Arc<AtomicBool>,
+}
+
+impl SessionController {
+    pub fn new(
+        initial: Option<String>,
+        cookie_jar: Arc<reqwest::cookie::Jar>,
+        cookie_domain: Option<String>,
+        start_url: Option<String>,
+        db: PgPool,
+        session_expired: Arc<AtomicBool>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            phpsessid: RwLock::new(initial),
+            cookie_jar,
+            cookie_domain,
+            start_url,
+            db,
+            session_expired,
+        })
+    }
+
+    /// The PHPSESSID a fresh browser should inject (None when unset).
+    pub async fn current(&self) -> Option<String> {
+        self.phpsessid.read().await.clone()
+    }
+
+    /// Whether the sticky session-expired flag is set (chapter workers
+    /// idle while true).
+    pub fn is_expired(&self) -> bool {
+        self.session_expired.load(Ordering::Acquire)
+    }
+
+    /// Clear the session-expired flag without changing the cookie — used
+    /// when the operator knows the session is fine and wants workers to
+    /// resume immediately.
+    pub fn clear_expired(&self) {
+        self.session_expired.store(false, Ordering::Release);
+    }
+
+    /// Update the session everywhere: reqwest jar, in-memory value, and
+    /// persisted `crawler_state`. Clears the session-expired flag. Does
+    /// NOT relaunch the browser — the caller triggers a coordinated
+    /// restart so `on_launch` re-injects + re-probes.
+    pub async fn update(&self, sid: &str) -> anyhow::Result<()> {
+        let sid = sid.trim().to_string();
+        anyhow::ensure!(!sid.is_empty(), "PHPSESSID must not be empty");
+        // The value is spliced into a cookie string and a CDP CookieParam.
+        // PHPSESSID values produced by PHP are URL-safe base64 alphanumerics
+        // plus a small set of punctuation depending on session.sid_bits_per_
+        // character. An allow-list (rather than a blocklist of control chars
+        // + `;,`) makes the check robust against future cookie syntax
+        // extensions and forces a paste that includes whitespace, quotes,
+        // backslashes, etc. — typical signs of a botched copy-paste — to
+        // be rejected early.
+        anyhow::ensure!(
+            sid.chars().all(is_phpsessid_char),
+            "PHPSESSID contains invalid characters"
+        );
+
+        if let (Some(domain), Some(start_url)) = (&self.cookie_domain, &self.start_url) {
+            let cookie_str = format!("PHPSESSID={sid}; Domain={domain}; Path=/");
+            let seed_url =
+                reqwest::Url::parse(start_url).context("parse start_url for cookie seed")?;
+            self.cookie_jar.add_cookie_str(&cookie_str, &seed_url);
+        }
+        *self.phpsessid.write().await = Some(sid.clone());
+        repo::crawler::runtime_session_persist(&self.db, &sid)
+            .await
+            .context("persist runtime session")?;
+        self.session_expired.store(false, Ordering::Release);
+        tracing::info!("crawler session updated at runtime");
+        Ok(())
+    }
+
+    /// Read a persisted runtime session (if any) from `crawler_state`.
+    /// Called at startup so a mid-day refresh survives a restart.
+    pub async fn load_persisted(db: &PgPool) -> Option<String> {
+        repo::crawler::runtime_session_load(db).await.ok().flatten()
+    }
+}
+
+/// Characters allowed in a PHPSESSID. PHP's session.sid_bits_per_character
+/// produces alphanumerics plus `-` and `,` in the lowest-bit mode, but our
+/// audit rejects `,` (cookie delimiter) — operators paste from a browser
+/// devtools snapshot, which never embeds raw commas in the SID itself.
+/// Underscore is allowed because some sources customise their session
+/// alphabet.
+fn is_phpsessid_char(c: char) -> bool {
+    c.is_ascii_alphanumeric() || matches!(c, '-' | '_')
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn controller(db: PgPool) -> Arc<SessionController> {
+        SessionController::new(
+            None,
+            Arc::new(reqwest::cookie::Jar::default()),
+            Some("example.com".into()),
+            Some("https://example.com/".into()),
+            db,
+            Arc::new(AtomicBool::new(true)),
+        )
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn update_rejects_empty_and_control_chars(pool: PgPool) {
+        let c = controller(pool);
+        assert!(c.update("   ").await.is_err(), "empty rejected");
+        assert!(c.update("abc\r\ndef").await.is_err(), "CRLF rejected");
+        assert!(c.update("ab;Domain=evil").await.is_err(), "semicolon rejected");
+        assert!(c.update("x,y").await.is_err(), "comma rejected");
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn update_rejects_non_alphanumeric_pastes(pool: PgPool) {
+        // Allow-list tightening (M6): pastes that include whitespace,
+        // quotes, slashes, backslashes, `=`, etc. are typical signs of a
+        // botched copy-paste and must be rejected outright.
+        let c = controller(pool);
+        for bad in ["ab cd", "ab\"cd", "ab=cd", "ab/cd", "ab\\cd", "ab+cd", "ab.cd"] {
+            assert!(c.update(bad).await.is_err(), "{bad:?} should be rejected");
+        }
+        // Allowed cases (sanity): plain alphanumerics, '-' and '_'.
+        assert!(c.update("abc_DEF-123").await.is_ok());
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn update_persists_and_clears_expired_then_round_trips(pool: PgPool) {
+        let c = controller(pool.clone());
+        c.update("good-sid-123").await.unwrap();
+        assert_eq!(c.current().await.as_deref(), Some("good-sid-123"));
+        assert!(!c.is_expired(), "update clears the expired flag");
+        // Persisted to crawler_state and readable by a fresh load.
+        assert_eq!(
+            SessionController::load_persisted(&pool).await.as_deref(),
+            Some("good-sid-123")
+        );
+    }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn clear_expired_flips_sticky_flag_without_touching_session(pool: PgPool) {
+        // The flag starts `true` per `controller(pool)`'s test wiring.
+        let c = controller(pool);
+        assert!(c.is_expired(), "test fixture starts with the flag set");
+        c.clear_expired();
+        assert!(!c.is_expired(), "clear_expired flips the sticky flag to false");
+        assert!(
+            c.current().await.is_none(),
+            "clear_expired does not invent a session"
+        );
+    }
+}

backend/src/crawler/status.rs

@@ -0,0 +1,456 @@
+//! Live, in-process crawler status.
+//!
+//! The metadata pass runs inline in the cron tick (it is not a
+//! `crawler_jobs` row), so without this surface "what is the crawler doing
+//! right now" is unanswerable from the dashboard. The daemon publishes its
+//! current [`Phase`], the chapters being crawled right now (with a live
+//! page count), and the cover being fetched into a shared [`StatusHandle`];
+//! the admin endpoint reads a [`CrawlerStatus`] snapshot and composes it
+//! with DB-derived counts + the session/browser flags.
+//!
+//! NOTE: this is per-process state. The deployment is a single server
+//! (see CLAUDE.md), so an in-memory handle is sufficient; durable signals
+//! (last-pass summary, runtime session) are persisted in `crawler_state`.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use tokio::sync::{watch, RwLock};
+use uuid::Uuid;
+
+use crate::crawler::pipeline::MetadataStats;
+
+/// What the daemon's metadata pass is doing right now. Serialised with an
+/// internal `state` tag so the frontend can switch on it.
+#[derive(Clone, Debug, Serialize)]
+#[serde(tag = "state", rename_all = "snake_case")]
+pub enum Phase {
+    /// Sleeping until the next scheduled metadata pass.
+    Idle { next_fire: Option<DateTime<Utc>> },
+    /// Walking the source catalog list pages.
+    WalkingList,
+    /// Fetching one manga's metadata. `index`/`total` drive a progress bar
+    /// (`total` is `None` when the source size is unknown / uncapped).
+    FetchingMetadata {
+        index: usize,
+        total: Option<usize>,
+        title: String,
+    },
+    /// Backfilling covers that failed on first attempt. `index`/`total`
+    /// track progress through this tick's batch.
+    CoverBackfill { index: usize, total: usize },
+}
+
+/// A chapter being downloaded right now, with a live page count. Keyed in
+/// the status by `chapter_id`; inserted by the dispatcher when a job starts
+/// and removed (via an RAII guard) when it finishes, panics, or times out.
+#[derive(Clone, Debug, Serialize)]
+pub struct ActiveChapter {
+    pub manga_id: Uuid,
+    pub manga_title: String,
+    pub chapter_id: Uuid,
+    pub chapter_number: i32,
+    pub pages_done: usize,
+    /// `None` until the chapter page list has been parsed.
+    pub pages_total: Option<usize>,
+}
+
+/// The manga whose cover is being downloaded right now.
+#[derive(Clone, Debug, Serialize)]
+pub struct CoverTarget {
+    pub manga_id: Uuid,
+    pub manga_title: String,
+}
+
+/// Summary of the most recent metadata pass (persisted across restarts in
+/// `crawler_state` by the cron; mirrored here for the live read).
+#[derive(Clone, Debug, Serialize, Default)]
+pub struct LastPass {
+    pub at: Option<DateTime<Utc>>,
+    pub discovered: usize,
+    pub upserted: usize,
+    pub covers_fetched: usize,
+    pub mangas_failed: usize,
+}
+
+/// A point-in-time snapshot returned by [`StatusHandle::snapshot`]. The
+/// session/browser/queue fields are composed at read time by the endpoint
+/// (they live elsewhere), so they are not stored here.
+#[derive(Clone, Debug, Serialize)]
+pub struct CrawlerStatus {
+    pub phase: Phase,
+    /// Number of configured chapter workers (for "N busy / M workers").
+    pub worker_count: usize,
+    /// Chapters being downloaded right now, with live page counts.
+    pub active_chapters: Vec<ActiveChapter>,
+    pub last_pass: LastPass,
+    /// The cover being downloaded right now, if any.
+    pub current_cover: Option<CoverTarget>,
+}
+
+/// Scalar status state held under the async `RwLock`. Active chapters and
+/// the current cover live in separate sync maps so per-page updates and
+/// RAII removal don't need to `.await` (removal happens in `Drop`).
+#[derive(Clone, Debug)]
+struct Scalar {
+    phase: Phase,
+    worker_count: usize,
+    last_pass: LastPass,
+}
+
+/// Cloneable handle the daemon tasks use to publish status. Cheap to clone
+/// (`Arc`). All writers funnel through the helper methods so locking stays
+/// localised. Every mutation bumps a `watch` version so SSE subscribers
+/// get pushed an update instead of polling.
+#[derive(Clone)]
+pub struct StatusHandle {
+    scalar: Arc<RwLock<Scalar>>,
+    /// Currently-downloading chapters keyed by `chapter_id`. A sync mutex so
+    /// the RAII [`ChapterGuard`]'s `Drop` can remove without `.await`.
+    active: Arc<Mutex<HashMap<Uuid, ActiveChapter>>>,
+    /// The cover being downloaded right now (if any). Sync mutex so the
+    /// RAII [`CoverGuard`]'s `Drop` can clear without `.await`, which is
+    /// what makes the cleared-on-panic guarantee hold.
+    current_cover: Arc<Mutex<Option<CoverTarget>>>,
+    /// Monotonic version bumped on every change. SSE handlers `subscribe()`
+    /// and `await .changed()` for instant pushes; `watch` has no
+    /// lost-wakeup so a change between snapshots is never missed.
+    version: Arc<watch::Sender<u64>>,
+}
+
+/// Lock the active map, recovering from a poisoned mutex. The map values
+/// are plain structs and we never hold the lock across a panic-prone
+/// section, so resuming on poison is safe — but log it so a real poison
+/// (which signals a panic-in-critical-section bug somewhere) doesn't pass
+/// in silence.
+fn lock_active(
+    m: &Mutex<HashMap<Uuid, ActiveChapter>>,
+) -> std::sync::MutexGuard<'_, HashMap<Uuid, ActiveChapter>> {
+    m.lock().unwrap_or_else(|e| {
+        tracing::warn!(
+            "status::lock_active recovered from a poisoned mutex — \
+             this implies a panic somewhere holding the lock"
+        );
+        e.into_inner()
+    })
+}
+
+/// Same shape as [`lock_active`] but for the single-slot cover mutex.
+fn lock_cover(
+    m: &Mutex<Option<CoverTarget>>,
+) -> std::sync::MutexGuard<'_, Option<CoverTarget>> {
+    m.lock().unwrap_or_else(|e| {
+        tracing::warn!(
+            "status::lock_cover recovered from a poisoned mutex — \
+             this implies a panic somewhere holding the lock"
+        );
+        e.into_inner()
+    })
+}
+
+impl StatusHandle {
+    pub fn new(num_workers: usize) -> Self {
+        let (version, _rx) = watch::channel(0u64);
+        Self {
+            scalar: Arc::new(RwLock::new(Scalar {
+                phase: Phase::Idle { next_fire: None },
+                worker_count: num_workers.max(1),
+                last_pass: LastPass::default(),
+            })),
+            active: Arc::new(Mutex::new(HashMap::new())),
+            current_cover: Arc::new(Mutex::new(None)),
+            version: Arc::new(version),
+        }
+    }
+
+    fn bump(&self) {
+        self.version.send_modify(|v| *v = v.wrapping_add(1));
+    }
+
+    /// A receiver whose `.changed()` resolves on the next status change.
+    pub fn subscribe(&self) -> watch::Receiver<u64> {
+        self.version.subscribe()
+    }
+
+    /// Signal a change without mutating in-memory state — used when an
+    /// *external* signal the live snapshot reflects (browser phase,
+    /// session-expired flag, queue counts) has changed, so subscribers
+    /// recompose promptly.
+    pub fn poke(&self) {
+        self.bump();
+    }
+
+    pub async fn set_phase(&self, phase: Phase) {
+        self.scalar.write().await.phase = phase;
+        self.bump();
+    }
+
+    /// Register a cover-fetch as in flight; returns a guard that clears
+    /// the current cover when dropped (on completion, panic-unwind, or
+    /// any future early-return). Last-writer-wins: a guard only clears
+    /// the slot when it still holds the cover it set (so overlapping
+    /// guards — not used today, but defensive — don't clobber each
+    /// other).
+    pub fn begin_cover(&self, target: CoverTarget) -> CoverGuard {
+        let manga_id = target.manga_id;
+        *lock_cover(&self.current_cover) = Some(target);
+        self.bump();
+        CoverGuard {
+            current_cover: Arc::clone(&self.current_cover),
+            version: Arc::clone(&self.version),
+            manga_id,
+        }
+    }
+
+    /// Register a chapter as crawling now; returns a guard that removes it
+    /// when dropped (on completion, panic-unwind, or timeout-drop).
+    pub fn begin_chapter(&self, chapter: ActiveChapter) -> ChapterGuard {
+        let id = chapter.chapter_id;
+        lock_active(&self.active).insert(id, chapter);
+        self.bump();
+        ChapterGuard {
+            active: Arc::clone(&self.active),
+            version: Arc::clone(&self.version),
+            chapter_id: id,
+        }
+    }
+
+    /// Update the live page count of an in-flight chapter. Sync (no
+    /// `.await`) so it's cheap to call once per stored page.
+    pub fn set_chapter_pages(&self, chapter_id: Uuid, done: usize, total: Option<usize>) {
+        {
+            let mut map = lock_active(&self.active);
+            if let Some(c) = map.get_mut(&chapter_id) {
+                c.pages_done = done;
+                c.pages_total = total;
+            }
+        }
+        self.bump();
+    }
+
+    /// Record a finished metadata pass. Stamps `at` with `now`.
+    pub async fn record_pass(&self, stats: &MetadataStats, at: DateTime<Utc>) {
+        self.scalar.write().await.last_pass = LastPass {
+            at: Some(at),
+            discovered: stats.discovered,
+            upserted: stats.upserted,
+            covers_fetched: stats.covers_fetched,
+            mangas_failed: stats.mangas_failed,
+        };
+        self.bump();
+    }
+
+    /// Seed the last-pass summary from a persisted `crawler_state` value on
+    /// startup so the dashboard isn't blank until the first tick.
+    pub async fn set_last_pass(&self, last: LastPass) {
+        self.scalar.write().await.last_pass = last;
+        self.bump();
+    }
+
+    pub async fn snapshot(&self) -> CrawlerStatus {
+        let scalar = self.scalar.read().await.clone();
+        let mut active_chapters: Vec<ActiveChapter> =
+            lock_active(&self.active).values().cloned().collect();
+        // Stable, readable order: by chapter number then id.
+        active_chapters.sort_by(|a, b| {
+            a.chapter_number
+                .cmp(&b.chapter_number)
+                .then(a.chapter_id.cmp(&b.chapter_id))
+        });
+        let current_cover = lock_cover(&self.current_cover).clone();
+        CrawlerStatus {
+            phase: scalar.phase,
+            worker_count: scalar.worker_count,
+            active_chapters,
+            last_pass: scalar.last_pass,
+            current_cover,
+        }
+    }
+}
+
+/// RAII handle clearing the [`CoverTarget`] from the live status when the
+/// cover-fetch finishes, panics, or is dropped on any early-return.
+pub struct CoverGuard {
+    current_cover: Arc<Mutex<Option<CoverTarget>>>,
+    version: Arc<watch::Sender<u64>>,
+    /// Manga id whose cover this guard registered. The drop only clears
+    /// the slot when the stored value still matches — defends against a
+    /// hypothetical newer guard clobbering this one's clear.
+    manga_id: Uuid,
+}
+
+impl Drop for CoverGuard {
+    fn drop(&mut self) {
+        let mut slot = lock_cover(&self.current_cover);
+        if slot.as_ref().map(|c| c.manga_id) == Some(self.manga_id) {
+            *slot = None;
+        }
+        self.version.send_modify(|v| *v = v.wrapping_add(1));
+    }
+}
+
+/// RAII handle removing an [`ActiveChapter`] from the live status when the
+/// chapter dispatch finishes, panics, or is dropped on timeout.
+pub struct ChapterGuard {
+    active: Arc<Mutex<HashMap<Uuid, ActiveChapter>>>,
+    version: Arc<watch::Sender<u64>>,
+    chapter_id: Uuid,
+}
+
+impl Drop for ChapterGuard {
+    fn drop(&mut self) {
+        lock_active(&self.active).remove(&self.chapter_id);
+        self.version.send_modify(|v| *v = v.wrapping_add(1));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sample_chapter(n: i32) -> ActiveChapter {
+        ActiveChapter {
+            manga_id: Uuid::new_v4(),
+            manga_title: "M".into(),
+            chapter_id: Uuid::new_v4(),
+            chapter_number: n,
+            pages_done: 0,
+            pages_total: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn begin_chapter_shows_in_snapshot_and_guard_removes_on_drop() {
+        let h = StatusHandle::new(2);
+        let chap = sample_chapter(7);
+        let cid = chap.chapter_id;
+        {
+            let _guard = h.begin_chapter(chap);
+            let snap = h.snapshot().await;
+            assert_eq!(snap.active_chapters.len(), 1);
+            assert_eq!(snap.active_chapters[0].chapter_id, cid);
+            assert_eq!(snap.worker_count, 2);
+        }
+        // Guard dropped → entry removed.
+        let snap = h.snapshot().await;
+        assert!(snap.active_chapters.is_empty());
+    }
+
+    #[tokio::test]
+    async fn set_chapter_pages_updates_live_count() {
+        let h = StatusHandle::new(1);
+        let chap = sample_chapter(1);
+        let cid = chap.chapter_id;
+        let _guard = h.begin_chapter(chap);
+        h.set_chapter_pages(cid, 3, Some(20));
+        let snap = h.snapshot().await;
+        assert_eq!(snap.active_chapters[0].pages_done, 3);
+        assert_eq!(snap.active_chapters[0].pages_total, Some(20));
+        // Updating an unknown chapter is a no-op, not a panic.
+        h.set_chapter_pages(Uuid::new_v4(), 9, Some(9));
+    }
+
+    #[tokio::test]
+    async fn snapshot_sorts_active_chapters_by_number() {
+        let h = StatusHandle::new(2);
+        let _g1 = h.begin_chapter(sample_chapter(5));
+        let _g2 = h.begin_chapter(sample_chapter(2));
+        let snap = h.snapshot().await;
+        assert_eq!(snap.active_chapters[0].chapter_number, 2);
+        assert_eq!(snap.active_chapters[1].chapter_number, 5);
+    }
+
+    #[tokio::test]
+    async fn cover_guard_sets_then_clears_on_drop() {
+        let h = StatusHandle::new(1);
+        let mid = Uuid::new_v4();
+        {
+            let _g = h.begin_cover(CoverTarget {
+                manga_id: mid,
+                manga_title: "One Piece".into(),
+            });
+            assert_eq!(
+                h.snapshot().await.current_cover.map(|c| c.manga_id),
+                Some(mid)
+            );
+        }
+        assert!(h.snapshot().await.current_cover.is_none());
+    }
+
+    #[tokio::test]
+    async fn cover_guard_clears_on_panic_drop() {
+        // Simulate a download_and_store_cover panic: the guard is on the
+        // stack, the panic unwinds, and the slot must still be cleared.
+        let h = StatusHandle::new(1);
+        let mid = Uuid::new_v4();
+        let h2 = h.clone();
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _g = h2.begin_cover(CoverTarget {
+                manga_id: mid,
+                manga_title: "K-On!".into(),
+            });
+            panic!("simulated cover-download panic");
+        }));
+        assert!(result.is_err());
+        assert!(h.snapshot().await.current_cover.is_none());
+    }
+
+    #[tokio::test]
+    async fn cover_guard_does_not_clobber_a_newer_target() {
+        // Defensive: if a *newer* begin_cover ran before the older
+        // guard's drop fires, the drop must not clear the newer
+        // target. (No code today produces overlapping guards, but the
+        // invariant prevents a future caller from quietly breaking the
+        // live-cover surface.)
+        let h = StatusHandle::new(1);
+        let older = Uuid::new_v4();
+        let newer = Uuid::new_v4();
+        let g_old = h.begin_cover(CoverTarget {
+            manga_id: older,
+            manga_title: "old".into(),
+        });
+        let _g_new = h.begin_cover(CoverTarget {
+            manga_id: newer,
+            manga_title: "new".into(),
+        });
+        drop(g_old);
+        assert_eq!(
+            h.snapshot().await.current_cover.map(|c| c.manga_id),
+            Some(newer)
+        );
+    }
+
+    #[tokio::test]
+    async fn record_pass_captures_stats_and_timestamp() {
+        let h = StatusHandle::new(1);
+        let stats = MetadataStats {
+            discovered: 5,
+            upserted: 3,
+            covers_fetched: 2,
+            mangas_failed: 1,
+        };
+        let at = Utc::now();
+        h.record_pass(&stats, at).await;
+        let snap = h.snapshot().await;
+        assert_eq!(snap.last_pass.discovered, 5);
+        assert_eq!(snap.last_pass.upserted, 3);
+        assert_eq!(snap.last_pass.at, Some(at));
+    }
+
+    #[tokio::test]
+    async fn subscribe_resolves_on_mutation_poke_and_chapter_change() {
+        let h = StatusHandle::new(1);
+        let mut rx = h.subscribe();
+        h.set_phase(Phase::WalkingList).await;
+        rx.changed().await.unwrap();
+        h.poke();
+        rx.changed().await.unwrap();
+        // begin_chapter + guard drop each bump the version.
+        let g = h.begin_chapter(sample_chapter(1));
+        rx.changed().await.unwrap();
+        drop(g);
+        rx.changed().await.unwrap();
+    }
+}