Fail only the faulted file when an HTTP transfer breaks, and port the chaos tests

.github/workflows/nightly.yml

@@ -0,0 +1,85 @@
+# Runs the chaos tests with many more random operations and fresh seeds.
+
+name: Nightly
+
+on:
+  schedule:
+  - cron: "23 6 * * *"
+  workflow_dispatch:
+    inputs:
+      steps:
+        description: Random operations per seed
+        default: "1000"
+      seeds:
+        description: Comma-separated seeds (blank derives them from the run)
+        default: ""
+  # Scheduled and manual runs start only from the default branch.
+  pull_request:
+    paths:
+    - .github/workflows/nightly.yml
+    - test/chaos.py
+    - test/chaos_child.py
+    - test/http_server.py
+    - test/survey.py
+    - test/test_chaos_*.py
+
+permissions:
+  contents: read
+
+concurrency:
+  group: nightly-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  chaos:
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 45
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+        - os: ubuntu-24.04
+          python-version: "3.9"
+          lane: 1
+        - os: ubuntu-24.04
+          python-version: "3.14"
+          lane: 2
+        - os: macos-latest
+          python-version: "3.14"
+          lane: 3
+    steps:
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@v7
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install Hallmark and test tools
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -e . --group dev
+    - name: Choose seeds and steps
+      env:
+        INPUT_SEEDS: ${{ inputs.seeds }}
+        INPUT_STEPS: ${{ inputs.steps }}
+        RUN_ID: ${{ github.run_id }}
+        LANE: ${{ matrix.lane }}
+        EVENT: ${{ github.event_name }}
+      run: |
+        seeds="${INPUT_SEEDS:-${RUN_ID}${LANE}1,${RUN_ID}${LANE}2,${RUN_ID}${LANE}3}"
+        if [ "$EVENT" = pull_request ]; then default_steps=300; else default_steps=1000; fi
+        steps="${INPUT_STEPS:-$default_steps}"
+        [[ "$seeds" =~ ^[0-9,]+$ && "$steps" =~ ^[0-9]+$ ]] || { echo "Invalid seeds or steps"; exit 1; }
+        echo "HALLMARK_CHAOS_SEED=$seeds" >> "$GITHUB_ENV"
+        echo "HALLMARK_CHAOS_STEPS=$steps" >> "$GITHUB_ENV"
+        echo "Chaos seeds: \`$seeds\`, steps per seed: $steps" >> "$GITHUB_STEP_SUMMARY"
+    - name: Run the chaos tests
+      run: python -m pytest -m chaos -ra --timeout=1800 --basetemp="$RUNNER_TEMP/pytest"
+    - name: Keep failed repositories for inspection
+      if: failure()
+      uses: actions/upload-artifact@v7
+      with:
+        name: chaos-${{ runner.os }}-py${{ matrix.python-version }}
+        path: ${{ runner.temp }}/pytest
+        include-hidden-files: true
+        retention-days: 14

mod/hallmark/cli.py

@@ -19,9 +19,11 @@
 from pathlib import Path
 
 import click
+import requests
 import yaml
 from click import ClickException
 from git.exc import GitError
+from urllib3.exceptions import HTTPError as Urllib3Error
 
 from . import Repo
 from .fmt_detection import detect_fmt
@@ -175,7 +177,8 @@ def _run_download(repo, plan, *, max_workers):
 # Errors a user causes or can fix, such as a typo in config.yml, a missing
 # file, a full disk or a git failure. Every command reports them in one line;
 # anything else is a bug and keeps its traceback.
-_USER_ERRORS = (HallmarkError, GitError, OSError, yaml.YAMLError, ValueError)
+_USER_ERRORS = (HallmarkError, GitError, OSError, yaml.YAMLError, ValueError,
+                requests.RequestException, Urllib3Error)
 
 
 class _HallmarkGroup(click.Group):

mod/hallmark/transport/http.py

@@ -14,6 +14,9 @@
 
 import requests
 from requests.auth import AuthBase
