feat: incremental crawl mode with seed-completion gate (0.33.0)

Daemon now auto-detects mode per source: Backfill until the first
full walk records `seed_completed:<source>` in `crawler_state`, then
Incremental (newest-first, stops after N consecutive Unchanged
upserts). `CRAWLER_MODE` overrides to a fixed mode; CLI rejects
`auto` since it has no pre-run DB state.

`Source::discover` returns a lazy `DiscoverWalk` so Incremental can
break out mid-walk without prefetching pages. The drop pass and seed
marker are now gated on a true full walk — fixes a latent soft-drop
of the index tail under partial sweeps.

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

backend/src/crawler/pipeline.rs

@@ -23,14 +23,34 @@ pub struct MetadataStats {
     pub mangas_failed: usize,
 }
 
+/// Decide whether the per-ref loop should stop based on the Incremental
+/// streak counter. Pulled out as a pure function so the rule is unit-
+/// testable without standing up the walker or DB.
+pub(crate) fn should_stop(mode: DiscoverMode, consecutive_unchanged: usize) -> bool {
+    match mode {
+        DiscoverMode::Backfill => false,
+        DiscoverMode::Incremental { stop_after_unchanged } => {
+            consecutive_unchanged >= stop_after_unchanged
+        }
+    }
+}
+
 /// 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.
 ///
