diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py index c3ae536..b3e553c 100644 --- a/backend/app/api/downloads.py +++ b/backend/app/api/downloads.py @@ -44,6 +44,9 @@ def _list_record(event: DownloadEvent, source: Source | None, artist: Artist | N "bytes_downloaded": event.bytes_downloaded, "error": event.error, "summary": _summary_from_metadata(event.metadata_), + # plan #709: mid-walk live counts for a RUNNING native-ingester event + # (None otherwise; phase 3 overwrites metadata with run_stats on finish). + "live": (event.metadata_ or {}).get("live"), } diff --git a/backend/app/services/download_backends.py b/backend/app/services/download_backends.py index 02095f8..082b92b 100644 --- a/backend/app/services/download_backends.py +++ b/backend/app/services/download_backends.py @@ -139,6 +139,8 @@ async def _run_native_ingester( resume_cursor=source_config.resume_cursor, time_budget_seconds=source_config.timeout, posts_base=int(overrides.get("_backfill_posts", 0)), + # plan #709: live progress writes to this running event mid-walk. + event_id=ctx.get("event_id"), ), ) return dl_result, resolved_campaign_id diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 2b5286c..210a801 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -25,6 +25,7 @@ Plain-HTTP homelab: no secure-context Web API. from __future__ import annotations +import json import logging import time from collections.abc import Callable @@ -48,6 +49,12 @@ DEAD_LETTER_THRESHOLD = 3 # last_error is Text but bound it so a giant traceback doesn't bloat the row. _ERROR_MAX = 1000 +# plan #709: throttle the live-progress write to the running DownloadEvent to one +# every ~5s — a steady cadence for the Downloads view regardless of how big/slow a +# page is (page boundaries can be minutes apart on image-dense backfills, so a +# page-tied update would lurch). Trivial churn (~one single-row UPDATE / 5s). +_LIVE_PROGRESS_INTERVAL = 5.0 + class Ingester: """Generic native-ingest orchestration. Subclass with a platform adapter @@ -92,6 +99,7 @@ class Ingester: time_budget_seconds: float = 870.0, seen_threshold: int = _TICK_SEEN_THRESHOLD, posts_base: int = 0, + event_id: int | None = None, ) -> DownloadResult: """Walk + download for one source, returning a gallery-dl-shaped result. @@ -109,6 +117,7 @@ class Ingester: checkpoint = mode in ("backfill", "recovery") ledger_key = self._ledger_key start = time.monotonic() + last_live = start # plan #709: last live-progress write timestamp log_lines: list[str] = [] written: list[str] = [] quarantined_paths: list[str] = [] @@ -288,6 +297,19 @@ class Ingester: if to_fail: self._record_failures(source_id, to_fail) + # plan #709: time-throttled live progress to the running event so + # the Downloads view ticks ~every 5s, independent of page size. + now = time.monotonic() + if event_id is not None and (now - last_live) >= _LIVE_PROGRESS_INTERVAL: + last_live = now + self._write_live_progress(event_id, { + "downloaded": downloaded, + "skipped": skipped_count, + "errors": errors, + "quarantined": quarantined, + "posts": posts_processed, + }) + if early_out: break else: @@ -474,6 +496,25 @@ class Ingester: ) session.commit() + def _write_live_progress(self, event_id: int, counts: dict) -> None: + """Throttled mid-walk write of live counts to the RUNNING download_event + (plan #709) so the Downloads view shows progress before the chunk + finishes. A short session (never held across the walk); the `status = + 'running'` guard avoids clobbering an event phase 3 already finalized. + `metadata` is JSONB — jsonb_set sets just the `live` key, leaving the rest + for phase 3 to overwrite with the final run_stats.""" + with self.session_factory() as session: + session.execute( + text( + "UPDATE download_event SET metadata = jsonb_set(" + " coalesce(metadata, '{}'::jsonb), '{live}'," + " cast(:live AS jsonb)) " + "WHERE id = :eid AND status = 'running'" + ), + {"live": json.dumps(counts), "eid": event_id}, + ) + session.commit() + def _still_running(self, source_id: int) -> bool: """True while the source is armed for a deep walk (plan #708 B4). diff --git a/frontend/src/components/subscriptions/ActiveDownloadsPanel.vue b/frontend/src/components/subscriptions/ActiveDownloadsPanel.vue index e35abd6..6f8fb43 100644 --- a/frontend/src/components/subscriptions/ActiveDownloadsPanel.vue +++ b/frontend/src/components/subscriptions/ActiveDownloadsPanel.vue @@ -23,6 +23,18 @@ {{ e.artist_name || '—' }} + + + ↓ {{ e.live.downloaded }} + + ⤼ {{ e.live.skipped }} + ✕ {{ e.live.errors }} + + {{ e.live.posts }} posts + {{ elapsed(e.started_at) }} @@ -123,6 +135,13 @@ function elapsed (startedIso) { font-size: 0.78rem; text-transform: uppercase; letter-spacing: 0.05em; color: rgb(var(--v-theme-on-surface-variant)); } +.fc-active__counts { + display: inline-flex; align-items: center; gap: 8px; + font-variant-numeric: tabular-nums; font-size: 0.78rem; +} +.fc-active__count { color: rgb(var(--v-theme-on-surface-variant)); } +.fc-active__count--err { color: rgb(var(--v-theme-error)); } +.fc-active__count--posts { opacity: 0.7; } @keyframes fc-active-pulse { 0%, 100% { opacity: 1; transform: scale(1); } 50% { opacity: 0.35; transform: scale(0.7); } diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 95da4bd..2e32075 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -421,6 +421,34 @@ async def test_preview_page_limit_caps_walk(source_id, sync_engine, tmp_path): assert result["has_more"] is True +@pytest.mark.asyncio +async def test_write_live_progress_updates_running_event( + source_id, sync_engine, tmp_path, db, +): + """plan #709: live counts land in the RUNNING event's metadata.live; a + finalized event is left alone (status guard).""" + from backend.app.models import DownloadEvent + + ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path)) + running = DownloadEvent(source_id=source_id, status="running") + done = DownloadEvent(source_id=source_id, status="ok") + db.add_all([running, done]) + await db.commit() + + counts = {"downloaded": 3, "skipped": 1, "errors": 0, "quarantined": 0, "posts": 4} + ing._write_live_progress(running.id, counts) + ing._write_live_progress(done.id, {"downloaded": 9}) # guarded: status != running + + live = (await db.execute( + select(DownloadEvent.metadata_).where(DownloadEvent.id == running.id) + )).scalar_one() + assert live["live"] == counts + done_meta = (await db.execute( + select(DownloadEvent.metadata_).where(DownloadEvent.id == done.id) + )).scalar_one() + assert "live" not in (done_meta or {}) + + @pytest.mark.asyncio async def test_still_running_reads_backfill_state(source_id, sync_engine, tmp_path, db): """plan #708 B4: _still_running reflects the source's `_backfill_state` — the