feat(download): smarter backfill — time-boxed chunks, run-until-done (backend)

Plan #693. Large-catalog backfill (Anduo) no longer sprints to the timeout
wall and dies as an error each run. Builds on the cursor checkpoint (#689).

- Time-boxed chunks: BACKFILL_TIMEOUT_SECONDS(1170)→BACKFILL_CHUNK_SECONDS(600),
  far under the 1350 soft limit. Hitting it = normal chunk boundary (the
  TimeoutExpired path already captures partial output + the cursor), not a
  near-wall death.
- Run-until-done state machine driven by config_overrides[_backfill_state]
  (running/complete/stalled). A running backfill auto-continues in chunks
  across ticks until gallery-dl exits cleanly (rc=0 = reached the bottom →
  'complete'); a safety-cap (BACKFILL_MAX_CHUNKS=200) + the #689 stall-guard
  pause a pathological walk as 'stalled'. Replaces the N-runs counter
  (backfill_runs_remaining repurposed as the cap countdown).
- Progress, not error: a chunk that timed out but advanced (cursor moved
  and/or files written) is reclassified TIMEOUT→PARTIAL (status 'ok').
- Retry storm tamed: gallery-dl retries 3→2, downloader timeout 120→60s, so
  one stuck CDN file fails in ~1-2 min not ~10 (Anduo #40838).
- API: POST /sources/{id}/backfill now takes {action: start|stop}; service
  start_backfill/stop_backfill; new enabled sources auto-arm run-until-done;
  source dict exposes backfill_state + backfill_chunks.

Frontend (Start/Stop control + state badge) lands in the next push.

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

tests/test_source_service.py

@@ -175,58 +175,75 @@ async def test_list_hides_sidecar_synthetic_anchors(db):
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_arms_source(db):
-    """The service method overrides backfill_runs_remaining (regardless of
-    the auto-arm-on-create starting value) and returns the updated record
-    so the API can echo it back."""
+async def test_start_backfill_arms_run_until_done(db):
+    """start_backfill sets state=running + the chunk cap and clears any prior
+    cursor/chunk state, returning the updated record for the API to echo."""
+    from backend.app.services.source_service import BACKFILL_MAX_CHUNKS
+
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice",
     )
-    updated = await svc.set_backfill_runs(rec.id, 5)
-    assert updated.backfill_runs_remaining == 5
+    # Simulate a prior, finished walk leaving stale checkpoint state.
+    await db.execute(
+        Source.__table__.update().where(Source.id == rec.id).values(
+            config_overrides={"_backfill_state": "complete", "_backfill_cursor": "old",
+                              "_backfill_chunks": 7},
+        )
+    )
+    await db.commit()
 
-    db_value = (await db.execute(
-        select(Source.backfill_runs_remaining).where(Source.id == rec.id)
+    updated = await svc.start_backfill(rec.id)
+    assert updated.backfill_state == "running"
+    assert updated.backfill_chunks == 0
+    assert updated.backfill_runs_remaining == BACKFILL_MAX_CHUNKS
+
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == rec.id)
     )).scalar_one()
-    assert db_value == 5
+    assert co.get("_backfill_state") == "running"
+    assert "_backfill_cursor" not in co
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_rejects_out_of_range(db):
+async def test_stop_backfill_returns_to_idle(db):
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice",
     )
-    with pytest.raises(ValueError):
-        await svc.set_backfill_runs(rec.id, 0)
-    with pytest.raises(ValueError):
-        await svc.set_backfill_runs(rec.id, 11)
+    await svc.start_backfill(rec.id)
+    updated = await svc.stop_backfill(rec.id)
+    assert updated.backfill_state is None
+    assert updated.backfill_runs_remaining == 0
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == rec.id)
+    )).scalar_one()
+    assert "_backfill_state" not in (co or {})
 
 
 @pytest.mark.asyncio
-async def test_set_backfill_runs_raises_when_source_missing(db):
+async def test_start_backfill_raises_when_source_missing(db):
     svc = SourceService(db)
     with pytest.raises(LookupError):
-        await svc.set_backfill_runs(99999, 3)
+        await svc.start_backfill(99999)
 
 
 @pytest.mark.asyncio
 async def test_new_enabled_source_starts_in_backfill_mode(db):
-    """Plan #544 follow-up: freshly added enabled sources have no archive
-    yet, so the first few polls would blow the wall-clock cap in tick
-    mode. Pre-arm backfill so the initial walks use the longer timeout."""
+    """Plan #693: freshly added enabled sources have no archive yet, so they
+    arm run-until-done backfill — state 'running' — to walk the full history
+    on the first ticks instead of blowing the wall-clock cap in tick mode."""
     artist = await _artist(db, "Alice")
     svc = SourceService(db)
     rec = await svc.create(
         artist_id=artist.id, platform="patreon",
         url="https://patreon.com/alice-new",
     )
-    assert rec.backfill_runs_remaining == 3
+    assert rec.backfill_state == "running"
 
 
 @pytest.mark.asyncio