-/// `limit == 0` means no cap (full backfill). `skip_chapters == true` is
-/// the "metadata-only" mode (parser doesn't extract chapters, and
-/// `sync_manga_chapters` is skipped — otherwise an empty chapter list
-/// would soft-drop existing rows).
+/// `limit == 0` means no cap (full sweep up to the source's own bound).
+/// `skip_chapters == true` is the "metadata-only" mode (parser doesn't
+/// extract chapters, and `sync_manga_chapters` is skipped — otherwise an
+/// empty chapter list would soft-drop existing rows).
+///
+/// `mode` controls the walk:
+/// - `Backfill` — oldest-first, no early exit. The only mode that runs
+///   the end-of-walk drop pass + writes `seed_completed_at`.
+/// - `Incremental { stop_after_unchanged }` — newest-first, breaks out
+///   after N consecutive Unchanged upserts. Drop pass is skipped (the
+///   tail of the index is never visited, so its `last_seen_at` is
+///   stale and using it to soft-drop would be unsafe).
 #[allow(clippy::too_many_arguments)]
 pub async fn run_metadata_pass(
     browser_manager: &BrowserManager,
@@ -41,6 +61,7 @@ pub async fn run_metadata_pass(
     start_url: &str,
     limit: usize,
     skip_chapters: bool,
+    mode: DiscoverMode,
 ) -> anyhow::Result<MetadataStats> {
     let lease = browser_manager
         .acquire()
@@ -74,123 +95,189 @@ pub async fn run_metadata_pass(
     let run_started_at = chrono::Utc::now();
     let max_refs = (limit > 0).then_some(limit);
 
-    tracing::info!(?max_refs, "discovering manga list");
-    let refs = source
-        .discover(&ctx, DiscoverMode::Backfill, max_refs)
+    tracing::info!(?mode, ?max_refs, "starting metadata pass");
+    let mut walker = source
+        .discover(&ctx, mode)
         .await
         .context("discover failed")?;
-    tracing::info!(count = refs.len(), "discovered manga list");
 
-    let mut stats = MetadataStats {
-        discovered: refs.len(),
-        ..MetadataStats::default()
-    };
+    let mut stats = MetadataStats::default();
+    let mut consecutive_unchanged: usize = 0;
+    let mut walked_to_completion = false;
+    let mut hit_limit = false;
+    let mut hit_incremental_stop = false;
 
-    for (i, r) in refs.iter().enumerate() {
-        tracing::info!(
-            idx = i + 1,
-            total = stats.discovered,
-            key = %r.source_manga_key,
-            "fetching metadata"
-        );
-        let manga = match source.fetch_manga(&ctx, r).await {
-            Ok(m) => m,
-            Err(e) => {
-                tracing::warn!(
-                    key = %r.source_manga_key,
-                    url = %r.url,
-                    error = ?e,
-                    "fetch_manga failed"
-                );
-                stats.mangas_failed += 1;
-                continue;
+    'outer: loop {
+        let batch = match walker.next_batch(&ctx).await? {
+            Some(b) => b,
+            None => {
+                walked_to_completion = true;
+                break;
             }
         };
-
-        let upsert = match repo::crawler::upsert_manga_from_source(db, source_id, &r.url, &manga)
-            .await
-        {
-            Ok(u) => u,
-            Err(e) => {
-                tracing::error!(
-                    key = %r.source_manga_key,
-                    error = ?e,
-                    "upsert_manga_from_source failed"
-                );
-                stats.mangas_failed += 1;
-                continue;
+        for r in batch {
+            if max_refs.map(|m| stats.discovered >= m).unwrap_or(false) {
+                hit_limit = true;
+                tracing::info!(cap = ?max_refs, "max_results reached; halting walk");
+                break 'outer;
             }
-        };
-        stats.upserted += 1;
-        tracing::info!(
-            key = %manga.source_manga_key,
-            manga_id = %upsert.manga_id,
-            status = ?upsert.status,
-            title = %manga.title,
-            "manga upserted"
-        );
-
-        // Cover image: download when missing in storage or when metadata
-        // signaled an update (cover URL is part of metadata_hash, so
-        // Updated implies the URL may have moved). Failures are non-fatal.
-        let needs_cover = upsert.cover_image_path.is_none()
-            || 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(
-                    db,
-                    storage,
-                    http,
-                    rate,
-                    &r.url,
-                    upsert.manga_id,
-                    cover_url,
-                )
-                .await
-                {
-                    Ok(()) => stats.covers_fetched += 1,
-                    Err(e) => tracing::warn!(
-                        manga_id = %upsert.manga_id,
+            stats.discovered += 1;
+            tracing::info!(
+                idx = stats.discovered,
+                key = %r.source_manga_key,
+                "fetching metadata"
+            );
+            let manga = match source.fetch_manga(&ctx, &r).await {
+                Ok(m) => m,
+                Err(e) => {
+                    tracing::warn!(
+                        key = %r.source_manga_key,
+                        url = %r.url,
                         error = ?e,
-                        "cover download failed"
-                    ),
+                        "fetch_manga failed"
+                    );
+                    stats.mangas_failed += 1;
+                    continue;
                 }
-            }
-        }
+            };
 
-        if !skip_chapters {
-            match repo::crawler::sync_manga_chapters(
-                db,
-                source_id,
-                upsert.manga_id,
-                &manga.chapters,
+            let upsert = match repo::crawler::upsert_manga_from_source(
+                db, source_id, &r.url, &manga,
             )
             .await
             {
-                Ok(diff) => tracing::info!(
-                    manga_id = %upsert.manga_id,
-                    new = diff.new,
-                    refreshed = diff.refreshed,
-                    dropped = diff.dropped,
-                    "chapters synced"
-                ),
-                Err(e) => tracing::warn!(
-                    manga_id = %upsert.manga_id,
-                    error = ?e,
-                    "chapter sync failed"
-                ),
+                Ok(u) => u,
+                Err(e) => {
+                    tracing::error!(
+                        key = %r.source_manga_key,
+                        error = ?e,
+                        "upsert_manga_from_source failed"
+                    );
+                    stats.mangas_failed += 1;
+                    continue;
+                }
+            };
+            stats.upserted += 1;
+            tracing::info!(
+                key = %manga.source_manga_key,
+                manga_id = %upsert.manga_id,
+                status = ?upsert.status,
+                title = %manga.title,
+                "manga upserted"
+            );
+
+            // Cover image: download when missing in storage or when metadata
+            // signaled an update (cover URL is part of metadata_hash, so
+            // Updated implies the URL may have moved). Failures are non-fatal.
+            let needs_cover = upsert.cover_image_path.is_none()
+                || 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(
+                        db,
+                        storage,
+                        http,
+                        rate,
+                        &r.url,
+                        upsert.manga_id,
+                        cover_url,
+                    )
+                    .await
+                    {
+                        Ok(()) => stats.covers_fetched += 1,
+                        Err(e) => tracing::warn!(
+                            manga_id = %upsert.manga_id,
+                            error = ?e,
+                            "cover download failed"
+                        ),
+                    }
+                }
+            }
+
+            if !skip_chapters {
+                match repo::crawler::sync_manga_chapters(
+                    db,
+                    source_id,
+                    upsert.manga_id,
+                    &manga.chapters,
+                )
+                .await
+                {
+                    Ok(diff) => tracing::info!(
+                        manga_id = %upsert.manga_id,
+                        new = diff.new,
+                        refreshed = diff.refreshed,
+                        dropped = diff.dropped,
+                        "chapters synced"
+                    ),
+                    Err(e) => tracing::warn!(
+                        manga_id = %upsert.manga_id,
+                        error = ?e,
+                        "chapter sync failed"
+                    ),
+                }
+            }
+
+            // Incremental stop: count consecutive Unchanged upserts and
+            // bail once the threshold is reached. New/Updated resets the
+            // streak so a fresh entry mid-batch doesn't accidentally trip
+            // the cutoff.
+            match upsert.status {
+                repo::crawler::UpsertStatus::Unchanged => {
+                    consecutive_unchanged += 1;
+                }
+                repo::crawler::UpsertStatus::New | repo::crawler::UpsertStatus::Updated => {
+                    consecutive_unchanged = 0;
+                }
+            }
+            if should_stop(mode, consecutive_unchanged) {
+                hit_incremental_stop = true;
+                tracing::info!(
+                    consecutive_unchanged,
+                    "incremental stop threshold reached; halting walk"
+                );
+                break 'outer;
             }
         }
     }
 
-    if limit == 0 {
+    // Drop pass: only when the walk truly covered everything the source
+    // surfaces. `last_seen_at` on un-visited rows is stale, so running
+    // the drop on a partial walk would soft-drop the tail of the index.
+    let full_walk = walked_to_completion && !hit_limit && !hit_incremental_stop;
+    let backfill_complete = full_walk && matches!(mode, DiscoverMode::Backfill);
+    if full_walk {
         match repo::crawler::mark_dropped_mangas(db, source_id, run_started_at).await {
             Ok(n) => tracing::info!(dropped = n, "marked unseen manga as dropped"),
             Err(e) => tracing::warn!(error = ?e, "drop-pass failed"),
         }
     } else {
-        tracing::info!(limit, "partial sync — skipping drop pass");
+        tracing::info!(
+            ?mode,
+            hit_limit,
+            hit_incremental_stop,
+            "partial sync — skipping drop pass"
+        );
     }
+    if backfill_complete {
+        if let Err(e) = repo::crawler::mark_seed_completed(db, source_id, run_started_at).await {
+            tracing::warn!(error = ?e, "mark_seed_completed failed");
+        } else {
+            tracing::info!(source_id, "seed marked complete");
+        }
+    }
+
+    tracing::info!(
+        ?mode,
+        discovered = stats.discovered,
+        upserted = stats.upserted,
+        covers_fetched = stats.covers_fetched,
+        mangas_failed = stats.mangas_failed,
+        walked_to_completion,
+        hit_limit,
+        hit_incremental_stop,
+        "metadata pass complete"
+    );
 
     drop(lease);
     Ok(stats)
@@ -345,3 +432,36 @@ fn origin_of(url: &str) -> Option<String> {
     let host = rest.split('/').next()?;
     Some(format!("{scheme}://{host}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn backfill_never_stops_regardless_of_streak() {
+        assert!(!should_stop(DiscoverMode::Backfill, 0));
+        assert!(!should_stop(DiscoverMode::Backfill, 100));
+        assert!(!should_stop(DiscoverMode::Backfill, usize::MAX));
+    }
+
+    #[test]
+    fn incremental_stops_when_streak_meets_threshold() {
+        let mode = DiscoverMode::Incremental {
+            stop_after_unchanged: 3,
+        };
+        assert!(!should_stop(mode, 0));
+        assert!(!should_stop(mode, 2));
+        assert!(should_stop(mode, 3), "stops at exactly the threshold");
+        assert!(should_stop(mode, 100), "stops at anything past threshold");
+    }
+
+    #[test]
+    fn incremental_with_zero_threshold_stops_immediately() {
+        // A nonsensical config (no Unchanged needed to stop) shouldn't
+        // panic — it just means the very first ref triggers the bail.
+        let mode = DiscoverMode::Incremental {
+            stop_after_unchanged: 0,
+        };
+        assert!(should_stop(mode, 0));
+    }
+}