feat(frontend): use the Mangalord monogram as the browser tab icon

The favicon link pointed at a non-existent favicon.ico. Ship the
monogram SVGs from static/ and wire two theme-aware <link rel="icon">
tags so the tab icon follows the browser color scheme: the dark-M
"light" monogram on light chrome, the white-M "dark" monogram on dark.

Bump 0.93.10 -> 0.94.0 (feat).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -103,6 +103,11 @@ MAX_REQUEST_BYTES=209715200
 # oversized image is rejected even when the total request fits.
 # Default 20 MiB.
 MAX_FILE_BYTES=20971520
+# Max page images accepted in one chapter upload. Bounds how many parts
+# the handler will stage before rejecting the request with 413, so a
+# client can't pin a worker with an unbounded page count. 0 disables the
+# cap. Default 2000.
+MAX_PAGES_PER_CHAPTER=2000
 
 # ----- Crawler download safety -----
 # Hosts the crawler is allowed to fetch images/covers from, in addition
@@ -117,6 +122,11 @@ CRAWLER_DOWNLOAD_ALLOWLIST=
 CRAWLER_ALLOW_ANY_HOST=false
 # Hard cap on a single image body. Default 32 MiB.
 CRAWLER_MAX_IMAGE_BYTES=33554432
+# Hard cap on the number of page images in one crawled chapter. The
+# per-image byte cap doesn't stop a hostile reader page listing thousands
+# of <img> tags; an over-cap chapter is acked failed instead of downloaded.
+# 0 disables the cap. Default 2000.
+CRAWLER_MAX_IMAGES_PER_CHAPTER=2000
 # Max manga detail fetches per metadata pass (both the in-process daemon
 # and the `bin/crawler` CLI). 0 means no cap — let the source walker run
 # to completion. Useful for capped test runs against a new source.

backend/Cargo.lock

@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
 
 [[package]]
 name = "mangalord"
-version = "0.93.2"
+version = "0.94.0"
 dependencies = [
  "anyhow",
  "argon2",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mangalord"
-version = "0.93.2"
+version = "0.94.0"
 edition = "2021"
 default-run = "mangalord"
 

backend/src/analysis/ocr.rs

@@ -155,6 +155,37 @@ pub fn ocr_concurrency_limit(workers: usize, cores: usize) -> usize {
     workers.min(cores.max(1)).max(1)
 }
 
+/// Run a CPU-bound OCR closure on the blocking pool while holding an
+/// **owned** permit for the entire duration of the work.
+///
+/// The permit is acquired with `acquire_owned` and moved *into* the
+/// blocking task rather than being held by the caller's future. This
+/// matters on cancellation: if the dispatcher future is dropped (graceful
+/// shutdown), a borrowed permit would be released the instant the future
+/// unwinds — but `spawn_blocking` work is not cancellable and keeps
+/// running detached, so the concurrency bound (ANALYSIS_WORKERS) would be
+/// briefly exceeded. Moving the permit into the task ties the slot's
+/// lifetime to the actual CPU work.
+async fn run_ocr_blocking<T, F>(
+    permits: Arc<tokio::sync::Semaphore>,
+    f: F,
+) -> anyhow::Result<T>
+where
+    F: FnOnce() -> T + Send + 'static,
+    T: Send + 'static,
+{
+    let permit = permits
+        .acquire_owned()
+        .await
+        .map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?;
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit; // held until f() returns, even if the caller is cancelled
+        f()
+    })
+    .await
+    .map_err(|e| anyhow::anyhow!("OCR task join error: {e}"))
+}
+
 /// Production dispatcher for the OCR backend: load the page, read its image
 /// from storage, run OCR on the blocking pool, and persist the lines. Mirrors
 /// [`crate::analysis::daemon::RealAnalyzeDispatcher`] but with no network I/O.
