From 28f9166e30630b9ec4da278075c928582d8b1da2 Mon Sep 17 00:00:00 2001 From: JD Bothma Date: Tue, 4 Aug 2026 16:38:30 +0100 Subject: [PATCH] Tolerate per-URL operational errors instead of aborting the run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single capture raising an operational error (e.g. Pravda's HAR processing timeout, which Pravda raises rather than persists) propagated out of asyncio.gather and aborted the entire run. Gather these per-URL failures instead: capture_urls now catches exceptions per URL, records them as OperationalError, and returns them alongside the successful captures; the extraction loop does the same. summarise_errors logs a count (by stage and exception type) plus the per-URL detail at the end of the run. The run now completes and writes outputs in spite of these errors. These are operational errors, so they go to the log only — not the database, which is reserved for the capture failures Pravda persists on the snapshot (HTTP status, browser failures). Co-Authored-By: Claude Opus 4.8 --- kolkhoz/capture.py | 77 ++++++++++++++++++++++++++++++++++++++++------ kolkhoz/cli.py | 24 +++++++++++++-- 2 files changed, 88 insertions(+), 13 deletions(-) diff --git a/kolkhoz/capture.py b/kolkhoz/capture.py index 56adef9..c63d872 100644 --- a/kolkhoz/capture.py +++ b/kolkhoz/capture.py @@ -16,6 +16,8 @@ import io import logging import os +from collections import Counter +from dataclasses import dataclass import fsspec from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper @@ -131,25 +133,80 @@ def spans(size: int) -> list[tuple[int, int]]: return tiles +@dataclass +class OperationalError: + """A per-URL failure that must not abort the whole run. + + Distinct from the capture failures Pravda *persists* on a snapshot with + ``error`` set (HTTP status, browser/context failures): those come back as + an errored Snapshot and are recorded in the database. An + ``OperationalError`` is one Pravda *raises* instead — e.g. an inner HAR + processing or storage timeout, which Pravda's own code notes is an + operational error rather than persisted evidence. We gather these, count + them, and let the run finish; they belong in the log, not the database. + """ + + stage: str + url: str + error: str + + +def summarise_errors(errors: list[OperationalError]) -> None: + """Log a summary of the operational errors gathered during the run.""" + if not errors: + log.info("operational errors: none") + return + by_stage = Counter(e.stage for e in errors) + by_type = Counter(e.error.split(":", 1)[0] for e in errors) + log.warning( + "operational errors: %d across %d URL(s) — by stage: %s", + len(errors), + len({e.url for e in errors}), + ", ".join(f"{stage}={n}" for stage, n in by_stage.most_common()), + ) + for etype, count in by_type.most_common(): + log.warning(" %s: %d", etype, count) + for error in errors: + log.warning(" [%s] %s — %s", error.stage, error.url, error.error) + + async def capture_urls( pravda: Pravda, urls: list[str], concurrency: int -) -> dict[str, Snapshot]: +) -> tuple[dict[str, Snapshot], list[OperationalError]]: """Capture each URL once, concurrently, through one Pravda instance. At most *concurrency* captures run at once (an ``asyncio.Semaphore`` bounds them); each is otherwise independent. Returns a mapping of the - requested URL to the Snapshot Pravda persisted for it. Pravda persists - capture failures with ``error`` set rather than raising, so the mapping - covers every requested URL — callers decide how to treat errored - snapshots. + requested URL to the Snapshot Pravda persisted for it, plus a list of + operational errors. Pravda persists capture failures with ``error`` set + rather than raising, so those still appear in the mapping as errored + snapshots. A URL that Pravda fails *operationally* (raising, e.g. a HAR + timeout) is absent from the mapping and recorded as an ``OperationalError`` + instead, so one such failure cannot abort the whole run. """ sem = asyncio.Semaphore(concurrency) + captures: dict[str, Snapshot] = {} + errors: list[OperationalError] = [] - async def snap(url: str) -> tuple[str, Snapshot]: + async def snap(url: str) -> None: async with sem: - snapshot = await pravda.snapshot(url) + try: + snapshot = await pravda.snapshot(url) + except Exception as exc: + # Never abort the run for one URL. CancelledError / + # KeyboardInterrupt are BaseException and still propagate. + log.warning( + "capture failed operationally for %s: %s: %s", + url, + type(exc).__name__, + exc, + ) + errors.append( + OperationalError("capture", url, f"{type(exc).__name__}: {exc}") + ) + return log.info("snapshotted %s has_error=%s", snapshot.url, snapshot.error is not None) - return url, snapshot + captures[url] = snapshot - pairs = await asyncio.gather(*(snap(url) for url in urls)) - return dict(pairs) + await asyncio.gather(*(snap(url) for url in urls)) + return captures, errors diff --git a/kolkhoz/cli.py b/kolkhoz/cli.py index a9343a1..93a3450 100644 --- a/kolkhoz/cli.py +++ b/kolkhoz/cli.py @@ -9,11 +9,13 @@ from pravda import Snapshot, migrate from kolkhoz.capture import ( + OperationalError, capture_urls, is_blank, pravda_client, read_artifact, storage_filesystem, + summarise_errors, ) from kolkhoz.config import Config, load_config from kolkhoz.export import holder_to_record, write_outputs @@ -123,12 +125,16 @@ async def _run_pipeline( fs = storage_filesystem(config.pravda) pravda = pravda_client(config.pravda) async with pravda: - captures = await capture_urls(pravda, urls, concurrency) + captures, errors = await capture_urls(pravda, urls, concurrency) extracted = 0 hits = 0 for dataset, url, organization in associations: - snapshot = captures[url] + snapshot = captures.get(url) + if snapshot is None: + # Operational capture failure: already logged and counted in + # `errors`; keep going rather than abort the run. + continue if snapshot.error is not None: log.warning(" skip %s — capture failed: %s", url, snapshot.error) continue @@ -149,7 +155,18 @@ async def _run_pipeline( ) continue - holders = await extract_snapshot(snapshot, fs, config, client) + try: + holders = await extract_snapshot(snapshot, fs, config, client) + except Exception as exc: + # One page's extraction failing (LLM/parse/IO) must not abort + # the run; log, count, and move on. + log.warning( + " extraction failed for %s: %s: %s", url, type(exc).__name__, exc + ) + errors.append( + OperationalError("extract", url, f"{type(exc).__name__}: {exc}") + ) + continue groups[dataset].extend( holder_to_record(dataset, url, organization, snapshot, holder) for holder in holders @@ -160,6 +177,7 @@ async def _run_pipeline( await write_outputs(groups, config.paths) log.info("extraction: %d hit, %d miss", hits, extracted - hits) + summarise_errors(errors) @cli.command(