daoudclarke · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026
diff --git a/mwmbl/crawl.py b/mwmbl/crawl.py
@@ -57,6 +57,8 @@
 )
 from mwmbl.crawler.politeness import abandon_domains, release_domains
 from mwmbl.crawler.retrieve import CRAWLER_VERSION, crawl_batch
+from mwmbl.indexer.blacklist_providers import CombinedBlacklistProvider
+from mwmbl.indexer.blacklist_snapshot import SnapshotBlacklist, refresh_snapshot
 from mwmbl.indexer.index_batches import index_batches, index_pages
 from mwmbl.indexer.update_urls import record_urls_in_database
 from mwmbl.rankeval.evaluation.remote_index import RemoteIndex
@@ -103,6 +105,9 @@
 # request's time - but not forever.
 RESULTS_POST_TIMEOUT_SECONDS = 120
 
+# Two downloads of up to 60 seconds each, then parsing and hashing.
+SNAPSHOT_PUBLISH_TIMEOUT_SECONDS = 600
+
 
 def _fetch_curated_domains() -> set[str]:
     global _curated_domains_cache, _curated_domains_fetched_at
@@ -304,6 +309,7 @@ def __init__(self, redis_url=None):
         self.redis_url = redis_url or settings.REDIS_URL
         self._redis = None
         self._url_queue = None
+        self._snapshot_blacklist = None
 
     @property
     def redis(self):
@@ -320,9 +326,44 @@ def redis(self):
     def url_queue(self):
         """Lazy initialization of URL queue."""
         if self._url_queue is None:
-            self._url_queue = RedisURLQueue(self.redis, _fetch_curated_domains)
+            self._url_queue = RedisURLQueue(self.redis, _fetch_curated_domains, self.new_blacklist_provider())
         return self._url_queue
 
+    @property
+    def snapshot_blacklist(self) -> SnapshotBlacklist:
+        """The remote lists as a hash array, ~11MB, where the sets of domains they come from take ~300MB.
+
+        Lazy so that each process loads its own after the fork. The client is not the one above
+        because the snapshot is binary and that one decodes responses as text.
+        """
+        if self._snapshot_blacklist is None:
+            self._snapshot_blacklist = SnapshotBlacklist(redis_client=Redis.from_url(self.redis_url))
+            self._snapshot_blacklist.load_now()
+        return self._snapshot_blacklist
+
+    def new_blacklist_provider(self) -> CombinedBlacklistProvider:
+        # A provider of its own because it holds the approved domains from when it first needs
+        # them, so one kept for the life of the process would never see a later approval.
+        return CombinedBlacklistProvider([self.snapshot_blacklist], get_exempt_domains=_fetch_curated_domains)
+
+    def publish_blacklist_snapshot(self):
+        refresh_snapshot(redis_client=Redis.from_url(self.redis_url))
+
+    def refresh_blacklist_snapshot(self):
+        """Download the blocklists and publish them for the workers, from a process that then exits.
+
+        Parsing them takes ~300MB, and exiting is what reliably gives that back to the OS.
+        """
+        process = Process(target=self.publish_blacklist_snapshot)
+        process.start()
+        process.join(SNAPSHOT_PUBLISH_TIMEOUT_SECONDS)
+        if process.is_alive():
+            process.kill()
+            process.join()
+            raise RuntimeError(f"Publishing the blacklist snapshot took over {SNAPSHOT_PUBLISH_TIMEOUT_SECONDS}s")
+        if process.exitcode != 0:
+            raise RuntimeError(f"Publishing the blacklist snapshot failed with exit code {process.exitcode}")
+
     def check_redis(self):
         """Check Redis connection health.
 
@@ -371,7 +412,7 @@ def process_batch(self):
                 "items": results,
             }
         )
-        record_urls_in_database([batch], self.url_queue)
+        record_urls_in_database([batch], self.url_queue, self.new_blacklist_provider())
 
         # Push the batch into the Redis queue
         batch_json = batch.json()
@@ -534,6 +575,7 @@ def run_once(self):
         """
         validate_environment()
         self.check_redis()
