fix(upload): stream chapter pages to storage instead of buffering the whole chapter

The chapter upload handler read every `page` part fully into a Vec before
writing any, so peak memory was the whole chapter (bounded only by the
200 MiB body limit and amplified by concurrent uploads). It also accepted
an unbounded number of pages.

Stream each page part to a `staging/{upload_id}/…` key as it arrives — at
most one page's bytes are held at a time — then, once the chapter row (and
its id) exists, promote each staged blob to its final key via a new
`Storage::rename` (LocalStorage: fs rename; default impl: stream+delete for
future backends). Finalization is all-or-nothing: on any failure the DB
rolls back and both staged and already-finalized blobs are cleaned up.

Add MAX_PAGES_PER_CHAPTER (UploadConfig, default 2000, 0 = disabled),
rejecting an over-cap upload with 413 before any DB write. Also document
the crawler-side CRAWLER_MAX_IMAGES_PER_CHAPTER (added earlier) in
.env.example + docker-compose so the env-coverage test passes.

Tests: LocalStorage rename unit tests; a 413 over-cap upload test; existing
rollback + happy-path upload tests still green (the fault-injecting storage
counts put/put_stream, so mid-upload failure still rolls back).

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

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,97 +93,173 @@ 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();
 
-    while let Some(field) = next_field(&mut multipart).await? {
-        match field.name() {
-            Some("metadata") => {
-                let bytes = read_field_bytes(field).await?;
-                metadata =
-                    Some(serde_json::from_slice(&bytes).map_err(|e| {
+    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 = Some(serde_json::from_slice(&bytes).map_err(|e| {
                         AppError::ValidationFailed {
                             message: "metadata is not valid JSON".into(),
                             details: json!({ "metadata": e.to_string() }),
                         }
                     })?);
+                }
+                Some("page") => {
+                    if state.upload.max_pages_per_chapter != 0
+                        && staged.len() >= state.upload.max_pages_per_chapter
+                    {
+                        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,
             }
-            Some("page") => {
-                let bytes = read_field_bytes(field).await?.to_vec();
-                let field_name = format!("page[{}]", pages.len());
-                pages.push(parse_image(bytes, state.upload.max_file_bytes, &field_name)?);
-            }
-            _ => continue,
         }
-    }
 
-    let metadata = metadata.ok_or_else(|| AppError::ValidationFailed {
-        message: "metadata part is required".into(),
-        details: json!({ "metadata": "required" }),
-    })?;
-    // Chapter number is 1-indexed everywhere (URLs, upload form,
-    // reader). Reject 0 / negative numbers up front so the row never
-    // makes it into the DB. Mirrors the page>=1 rule on bookmarks.
-    if metadata.number < 1 {
-        return Err(AppError::ValidationFailed {
-            message: "chapter number must be 1 or greater".into(),
-            details: json!({ "number": "must be >= 1" }),
-        });
-    }
-    if pages.is_empty() {
-        return Err(AppError::ValidationFailed {
-            message: "at least one page is required".into(),
-            details: json!({ "page": "at least one required" }),
-        });
+        let metadata = metadata.take().ok_or_else(|| AppError::ValidationFailed {
+            message: "metadata part is required".into(),
+            details: json!({ "metadata": "required" }),
+        })?;
+        // Chapter number is 1-indexed everywhere (URLs, upload form,
+        // reader). Reject 0 / negative numbers up front so the row never
+        // makes it into the DB. Mirrors the page>=1 rule on bookmarks.
+        if metadata.number < 1 {
+            return Err(AppError::ValidationFailed {
+                message: "chapter number must be 1 or greater".into(),
+                details: json!({ "number": "must be >= 1" }),
+            });
+        }
+        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
-    // fails mid-loop, the chapter row + any earlier page rows are
-    // rolled back so we don't leave a chapter with stale page_count=0
-    // and orphaned page rows. Bytes already written to storage on a
-    // rolled-back transaction become orphans on disk; a future reaper
-    // can sweep them. DB consistency wins over storage tidiness here.
-    let mut tx = state.db.begin().await?;
-    let mut chapter = repo::chapter::create(
+    let metadata = match stage_result {
+        Ok(m) => m,
+        Err(e) => {
+            // Reject before any DB write — remove every page we staged.
+            cleanup_staging(state.storage.as_ref(), &staged).await;
+            return Err(e);
+        }
+    };
+
+    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());
-    for (idx, page) in pages.iter().enumerate() {
+    let mut page_ids: Vec<Uuid> = Vec::with_capacity(staged.len());
+    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 key = format!(
-            "mangas/{}/chapters/{}/pages/{}.{}",
-            manga_id, chapter.id, nnnn, page.ext
+        let final_key = format!(
+            "mangas/{}/chapters/{}/pages/{:04}.{}",
+            manga_id, chapter.id, page_number, page.ext
         );
-        state.storage.put(&key, &page.bytes).await?;
-        let created = repo::page::create(
+        if let Err(e) = storage.rename(&page.staging_key, &final_key).await {
+            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?;
-        page_ids.push(created.id);
+        .await
+        {
+            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;
-    repo::chapter::set_page_count(&mut *tx, chapter.id, page_count).await?;
+    let page_count = staged.len() as i32;
+    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
-    // captured, so the true total is the sum of their byte lengths — set it
-    // on the response so the 201 body matches the persisted state.
-    chapter.size_bytes = Some(pages.iter().map(|p| p.bytes.len() as i64).sum());
+    // its `size_bytes` is a stale 0. Each staged page carried its byte
+    // length, so their sum is the chapter's true storage — set it on the
+    // response so the 201 body matches the persisted state.
+    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) =
-                repo::page_analysis::enqueue_for_page(&state.db, page_id, false).await
-            {
+            if let Err(e) = 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/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,
         }
     }
 }
@@ -533,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()

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!(