From d27c76e99f847cac7089e12ee31f7b3b1dadf1fb Mon Sep 17 00:00:00 2001 From: mfro Date: Mon, 7 Sep 2026 19:11:27 -0500 Subject: [PATCH 1/2] feat: throw error on partial downloads --- tests/test_end_to_end.py | 16 ++++++++++++++++ warc2zip.py | 20 +++++++++++++++----- 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index c776624..d761f3d 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -519,3 +519,19 @@ def test_limit_counts_revisit_captures(warc_path, tmp_path): manifest = [dict(zip(MANIFEST_COLUMNS, row)) for row in read_rows(zf, "manifest.csv")] assert len(manifest) == 4 assert manifest[-1]["warc_type"] == "revisit" + + +def test_short_download_is_reported(warc_path, tmp_path, monkeypatch, capsys): + class SizedBytesIO(io.BytesIO): + def __init__(self, data, size): + super().__init__(data) + self.size = size + + data = warc_path.read_bytes() + monkeypatch.setattr( + "warc2zip.fsspec_open", + lambda *_args, **_kwargs: SizedBytesIO(data, len(data) + 1), + ) + + assert main("https://example.test/test.warc.gz", str(tmp_path / "out.zip")) == 1 + assert f"read {len(data)} bytes; expected {len(data) + 1} bytes" in capsys.readouterr().err diff --git a/warc2zip.py b/warc2zip.py index d6b1481..b4cb580 100644 --- a/warc2zip.py +++ b/warc2zip.py @@ -757,14 +757,17 @@ def main(input_file, output_path, dry_run=False, limit=None, output_format="flat record_types = Counter() # every record read, by WARC-Type — extracted or not limit_reached = False + download_size_mismatch = False # response -> Record id <-> metadata - with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as outer_zip: # Pass 1: Read WARC, write payloads immediately, buffer only headers - with fsspec_open(input_file, "rb", default_fh=sys.stdin.buffer) as stream: - if not hasattr(stream, "tell") or stream == sys.stdin.buffer: - # sys.stdin.buffer has a tell() method but it crashes - stream = CountingStream(stream) + with fsspec_open(input_file, "rb", default_fh=sys.stdin.buffer) as raw_stream: + # fsspec HTTP/S3 handles expose the response size when the server provides one. + # Count reads ourselves so an early EOF is not mistaken for a complete archive. + # if we ever decide to add partial-range downloads, this needs to account for that + expected_size = getattr(raw_stream, "size", None) or file_size + stream = CountingStream(raw_stream) pbar = tqdm(total=file_size, unit="B", unit_scale=True, desc="Reading WARC") # Held by name rather than iterated anonymously: get_record_offset() / # get_record_length() hang off the iterator, not the record. @@ -896,6 +899,13 @@ def main(input_file, output_path, dry_run=False, limit=None, output_format="flat pbar.update(stream.tell() - pbar.n) pbar.close() + if limit is None and expected_size is not None and stream.tell() != expected_size: + download_size_mismatch = True + print( + f"Warning: read {stream.tell()} bytes; expected {expected_size} bytes", + file=sys.stderr, + ) + # Link pending requests to their capture groups. A response names its request in # WARC-Concurrent-To; a CC revisit names it in WARC-Refers-To instead, so both are # tried (only ids that actually belong to a request record can match). @@ -1027,7 +1037,7 @@ def main(input_file, output_path, dry_run=False, limit=None, output_format="flat if skipped: print(f"warning: {skipped} CSV row(s) could not be written (see warnings above)", file=sys.stderr) - return skipped + return skipped + int(download_size_mismatch) def cli(): From 05c007dba76c4b3d07ad786243f6e4fb69d4598f Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Tue, 8 Sep 2026 11:35:37 +0100 Subject: [PATCH 2/2] feat: Implement transient failure handling in CountingStream with retry logic --- README.md | 9 ++ tests/test_read_retry.py | 222 +++++++++++++++++++++++++++++++++++++++ warc2zip.py | 117 +++++++++++++++++++-- 3 files changed, 337 insertions(+), 11 deletions(-) create mode 100644 tests/test_read_retry.py diff --git a/README.md b/README.md index 6ab7c0d..a96014a 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,15 @@ Causes: Warnings never change the exit status; warc2zip exits 1 only when a CSV row could not be written. +``` +warning: : attempt 1/9 failed (503, message='Service Unavailable', ...), retrying in 2.6 s +``` + +A transient failure while reading a remote input — throttling, a 5xx, a connection dropped +mid-block. The read resumes exactly where it stopped, so nothing is duplicated or lost. Up to +8 retries with exponential backoff (capped at 60 s, `Retry-After` honoured); after that the +error is raised. A pipe (`-` on stdin) cannot be rewound, so it is not retried. + ## Output Formats All files are placed under a unique root directory inside the zip to prevent collisions when extracting multiple archives into the same folder. The directory name is derived from the WARC-Filename header (in the `warcinfo` record), the current timestamp, and a random suffix: `{crawl_name}_{YYYYMMDDTHHMMSS}_{hex}`. diff --git a/tests/test_read_retry.py b/tests/test_read_retry.py new file mode 100644 index 0000000..8fec489 --- /dev/null +++ b/tests/test_read_retry.py @@ -0,0 +1,222 @@ +"""Transient failures while reading the input are retried by CountingStream. + +The policy is unit-tested with fakes (which errors, where the stream resumes, when it gives +up), then exercised for real: a conversion over a local HTTP server that answers the first +Range request with 503 and cuts the connection halfway through the second. That second case is +the one a per-request retry client would miss — the failure surfaces from the body read, after +the request has already "succeeded". +""" + +import functools +import io +import re +import threading +import zipfile +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest +from warcio.statusandheaders import StatusAndHeaders +from warcio.warcwriter import WARCWriter + +import warc2zip +from warc2zip import FETCH_DEFAULT_RETRIES, CountingStream, is_retryable, main + +PAYLOAD_RE = re.compile(r"^\d+\.[A-Za-z0-9]+$") + + +class FakeHTTPError(Exception): + def __init__(self, status, headers=None): + super().__init__(f"HTTP {status}") + self.status = status + self.headers = headers or {} + + +class FlakyStream(io.BytesIO): + """BytesIO whose read() raises the queued errors first — *after* advancing the position, + the way a partially transferred block leaves a real file object.""" + + def __init__(self, data, errors, seekable=True): + super().__init__(data) + self.errors = list(errors) + self._seekable = seekable + + def seekable(self): + return self._seekable + + def read(self, size=-1): + if self.errors: + super().read(7) + raise self.errors.pop(0) + return super().read(size) + + +def read_all(stream, chunk=64): + out = b"" + while True: + data = stream.read(chunk) + if not data: + return out + out += data + + +def test_read_resumes_at_the_byte_count_after_a_transient_error(capsys): + data = bytes(range(256)) * 4 + sleeps = [] + stream = CountingStream(FlakyStream(data, [FakeHTTPError(503), OSError("reset")]), label="x", sleep=sleeps.append) + + assert read_all(stream) == data # nothing duplicated, nothing lost + assert stream.tell() == len(data) + assert len(sleeps) == 2 + err = capsys.readouterr().err + assert f"warning: x: attempt 1/{FETCH_DEFAULT_RETRIES + 1} failed (HTTP 503)" in err + assert f"attempt 2/{FETCH_DEFAULT_RETRIES + 1} failed (reset)" in err + + +def test_a_pipe_cannot_be_rewound_so_it_is_not_retried(): + sleeps = [] + stream = CountingStream(FlakyStream(b"abc", [FakeHTTPError(503)], seekable=False), sleep=sleeps.append) + with pytest.raises(FakeHTTPError): + stream.read(3) + assert sleeps == [] + + +def test_deterministic_errors_are_raised_at_once(): + sleeps = [] + stream = CountingStream(FlakyStream(b"abc", [PermissionError("403")]), sleep=sleeps.append) + with pytest.raises(PermissionError): + stream.read(3) + assert sleeps == [] + + +def test_gives_up_after_the_configured_retries(): + sleeps = [] + stream = CountingStream(FlakyStream(b"abc", [FakeHTTPError(503)] * 5), retries=3, sleep=sleeps.append) + with pytest.raises(FakeHTTPError): + stream.read(3) + assert len(sleeps) == 3 + + +@pytest.mark.parametrize( + "exc, expected", + [ + (FakeHTTPError(503), True), + (FakeHTTPError(429), True), + (FakeHTTPError(416), False), + (FileNotFoundError("404"), False), + (PermissionError("403"), False), + (ConnectionResetError(), True), + (TimeoutError(), True), + ], +) +def test_is_retryable(exc, expected): + assert is_retryable(exc) is expected + + +# --- a conversion over a server that throttles and drops connections ---------------------- + + +def build_warc(path): + with open(path, "wb") as fh: + writer = WARCWriter(fh, gzip=True) + writer.write_record(writer.create_warcinfo_record("flaky.warc.gz", {"software": "warc2zip-tests"})) + for i in range(3): + body = f"page {i}".encode() * 40 + record = writer.create_warc_record( + f"http://example.com/{i}", + "response", + payload=io.BytesIO(body), + length=len(body), + http_headers=StatusAndHeaders("200 OK", [("Content-Type", "text/html")], protocol="HTTP/1.1"), + ) + writer.write_record(record) + + +class _FaultyRangeHandler(BaseHTTPRequestHandler): + """Serves `data` with Range support. Each ranged GET consumes one entry of `faults`: + "503" answers with 503, "drop" sends the headers and half the body then closes.""" + + data = b"" + faults = [] + ranged_gets = [] + + def log_message(self, *args): + pass + + def _range(self): + header = self.headers.get("Range") + if not header: + return 0, len(self.data) - 1 + first, _, last = header[len("bytes=") :].partition("-") + return int(first), min(int(last), len(self.data) - 1) if last else len(self.data) - 1 + + def do_HEAD(self): + self.send_response(200) + self.send_header("Content-Length", str(len(self.data))) + self.send_header("Accept-Ranges", "bytes") + self.end_headers() + + def do_GET(self): + start, end = self._range() + if start >= len(self.data): + self.send_response(416) + self.send_header("Content-Range", f"bytes */{len(self.data)}") + self.send_header("Content-Length", "0") + self.end_headers() + return + fault = None + if self.headers.get("Range"): + fault = self.faults.pop(0) if self.faults else None + self.ranged_gets.append(fault) + if fault == "503": + self.send_response(503) + self.send_header("Content-Length", "0") + self.end_headers() + return + body = self.data[start : end + 1] + self.send_response(206) + self.send_header("Content-Range", f"bytes {start}-{end}/{len(self.data)}") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if fault == "drop": + self.wfile.write(body[: len(body) // 2]) + self.wfile.flush() + self.connection.close() + return + self.wfile.write(body) + + +@pytest.fixture +def faulty_server(tmp_path): + warc = tmp_path / "flaky.warc.gz" + build_warc(warc) + handler = type("Handler", (_FaultyRangeHandler,), {"data": warc.read_bytes(), "faults": [], "ranged_gets": []}) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + handler.url = f"http://127.0.0.1:{server.server_port}/flaky.warc.gz" + handler.local = warc + try: + yield handler + finally: + server.shutdown() + server.server_close() + + +def payloads(zip_path): + with zipfile.ZipFile(zip_path) as zf: + return sorted(zf.read(n) for n in zf.namelist() if PAYLOAD_RE.match(n.rsplit("/", 1)[-1])) + + +def test_conversion_survives_a_503_and_a_dropped_connection(faulty_server, tmp_path, monkeypatch, capsys): + faulty_server.faults[:] = ["503", "drop"] + sleeps = [] + monkeypatch.setattr(warc2zip, "CountingStream", functools.partial(CountingStream, sleep=sleeps.append)) + + assert main(faulty_server.url, str(tmp_path / "remote.zip")) == 0 + assert main(str(faulty_server.local), str(tmp_path / "local.zip")) == 0 + + assert faulty_server.ranged_gets[:3] == ["503", "drop", None] + assert len(sleeps) == 2 + err = capsys.readouterr().err + assert err.count("retrying in") == 2 + assert payloads(tmp_path / "remote.zip") == payloads(tmp_path / "local.zip") + assert len(payloads(tmp_path / "remote.zip")) == 3 diff --git a/warc2zip.py b/warc2zip.py index b4cb580..62cdc31 100644 --- a/warc2zip.py +++ b/warc2zip.py @@ -1,12 +1,15 @@ import argparse +import asyncio import csv import io import json import mimetypes import posixpath +import random import re import secrets import sys +import time import zipfile from collections import Counter from datetime import datetime, timezone @@ -55,26 +58,120 @@ EXTRACTED_RECORD_TYPES = ("warcinfo", "response", "revisit", "request", "metadata") +# Transient-failure policy for reading a remote input. Names match the --fetch helpers so the +# two definitions merge into one. +FETCH_DEFAULT_RETRIES = 8 # retries after the first attempt +FETCH_MAX_BACKOFF = 60.0 # seconds; the cap on one exponential-backoff wait +FETCH_RETRYABLE_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504}) + + +def error_status(exc): + """HTTP status carried by an exception (aiohttp's ClientResponseError), else None.""" + status = getattr(exc, "status", None) + return status if isinstance(status, int) else None + + +def retry_after_seconds(exc): + """The Retry-After delay the server asked for, in seconds; only the delta-seconds form.""" + headers = getattr(exc, "headers", None) + if not headers: + return None + try: + return max(0.0, float(headers.get("Retry-After"))) + except (TypeError, ValueError): + return None + + +def is_retryable(exc): + """Transient failures only: throttling, server errors, timeouts, dropped connections. + + A 404/403 (fsspec raises FileNotFoundError / PermissionError) and any other 4xx are + deterministic and re-raised at once. aiohttp's non-OSError exceptions (payload/disconnect) + are recognised by module, so aiohttp is not imported here. + """ + status = error_status(exc) + if status is not None: + return status in FETCH_RETRYABLE_STATUSES + if isinstance(exc, (FileNotFoundError, PermissionError)): + return False + if isinstance(exc, (OSError, TimeoutError, asyncio.TimeoutError)): + return True + return type(exc).__module__.split(".")[0] == "aiohttp" + + +def fetch_with_retry(fetch, *, retries=FETCH_DEFAULT_RETRIES, label="", sleep=time.sleep, rng=random.random): + """Call the zero-argument `fetch` until it returns, retrying transient failures. + + Waits Retry-After when the server sends one, otherwise an exponential backoff (2^attempt + seconds, capped at FETCH_MAX_BACKOFF) with jitter in [0.5, 1.5). One stderr line per retry. + Re-raises on a non-retryable error or once `retries` retries are used up. + """ + attempts = max(1, retries + 1) + for attempt in range(1, attempts + 1): + try: + return fetch() + except Exception as exc: + if attempt == attempts or not is_retryable(exc): + raise + delay = retry_after_seconds(exc) + if delay is None: + delay = min(FETCH_MAX_BACKOFF, 2.0**attempt) * (0.5 + rng()) + print( + f"warning: {label}: attempt {attempt}/{attempts} failed ({exc}), retrying in {delay:.1f} s", + file=sys.stderr, + ) + sleep(delay) + + class CountingStream(io.IOBase): - def __init__(self, raw_stream): + """Sequential read-through wrapper: counts the bytes handed out and retries transient failures. + + tell() is the count, which is what the progress bar reads (sys.stdin.buffer has a tell() + that crashes on a pipe). A read that fails with a transient error — throttling, a 5xx, a + connection dropped mid-block — is retried after seeking the underlying stream back to the + count, so the caller sees one contiguous byte stream and never a duplicated or missing + block. A stream that cannot seek (a pipe) gets no retry: there is nothing to rewind to, + so the error propagates as before. + """ + + def __init__(self, raw_stream, *, label="", retries=FETCH_DEFAULT_RETRIES, sleep=time.sleep): self._stream = raw_stream self._bytes_read = 0 + self._label = label + self._retries = retries + self._sleep = sleep + try: + self._seekable = bool(raw_stream.seekable()) + except (AttributeError, OSError, ValueError): + self._seekable = False def tell(self): """Acts as the progress tracker for progress bar libraries.""" return self._bytes_read - def read(self, size=-1): - data = self._stream.read(size) + def _read_with_retry(self, method, size): + attempts = 0 + + def once(): + nonlocal attempts + if attempts: # a failed attempt may have moved the underlying position + self._stream.seek(self._bytes_read) + attempts += 1 + return method(size) + + if self._seekable: + data = fetch_with_retry(once, retries=self._retries, label=self._label, sleep=self._sleep) + else: + data = once() if data: self._bytes_read += len(data) return data + def read(self, size=-1): + return self._read_with_retry(self._stream.read, size) + def readline(self, size=-1): - data = self._stream.readline(size) - if data: - self._bytes_read += len(data) - return data + return self._read_with_retry(self._stream.readline, size) # Forward other essential methods to the underlying stream def readable(self): @@ -710,9 +807,7 @@ def main(input_file, output_path, dry_run=False, limit=None, output_format="flat limit_reached = False with fsspec_open(input_file, "rb", default_fh=sys.stdin.buffer) as stream: - if not hasattr(stream, "tell") or stream == sys.stdin.buffer: - # sys.stdin.buffer has a tell() method but it crashes - stream = CountingStream(stream) + stream = CountingStream(stream, label=input_file) with tqdm(total=file_size, unit="B", unit_scale=True, desc="Scanning") as pbar: for record in open_archive_iterator(stream): if limit_reached and record.rec_type in ("response", "revisit"): @@ -767,7 +862,7 @@ def main(input_file, output_path, dry_run=False, limit=None, output_format="flat # Count reads ourselves so an early EOF is not mistaken for a complete archive. # if we ever decide to add partial-range downloads, this needs to account for that expected_size = getattr(raw_stream, "size", None) or file_size - stream = CountingStream(raw_stream) + stream = CountingStream(raw_stream, label=input_file) pbar = tqdm(total=file_size, unit="B", unit_scale=True, desc="Reading WARC") # Held by name rather than iterated anonymously: get_record_offset() / # get_record_length() hang off the iterator, not the record.