+from urllib3.exceptions import (HTTPError as Urllib3Error, ProtocolError,
+                                ReadTimeoutError)
+from urllib3.response import HTTPResponse as Urllib3Response
 
 from ..utils import REMOTE_REQUEST_TIMEOUT
 from .base import (DownloadError, RemoteConfigurationError, RemoteEntry,
@@ -23,6 +26,12 @@
 from .index import _parse_directory_listing, _redirected_directory
 
 
+def _refuse_short_body(response):
+    """Make urllib3 1.x refuse a body shorter than its Content-Length, as 2.x does."""
+    if isinstance(response.raw, Urllib3Response):
+        response.raw.enforce_content_length = True
+    return response
+
 
 def _declared_charset(content_type):
     """
@@ -142,20 +151,27 @@ def _open_file_response(self, path):
         Asking for the identity encoding keeps servers from compressing a file
         on the fly, so its checksums still match.
         """
-        return self.context.session().get(
+        return _refuse_short_body(self.context.session().get(
             self.context.remote.file_url(path),
             stream=True,
             timeout=REMOTE_REQUEST_TIMEOUT,
             headers={"Accept-Encoding": "identity"},
             **self._auth_option(),
-        )
+        ))
 
     def _download_error(self, exc):
-        # Requests exceptions can embed credentials in their URL or response body.
         """Describe an HTTP failure without exposing response contents."""
+        # Requests exceptions can embed credentials in their URL or response body.
         status = getattr(getattr(exc, "response", None), "status_code", None)
         cls = RemoteObjectMissing if status == 404 else DownloadError
-        detail = f"HTTP {status}" if isinstance(status, int) else type(exc).__name__
+        if isinstance(status, int):
+            detail = f"HTTP {status}"
+        elif isinstance(exc, ReadTimeoutError):
+            detail = "the connection stalled"
+        elif isinstance(exc, ProtocolError):
+            detail = "the connection was reset or the body was truncated"
+        else:
+            detail = type(exc).__name__
         if status in {401, 403}:
             detail += (f"; check auth profile {self.context.remote.auth!r}"
                        if self.auth is not None else
@@ -202,7 +218,7 @@ def endpoint(parsed):
                 response = self.context.session().request(
                     method, url, headers=headers, data=data, **options)
             if response.status_code not in {301, 302, 303, 307, 308}:
-                return response, url
+                return _refuse_short_body(response), url
             location = response.headers.get("Location")
             response.close()
             if not isinstance(location, str) or not location:
@@ -224,8 +240,11 @@ def fetch(self, relative_path, destination, *, chunk_size=8192):
                             handle.write(chunk)
                             if self.context.on_bytes is not None:
                                 self.context.on_bytes(len(chunk))
-        except requests.RequestException as exc:
+        # Reading response.raw directly bypasses Requests' error wrapping, so
+        # a reset, truncated or stalled transfer arrives as a urllib3 error.
+        except (requests.RequestException, Urllib3Error) as exc:
             raise self._download_error(exc) from None
+
     def read_text(self, relative_path, limit):
         """Read metadata within the source root and byte limit."""
         content = bytearray()

pyproject.toml

@@ -69,6 +69,9 @@ addopts = "--strict-markers"
 markers = [
     "ssh_client: exercise the actual local OpenSSH SFTP command parser",
     "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)",
+    "chaos: fault injection, crash consistency and seeded random operations",
+    "survey: end-to-end survey workflows against loopback HTTP and SSH servers",
+    "survey_live: tiny downloads from public archives (HALLMARK_RUN_LIVE_TESTS=1 required)",
 ]
 
 [tool.coverage.run]

test/chaos.py

@@ -0,0 +1,189 @@
+"""Helpers for fault-injection ("chaos monkey") tests.
+
+``check_invariants`` describes what must hold for a repository after any
+operation, failed or not. Problems are grouped by severity:
+
+* corruption: the repository cannot be opened, an object does not match
+  its checksum, a committed file has no object, or Git reports damage;
+* debris: temporary files that a finished operation left behind.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import errno
+import hashlib
+from io import StringIO
+import os
+import pathlib
+from pathlib import Path
+import subprocess
+
+import pandas as pd
+import pytest
+
+from hallmark import Repo
+
+STATE_FILES = ("config.yml", "meta.yml", "data.tsv")
+DEBRIS_PATTERNS = ("**/.*.part", "**/.*.tmp", "**/.hallmark-checkout-*",
+                   "**/objects/tmp/*")
+
+
+@dataclass
+class Violations:
+    """Invariant violations found in a repository."""
+    corruption: list = field(default_factory=list)
+    debris: list = field(default_factory=list)
+
+    def __bool__(self):
+        return bool(self.corruption or self.debris)
+
+    def __str__(self):
+        lines = [f"corruption: {item}" for item in self.corruption]
+        lines += [f"debris: {item}" for item in self.debris]
+        return "\n".join(lines) or "no violations"
+
+
+def skip_if_root():
+    """Skip permission tests when running as root, which ignores modes."""
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        pytest.skip("root ignores file permissions")
+
+
+def fail_nth_replace(monkeypatch, nth, *, code=errno.ENOSPC, match=lambda target: True):
+    """Make the ``nth`` matching ``Path.replace`` call raise ``OSError(code)``.
+
+    Patch the method on the class: on Python 3.9 and 3.10, pathlib calls a
+    reference to ``os.replace`` captured at import, so patching ``os`` misses it.
+
+    Returns:
+        list: Targets of the matching calls, in order, for assertions.
+    """
+    original = pathlib.Path.replace
+    calls = []
+
+    def replace(self, target):
+        if match(Path(target)):
+            calls.append(Path(target))
+            if len(calls) == nth:
+                raise OSError(code, os.strerror(code), str(target))
+        return original(self, target)
+
+    monkeypatch.setattr(pathlib.Path, "replace", replace)
+    return calls
+
+
+class _FailingWriter:
+    """File proxy whose writes fail once ``limit`` bytes have been written."""
+
+    def __init__(self, handle, limit, code):
+        self._handle, self._limit, self._code = handle, limit, code
+        self._written = 0
+
+    def write(self, data):
+        if self._written + len(data) > self._limit:
+            raise OSError(self._code, os.strerror(self._code))
+        self._written += len(data)
+        return self._handle.write(data)
+
+    def __getattr__(self, name):
+        return getattr(self._handle, name)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return self._handle.__exit__(*exc)
+
+
+def fail_writes_after(monkeypatch, limit, *, code=errno.ENOSPC,
+                      match=lambda path: True):
+    """Make binary writes to matching paths fail after ``limit`` bytes."""
+    original = pathlib.Path.open
+
+    def open_(self, mode="r", *args, **kwargs):
+        handle = original(self, mode, *args, **kwargs)
+        if "w" in mode and "b" in mode and match(self):
+            return _FailingWriter(handle, limit, code)
+        return handle
+
+    monkeypatch.setattr(pathlib.Path, "open", open_)
+
+
+def _git(dothm, *args):
+    return subprocess.run(["git", "-C", str(dothm), *args], check=True,
+                          capture_output=True, text=True).stdout
+
+
+def snapshot_hm(repo_path):
+    """Capture a repository's tracked state: state files, index and refs."""
+    repo_path = Path(repo_path)
+    dothm = repo_path if repo_path.name.endswith(".hm") else repo_path / ".hm"
+    return {
+        "files": {name: (dothm / name).read_bytes() for name in STATE_FILES
+                  if (dothm / name).is_file()},
+        "index": _git(dothm, "ls-files", "--stage"),
+        "refs": _git(dothm, "for-each-ref", "--format=%(refname) %(objectname)"),
+        "head": subprocess.run(["git", "-C", str(dothm), "symbolic-ref", "-q", "HEAD"],
+                               capture_output=True, text=True).stdout.strip()
+        or _git(dothm, "rev-parse", "HEAD").strip(),
+    }
+
+
+def check_invariants(repo_path, *, fsck=True):
+    """Check a repository for corruption and leftover temporary files.
+
+    Args:
+        repo_path: Worktree or bare ``.hm`` path.
+        fsck (bool): Also run ``git fsck``, which is slower.
+
+    Returns:
+        Violations: Empty (falsy) when every invariant holds.
+    """
+    found = Violations()
+    try:
+        repo = Repo(repo_path)
+        repo.status()
+    except Exception as exc:
+        found.corruption.append(
+            f"repository does not open: {type(exc).__name__}: {exc}")
+        return found
+
+    objects = repo.objects.root
+    if objects.is_dir():
+        for path in objects.glob("??/*"):
+            if path.name.startswith("."):
+                found.debris.append(str(path))
+                continue
+            digest = hashlib.sha1(path.read_bytes()).hexdigest()
+            if digest != path.parent.name + path.name:
+                found.corruption.append(f"object {path} has checksum {digest}")
+
+    dothm = Path(repo.dothm.path)
+    for head in repo.dothm.heads:
+        try:
+            text = _git(dothm, "show", f"{head.name}:data.tsv")
+        except subprocess.CalledProcessError:
+            continue
+        frame = pd.read_csv(StringIO(text), sep="\t", dtype=str,
+                            keep_default_na=False)
+        if "sha1" in frame.columns:
+            for sha1 in frame.sha1:
+                if sha1 and not repo.objects.contains(sha1):
+                    found.corruption.append(
+                        f"branch {head.name} references missing object {sha1}")
+
+    if fsck:
+        result = subprocess.run(["git", "-C", str(dothm), "fsck", "--no-dangling"],
+                                capture_output=True, text=True)
+        if result.returncode != 0:
+            found.corruption.append(f"git fsck: {result.stdout}{result.stderr}")
+
+    roots = {Path(repo_path)}
+    if repo.worktree is not None:
+        roots.add(Path(repo.worktree))
+    for root in roots:
+        for pattern in DEBRIS_PATTERNS:
+            found.debris.extend(str(path) for path in root.glob(pattern))
+    found.debris = sorted(set(found.debris))
+    return found