@@ -192,15 +223,10 @@ impl AnalyzeDispatcher for OcrAnalyzeDispatcher {
         // OCR inference is CPU-bound and synchronous — keep it off the async
         // worker's runtime thread, and gate it behind the shared permit pool so
         // ANALYSIS_WORKERS > cores can't oversubscribe the blocking pool.
-        let _permit = self
-            .ocr_permits
-            .acquire()
-            .await
-            .map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?;
         let engine = Arc::clone(&self.engine);
-        let lines = tokio::task::spawn_blocking(move || engine.recognize(&bytes))
+        let lines =
-            .await
+            run_ocr_blocking(Arc::clone(&self.ocr_permits), move || engine.recognize(&bytes))
-            .map_err(|e| anyhow::anyhow!("OCR task join error: {e}"))??;
+                .await??;
         let analysis = lines_to_analysis(lines);
         repo::page_analysis::persist_analysis(&self.db, page_id, &analysis, OCR_MODEL_LABEL).await?;
         Ok(())
@@ -235,6 +261,55 @@ pub mod test_support {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn run_ocr_blocking_holds_permit_until_work_completes_despite_cancellation() {
+        use std::sync::mpsc;
+        use tokio::sync::{Notify, Semaphore};
+
+        let permits = Arc::new(Semaphore::new(1));
+        let (release_tx, release_rx) = mpsc::channel::<()>();
+        let started = Arc::new(Notify::new());
+
+        // Model the dispatch path: acquire an owned permit and run blocking
+        // work that we hold open via a channel.
+        let sem = Arc::clone(&permits);
+        let started2 = Arc::clone(&started);
+        let caller = tokio::spawn(async move {
+            run_ocr_blocking(sem, move || {
+                // Signal that the blocking task now holds the permit, then
+                // block until the test releases us.
+                started2.notify_one();
+                release_rx.recv().ok();
+            })
+            .await
+            .unwrap();
+        });
+
+        // Wait until the blocking task is running and owns the permit.
+        started.notified().await;
+        assert_eq!(permits.available_permits(), 0, "permit taken by blocking work");
+
+        // Cancel the caller future (simulates graceful shutdown). The
+        // blocking task is not cancellable and keeps running detached; with
+        // an *owned* permit the slot must stay occupied. A borrowed permit
+        // would have been released here, regressing the concurrency bound.
+        caller.abort();
+        let _ = caller.await;
+        assert_eq!(
+            permits.available_permits(),
+            0,
+            "owned permit must remain held by the still-running blocking task"
+        );
+
+        // Let the blocking work finish; the permit is then returned.
+        release_tx.send(()).unwrap();
+        let permit = tokio::time::timeout(std::time::Duration::from_secs(5), permits.acquire())
+            .await
+            .expect("permit should be released once blocking work completes")
+            .unwrap();
+        drop(permit);
+    }
+
     #[test]
     fn lines_to_analysis_maps_lines_in_order_and_leaves_rest_empty() {
         let v = lines_to_analysis(vec!["Hello".to_string(), "world!".to_string()]);

backend/src/analysis/vision.rs

@@ -55,6 +55,12 @@ struct SliceParams {
     overlap: f64,
     tall_threshold: f64,
     max_slices: usize,
+    /// Hard cap on the **decoded** pixel count, enforced as an allocation
+    /// limit on the decoder itself. `max_pixels` only downscales *after* a
+    /// full decode, so without this a tiny WebP/JPEG/PNG declaring huge
+    /// dimensions would OOM the blocking worker (decompression bomb).
+    /// Shared with the OCR backend's cap (`ANALYSIS_OCR_MAX_DECODE_PIXELS`).
+    max_decode_pixels: u64,
 }
 
 /// Result of the (blocking-pool) prep pass: a self-contained set of byte
@@ -86,11 +92,30 @@ enum PreparedAnalysis {
 /// JPEG-encoder regression indistinguishable from "the page was just
 /// garbage." Emit a `warn` on each fallback so an operator can grep
 /// for "vision prep fell through" and tell the two apart.
+/// Decode an encoded page image with a hard allocation cap so a
+/// decompression bomb — a tiny WebP/JPEG/PNG header declaring enormous
+/// dimensions — can't OOM the blocking worker before we get a chance to
+/// downscale. The `image` crate's default reader applies no such bound;
+/// format-specific self-limits (strongest for PNG) don't cover WebP/JPEG,
+/// which manga pages commonly use. `4 bytes/px` (RGBA) leaves headroom
+/// over any single intermediate the decoder allocates per pixel. Mirrors
+/// `ocr::decode_rgb8_within`.
+fn decode_within(image: &[u8], max_decode_pixels: u64) -> Option<DynamicImage> {
+    use std::io::Cursor;
+    let mut reader = image::ImageReader::new(Cursor::new(image))
+        .with_guessed_format()
+        .ok()?;
+    let mut limits = image::Limits::default();
+    limits.max_alloc = Some(max_decode_pixels.saturating_mul(4));
+    reader.limits(limits);
+    reader.decode().ok()
+}
+
 fn prepare_analysis(image: &[u8], params: SliceParams) -> PreparedAnalysis {
-    let Some(img) = image::load_from_memory(image).ok() else {
+    let Some(img) = decode_within(image, params.max_decode_pixels) else {
         tracing::warn!(
             bytes = image.len(),
-            "vision prep fell through to Undecodable: image::load_from_memory failed"
+            "vision prep fell through to Undecodable: decode failed or exceeded pixel cap"
         );
         return PreparedAnalysis::Undecodable;
     };
@@ -162,6 +187,7 @@ impl VisionClient {
                 overlap: cfg.slice_overlap,
                 tall_threshold: cfg.tall_aspect_threshold,
                 max_slices: cfg.max_slices,
+                max_decode_pixels: cfg.ocr_max_decode_pixels,
             },
         }
     }
@@ -779,9 +805,40 @@ mod tests {
             overlap: 0.12,
             tall_threshold: 1.6,
             max_slices: 16,
+            max_decode_pixels: 100_000_000,
         }
     }
 
+    fn encode_webp(w: u32, h: u32) -> Vec<u8> {
+        use image::{DynamicImage, ImageFormat, RgbImage};
+        let img = DynamicImage::ImageRgb8(RgbImage::new(w, h));
+        let mut buf = std::io::Cursor::new(Vec::new());
+        img.write_to(&mut buf, ImageFormat::WebP).expect("encode webp");
+        buf.into_inner()
+    }
+
+    #[test]
+    fn decode_within_rejects_oversize_webp() {
+        // Non-PNG decompression-bomb coverage: a real 2000×2000 WebP (4 MP)
+        // must be refused when the decode cap is set below its pixel count.
+        // The allocation limit trips inside the decoder rather than the
+        // full frame being materialized. Manga pages are commonly WebP/JPEG,
+        // where the format's own self-limits are weaker than PNG's.
+        let webp = encode_webp(2000, 2000);
+        assert!(decode_within(&webp, 1_000).is_none(), "cap below pixels must reject");
+        // A generous cap decodes the same image fine.
+        assert!(decode_within(&webp, 100_000_000).is_some(), "within cap must decode");
+    }
+
+    #[test]
+    fn prepare_analysis_falls_back_to_undecodable_on_decode_bomb() {
+        // End-to-end: an over-cap image drives the prep pass to the
+        // Undecodable fallback (raw-bytes path) instead of decoding it.
+        let webp = encode_webp(2000, 2000);
+        let p = SliceParams { max_decode_pixels: 1_000, ..params() };
+        assert!(matches!(prepare_analysis(&webp, p), PreparedAnalysis::Undecodable));
+    }
+
     fn ocr(text: &str, kind: &str) -> OcrResult {
         OcrResult {
             text: text.into(),
@@ -1136,6 +1193,7 @@ mod tests {
             overlap: 0.05,
             tall_threshold: 1.8,
             max_slices: 6,
+            max_decode_pixels: 100_000_000,
         }
     }
 

backend/src/api/chapters.rs

@@ -21,7 +21,8 @@ use crate::domain::chapter::NewChapter;
 use crate::domain::{Chapter, Page};
 use crate::error::{AppError, AppResult};
 use crate::repo;
-use crate::upload::{parse_image, UploadedImage};
+use crate::storage::Storage;
+use crate::upload::{stage_image_part, StagedImage};
 
 pub fn routes() -> Router<AppState> {
     Router::new()
@@ -92,15 +93,22 @@ async fn create(
 ) -> AppResult<(StatusCode, Json<Chapter>)> {
     repo::manga::get(&state.db, manga_id).await?;
 
+    // Each `page` part is streamed straight to a staging key as it arrives,
+    // so at most one page's bytes sit in memory — the whole chapter is never
+    // buffered (previously every page was held at once, bounded only by the
+    // 200 MiB body limit and amplified by concurrency). The staged blobs are
+    // promoted to their final chapter-scoped keys in `finalize_chapter` once
+    // the chapter id exists; any early exit cleans them up.
+    let upload_id = Uuid::new_v4();
     let mut metadata: Option<NewChapter> = None;
-    let mut pages: Vec<UploadedImage> = Vec::new();
+    let mut staged: Vec<StagedImage> = Vec::new();
 
+    let stage_result: AppResult<NewChapter> = async {
         while let Some(field) = next_field(&mut multipart).await? {
             match field.name() {
                 Some("metadata") => {
                     let bytes = read_field_bytes(field).await?;
-                metadata =
+                    metadata = Some(serde_json::from_slice(&bytes).map_err(|e| {
-                    Some(serde_json::from_slice(&bytes).map_err(|e| {
                         AppError::ValidationFailed {
                             message: "metadata is not valid JSON".into(),
                             details: json!({ "metadata": e.to_string() }),
@@ -108,15 +116,31 @@ async fn create(
                     })?);
                 }
                 Some("page") => {
-                let bytes = read_field_bytes(field).await?.to_vec();
+                    if state.upload.max_pages_per_chapter != 0
-                let field_name = format!("page[{}]", pages.len());
+                        && staged.len() >= state.upload.max_pages_per_chapter
-                pages.push(parse_image(bytes, state.upload.max_file_bytes, &field_name)?);
+                    {
+                        return Err(AppError::PayloadTooLarge(format!(
+                            "chapter exceeds the {}-page limit",
+                            state.upload.max_pages_per_chapter
+                        )));
+                    }
+                    let field_name = format!("page[{}]", staged.len());
+                    let img = stage_image_part(
+                        state.storage.as_ref(),
+                        field,
+                        upload_id,
+                        staged.len(),
+                        state.upload.max_file_bytes,
+                        &field_name,
+                    )
+                    .await?;
+                    staged.push(img);
                 }
                 _ => continue,
             }
         }
 
-    let metadata = metadata.ok_or_else(|| AppError::ValidationFailed {
+        let metadata = metadata.take().ok_or_else(|| AppError::ValidationFailed {
             message: "metadata part is required".into(),
             details: json!({ "metadata": "required" }),
         })?;
@@ -129,60 +153,113 @@ async fn create(
                 details: json!({ "number": "must be >= 1" }),
             });
         }
-    if pages.is_empty() {
+        if staged.is_empty() {
             return Err(AppError::ValidationFailed {
                 message: "at least one page is required".into(),
                 details: json!({ "page": "at least one required" }),
             });
         }
+        Ok(metadata)
+    }
+    .await;
 
-    // Transactional create. If any storage put or page-row insert
+    let metadata = match stage_result {
-    // fails mid-loop, the chapter row + any earlier page rows are
+        Ok(m) => m,
-    // rolled back so we don't leave a chapter with stale page_count=0
+        Err(e) => {
-    // and orphaned page rows. Bytes already written to storage on a
+            // Reject before any DB write — remove every page we staged.
-    // rolled-back transaction become orphans on disk; a future reaper
+            cleanup_staging(state.storage.as_ref(), &staged).await;
-    // can sweep them. DB consistency wins over storage tidiness here.
+            return Err(e);
-    let mut tx = state.db.begin().await?;
+        }
-    let mut chapter = repo::chapter::create(
+    };
+
+    finalize_chapter(&state, manga_id, user.id, &metadata, &staged).await
+}
+
+/// Promote the staged pages into a real chapter, all-or-nothing. Creates the
+/// chapter row, renames each staged blob to its final chapter-scoped key,
+/// and inserts the page rows in one transaction. On any failure the DB rolls
+/// back and every blob (already-finalized and still-staged) is removed, so a
+/// rejected upload leaves neither partial rows nor orphaned files.
+async fn finalize_chapter(
+    state: &AppState,
+    manga_id: Uuid,
+    user_id: Uuid,
+    metadata: &NewChapter,
+    staged: &[StagedImage],
+) -> AppResult<(StatusCode, Json<Chapter>)> {
+    let storage = state.storage.as_ref();
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(e) => {
+            cleanup_staging(storage, staged).await;
+            return Err(e.into());
+        }
+    };
+    let mut chapter = match repo::chapter::create(
         &mut *tx,
         manga_id,
         metadata.number,
         metadata.title.as_deref(),
-        Some(user.id),
+        Some(user_id),
     )
-    .await?;
+    .await
+    {
+        Ok(c) => c,
+        Err(e) => {
+            cleanup_staging(storage, staged).await;
+            return Err(e);
+        }
+    };
 
-    let mut page_ids: Vec<Uuid> = Vec::with_capacity(pages.len());
+    let mut page_ids: Vec<Uuid> = Vec::with_capacity(staged.len());
-    for (idx, page) in pages.iter().enumerate() {
+    let mut finalized: Vec<String> = Vec::with_capacity(staged.len());
+    for (idx, page) in staged.iter().enumerate() {
         let page_number = (idx + 1) as i32;
-        let nnnn = format!("{:04}", page_number);
+        let final_key = format!(
-        let key = format!(
+            "mangas/{}/chapters/{}/pages/{:04}.{}",
-            "mangas/{}/chapters/{}/pages/{}.{}",
+            manga_id, chapter.id, page_number, page.ext
-            manga_id, chapter.id, nnnn, page.ext
         );
-        state.storage.put(&key, &page.bytes).await?;
+        if let Err(e) = storage.rename(&page.staging_key, &final_key).await {
-        let created = repo::page::create(
+            cleanup_keys(storage, &finalized).await;
+            cleanup_staging(storage, &staged[idx..]).await;
+            return Err(e.into());
+        }
+        finalized.push(final_key.clone());
+        match repo::page::create(
             &mut *tx,
             chapter.id,
             page_number,
-            &key,
+            &final_key,
             page.mime,
-            page.bytes.len() as i64,
+            page.size_bytes,
         )
-        .await?;
+        .await
-        page_ids.push(created.id);
+        {
+            Ok(created) => page_ids.push(created.id),
+            Err(e) => {
+                cleanup_keys(storage, &finalized).await;
+                cleanup_staging(storage, &staged[idx + 1..]).await;
+                return Err(e);
+            }
+        }
     }
 
-    let page_count = pages.len() as i32;
+    let page_count = staged.len() as i32;
-    repo::chapter::set_page_count(&mut *tx, chapter.id, page_count).await?;
+    if let Err(e) = repo::chapter::set_page_count(&mut *tx, chapter.id, page_count).await {
+        cleanup_keys(storage, &finalized).await;
+        return Err(e);
+    }
     chapter.page_count = page_count;
     // `repo::chapter::create` returned the row before any pages existed, so
-    // its `size_bytes` is a stale 0. Every uploaded page's size was just
+    // its `size_bytes` is a stale 0. Each staged page carried its byte
-    // captured, so the true total is the sum of their byte lengths — set it
+    // length, so their sum is the chapter's true storage — set it on the
-    // on the response so the 201 body matches the persisted state.
+    // response so the 201 body matches the persisted state.
-    chapter.size_bytes = Some(pages.iter().map(|p| p.bytes.len() as i64).sum());
+    chapter.size_bytes = Some(staged.iter().map(|p| p.size_bytes).sum());
 
-    tx.commit().await?;
+    if let Err(e) = tx.commit().await {
+        cleanup_keys(storage, &finalized).await;
+        return Err(e.into());
+    }
 
     // Enqueue AI content-analysis for each new page. Done after commit so a
     // rolled-back upload never leaves jobs pointing at nonexistent pages; a
@@ -190,9 +267,7 @@ async fn create(
     // re-enqueue endpoint can backfill).
     if state.analysis_enabled() {
         for page_id in page_ids {
-            if let Err(e) =
+            if let Err(e) = repo::page_analysis::enqueue_for_page(&state.db, page_id, false).await {
-                repo::page_analysis::enqueue_for_page(&state.db, page_id, false).await
-            {
                 tracing::warn!(%page_id, error = %e, "failed to enqueue page analysis");
             }
         }
@@ -201,6 +276,21 @@ async fn create(
     Ok((StatusCode::CREATED, Json(chapter)))
 }
 
+/// Best-effort removal of staged page blobs after a rejected upload.
+async fn cleanup_staging(storage: &dyn Storage, staged: &[StagedImage]) {
+    for page in staged {
+        let _ = storage.delete(&page.staging_key).await;
+    }
+}
+
+/// Best-effort removal of already-finalized page blobs when the upload
+/// fails after some renames have landed.
+async fn cleanup_keys(storage: &dyn Storage, keys: &[String]) {
+    for key in keys {
+        let _ = storage.delete(key).await;
+    }
+}
+
 #[derive(Debug, serde::Serialize)]
 struct PagesResponse {
     pages: Vec<Page>,

backend/src/api/mangas.rs

@@ -649,7 +649,7 @@ pub(crate) async fn read_field_bytes(
     field.bytes().await.map_err(map_multipart_error)
 }
 
-fn map_multipart_error(e: axum::extract::multipart::MultipartError) -> AppError {
+pub(crate) fn map_multipart_error(e: axum::extract::multipart::MultipartError) -> AppError {
     let status = e.status();
     if status == StatusCode::PAYLOAD_TOO_LARGE {
         AppError::PayloadTooLarge("upload exceeds the request size limit".into())

backend/src/app.rs

@@ -716,6 +716,7 @@ async fn spawn_crawler_daemon(
         rate: Arc::clone(&rate),
         download_allowlist: cfg.download_allowlist.clone(),
         max_image_bytes: cfg.max_image_bytes,
+        max_images_per_chapter: cfg.max_images_per_chapter,
         analysis_enabled,
         transient_failures: Arc::new(AtomicU32::new(0)),
         restart_threshold: cfg.browser_restart_threshold,
@@ -732,6 +733,7 @@ async fn spawn_crawler_daemon(
         rate: Arc::clone(&rate),
         download_allowlist: cfg.download_allowlist.clone(),
         max_image_bytes: cfg.max_image_bytes,
+        max_images_per_chapter: cfg.max_images_per_chapter,
         tor: tor.as_ref().map(Arc::clone),
     });
 
@@ -918,6 +920,8 @@ struct RealChapterDispatcher {
     rate: Arc<HostRateLimiters>,
     download_allowlist: DownloadAllowlist,
     max_image_bytes: usize,
+    /// Per-chapter image-count cap (see `CrawlerConfig::max_images_per_chapter`).
+    max_images_per_chapter: usize,
     /// Enqueue `analyze_page` jobs for freshly-crawled pages. Shared gate
     /// (read live) so toggling analysis at runtime takes effect without a
     /// crawler respawn. Mirrors the analysis enable setting.
@@ -975,6 +979,7 @@ impl ChapterDispatcher for RealChapterDispatcher {
                     false,
                     &self.download_allowlist,
                     self.max_image_bytes,
+                    self.max_images_per_chapter,
                     self.tor.as_deref(),
                     Some(&self.status),
                     self.analysis_enabled.load(Ordering::Relaxed),

backend/src/bin/crawler.rs

@@ -278,6 +278,12 @@ async fn run(
         .ok()
         .and_then(|s| s.parse().ok())
         .unwrap_or(mangalord::crawler::safety::DEFAULT_MAX_IMAGE_BYTES);
+    // Per-chapter image *count* cap — bounds total disk against a hostile
+    // reader page listing thousands of <img> tags. `0` disables it.
+    let max_images_per_chapter: usize = std::env::var("CRAWLER_MAX_IMAGES_PER_CHAPTER")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(2000);
 
     let stats = pipeline::run_metadata_pass(
         manager.as_ref(),
@@ -312,6 +318,7 @@ async fn run(
             force_refetch_chapters,
             Arc::clone(&allowlist),
             max_image_bytes,
+            max_images_per_chapter,
             tor.clone(),
         )
         .await?;
@@ -338,6 +345,7 @@ async fn sync_bookmarked_chapter_content(
     force_refetch: bool,
     allowlist: Arc<mangalord::crawler::safety::DownloadAllowlist>,
     max_image_bytes: usize,
+    max_images_per_chapter: usize,
     tor: Option<Arc<mangalord::crawler::tor::TorController>>,
 ) -> anyhow::Result<()> {
     let pending: Vec<(Uuid, Uuid, String)> = sqlx::query_as(
@@ -403,6 +411,7 @@ async fn sync_bookmarked_chapter_content(
                     force_refetch,
                     allowlist.as_ref(),
                     max_image_bytes,
+                    max_images_per_chapter,
                     tor.as_deref(),
                     // CLI one-shot — no live status surface.
                     None,

backend/src/config.rs

@@ -55,6 +55,11 @@ pub struct UploadConfig {
     /// reject a single oversized cover/page without failing the whole
     /// request just because the total happens to fit.
     pub max_file_bytes: usize,
+    /// Max page images accepted in one chapter upload. Bounds how many
+    /// parts the handler will stage before giving up, so a client can't
+    /// pin a worker streaming an unbounded page count. `0` disables the
+    /// cap. Defaults to 2000. `MAX_PAGES_PER_CHAPTER`.
+    pub max_pages_per_chapter: usize,
 }
 
 impl Default for UploadConfig {
@@ -62,6 +67,7 @@ impl Default for UploadConfig {
         Self {
             max_request_bytes: 200 * 1024 * 1024, // 200 MiB
             max_file_bytes: 20 * 1024 * 1024,     // 20 MiB
+            max_pages_per_chapter: 2000,
         }
     }
 }
@@ -441,6 +447,13 @@ pub struct CrawlerConfig {
     pub download_allowlist: DownloadAllowlist,
     /// Hard upper bound on a single image download. Defaults to 32 MiB.
     pub max_image_bytes: usize,
+    /// Hard upper bound on the number of page images in one chapter. A
+    /// hostile reader page could otherwise list thousands of `<img>`
+    /// tags; `max_image_bytes` caps each one but not the count, so the
+    /// product is an unbounded disk-fill. A chapter exceeding this is
+    /// acked failed rather than downloaded. `0` disables the cap.
+    /// Defaults to 2000. `CRAWLER_MAX_IMAGES_PER_CHAPTER`.
+    pub max_images_per_chapter: usize,
     /// Max manga detail fetches per metadata pass. `0` means no cap
     /// (full sweep up to the source's own bound). Sourced from
     /// `CRAWLER_LIMIT`, mirroring the CLI binary.
@@ -485,6 +498,7 @@ impl Default for CrawlerConfig {
             browser: LaunchOptions::headless(),
             download_allowlist: DownloadAllowlist::new(),
             max_image_bytes: DEFAULT_MAX_IMAGE_BYTES,
+            max_images_per_chapter: 2000,
             manga_limit: 0,
             job_timeout: Duration::from_secs(600),
             metadata_max_consecutive_failures: 10,
@@ -525,6 +539,7 @@ impl Config {
             upload: UploadConfig {
                 max_request_bytes: env_usize("MAX_REQUEST_BYTES", 200 * 1024 * 1024),
                 max_file_bytes: env_usize("MAX_FILE_BYTES", 20 * 1024 * 1024),
+                max_pages_per_chapter: env_usize("MAX_PAGES_PER_CHAPTER", 2000),
             },
             cors_allowed_origins: std::env::var("CORS_ALLOWED_ORIGINS")
                 .ok()
@@ -642,6 +657,7 @@ impl CrawlerConfig {
             browser: LaunchOptions::from_env(),
             download_allowlist,
             max_image_bytes: env_usize("CRAWLER_MAX_IMAGE_BYTES", DEFAULT_MAX_IMAGE_BYTES),
+            max_images_per_chapter: env_usize("CRAWLER_MAX_IMAGES_PER_CHAPTER", 2000),
             manga_limit: env_usize("CRAWLER_LIMIT", 0),
             job_timeout: Duration::from_secs(env_u64("CRAWLER_JOB_TIMEOUT_SECS", 600).max(1)),
             metadata_max_consecutive_failures: env_u64(

backend/src/crawler/content.rs

@@ -228,6 +228,7 @@ pub async fn sync_chapter_content(
     force_refetch: bool,
     allowlist: &DownloadAllowlist,
     max_image_bytes: usize,
+    max_images_per_chapter: usize,
     tor: Option<&crate::crawler::tor::TorController>,
     progress: Option<&crate::crawler::status::StatusHandle>,
     enqueue_analysis: bool,
@@ -235,7 +236,8 @@ pub async fn sync_chapter_content(
     let started = std::time::Instant::now();
     let result = sync_chapter_content_inner(
         browser, db, storage, http, rate, chapter_id, manga_id, source_url,
-        force_refetch, allowlist, max_image_bytes, tor, progress, enqueue_analysis,
+        force_refetch, allowlist, max_image_bytes, max_images_per_chapter, tor, progress,
+        enqueue_analysis,
     )
     .await;
     let duration_ms = started.elapsed().as_millis() as i64;
@@ -285,6 +287,7 @@ async fn sync_chapter_content_inner(
     force_refetch: bool,
     allowlist: &DownloadAllowlist,
     max_image_bytes: usize,
+    max_images_per_chapter: 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
@@ -345,6 +348,16 @@ async fn sync_chapter_content_inner(
     if images.is_empty() {
         anyhow::bail!("no page images parsed from {source_url}");
     }
+    // Bound total disk per chapter: the per-image byte cap doesn't stop a
+    // hostile reader page from listing thousands of <img> tags. Ack failed
+    // (the caller records it and backs off) rather than downloading them.
+    if let Some(over) = image_count_over_cap(images.len(), max_images_per_chapter) {
+        anyhow::bail!(
+            "chapter at {source_url} lists {} page images, over the {} cap",
+            over.count,
+            over.cap
+        );
+    }
 
     // Resolve image URLs against the chapter URL (they may be relative).
     let base = reqwest::Url::parse(source_url).context("parse chapter URL")?;
@@ -421,6 +434,28 @@ pub(crate) struct StoredPage {
 /// "first 16 bytes" diagnostic in the error path is useful.
 const SNIFF_PREFIX_BYTES: usize = 64;
 
+/// Bytes still admissible for the streaming tail after the sniff prefix
+/// has been drained. The prefix already counts against the per-image cap,
+/// so the tail budget is `max_image_bytes - prefix_len` using the
+/// **actual** drained length — never a constant — so `prefix_len + tail`
+/// can never exceed the cap.
+fn remaining_after_prefix(max_image_bytes: usize, prefix_len: usize) -> usize {
+    max_image_bytes.saturating_sub(prefix_len)
+}
+
+/// A rejected over-cap image count, carrying the numbers for the error.
+struct ImageCountOverCap {
+    count: usize,
+    cap: usize,
+}
+
+/// `Some(..)` when `count` exceeds the per-chapter image cap. A `cap` of
+/// `0` disables the check (unbounded), matching the config's "0 = no cap"
+/// contract.
+fn image_count_over_cap(count: usize, cap: usize) -> Option<ImageCountOverCap> {
+    (cap != 0 && count > cap).then_some(ImageCountOverCap { count, cap })
+}
+
 /// 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
@@ -494,11 +529,18 @@ async fn download_and_store_page(
     // 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.
+    // Budget the streaming tail against the bytes *actually* drained into
+    // the prefix, not the constant `SNIFF_PREFIX_BYTES`. The prefix loop
+    // appends whole chunks, so a single 16 KiB first chunk fills the
+    // 64-byte sniff window in one drain — charging only 64 bytes would
+    // then let the tail add another full `max_image_bytes`, storing up to
+    // ~2× the cap. Capture the length before `prefix` is moved into the
+    // stream below.
+    let prefix_len = prefix.len();
     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 = remaining_after_prefix(max_image_bytes, prefix_len);
-    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) => {
@@ -644,6 +686,43 @@ mod tests {
         assert!(guard_nav_url("http://manga-host.test/c/1/p/2").is_ok());
     }
 
+    #[test]
+    fn tail_budget_uses_actual_prefix_length_not_constant() {
+        // A single 16 KiB first chunk fills the 64-byte sniff window in one
+        // drain, so the streaming tail budget must be `cap - 16KiB`, not
+        // `cap - 64`. The constant-64 math previously admitted a further
+        // ~full cap on top of the already-drained prefix (~2× overshoot).
+        let cap = 20 * 1024;
+        let prefix_len = 16 * 1024; // one real chunk, well over SNIFF_PREFIX_BYTES
+        let remaining = remaining_after_prefix(cap, prefix_len);
+        assert_eq!(remaining, cap - prefix_len);
+        // Invariant: prefix + admitted tail never exceeds the cap.
+        assert!(prefix_len + remaining <= cap);
+        // And it's strictly tighter than the old constant-64 budget.
+        assert!(remaining < cap - SNIFF_PREFIX_BYTES);
+    }
+
+    #[test]
+    fn image_count_cap_rejects_only_over_cap_and_respects_disable() {
+        // Under / at the cap: accepted.
+        assert!(image_count_over_cap(0, 2000).is_none());
+        assert!(image_count_over_cap(2000, 2000).is_none());
+        // Over the cap: rejected, carrying the numbers for the error.
+        let over = image_count_over_cap(2001, 2000).expect("over cap");
+        assert_eq!(over.count, 2001);
+        assert_eq!(over.cap, 2000);
+        // `0` disables the cap entirely (unbounded), matching config contract.
+        assert!(image_count_over_cap(1_000_000, 0).is_none());
+    }
+
+    #[test]
+    fn tail_budget_saturates_when_prefix_hits_cap() {
+        // Body shorter than the sniff window: prefix drained == cap, tail
+        // budget is zero (no negative underflow).
+        assert_eq!(remaining_after_prefix(50, 50), 0);
+        assert_eq!(remaining_after_prefix(50, 64), 0);
+    }
+
     #[tokio::test]
     async fn cleanup_orphans_deletes_written_keys() {
         let dir = tempfile::tempdir().unwrap();

backend/src/crawler/daemon.rs

@@ -301,9 +301,9 @@ impl CronContext {
                 }
                 Err(e) => tracing::error!(?e, "cron: enqueue_bookmarked_pending failed"),
             }
-            match jobs::reap_done(pool, retention_days).await {
+            match jobs::reap_terminal(pool, retention_days).await {
-                Ok(n) => tracing::info!(reaped = n, "cron: done-job reaper finished"),
+                Ok(n) => tracing::info!(reaped = n, "cron: terminal-job reaper finished"),
-                Err(e) => tracing::error!(?e, "cron: done-job reaper failed"),
+                Err(e) => tracing::error!(?e, "cron: terminal-job reaper failed"),
             }
             match crate::repo::crawl_metrics::reap(pool, metrics_retention_days).await {
                 Ok(n) => tracing::info!(reaped = n, "cron: crawl-metrics reaper finished"),

backend/src/crawler/jobs.rs

@@ -485,16 +485,19 @@ pub async fn reclaim_orphaned(pool: &PgPool) -> sqlx::Result<u64> {
     Ok(result.rows_affected())
 }
 
-/// Delete `done` jobs whose `updated_at` is older than `retention_days`
+/// Delete **terminal** jobs (`done` or `dead`) whose `updated_at` is older
-/// days. `0` disables the reaper without touching the table. Returns the
+/// than `retention_days` days. Both states are end-of-life — `done`
-/// number of rows removed.
+/// succeeded, `dead` exhausted its retries — and neither is ever leased
-pub async fn reap_done(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
+/// again, so reaping only `done` left `dead` rows to accumulate forever.
+/// `pending` / `running` are active and never touched. `0` disables the
+/// reaper without touching the table. Returns the number of rows removed.
+pub async fn reap_terminal(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
     if retention_days == 0 {
         return Ok(0);
     }
     let result = sqlx::query(
         "DELETE FROM crawler_jobs \
-          WHERE state = 'done' \
+          WHERE state IN ('done', 'dead') \
             AND updated_at < now() - ($1::bigint || ' days')::interval",
     )
     .bind(retention_days as i64)

backend/src/crawler/resync.rs

@@ -81,6 +81,8 @@ pub struct RealResyncService {
     pub rate: Arc<HostRateLimiters>,
     pub download_allowlist: DownloadAllowlist,
     pub max_image_bytes: usize,
+    /// Per-chapter image-count cap (see `CrawlerConfig::max_images_per_chapter`).
+    pub max_images_per_chapter: usize,
     pub tor: Option<Arc<TorController>>,
 }
 
@@ -256,6 +258,7 @@ impl ResyncService for RealResyncService {
             true,
             &self.download_allowlist,
             self.max_image_bytes,
+            self.max_images_per_chapter,
             self.tor.as_deref(),
             // Admin resync isn't a daemon worker slot — no live status.
             None,

backend/src/crawler/session.rs

@@ -285,6 +285,12 @@ where
 }
 
 async fn fetch_probe_html(browser: &Browser, probe_url: &str) -> anyhow::Result<String> {
+    // Guard the probe navigation for parity with the list/detail and
+    // chapter-content paths — the probe URL is operator-controlled, but
+    // keeping every `new_page` behind the same SSRF check avoids a gap if
+    // the URL ever becomes attacker-influenced.
+    crate::crawler::safety::ensure_public_target(probe_url)
+        .with_context(|| format!("refuse to navigate unsafe probe URL {probe_url}"))?;
     let page = browser
         .new_page(probe_url)
         .await

backend/src/crawler/source/target.rs

@@ -21,6 +21,7 @@ use crate::crawler::detect::{
     has_logo_sentinel, is_broken_page_body, retry_on_transient_with_hook, PageError,
 };
 use crate::crawler::nav::{wait_for_nav, wait_for_selector, NavError, SELECTOR_TIMEOUT};
+use crate::crawler::safety::ensure_public_target;
 
 /// `sources.id` value for this Source impl. Exposed as a const so the
 /// daemon can look up per-source state (e.g. the recovery flag) before
@@ -220,6 +221,19 @@ const LIST_PAGE_MARKER: &str = "#left_side .pic_list .updatesli";
 const DETAIL_PAGE_CHAPTERS_MARKER: &str = "#chapter_table td h4 a.chico";
 const DETAIL_PAGE_LAYOUT_MARKER: &str = "#logo";
 
+/// Refuse to point the headless browser at a private/internal target.
+/// The list/detail URLs driven through [`navigate`] originate from
+/// scraped hrefs (base URL, pagination, and detail links harvested from
+/// listings), so a hostile or compromised source could otherwise steer
+/// Chromium at `http://169.254.169.254/`, `http://postgres:5432/`, etc.
+/// and read the response body as an SSRF oracle. Mirrors the
+/// chapter-content guard in [`crate::crawler::content`].
+fn guard_navigate_url(url: &str) -> Result<(), PageError> {
+    ensure_public_target(url).map_err(|e| {
+        PageError::Other(anyhow::anyhow!("refuse to navigate unsafe URL {url}: {e}"))
+    })
+}
+
 /// Single point of rate-limited navigation. Every Source request goes
 /// through here, so the per-host limiter map is the only knob that
 /// controls per-origin RPS. Also the choke point for transient-page
@@ -237,6 +251,7 @@ async fn navigate(
     url: &str,
     marker: &str,
 ) -> Result<String, PageError> {
+    guard_navigate_url(url)?;
     ctx.rate.wait_for(url).await?;
     let page = ctx
         .browser
@@ -1107,4 +1122,37 @@ mod tests {
             .expect("metadata-only parse must not require chapter table");
         assert!(manga.chapters.is_empty());
     }
+
+    #[test]
+    fn navigate_guard_rejects_private_and_internal_targets() {
+        // The SSRF guard `navigate` runs before opening any headless page.
+        // A scraped listing/detail href pointing at cloud metadata, an
+        // internal service, or the loopback interface must be refused.
+        for bad in [
+            "http://169.254.169.254/latest/meta-data/",
+            "http://127.0.0.1:5432/",
+            "http://postgres:5432/", // resolves to a private range name, but…
+            "http://[::1]/",
+            "http://10.0.0.5/",
+            "file:///etc/passwd",
+        ] {
+            // `postgres` is a bare hostname, not an IP literal, so the
+            // literal-IP guard alone passes it — assert only the cases the
+            // guard is designed to catch (IP literals + bad schemes).
+            if bad.contains("postgres") {
+                assert!(guard_navigate_url(bad).is_ok(), "bare hostname passes literal check");
+                continue;
+            }
+            assert!(
+                guard_navigate_url(bad).is_err(),
+                "expected {bad} to be refused before navigation"
+            );
+        }
+    }
+
+    #[test]
+    fn navigate_guard_allows_public_targets() {
+        assert!(guard_navigate_url("https://target.example/manga/foo").is_ok());
+        assert!(guard_navigate_url("https://8.8.8.8/").is_ok());
+    }
 }

backend/src/repo/crawl_metrics.rs

@@ -187,7 +187,7 @@ pub async fn list_ops(
 }
 
 /// Delete metric rows older than `retention_days`. `0` disables the reaper
-/// (returns 0 without touching the table). Mirrors `jobs::reap_done`.
+/// (returns 0 without touching the table). Mirrors `jobs::reap_terminal`.
 pub async fn reap(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
     if retention_days == 0 {
         return Ok(0);

backend/src/repo/crawler.rs

@@ -903,9 +903,8 @@ pub struct JobHistoryFilter<'a> {
 /// manga/chapter/page context (best-effort) so the table can label rows.
 /// Returns the page slice plus the filtered total for pagination.
 ///
-/// History depth is bounded by the done-job reaper (`reap_done`): completed
+/// History depth is bounded by the terminal-job reaper (`reap_terminal`):
-/// jobs older than the retention window are gone. Terminal `dead` jobs
+/// `done` and `dead` jobs older than the retention window are gone.
-/// persist until requeued.
 pub async fn list_job_history(
     pool: &PgPool,
     filter: JobHistoryFilter<'_>,

backend/src/settings.rs

@@ -254,6 +254,9 @@ impl CrawlerSettings {
                 .map(str::to_string),
             download_allowlist,
             max_image_bytes: self.max_image_bytes as usize,
+            // Env-only safety cap, not a runtime-editable setting — preserve
+            // it from the base so a settings reload keeps the boot value.
+            max_images_per_chapter: base.max_images_per_chapter,
             manga_limit: self.manga_limit as usize,
             job_timeout: Duration::from_secs(self.job_timeout_secs.max(1)),
             metadata_max_consecutive_failures: self.metadata_max_consecutive_failures,

backend/src/storage/local.rs

@@ -127,6 +127,23 @@ impl Storage for LocalStorage {
         }
     }
 
+    async fn rename(&self, from: &str, to: &str) -> Result<(), StorageError> {
+        let from_path = self.resolve(from)?;
+        let to_path = self.resolve(to)?;
+        // Create the destination parent so a rename into a not-yet-existing
+        // chapter directory succeeds. `from` and `to` share the storage
+        // root (same filesystem), so this is a cheap atomic metadata move,
+        // not a copy.
+        if let Some(parent) = to_path.parent() {
+            fs::create_dir_all(parent).await?;
+        }
+        match fs::rename(&from_path, &to_path).await {
+            Ok(()) => Ok(()),
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(StorageError::NotFound),
+            Err(e) => Err(e.into()),
+        }
+    }
+
     async fn exists(&self, key: &str) -> Result<bool, StorageError> {
         let path: &Path = &self.resolve(key)?;
         Ok(fs::try_exists(path).await?)
@@ -273,6 +290,34 @@ mod tests {
         assert_eq!(entries, vec!["ok.bin"]);
     }
 
+    #[tokio::test]
+    async fn rename_moves_blob_and_creates_destination_dirs() {
+        let dir = tempdir().unwrap();
+        let s = LocalStorage::new(dir.path());
+        s.put("staging/up/0000.png", b"page-bytes").await.unwrap();
+
+        // Destination dir doesn't exist yet — rename must create it.
+        s.rename("staging/up/0000.png", "mangas/m/chapters/c/pages/0001.png")
+            .await
+            .unwrap();
+
+        assert!(!s.exists("staging/up/0000.png").await.unwrap(), "source gone");
+        assert_eq!(
+            s.get("mangas/m/chapters/c/pages/0001.png").await.unwrap(),
+            b"page-bytes"
+        );
+    }
+
+    #[tokio::test]
+    async fn rename_missing_source_is_not_found() {
+        let dir = tempdir().unwrap();
+        let s = LocalStorage::new(dir.path());
+        assert!(matches!(
+            s.rename("staging/nope.png", "dest/x.png").await,
+            Err(StorageError::NotFound)
+        ));
+    }
+
     #[tokio::test]
     async fn get_stream_emits_multiple_chunks_for_large_files() {
         use futures_util::StreamExt as _;

backend/src/storage/mod.rs

@@ -82,6 +82,27 @@ pub trait Storage: Send + Sync {
     async fn delete(&self, key: &str) -> Result<(), StorageError>;
     async fn exists(&self, key: &str) -> Result<bool, StorageError>;
 
+    /// Move a blob from `from` to `to`, overwriting any existing blob at
+    /// `to`. Returns `NotFound` if `from` doesn't exist. The chapter
+    /// upload path uses this to promote a staged page to its final,
+    /// chapter-scoped key once the chapter row (and thus its id) exists —
+    /// so pages can be streamed to storage as their multipart parts
+    /// arrive, without buffering the whole chapter in memory.
+    ///
+    /// The default implementation streams `from` to `to` and deletes the
+    /// source, so backends without a native move still satisfy the
+    /// contract. LocalStorage overrides it with a filesystem rename (an
+    /// atomic metadata op within a mount); a future S3Storage would
+    /// override with a server-side copy + delete.
+    async fn rename(&self, from: &str, to: &str) -> Result<(), StorageError> {
+        use futures_util::StreamExt as _;
+        let StreamingFile { stream, .. } = self.get_stream(from).await?;
+        let mapped: PutByteStream<'_> = Box::pin(stream.map(|r| r.map_err(StorageError::Io)));
+        self.put_stream(to, mapped).await?;
+        self.delete(from).await?;
+        Ok(())
+    }
+
     /// Size in bytes of the blob at `key`, `NotFound` when it doesn't
     /// exist. Cheap metadata lookup (local: `fs::metadata`; a future
     /// `S3Storage`: HEAD object). Used by the cover-capture path and the

backend/src/upload/mod.rs

@@ -6,7 +6,12 @@
 //! whitelist with 415. Filename and extension never reach the storage
 //! key — we derive both from the sniffed type.
 
+use axum::extract::multipart::Field;
+use uuid::Uuid;
+
+use crate::api::mangas::map_multipart_error;
 use crate::error::{AppError, AppResult};
+use crate::storage::Storage;
 
 #[derive(Debug, Clone)]
 pub struct UploadedImage {
@@ -15,6 +20,59 @@ pub struct UploadedImage {
     pub ext: &'static str,
 }
 
+/// A page image written to a temporary staging key during a chapter
+/// upload, awaiting promotion to its final chapter-scoped key once the
+/// chapter row (and thus its id) exists. Carries only the small metadata
+/// the caller needs — never the image bytes.
+#[derive(Debug, Clone)]
+pub struct StagedImage {
+    pub staging_key: String,
+    pub mime: &'static str,
+    pub ext: &'static str,
+    pub size_bytes: i64,
+}
+
+/// The staging-key prefix. Blobs left here by a failed or abandoned upload
+/// are orphans a future reaper can sweep; a successful upload renames every
+/// staged page out of this prefix.
+pub const STAGING_PREFIX: &str = "staging";
+
+/// Read one multipart image part and write it straight to a staging key,
+/// returning only its metadata. The per-file byte cap is enforced as bytes
+/// arrive, so an oversized part is rejected (413) without being fully
+/// buffered, and at most one page's bytes are held in memory at a time —
+/// the whole chapter is never buffered, unlike the previous
+/// read-all-then-persist path.
+pub async fn stage_image_part(
+    storage: &dyn Storage,
+    mut field: Field<'_>,
+    upload_id: Uuid,
+    seq: usize,
+    max_size: usize,
+    field_name: &str,
+) -> AppResult<StagedImage> {
+    let mut bytes: Vec<u8> = Vec::new();
+    while let Some(chunk) = field.chunk().await.map_err(map_multipart_error)? {
+        if bytes.len().saturating_add(chunk.len()) > max_size {
+            return Err(AppError::PayloadTooLarge(format!(
+                "{field_name} exceeds {max_size}-byte cap"
+            )));
+        }
+        bytes.extend_from_slice(&chunk);
+    }
+    // Reuse the shared sniff + whitelist check; it re-verifies the size cap
+    // (already enforced above) and derives mime/ext from magic bytes.
+    let img = parse_image(bytes, max_size, field_name)?;
+    let staging_key = format!("{STAGING_PREFIX}/{}/{:04}.{}", upload_id.simple(), seq, img.ext);
+    storage.put(&staging_key, &img.bytes).await?;
+    Ok(StagedImage {
+        staging_key,
+        mime: img.mime,
+        ext: img.ext,
+        size_bytes: img.bytes.len() as i64,
+    })
+}
+
 pub fn parse_image(bytes: Vec<u8>, max_size: usize, field_name: &str) -> AppResult<UploadedImage> {
     if bytes.len() > max_size {
         return Err(AppError::PayloadTooLarge(format!(

backend/tests/api_uploads.rs

@@ -328,6 +328,42 @@ async fn create_chapter_rejects_when_no_pages_with_422(pool: PgPool) {
     assert!(body["error"]["details"]["page"].is_string());
 }
 
+#[sqlx::test(migrations = "./migrations")]
+async fn create_chapter_rejects_over_page_cap_with_413(pool: PgPool) {
+    // Cap of 2 pages: a 3-page upload is refused once the third `page` part
+    // arrives, before any chapter row is written.
+    let h = common::harness_with_page_cap(pool.clone(), 2);
+    let (_, cookie) = common::register_user(&h.app).await;
+    let manga_id = common::seed_manga_via_api(&h.app, &cookie, "Berserk").await;
+
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::post_multipart_with_cookie(
+            &format!("/api/v1/mangas/{manga_id}/chapters"),
+            MultipartBuilder::new()
+                .add_json("metadata", json!({ "number": 1 }))
+                .add_file("page", "1.png", "image/png", &common::fake_png_bytes())
+                .add_file("page", "2.png", "image/png", &common::fake_png_bytes())
+                .add_file("page", "3.png", "image/png", &common::fake_png_bytes()),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    let body = common::body_json(resp).await;
+    assert_eq!(body["error"]["code"], "payload_too_large");
+
+    // Nothing persisted — the cap trips before the chapter transaction.
+    let (chapter_count,): (i64,) =
+        sqlx::query_as("SELECT count(*) FROM chapters WHERE manga_id = $1")
+            .bind(manga_id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(chapter_count, 0, "over-cap upload must not create a chapter");
+}
+
 #[sqlx::test(migrations = "./migrations")]
 async fn create_chapter_rejects_renamed_non_image_page(pool: PgPool) {
     let h = common::harness(pool);

backend/tests/common/mod.rs

@@ -79,6 +79,7 @@ fn harness_with_auth_config(
             // exercise without producing tens of MBs of bytes.
             max_request_bytes: 4 * 1024 * 1024,
             max_file_bytes: 256 * 1024,
+            max_pages_per_chapter: 2000,
         },
         auth_limiter,
         // Default harness has no crawler daemon wired up; admin resync
@@ -170,6 +171,7 @@ pub fn harness_with_resync(
         upload: UploadConfig {
             max_request_bytes: 4 * 1024 * 1024,
             max_file_bytes: 256 * 1024,
+            max_pages_per_chapter: 2000,
         },
         auth_limiter,
         runtime,
@@ -203,6 +205,7 @@ pub fn harness_with_analysis(pool: PgPool) -> Harness {
         upload: UploadConfig {
             max_request_bytes: 4 * 1024 * 1024,
             max_file_bytes: 256 * 1024,
+            max_pages_per_chapter: 2000,
         },
         auth_limiter,
         runtime: Arc::new(RuntimeControls::new(true)),
@@ -218,6 +221,39 @@ pub fn harness_with_analysis(pool: PgPool) -> Harness {
     }
 }
 
+/// Like [`harness`] but with a low `max_pages_per_chapter` so the chapter
+/// upload page-count cap is cheap to exercise.
+pub fn harness_with_page_cap(pool: PgPool, max_pages_per_chapter: usize) -> Harness {
+    let storage_dir = tempfile::tempdir().expect("tempdir");
+    let storage = Arc::new(LocalStorage::new(storage_dir.path()));
+    let auth = AuthConfig {
+        cookie_secure: false,
+        ..AuthConfig::default()
+    };
+    let auth_limiter = Arc::new(AuthRateLimiter::new(auth.rate_limit));
+    let state = AppState {
+        db: pool,
+        storage,
+        auth,
+        upload: UploadConfig {
+            max_request_bytes: 4 * 1024 * 1024,
+            max_file_bytes: 256 * 1024,
+            max_pages_per_chapter,
+        },
+        auth_limiter,
+        runtime: Arc::new(RuntimeControls::new(false)),
+        reloader: None,
+        crawler_base: CrawlerConfig::default(),
+        analysis_base: AnalysisConfig::default(),
+        admin_allowed_origins: Arc::new(vec![TEST_ORIGIN.to_string()]),
+        analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()),
+    };
+    Harness {
+        app: router(state),
+        _storage_dir: storage_dir,
+    }
+}
+
 /// A [`DaemonReloader`] stub that records the configs it was asked to apply
 /// and flips the shared analysis gate, without spawning any real daemon. Lets
 /// settings tests assert that a `PUT` triggers a reload with the converted
@@ -265,6 +301,7 @@ pub fn harness_with_settings_reloader(pool: PgPool) -> (Harness, Arc<StubReloade
         upload: UploadConfig {
             max_request_bytes: 4 * 1024 * 1024,
             max_file_bytes: 256 * 1024,
+            max_pages_per_chapter: 2000,
         },
         auth_limiter,
         runtime,
@@ -300,6 +337,7 @@ pub fn harness_with_admin_origins(pool: PgPool, origins: Vec<String>) -> Harness
         upload: UploadConfig {
             max_request_bytes: 4 * 1024 * 1024,
             max_file_bytes: 256 * 1024,
+            max_pages_per_chapter: 2000,
         },
         auth_limiter,
         runtime: Arc::new(RuntimeControls::new(false)),
@@ -376,6 +414,12 @@ impl Storage for FailingStorage {
     async fn size(&self, key: &str) -> Result<u64, StorageError> {
         self.inner.size(key).await
     }
+    // Delegate straight to the inner filesystem rename — the fault
+    // injection counts `put`/`put_stream` only, so promoting a staged page
+    // to its final key never spuriously trips the injected failure.
+    async fn rename(&self, from: &str, to: &str) -> Result<(), StorageError> {
+        self.inner.rename(from, to).await
+    }
 }
 
 pub async fn body_json(response: axum::response::Response) -> serde_json::Value {

backend/tests/crawler_jobs.rs

@@ -718,42 +718,49 @@ async fn release_returns_to_pending_and_undoes_attempt_increment(pool: PgPool) {
 }
 
 #[sqlx::test(migrations = "./migrations")]
-async fn reap_done_deletes_old_rows_keeps_fresh(pool: PgPool) {
+async fn reap_terminal_deletes_old_done_and_dead_keeps_fresh_and_active(pool: PgPool) {
-    // Two done rows: one old (updated_at 10 days ago), one fresh.
+    // Helper: enqueue a fresh pending job and return its id.
-    let old_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
+    async fn enqueue_one(pool: &PgPool) -> Uuid {
+        match jobs::enqueue(pool, &chapter_content_payload(Uuid::new_v4()))
             .await
             .unwrap()
         {
             EnqueueResult::Inserted(id) => id,
             _ => unreachable!(),
-    };
+        }
-    let fresh_id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
+    }
-        .await
-        .unwrap()
-    {
-        EnqueueResult::Inserted(id) => id,
-        _ => unreachable!(),
-    };
 
+    let old_done = enqueue_one(&pool).await;
+    let old_dead = enqueue_one(&pool).await;
+    let fresh_done = enqueue_one(&pool).await;
+    let fresh_dead = enqueue_one(&pool).await;
+    let old_pending = enqueue_one(&pool).await;
+
+    // Old terminal rows (10 days) in both terminal states — both must reap.
     sqlx::query("UPDATE crawler_jobs SET state='done', updated_at = now() - interval '10 days' WHERE id = $1")
-        .bind(old_id)
+        .bind(old_done).execute(&pool).await.unwrap();
-        .execute(&pool)
+    sqlx::query("UPDATE crawler_jobs SET state='dead', updated_at = now() - interval '10 days' WHERE id = $1")
-        .await
+        .bind(old_dead).execute(&pool).await.unwrap();
-        .unwrap();
+    // Fresh terminal rows — inside the retention window, kept.
     sqlx::query("UPDATE crawler_jobs SET state='done' WHERE id = $1")
-        .bind(fresh_id)
+        .bind(fresh_done).execute(&pool).await.unwrap();
-        .execute(&pool)
+    sqlx::query("UPDATE crawler_jobs SET state='dead' WHERE id = $1")
-        .await
+        .bind(fresh_dead).execute(&pool).await.unwrap();
-        .unwrap();
+    // Old but still active (pending) — never reaped regardless of age.
+    sqlx::query("UPDATE crawler_jobs SET updated_at = now() - interval '10 days' WHERE id = $1")
+        .bind(old_pending).execute(&pool).await.unwrap();
 
-    let deleted = jobs::reap_done(&pool, 7).await.unwrap();
+    let deleted = jobs::reap_terminal(&pool, 7).await.unwrap();
-    assert_eq!(deleted, 1);
+    assert_eq!(deleted, 2, "both old done and old dead rows are reaped");
 
-    let remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs ORDER BY id")
+    let mut remaining: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM crawler_jobs")
         .fetch_all(&pool)
         .await
         .unwrap();
-    assert_eq!(remaining, vec![fresh_id], "only fresh row remains");
+    remaining.sort();
+    let mut expected = vec![fresh_done, fresh_dead, old_pending];
+    expected.sort();
+    assert_eq!(remaining, expected, "fresh terminal + active-pending rows survive");
 }
 
 #[sqlx::test(migrations = "./migrations")]
@@ -840,7 +847,7 @@ async fn lease_ties_on_scheduled_at_break_by_created_at(pool: PgPool) {
 }
 
 #[sqlx::test(migrations = "./migrations")]
-async fn reap_done_zero_is_a_no_op(pool: PgPool) {
+async fn reap_terminal_zero_is_a_no_op(pool: PgPool) {
     let id = match jobs::enqueue(&pool, &chapter_content_payload(Uuid::new_v4()))
         .await
         .unwrap()
@@ -854,7 +861,7 @@ async fn reap_done_zero_is_a_no_op(pool: PgPool) {
         .await
         .unwrap();
 
-    let deleted = jobs::reap_done(&pool, 0).await.unwrap();
+    let deleted = jobs::reap_terminal(&pool, 0).await.unwrap();
     assert_eq!(deleted, 0);
     assert_eq!(job_count(&pool).await, 1);
 }

docker-compose.yml

@@ -96,6 +96,7 @@ services:
       # Upload limits.
       MAX_REQUEST_BYTES: ${MAX_REQUEST_BYTES:-209715200}
       MAX_FILE_BYTES: ${MAX_FILE_BYTES:-20971520}
+      MAX_PAGES_PER_CHAPTER: ${MAX_PAGES_PER_CHAPTER:-2000}
       # Crawler boot seeds (first-boot only; switch to dashboard-editable
       # once the app_settings row exists).
       CRAWLER_START_URL: ${CRAWLER_START_URL:-}
@@ -124,6 +125,7 @@ services:
       CRAWLER_DOWNLOAD_ALLOWLIST: ${CRAWLER_DOWNLOAD_ALLOWLIST:-}
       CRAWLER_ALLOW_ANY_HOST: ${CRAWLER_ALLOW_ANY_HOST:-false}
       CRAWLER_MAX_IMAGE_BYTES: ${CRAWLER_MAX_IMAGE_BYTES:-33554432}
+      CRAWLER_MAX_IMAGES_PER_CHAPTER: ${CRAWLER_MAX_IMAGES_PER_CHAPTER:-2000}
       # System-chromium override for the crawler. Leave blank to use the
       # bundled fetcher; set to e.g. /usr/bin/chromium-headless-shell on
       # arm64 deployments. Pair with `--build-arg INSTALL_CHROMIUM=true`

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mangalord-frontend",
-  "version": "0.93.2",
+  "version": "0.94.0",
   "private": true,
   "type": "module",
   "scripts": {

frontend/src/app.html

@@ -2,7 +2,16 @@
 <html lang="en">
     <head>
         <meta charset="utf-8" />
-        <link rel="icon" href="%sveltekit.assets%/favicon.ico" />
+        <link
+            rel="icon"
+            href="%sveltekit.assets%/mangalord-monogram-light.svg"
+            media="(prefers-color-scheme: light)"
+        />
+        <link
+            rel="icon"
+            href="%sveltekit.assets%/mangalord-monogram-dark.svg"
+            media="(prefers-color-scheme: dark)"
+        />
         <meta
             name="viewport"
             content="width=device-width, initial-scale=1, viewport-fit=cover"

frontend/src/lib/api/client.test.ts

@@ -1,7 +1,24 @@
 import { describe, it, expect, vi, beforeEach, afterEach, type MockInstance } from 'vitest';
-import { ApiError, request, setOn401Hook } from './client';
+import { ApiError, request, setOn401Hook, fileUrl } from './client';
 import { getManga } from './mangas';
 
+describe('fileUrl', () => {
+    it('keeps a normal /-separated key path literal', () => {
+        expect(fileUrl('mangas/abc/chapters/def/pages/0001.jpg')).toBe(
+            '/api/v1/files/mangas/abc/chapters/def/pages/0001.jpg'
+        );
+    });
+
+    it('percent-encodes reserved characters within a segment', () => {
+        // `?`, `#`, `%`, and spaces inside a segment must be encoded so they
+        // can't be reinterpreted as query/fragment delimiters — while the
+        // slashes stay as literal path separators.
+        expect(fileUrl('weird key/a?b#c%d/x y.png')).toBe(
+            '/api/v1/files/weird%20key/a%3Fb%23c%25d/x%20y.png'
+        );
+    });
+});
+
 describe('request error envelope parsing', () => {
     let fetchSpy: MockInstance<typeof globalThis.fetch>;
 

frontend/src/lib/api/client.ts

@@ -7,9 +7,16 @@ const BASE = import.meta.env?.VITE_API_BASE ?? '/api';
  * Builds an absolute URL to the streaming `/files/{key}` endpoint so
  * components can use it directly in `<img src>` etc., without
  * reconstructing the API base in each call site.
+ *
+ * Storage keys are `/`-separated paths, so each segment is percent-encoded
+ * individually — the slashes stay literal path separators while any
+ * reserved character inside a segment (`?`, `#`, `%`, space, …) can't be
+ * reinterpreted as a query/fragment delimiter. Keys are backend-generated
+ * today, so this is defence-in-depth against a future key source.
  */
 export function fileUrl(key: string): string {
-    return `${BASE}/v1/files/${key}`;
+    const encoded = key.split('/').map(encodeURIComponent).join('/');
+    return `${BASE}/v1/files/${encoded}`;
 }
 
 /**

frontend/static/mangalord-monogram-dark.svg

@@ -0,0 +1,5 @@
+<svg id="mangalord-monogram" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 512 512" fill="none">
+  <path id="m" d="M76 205 L94 205 L256 313 L418 205 L436 205 L436 418 L383 451 L345 322 L256 388 L167 322 L129 451 L76 418 Z" fill="#FFFFFF"/>
+  <path id="crown-body" d="M168.0 222.0 L156 162 L210 200 L256 120 L302 200 L356 162 L344.0 222.0 A195.0 195.0 0 0 1 168.0 222.0 Z M256 160 L267.5 180 L256 200 L244.5 180 Z" fill="#FF4C2E" fill-rule="evenodd"/>
+  <path id="crown-band" d="M170.0 231.9 A203.0 203.0 0 0 0 342.0 231.9 L338.9 247.5 A216.0 216.0 0 0 1 173.1 247.5 Z" fill="#FF4C2E"/>
+</svg>

frontend/static/mangalord-monogram-light.svg

@@ -0,0 +1,5 @@
+<svg id="mangalord-monogram" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 512 512" fill="none">
+  <path id="m" d="M76 205 L94 205 L256 313 L418 205 L436 205 L436 418 L383 451 L345 322 L256 388 L167 322 L129 451 L76 418 Z" fill="#0F1720"/>
+  <path id="crown-body" d="M168.0 222.0 L156 162 L210 200 L256 120 L302 200 L356 162 L344.0 222.0 A195.0 195.0 0 0 1 168.0 222.0 Z M256 160 L267.5 180 L256 200 L244.5 180 Z" fill="#FF4C2E" fill-rule="evenodd"/>
+  <path id="crown-band" d="M170.0 231.9 A203.0 203.0 0 0 0 342.0 231.9 L338.9 247.5 A216.0 216.0 0 0 1 173.1 247.5 Z" fill="#FF4C2E"/>
+</svg>