crawler: bridge to ingest + fast recrawl-check via conditional HEAD

Glues the verbatim crawler lift to aborist's existing ingest pipeline, plus adds a freshness probe that asks "does this need recrawling?" without paying for a body fetch when the answer is no. Bridge (aborist/sources/crawler/bridge.py): - crawl_seed(seed, depth, max_pages): BFS-discovers same-domain URLs via AsyncWebFetcher. Exact-hostname filter so subdomain crossover is off by default. Caps at max_pages. URL dedup, fragment-strip canonicalization. Sync wrapper around an async core so the CLI remains a one-liner. - ingest_crawled(conn, urls): runs the URL list through the standard ingest_source() path so chunking_version / canonicalization_version / Merkle commitment all stay consistent. Captures ETag + Last-Modified per response and writes one row per document_root into the new document_http_meta table. Schema (idempotent migration, mirrors _migrate_mesh_peer_chains): document_http_meta ( document_root TEXT PRIMARY KEY, etag TEXT, last_modified TEXT, last_fetched_at INTEGER NOT NULL, last_status INTEGER, last_checked_at INTEGER, FK(document_root) -> documents(document_root) ON DELETE CASCADE ) Recrawl-check (recrawl_check): - Sends conditional HEAD with If-None-Match + If-Modified-Since. 304 -> fresh (no body transfer needed). 200 -> stale (server says the body changed). 404/410 -> gone. Other -> unreachable. - Updates last_status + last_checked_at so consecutive runs visit oldest-first (NULLS FIRST in ORDER BY). - Optional --domain filter so operators can probe one site at a time. CLI: aborist crawl --seed-url URL [--depth 2] [--max-pages 20] [--ingest] aborist crawler recrawl-check [--domain DOMAIN] [--limit 100] Makefile: crawl-ingest URL=https://example.com [DEPTH=2 MAX=20] recrawl-check [DOMAIN=example.com LIMIT=100] Both targets gate on bootstrap-crawler so [crawler] extras land first. Tests: 17 new in tests/crawler/test_bridge.py — same-domain filter, BFS bounds (depth + max_pages), dedup, fragment normalization, ETag capture, content-type filter, HEAD classification (304/200/404), domain filter, last_checked_at update. All stub the network via a fake httpx.Client; default suite still 273 passed, 1 skipped. End-to-end test path: make bootstrap-crawler make crawl-ingest URL=https://russell.ballestrini.net DEPTH=2 MAX=20 make query Q="who is russell ballestrini?" make recrawl-check DOMAIN=russell.ballestrini.net
2026-04-28 20:34:54 -04:00 · 2026-04-28 20:34:54 -04:00 · 5f060ff326
commit 5f060ff326
parent a5e709a470
5 changed files with 885 additions and 1 deletions
--- a/tests/crawler/test_bridge.py
+++ b/tests/crawler/test_bridge.py
@ -0,0 +1,387 @@
+"""Bridge tests: crawler discovery + ingest + recrawl-check.
+
+These tests stub the network. The verbatim crawler tests in this same
+directory exercise the underlying AsyncWebFetcher; here we exercise
+the aborist bridge logic on top of it (BFS bounds, same-domain filter,
+ETag capture, conditional HEAD classification).
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from aborist.sources.crawler.bridge import (
+    _normalize,
+    _same_domain,
+    crawl_seed,
+    ingest_crawled,
+    recrawl_check,
+)
+from aborist.store import connect
+
+
+# ---------------------------------------------------------------------------
+# URL helpers (pure)
+# ---------------------------------------------------------------------------
+
+
+def test_same_domain_exact_hostname():
+    assert _same_domain("https://x.com/a", "https://x.com/b") is True
+
+
+def test_same_domain_rejects_subdomain():
+    assert _same_domain("https://x.com/a", "https://docs.x.com/b") is False
+
+
+def test_same_domain_rejects_other_host():
+    assert _same_domain("https://x.com/a", "https://y.com/a") is False
+
+
+def test_normalize_strips_fragment():
+    assert _normalize("https://x.com/a#section") == "https://x.com/a"
+
+
+def test_normalize_preserves_query():
+    assert _normalize("https://x.com/a?q=1") == "https://x.com/a?q=1"
+
+
+# ---------------------------------------------------------------------------
+# crawl_seed: BFS discovery (no real network — mock the fetcher)
+# ---------------------------------------------------------------------------
+
+
+class _StubFetcher:
+    """Stand-in for AsyncWebFetcher that returns pre-canned (html, links)
+    based on a URL→links map. Enough to drive the bridge's BFS.
+    """
+
+    def __init__(self, page_links: dict[str, list[str]]):
+        self.page_links = page_links
+
+    async def fetch_webpage(self, url: str):
+        return "<html></html>", self.page_links.get(url, [])
+
+
+def _run_crawl(page_links, seed, depth, max_pages):
+    """Helper: drive _crawl_seed_async with a stub fetcher."""
+    import asyncio
+    from aborist.sources.crawler import bridge
+
+    return asyncio.run(
+        bridge._crawl_seed_async(
+            seed, max_depth=depth, max_pages=max_pages, fetcher=_StubFetcher(page_links)
+        )
+    )
+
+
+def test_crawl_seed_breadth_first_within_limit():
+    page_links = {
+        "https://x.com/": ["/a", "/b"],
+        "https://x.com/a": ["/c"],
+        "https://x.com/b": [],
+        "https://x.com/c": [],
+    }
+    out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10)
+    # Seed first, then both depth-1 pages, then the depth-2 page.
+    assert out[0] == "https://x.com/"
+    assert "https://x.com/a" in out
+    assert "https://x.com/b" in out
+    assert "https://x.com/c" in out
+
+
+def test_crawl_seed_respects_max_pages():
+    page_links = {f"https://x.com/p{i}": [f"/p{i+1}"] for i in range(50)}
+    page_links["https://x.com/"] = ["/p0"]
+    out = _run_crawl(page_links, "https://x.com/", depth=10, max_pages=5)
+    assert len(out) == 5
+
+
+def test_crawl_seed_respects_max_depth():
+    page_links = {
+        "https://x.com/": ["/a"],
+        "https://x.com/a": ["/b"],
+        "https://x.com/b": ["/c"],
+        "https://x.com/c": [],
+    }
+    out = _run_crawl(page_links, "https://x.com/", depth=1, max_pages=10)
+    # Seed (depth 0) + /a (depth 1) only. /b is depth 2, skipped.
+    assert "https://x.com/" in out
+    assert "https://x.com/a" in out
+    assert "https://x.com/b" not in out
+
+
+def test_crawl_seed_drops_off_domain_links():
+    page_links = {
+        "https://x.com/": ["https://other.com/foo", "/local"],
+        "https://x.com/local": [],
+    }
+    out = _run_crawl(page_links, "https://x.com/", depth=2, max_pages=10)
+    assert "https://other.com/foo" not in out
+    assert "https://x.com/local" in out
+
+
+def test_crawl_seed_dedups():
+    page_links = {
+        "https://x.com/": ["/a", "/a", "/b"],
+        "https://x.com/a": ["/b"],
+        "https://x.com/b": ["/a"],
+    }
+    out = _run_crawl(page_links, "https://x.com/", depth=3, max_pages=10)
+    # Each URL appears at most once even with cycles.
+    assert len(out) == len(set(out))
+
+
+# ---------------------------------------------------------------------------
+# ingest_crawled: stores ETag + Last-Modified per document_root
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def http_meta_db(tmp_path):
+    db = tmp_path / "meta.db"
+    # Touch the DB so the schema is initialised, then close — we'll
+    # reopen as the test needs.
+    conn = connect(db)
+    try:
+        pass
+    finally:
+        conn.close()
+    return db
+
+
+def _patched_httpx_get(url_to_response: dict[str, dict]):
+    """Return a context-manager class that fakes httpx.Client.get."""
+
+    class _FakeClient:
+        def __init__(self, *a, **kw):
+            self._responses = url_to_response
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            pass
+
+        def get(self, url):
+            spec = self._responses.get(url)
+            if spec is None:
+                raise httpx.HTTPError(f"no stub for {url}")
+            req = httpx.Request("GET", url)
+            return httpx.Response(
+                spec.get("status", 200),
+                headers=spec.get("headers", {"content-type": "text/html"}),
+                content=spec.get("body", "<html><body>hi</body></html>").encode(),
+                request=req,
+            )
+
+        def head(self, url, headers=None):
+            spec = self._responses.get(url)
+            if spec is None:
+                raise httpx.HTTPError(f"no stub for {url}")
+            head_status = spec.get("head_status", 200)
+            req = httpx.Request("HEAD", url, headers=headers or {})
+            return httpx.Response(
+                head_status,
+                headers=spec.get("head_headers", {}),
+                request=req,
+            )
+
+    return _FakeClient
+
+
+def test_ingest_crawled_writes_http_meta(http_meta_db):
+    url = "https://x.com/page"
+    responses = {
+        url: {
+            "status": 200,
+            "headers": {
+                "content-type": "text/html",
+                "etag": '"abc123"',
+                "last-modified": "Tue, 28 Apr 2026 00:00:00 GMT",
+            },
+            "body": "<html><body><h1>Title</h1>Hello world.</body></html>",
+        }
+    }
+
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(http_meta_db)
+        try:
+            result = ingest_crawled(conn, [url])
+        finally:
+            conn.close()
+
+    assert result["http_meta_written"] == 1
+    assert len(result["documents"]) == 1
+    document_root = result["documents"][0]["document_root"]
+
+    conn = connect(http_meta_db)
+    try:
+        row = conn.execute(
+            "SELECT etag, last_modified, last_status, last_fetched_at, last_checked_at "
+            "FROM document_http_meta WHERE document_root = ?",
+            (document_root,),
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row["etag"] == '"abc123"'
+    assert row["last_modified"] == "Tue, 28 Apr 2026 00:00:00 GMT"
+    assert row["last_status"] == 200
+    assert row["last_fetched_at"] > 0
+
+
+def test_ingest_crawled_skips_non_html(http_meta_db):
+    url = "https://x.com/file.pdf"
+    responses = {
+        url: {
+            "status": 200,
+            "headers": {"content-type": "application/pdf"},
+            "body": "%PDF-1.4 binary",
+        }
+    }
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(http_meta_db)
+        try:
+            result = ingest_crawled(conn, [url])
+        finally:
+            conn.close()
+    assert result["http_meta_written"] == 0
+    assert result["inserted"] == 0
+
+
+# ---------------------------------------------------------------------------
+# recrawl_check: conditional HEAD classification
+# ---------------------------------------------------------------------------
+
+
+def _seed_meta(conn, *, document_root: str, document_uri: str, etag: str | None = None):
+    """Insert a documents row + http_meta row for recheck tests."""
+    from aborist.store import transaction
+
+    with transaction(conn):
+        conn.execute(
+            "INSERT INTO documents "
+            "(document_root, document_uri, source_type, kind, compression_depth, "
+            " title, chunking_version, canonicalization_version, schema_version, "
+            " ingest_ts, hit_count) "
+            "VALUES (?, ?, 'html', 'surface', 0, NULL, 'tok-512-v1', 'norm-v1', 'v9.8.0', ?, 0)",
+            (document_root, document_uri, int(time.time())),
+        )
+        conn.execute(
+            "INSERT INTO document_http_meta "
+            "(document_root, etag, last_modified, last_fetched_at, last_status, last_checked_at) "
+            "VALUES (?, ?, NULL, ?, 200, NULL)",
+            (document_root, etag, int(time.time())),
+        )
+
+
+def test_recrawl_check_classifies_304_as_fresh(tmp_path):
+    db = tmp_path / "rc.db"
+    conn = connect(db)
+    try:
+        _seed_meta(conn, document_root="aa" * 32, document_uri="https://x.com/p", etag='"v1"')
+    finally:
+        conn.close()
+
+    responses = {"https://x.com/p": {"head_status": 304}}
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(db)
+        try:
+            result = recrawl_check(conn, limit=10)
+        finally:
+            conn.close()
+    assert result["fresh"] == 1
+    assert result["stale"] == 0
+    assert result["items"][0]["verdict"] == "fresh"
+
+
+def test_recrawl_check_classifies_200_as_stale(tmp_path):
+    db = tmp_path / "rc.db"
+    conn = connect(db)
+    try:
+        _seed_meta(conn, document_root="bb" * 32, document_uri="https://x.com/q", etag='"v1"')
+    finally:
+        conn.close()
+
+    responses = {"https://x.com/q": {"head_status": 200}}
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(db)
+        try:
+            result = recrawl_check(conn, limit=10)
+        finally:
+            conn.close()
+    assert result["stale"] == 1
+    assert result["items"][0]["verdict"] == "stale"
+
+
+def test_recrawl_check_classifies_404_as_gone(tmp_path):
+    db = tmp_path / "rc.db"
+    conn = connect(db)
+    try:
+        _seed_meta(conn, document_root="cc" * 32, document_uri="https://x.com/r", etag='"v1"')
+    finally:
+        conn.close()
+
+    responses = {"https://x.com/r": {"head_status": 404}}
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(db)
+        try:
+            result = recrawl_check(conn, limit=10)
+        finally:
+            conn.close()
+    assert result["gone"] == 1
+
+
+def test_recrawl_check_domain_filter(tmp_path):
+    db = tmp_path / "rc.db"
+    conn = connect(db)
+    try:
+        _seed_meta(conn, document_root="11" * 32, document_uri="https://keep.com/a", etag='"v"')
+        _seed_meta(conn, document_root="22" * 32, document_uri="https://drop.com/a", etag='"v"')
+    finally:
+        conn.close()
+
+    responses = {
+        "https://keep.com/a": {"head_status": 304},
+        # drop.com would error if the filter is broken; we expect it to be skipped.
+    }
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(db)
+        try:
+            result = recrawl_check(conn, domain="keep.com", limit=10)
+        finally:
+            conn.close()
+    assert result["examined"] == 1
+    assert result["items"][0]["document_uri"] == "https://keep.com/a"
+
+
+def test_recrawl_check_updates_last_checked_at(tmp_path):
+    db = tmp_path / "rc.db"
+    conn = connect(db)
+    try:
+        _seed_meta(conn, document_root="33" * 32, document_uri="https://x.com/s", etag='"v1"')
+    finally:
+        conn.close()
+
+    responses = {"https://x.com/s": {"head_status": 304}}
+    with patch("aborist.sources.crawler.bridge.httpx.Client", _patched_httpx_get(responses)):
+        conn = connect(db)
+        try:
+            recrawl_check(conn, limit=10)
+        finally:
+            conn.close()
+
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT last_status, last_checked_at FROM document_http_meta WHERE document_root='33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33' || '33'"
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row["last_status"] == 304
+    assert row["last_checked_at"] is not None