+        self.refresh_blacklist_snapshot()
         self.process_batch()
         self.run_indexing()
 
@@ -549,6 +591,9 @@ def run(self):
         workers: int = CRAWLER_WORKERS
         assert workers > 0, f"Invalid value for CRAWLER_WORKERS: {workers}"
 
+        self.refresh_blacklist_snapshot()
+        last_snapshot_refresh = time.monotonic()
+
         batch_processes: list[Process] = []
         for i in range(workers):
             process = Process(target=self.process_batch_continuously)
@@ -563,6 +608,14 @@ def run(self):
         index_crash_history: list[tuple[datetime, int, int]] = []
 
         while True:
+            if time.monotonic() - last_snapshot_refresh >= settings.BLACKLIST_SNAPSHOT_REFRESH_SECONDS:
+                last_snapshot_refresh = time.monotonic()
+                # The workers keep the snapshot they have, so a failed refresh costs only staleness.
+                try:
+                    self.refresh_blacklist_snapshot()
+                except RuntimeError:
+                    logger.exception("Could not refresh the blacklist snapshot, keeping the current one")
+
             if not index_process.is_alive():
                 crash_time = datetime.now(timezone.utc)
                 exit_code = index_process.exitcode or -1

diff --git a/mwmbl/crawler/README.md b/mwmbl/crawler/README.md
@@ -16,6 +16,8 @@ The crawler can be configured using the following environment variables:
 
 **Process Architecture**: The crawler spawns `CRAWLER_WORKERS` processes for crawling work, each using `CRAWL_THREADS` threads for concurrent URL fetching. Additionally, one separate process handles indexing work. For example, with default settings you get 2 crawler processes × 20 threads = 40 total crawling threads, plus 1 indexing process. The queue gives each batch at most one URL per domain, so the threads of one batch never fetch pages from the same site at once.
 
+**Memory**: Memory use scales with `CRAWLER_WORKERS`, not `CRAWL_THREADS`, so on a small machine prefer fewer workers with more threads. The blocklists (about 1.3 million domains) are downloaded once when the crawler starts, and again every 6 hours, by a short-lived process that publishes them to Redis as an ~11MB hash array that every worker loads. Because that first download has to succeed before any worker starts, the crawler exits at startup if it cannot reach the lists. If a later refresh fails, the crawler keeps the lists it has.
+
 Batches crawled by different workers at the same time can share domains, since each worker samples its domains independently. The core domains (Wikipedia, GitHub and so on) are in every batch, so they are the usual overlap. With the default 2 workers, about 0.7% of fetches start while another fetch to the same domain is in flight. Overlap grows with `CRAWLER_WORKERS`. The crawler records it in Redis, and `redis-cli HGETALL crawl-domain-overlap` gives the totals: `fetches`, `overlapped` (fetches that started while another fetch to their domain was in flight) and `duplicate_domains` (should stay 0).
 
 ### Rate Limiting Configuration

diff --git a/mwmbl/crawler/version.py b/mwmbl/crawler/version.py
@@ -1,4 +1,4 @@
 # No imports: the crawler release workflow reads this without installing the project. Raising
 # it on main publishes a release; CI fails a pull request that changes the crawler without
 # raising it (mwmbl/crawler/release_check.py).
-CRAWLER_VERSION: str = "1.0.0"
+CRAWLER_VERSION: str = "1.0.1"
diff --git a/mwmbl/indexer/blacklist_snapshot.py b/mwmbl/indexer/blacklist_snapshot.py
@@ -167,7 +167,7 @@ def refresh_snapshot(provider: Optional[BlacklistProvider] = None, redis_client:
 # ---------------------------------------------------------------------------
 
 
-class SnapshotBlacklist:
+class SnapshotBlacklist(BlacklistProvider):
     """Membership queries against the published snapshot, plus the local built-in rules.
 
     One instance per process. Nothing here ever blocks a request on the network: the

diff --git a/mwmbl/indexer/update_urls.py b/mwmbl/indexer/update_urls.py
@@ -8,24 +8,19 @@
 from mwmbl.crawler.batch import HashedBatch, Link
 from mwmbl.crawler.domains import DomainLinkDatabase
 from mwmbl.crawler.urls import FoundURL, URLDatabase, URLStatus
-from mwmbl.indexer.blacklist import get_default_blacklist_provider
+from mwmbl.indexer.blacklist_providers import BlacklistProvider
 from mwmbl.indexer.index_batches import get_url_error_status
 from mwmbl.redis_url_queue import RedisURLQueue, get_domain_max_urls
 from mwmbl.utils import get_domain, parse_url
 
 logger = getLogger(__name__)
 
 
-def record_urls_in_database(batches: Collection[HashedBatch], new_item_queue: RedisURLQueue):
+def record_urls_in_database(
+    batches: Collection[HashedBatch], new_item_queue: RedisURLQueue, blacklist_provider: BlacklistProvider
+):
     start = datetime.now(timezone.utc)
-    # Take the approved domains from the queue rather than the database: the standalone
-    # crawler is the live caller of this, and it fetches them over HTTP.
-    blacklist_provider = get_default_blacklist_provider(new_item_queue.get_curated_domains_function)
-    blacklist_retrieval_time = datetime.now(timezone.utc) - start
-    logger.info(
-        f"Recording URLs in database for {len(batches)} batches, blacklist provider ready "
-        f"in {blacklist_retrieval_time.total_seconds()} seconds"
-    )
+    logger.info(f"Recording URLs in database for {len(batches)} batches")
 
     url_users = {}
     url_timestamps = {}

diff --git a/test/test_crawl_supervisor.py b/test/test_crawl_supervisor.py
@@ -7,6 +7,7 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from django.test import override_settings
 
 from mwmbl import crawl
 from mwmbl.crawl import Crawler
@@ -29,6 +30,7 @@ def supervisor_environment():
     with (
         patch.object(crawl, "validate_environment"),
         patch.object(Crawler, "check_redis"),
+        patch.object(Crawler, "refresh_blacklist_snapshot"),
         patch.object(crawl, "CRAWLER_WORKERS", 2),
     ):
         yield
@@ -81,18 +83,64 @@ def test_run_rejects_zero_workers():
             Crawler().run()
 
 
+def test_run_keeps_crawling_when_a_snapshot_refresh_fails(supervisor_environment):
+    def stop_after_first_pass(seconds):
+        if seconds == 10:
+            raise LoopStopped
+
+    with (
+        override_settings(BLACKLIST_SNAPSHOT_REFRESH_SECONDS=0),
+        patch.object(Crawler, "refresh_blacklist_snapshot", side_effect=[None, RuntimeError("no network")]) as refresh,
+        patch.object(crawl, "Process", side_effect=lambda target: _process(alive=True)),
+        patch.object(crawl.time, "sleep", side_effect=stop_after_first_pass),
+    ):
+        with pytest.raises(LoopStopped):
+            Crawler().run()
+
+    assert refresh.call_count == 2
+
+
+def test_a_failed_snapshot_publisher_is_an_error():
+    with patch.object(crawl, "Process", return_value=_process(alive=False, exitcode=1)):
+        with pytest.raises(RuntimeError, match="exit code 1"):
+            Crawler().refresh_blacklist_snapshot()
+
+
+def test_a_snapshot_publisher_that_hangs_is_killed():
+    process = _process(alive=True)
+    with patch.object(crawl, "Process", return_value=process):
+        with pytest.raises(RuntimeError, match="took over"):
+            Crawler().refresh_blacklist_snapshot()
+
+    process.join.assert_any_call(crawl.SNAPSHOT_PUBLISH_TIMEOUT_SECONDS)
+    process.kill.assert_called_once()
+
+
+def test_the_queue_and_each_batch_share_one_snapshot():
+    crawler = Crawler()
+    with patch.object(crawl, "SnapshotBlacklist") as snapshot_blacklist:
+        first = crawler.new_blacklist_provider()
+        second = crawler.new_blacklist_provider()
+
+    snapshot_blacklist.assert_called_once()
+    assert first is not second
+    assert first.providers == second.providers == [snapshot_blacklist.return_value]
+
+
 def test_run_once_checks_the_environment_then_crawls_and_indexes():
     crawler = Crawler()
     with (
         patch.object(crawl, "validate_environment") as validate,
         patch.object(Crawler, "check_redis") as check_redis,
+        patch.object(Crawler, "refresh_blacklist_snapshot") as refresh_blacklist_snapshot,
         patch.object(Crawler, "process_batch") as process_batch,
         patch.object(Crawler, "run_indexing") as run_indexing,
     ):
         crawler.run_once()
 
     validate.assert_called_once()
     check_redis.assert_called_once()
+    refresh_blacklist_snapshot.assert_called_once()
     process_batch.assert_called_once()
     run_indexing.assert_called_once()
 

diff --git a/test/test_crawler_ci_options.py b/test/test_crawler_ci_options.py
@@ -193,6 +193,7 @@ def test_run_once_does_one_pass_and_returns():
     with (
         patch("mwmbl.crawl.validate_environment"),
         patch.object(Crawler, "check_redis"),
+        patch.object(Crawler, "refresh_blacklist_snapshot"),
         patch.object(Crawler, "process_batch") as process_batch,
         patch.object(Crawler, "run_indexing") as run_indexing,
     ):
@@ -209,6 +210,7 @@ def test_run_once_propagates_errors_instead_of_restarting():
     with (
         patch("mwmbl.crawl.validate_environment"),
         patch.object(Crawler, "check_redis"),
+        patch.object(Crawler, "refresh_blacklist_snapshot"),
         patch.object(Crawler, "process_batch", side_effect=RuntimeError("crawl failed")),
         patch.object(Crawler, "run_indexing") as run_indexing,
     ):

diff --git a/test/test_update_urls.py b/test/test_update_urls.py
@@ -12,12 +12,13 @@
 from mwmbl.crawler.batch import HashedBatch, Item, ItemContent, ItemError, Link
 from mwmbl.crawler.urls import URLStatus
 from mwmbl.indexer import update_urls
+from mwmbl.indexer.blacklist_providers import BlacklistProvider
 from mwmbl.indexer.update_urls import get_datetime_from_timestamp, process_link, record_urls_in_database
 
 TIMESTAMP = datetime(2024, 1, 1, tzinfo=timezone.utc)
 
 
-class FakeBlacklist:
+class FakeBlacklist(BlacklistProvider):
     def is_domain_blacklisted(self, domain: str) -> bool:
         return domain == "blocked.example"
 
@@ -55,12 +56,11 @@ def test_unparseable_and_blacklisted_links_are_dropped(link_url):
 
 @pytest.fixture
 def recorded(monkeypatch):
-    """Patch the database and blacklist, returning the URLs that would have been queued."""
+    """Patch the database, returning the URLs that would have been queued."""
     found_urls = []
     domain_link_db = MagicMock()
     domain_link_db.__enter__.return_value = domain_link_db
 
-    monkeypatch.setattr(update_urls, "get_default_blacklist_provider", lambda curated: FakeBlacklist())
     monkeypatch.setattr(update_urls, "_add_found_urls_to_db_and_queue", lambda urls, queue: found_urls.extend(urls))
     monkeypatch.setattr(update_urls, "DomainLinkDatabase", lambda: domain_link_db)
     return found_urls, domain_link_db
@@ -92,7 +92,7 @@ def test_recording_a_batch_classifies_every_url(recorded):
         ],
     )
 
-    record_urls_in_database([batch], MagicMock())
+    record_urls_in_database([batch], MagicMock(), FakeBlacklist())
 
     statuses = {found.url: found.status for found in found_urls}
     assert statuses == {
@@ -115,7 +115,7 @@ def test_a_crawled_page_with_an_unparseable_url_contributes_no_links(recorded):
     content = ItemContent(title="T", extract="E", links=["https://linked.example/a"])
     batch = HashedBatch(user_id_hash="hash", timestamp=1704067200.0, items=[_item("no-scheme", content=content)])
 
-    record_urls_in_database([batch], MagicMock())
+    record_urls_in_database([batch], MagicMock(), FakeBlacklist())
 
     assert [found.url for found in found_urls] == ["no-scheme"]
     domain_link_db.update_domain_links.assert_not_called()