diff --git a/docker-compose.yml b/docker-compose.yml
index 4f90327..752c2d7 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -11,7 +11,7 @@ services:
command: >
bash -c "
cargo install cargo-watch &&
- cargo watch --poll -i logs/ -x 'run -- --properties /app/property-data/properties.parquet --postcode-features /app/property-data/postcode.parquet --pois /app/property-data/filtered_uk_pois.parquet --places /app/property-data/places.parquet --tiles /app/property-data/uk.pmtiles --postcodes /app/property-data/postcode_boundaries --travel-times /app/property-data/travel-times --satellite-tiles /app/property-data/satellite.pmtiles --satellite-highres-tiles /app/property-data/satellite_highres.pmtiles --noise-overlay-tiles /app/property-data/noise_lden_10m.pmtiles --crime-hotspot-tiles /app/property-data/crime_hotspots.pmtiles --tree-overlay-tiles /app/property-data/trees_outside_woodlands.pmtiles --property-border-tiles /app/property-data/property_borders.pmtiles --crime-by-year-path /app/property-data/crime_by_postcode_by_year.parquet --crime-records-path /app/property-data/crime_records.parquet --area-crime-averages-path /app/property-data/area_crime_averages.parquet --population-path /app/property-data/population_by_postcode.parquet'
+ cargo watch --poll -i logs/ -x 'run -- --properties /app/property-data-snapshot-2026-07-14/properties.parquet --postcode-features /app/property-data-snapshot-2026-07-14/postcode.parquet --pois /app/property-data-snapshot-2026-07-14/filtered_uk_pois.parquet --places /app/property-data-snapshot-2026-07-14/places.parquet --tiles /app/property-data-snapshot-2026-07-14/uk.pmtiles --postcodes /app/property-data-snapshot-2026-07-14/postcode_boundaries --travel-times /app/property-data-snapshot-2026-07-14/travel-times --satellite-tiles /app/property-data-snapshot-2026-07-14/satellite.pmtiles --satellite-highres-tiles /app/property-data-snapshot-2026-07-14/satellite_highres.pmtiles --noise-overlay-tiles /app/property-data-snapshot-2026-07-14/noise_lden_10m.pmtiles --crime-hotspot-tiles /app/property-data-snapshot-2026-07-14/crime_hotspots.pmtiles --tree-overlay-tiles /app/property-data-snapshot-2026-07-14/trees_outside_woodlands.pmtiles --property-border-tiles /app/property-data-snapshot-2026-07-14/property_borders.pmtiles --crime-by-year-path /app/property-data-snapshot-2026-07-14/crime_by_postcode_by_year.parquet --crime-records-path /app/property-data-snapshot-2026-07-14/crime_records.parquet --area-crime-averages-path /app/property-data-snapshot-2026-07-14/area_crime_averages.parquet --population-path /app/property-data-snapshot-2026-07-14/population_by_postcode.parquet'
"
ports:
- "8001:8001"
@@ -64,7 +64,7 @@ services:
BUGSINK_ENVIRONMENT: ${BUGSINK_ENVIRONMENT:-development}
BUGSINK_RELEASE: ${BUGSINK_RELEASE:-}
ACTUAL_LISTINGS_PATH: /app/finder/data/online_listings_buy_enriched.parquet
- DEVELOPMENTS_PATH: /app/property-data/development_sites.parquet
+ DEVELOPMENTS_PATH: /app/property-data-snapshot-2026-07-14/development_sites.parquet
BUGSINK_SEND_DEFAULT_PII: ${BUGSINK_SEND_DEFAULT_PII:-false}
depends_on:
screenshot:
diff --git a/finder/README.md b/finder/README.md
index bea1384..393643d 100644
--- a/finder/README.md
+++ b/finder/README.md
@@ -187,6 +187,9 @@ Environment variables (override the defaults in `constants.py`):
| `ZOOPLA_OUTCODE_TIMEOUT_SECONDS` | `300` | Per-outcode wall-clock budget for Zoopla. |
| `DETAIL_FETCH_CONCURRENCY` | `8` | Parallel detail fetches (Rightmove/OTM). |
| `REQUESTS_PER_SECOND` | `10` | Global request-rate cap. Lower it if you see `429`/`403`. |
+| `BLOCK_403_THRESHOLD` | `5` | Consecutive 403s from one host before the egress IP is rotated. |
+| `BLOCK_MAX_ROTATIONS` | `3` | Egress-IP rotations allowed per run. `0` disables rotation. |
+| `MAX_ROW_DROP_RATIO` | `0.25` | Reject the run if the merged total falls this far below the previous parquet. |
| `RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS` | `1` | Inert today (see note above). |
Non-env code constants worth knowing (`constants.py` / `onthemarket.py`):
@@ -216,7 +219,8 @@ uv run --with pytest pytest -q
| `scraper.py` | Orchestration: per-source runners, provider parallelism, cache load/save, merge + write. |
| `rightmove.py` / `onthemarket.py` / `zoopla.py` | Per-portal search + detail scraping and parsing. |
| `transform.py` | Raw listing → output schema; postcode trust rules. |
-| `http_client.py` | Shared httpx client, retry/backoff, and the global `RATE_LIMITER`. |
+| `http_client.py` | Shared httpx client, retry/backoff, the global `RATE_LIMITER`, and egress-block detection (`BlockedError`). |
+| `gluetun.py` | Gluetun control-API client: reads the public IP and rotates the (shared) VPN tunnel. |
| `postcode_cache.py` | Persistent (cross-run) detail-cache load/save. |
| `spatial.py` | Grid spatial index for coordinate → nearest postcode. |
| `storage.py` | Parquet writer (server-ready column names). |
diff --git a/finder/constants.py b/finder/constants.py
index 64924f5..1507cb8 100644
--- a/finder/constants.py
+++ b/finder/constants.py
@@ -25,6 +25,32 @@ RETRY_BASE_DELAY = 2.0
# down if the portals start returning 429/403.
DETAIL_FETCH_CONCURRENCY = int(os.environ.get("DETAIL_FETCH_CONCURRENCY", "8"))
REQUESTS_PER_SECOND = float(os.environ.get("REQUESTS_PER_SECOND", "10"))
+
+# Egress-block detection. A portal that 403s one URL is a quirk; a portal that
+# 403s a whole HOST in a row is refusing our egress IP, which no amount of
+# per-request retrying fixes. On 2026-07-15 Rightmove's Fastly edge did exactly
+# this and every one of the 366 outcodes 403'd on the typeahead call, which the
+# scraper silently recorded as "no such outcode" and published a Rightmove-less
+# dataset. So: after this many consecutive 403s from one host, rotate the egress
+# IP (finder/gluetun.py) and retry; once the rotation budget is spent and the
+# host still 403s, raise http_client.BlockedError and fail the run loudly.
+#
+# Rotation is DISRUPTIVE (see gluetun.py: it restarts the tunnel shared with the
+# media stack), so the budget is deliberately small. It is per-run, not per-host.
+BLOCK_403_THRESHOLD = int(os.environ.get("BLOCK_403_THRESHOLD", "5"))
+BLOCK_MAX_ROTATIONS = int(os.environ.get("BLOCK_MAX_ROTATIONS", "3"))
+
+# Sanity gates on a finished scrape, checked before the parquet is overwritten
+# (finder/scraper.py). A selected source yielding nothing, or the merged total
+# collapsing against the previous run, means something is broken upstream rather
+# than the market having moved; publishing that would quietly gut production.
+#
+# 10% is loose against real movement and tight against breakage: consecutive
+# healthy cycles land within ~0.15% of each other (103,087 / 103,142 / 103,008),
+# so a 10% fall is already a ~65x anomaly. It is deliberately BELOW the
+# 2026-07-15 incident's own 18.7% drop (103,008 -> 83,785), which a 25% gate
+# would have waved through.
+MAX_ROW_DROP_RATIO = float(os.environ.get("MAX_ROW_DROP_RATIO", "0.10"))
GRID_CELL_SIZE = 0.01 # degrees for postcode spatial index
MAX_BEDROOMS = 20 # sanity cap: values above this are almost certainly parsing errors
diff --git a/finder/gluetun.py b/finder/gluetun.py
new file mode 100644
index 0000000..4e3ae78
--- /dev/null
+++ b/finder/gluetun.py
@@ -0,0 +1,130 @@
+"""Gluetun VPN control-server client, shared by every scraper.
+
+The scrapers egress through a Gluetun container's network namespace
+(docker-compose.yml, network_mode "container:media_gluetun"), so when a portal
+starts refusing our egress IP the cheapest unblocker is to make Gluetun
+reconnect to a different VPN server. This module wraps Gluetun's HTTP control
+API so both callers share one implementation:
+
+ * http_client.py rotates when a portal 403s every request (an egress block).
+ * zoopla.py rotates when Cloudflare Turnstile fires.
+
+Rotation is SHARED and DISRUPTIVE: every container joined to Gluetun's netns
+(here also qbittorrent/sonarr/radarr/prowlarr/seerr/jellyfin) loses
+connectivity for the few seconds the tunnel takes to come back. It is therefore
+serialised behind a module lock, so N blocked worker threads trigger ONE
+rotation rather than N, and callers must budget rotations rather than retry
+them freely.
+"""
+
+import logging
+import threading
+import time
+
+import httpx
+
+from constants import GLUETUN_API_KEY, GLUETUN_CONTROL_URL
+
+log = logging.getLogger("rightmove")
+
+# Serialises rotation across worker threads: a rotation tears down the shared
+# tunnel, so two concurrent ones would fight (and needlessly double the outage).
+_ROTATION_LOCK = threading.Lock()
+
+
+def _base_url() -> str:
+ return GLUETUN_CONTROL_URL.rstrip("/")
+
+
+def _client() -> httpx.Client:
+ # Talks to the control server directly (not through the VPN proxy).
+ headers = {}
+ if GLUETUN_API_KEY:
+ headers["X-API-Key"] = GLUETUN_API_KEY
+ return httpx.Client(headers=headers)
+
+
+def public_ip(client: httpx.Client) -> str | None:
+ try:
+ resp = client.get(f"{_base_url()}/v1/publicip/ip", timeout=5.0)
+ if resp.status_code != 200:
+ return None
+ data = resp.json()
+ except (httpx.HTTPError, ValueError):
+ return None
+ return data.get("public_ip") or data.get("ip")
+
+
+def _set_vpn_status(client: httpx.Client, status: str) -> bool:
+ """PUT /v1/vpn/status with {'status': status}. Returns True on 2xx."""
+ try:
+ resp = client.put(
+ f"{_base_url()}/v1/vpn/status",
+ json={"status": status},
+ timeout=15.0,
+ )
+ except httpx.HTTPError as exc:
+ log.warning("Gluetun vpn/status %s failed: %s", status, exc)
+ return False
+ if resp.status_code == 401:
+ log.warning(
+ "Gluetun vpn/status %s: 401 Unauthorized. The API key must be "
+ "authorised for 'PUT /v1/vpn/status' in Gluetun's auth config.toml",
+ status,
+ )
+ return False
+ if resp.status_code >= 400:
+ log.warning(
+ "Gluetun vpn/status %s returned HTTP %d: %s",
+ status,
+ resp.status_code,
+ resp.text[:200],
+ )
+ return False
+ return True
+
+
+def rotate_ip(wait_seconds: int = 45) -> bool:
+ """Restart Gluetun's VPN and wait for the public IP to change.
+
+ Returns True if a new IP was observed within ``wait_seconds``. Serialised:
+ while one thread rotates, others block here and then see the already-rotated
+ IP, so a 403 storm across many threads costs one tunnel restart. A failed
+ rotation always attempts to bring the tunnel back up, because leaving it
+ stopped would strand every container sharing the netns.
+ """
+ with _ROTATION_LOCK, _client() as client:
+ old_ip = public_ip(client)
+ log.info("Requesting Gluetun IP rotation (current IP: %s)", old_ip or "unknown")
+
+ stop_attempted = False
+ restart_confirmed = False
+ try:
+ stop_attempted = True
+ if not _set_vpn_status(client, "stopped"):
+ return False
+ time.sleep(2)
+ restart_confirmed = _set_vpn_status(client, "running")
+ if not restart_confirmed:
+ return False
+
+ deadline = time.monotonic() + wait_seconds
+ while time.monotonic() < deadline:
+ time.sleep(2)
+ new_ip = public_ip(client)
+ if new_ip and new_ip != old_ip:
+ log.info("Gluetun rotated IP: %s -> %s", old_ip or "?", new_ip)
+ return True
+ finally:
+ if stop_attempted and not restart_confirmed:
+ log.warning(
+ "Gluetun VPN may be stopped after failed rotation; "
+ "attempting recovery start"
+ )
+ if not _set_vpn_status(client, "running"):
+ log.error(
+ "Gluetun VPN recovery start failed; manual intervention required"
+ )
+
+ log.warning("Gluetun IP did not change within %ds", wait_seconds)
+ return False
diff --git a/finder/http_client.py b/finder/http_client.py
index d1cb76d..146e66b 100644
--- a/finder/http_client.py
+++ b/finder/http_client.py
@@ -2,12 +2,16 @@ import logging
import random
import threading
import time
+from urllib.parse import urlsplit
import httpx
from fake_useragent import UserAgent
+import gluetun
import shutdown
from constants import (
+ BLOCK_403_THRESHOLD,
+ BLOCK_MAX_ROTATIONS,
GLUETUN_PROXY,
MAX_RETRIES,
REQUESTS_PER_SECOND,
@@ -17,6 +21,16 @@ from constants import (
log = logging.getLogger("rightmove")
+class BlockedError(RuntimeError):
+ """A host is refusing our egress IP and rotating away from it did not help.
+
+ Raised rather than returned so a block can never be mistaken for "this query
+ has no results": that conflation is precisely what turned Rightmove's
+ 2026-07-15 edge block into a silent 0-listing run (see BLOCK_403_THRESHOLD).
+ Callers should abandon the source, not the outcode.
+ """
+
+
class RateLimiter:
"""Thread-safe global limiter: spaces request starts by a minimum interval.
@@ -51,6 +65,74 @@ class RateLimiter:
# providers). Spacing is global, so politeness is decoupled from concurrency.
RATE_LIMITER = RateLimiter(REQUESTS_PER_SECOND)
+
+class _EgressBlockTracker:
+ """Escalates a run of 403s from one host: rotate the egress IP, then give up.
+
+ Counts CONSECUTIVE 403s per host, reset by any 200 from that host, so one
+ forbidden URL never trips it while a refused egress IP does within a couple
+ of calls. The rotation budget is per-run and shared across hosts and
+ threads, because the tunnel being rotated is shared too.
+ """
+
+ def __init__(self, threshold: int, max_rotations: int):
+ self._threshold = max(threshold, 1)
+ self._max_rotations = max(max_rotations, 0)
+ self._lock = threading.Lock()
+ self._consecutive: dict[str, int] = {}
+ self._rotations = 0
+
+ def record_success(self, host: str) -> None:
+ with self._lock:
+ self._consecutive.pop(host, None)
+
+ def record_403(self, host: str) -> str:
+ """Register a 403 and decide what the caller should do next.
+
+ Returns "retry" (below the threshold, back off normally), "rotated"
+ (the egress IP changed, so the call deserves a fresh retry budget) or
+ "blocked" (the rotation budget is spent, or rotating failed).
+
+ Rotation runs under the lock: concurrent threads caught in the same 403
+ storm queue behind ONE tunnel restart and then re-read the reset
+ counters, rather than each spending a slice of the budget.
+ """
+ with self._lock:
+ count = self._consecutive.get(host, 0) + 1
+ self._consecutive[host] = count
+ if count < self._threshold:
+ return "retry"
+ if self._rotations >= self._max_rotations:
+ return "blocked"
+
+ self._rotations += 1
+ log.warning(
+ "%d consecutive 403s from %s; rotating egress IP (rotation %d/%d)",
+ count,
+ host,
+ self._rotations,
+ self._max_rotations,
+ )
+ rotated = gluetun.rotate_ip()
+ # Either way the counters restart: on success the next 403s are
+ # evidence about the NEW IP, and on failure we must not re-enter
+ # rotation on the very next 403.
+ self._consecutive.clear()
+
+ if not rotated:
+ log.error("Egress IP rotation failed; treating %s as blocked", host)
+ return "blocked"
+ return "rotated"
+
+ def reset(self) -> None:
+ """Forget all counters and refund the rotation budget (tests only)."""
+ with self._lock:
+ self._consecutive.clear()
+ self._rotations = 0
+
+
+BLOCK_TRACKER = _EgressBlockTracker(BLOCK_403_THRESHOLD, BLOCK_MAX_ROTATIONS)
+
_ua = UserAgent(
browsers=["Chrome", "Edge"], os=["Windows", "Mac OS X"], min_version=120.0
)
@@ -69,22 +151,48 @@ def make_client() -> httpx.Client:
def fetch_with_retry(
client: httpx.Client, url: str, params: dict | None = None, on_403: bool = True
) -> dict | None:
- """GET JSON with retries on 429/5xx/connection errors.
+ """GET JSON with retries on 403/429/5xx/connection errors.
- Returns None on permanent failure. The on_403 argument is kept for
- compatibility with older callers; 403 is now treated as non-retryable.
+ Returns None on permanent failure, EXCEPT when the host is refusing our
+ egress IP, which raises BlockedError instead. A 403 from a shared CDN edge
+ is usually transient and is retried with backoff; once one host has 403'd
+ BLOCK_403_THRESHOLD times in a row the egress IP is rotated (and this call
+ gets a fresh retry budget on the new IP), and once the rotation budget is
+ spent the block is raised rather than degraded to a None the caller would
+ read as "no results". ``on_403=False`` opts out for callers where a 403 is
+ an expected per-URL answer rather than a verdict on our IP.
"""
- for attempt in range(MAX_RETRIES):
+ host = urlsplit(url).netloc
+ attempt = 0
+ while attempt < MAX_RETRIES:
if shutdown.stop_requested():
return None
try:
RATE_LIMITER.acquire()
resp = client.get(url, params=params)
if resp.status_code == 200:
+ BLOCK_TRACKER.record_success(host)
return resp.json()
if resp.status_code == 403 and on_403:
- log.error("HTTP 403 from %s (forbidden)", url)
- return None
+ action = BLOCK_TRACKER.record_403(host)
+ if action == "blocked":
+ raise BlockedError(
+ f"HTTP 403 from {url}: {host} is refusing this egress IP "
+ f"and rotating away from it did not help"
+ )
+ delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
+ log.warning(
+ "HTTP 403 from %s, retry %d/%d in %.1fs",
+ url,
+ attempt + 1,
+ MAX_RETRIES,
+ delay,
+ )
+ # A rotation means a different egress IP, so the failures this
+ # call already accrued say nothing about the new one.
+ attempt = 0 if action == "rotated" else attempt + 1
+ shutdown.sleep(delay)
+ continue
if resp.status_code in (429, 500, 502, 503, 504):
delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
log.warning(
@@ -95,6 +203,7 @@ def fetch_with_retry(
MAX_RETRIES,
delay,
)
+ attempt += 1
shutdown.sleep(delay)
continue
log.error("HTTP %d from %s (non-retryable)", resp.status_code, url)
@@ -114,6 +223,7 @@ def fetch_with_retry(
MAX_RETRIES,
delay,
)
+ attempt += 1
shutdown.sleep(delay)
log.error("All %d retries exhausted for %s", MAX_RETRIES, url)
return None
diff --git a/finder/main.py b/finder/main.py
index 92d3790..5c21521 100644
--- a/finder/main.py
+++ b/finder/main.py
@@ -218,6 +218,13 @@ def main() -> int:
return 130 # 128 + SIGINT, the conventional Ctrl+C exit code.
log.info("Scrape finished in %.1fs", elapsed)
log.info("Result: %s", result)
+ if result.get("failures"):
+ # Nothing was written, so the previous parquet is still the good one.
+ # Exiting non-zero keeps the caller (scripts/scrape-loop.sh) from
+ # enriching and publishing on top of a rejected run.
+ for failure in result["failures"]:
+ log.error("Scrape rejected: %s", failure)
+ return 1
if args.test and result.get("errors"):
raise SystemExit("Test scrape failed; see errors in the result above.")
return 0
diff --git a/finder/scraper.py b/finder/scraper.py
index a9c7f60..1c1c3ad 100644
--- a/finder/scraper.py
+++ b/finder/scraper.py
@@ -19,6 +19,7 @@ from constants import (
DATA_DIR,
DELAY_BETWEEN_OUTCODES,
LONDON_OUTCODE_PREFIXES,
+ MAX_ROW_DROP_RATIO,
ZOOPLA_DETAIL_BUDGET_FRACTION,
ZOOPLA_FETCH_DETAILS,
ZOOPLA_FETCHER,
@@ -29,7 +30,7 @@ import onthemarket
import rightmove
import shutdown
import zoopla
-from http_client import make_client
+from http_client import BlockedError, make_client
from onthemarket import search_outcode as onthemarket_search_outcode
from postcode_cache import load_cache, save_cache
from rightmove import resolve_outcode_id
@@ -316,6 +317,72 @@ def _record_error(
log.warning(message)
+def _previous_row_count(path: Path) -> int | None:
+ """Rows in the parquet we are about to replace, or None if there isn't one.
+
+ Reads footer metadata only, so it costs nothing on a 30MB file."""
+ if not path.exists():
+ return None
+ try:
+ return pl.scan_parquet(path).select(pl.len()).collect().item()
+ except Exception as exc: # noqa: BLE001 - an unreadable baseline must not fail the run
+ log.warning("Could not read row count from %s: %s", path, exc)
+ return None
+
+
+def _sanity_failures(
+ results: dict[str, list[dict]],
+ selected_sources: list[str],
+ merged: list[dict],
+ output_path: Path,
+ abandoned: set[str],
+) -> list[str]:
+ """Reasons this run's data is too broken to publish, empty when it's fine.
+
+ Catches upstream breakage that the per-outcode error path cannot see, via
+ three gates, because no one of them is sufficient:
+
+ * ABANDONED. A source that hit an egress block stopped partway, so its
+ listings cover only the outcodes it reached. This gate is what catches a
+ block that starts MID-RUN: the source is then non-empty, so the zero-yield
+ gate below would wave it through on a single banked listing.
+ * ZERO YIELD. A selected source that returned nothing at all, whatever the
+ cause (403 storm from the first outcode, markup change, DNS).
+ * COLLAPSE. The merged total against the previous run, for degradations that
+ are neither clean-zero nor a raised block.
+
+ Cross-source dedup is why the merged total alone cannot be trusted: when
+ Rightmove vanished on 2026-07-15, OnTheMarket stopped losing dedup ties to
+ it and backfilled the total from ~9k unique to ~84k, so losing an entire
+ source showed up as an 18.7% dip.
+ """
+ failures = []
+ for source in selected_sources:
+ if source in abandoned:
+ failures.append(
+ f"{source} was abandoned mid-run after an egress block, so its "
+ f"{_source_total(results, source)} listings cover only part of "
+ f"the outcode list"
+ )
+ elif _source_total(results, source) == 0:
+ failures.append(
+ f"{source} yielded 0 listings; it was selected for this run, so "
+ f"treat this as a failure rather than an empty market"
+ )
+ if not merged:
+ failures.append("no listings survived the merge")
+
+ previous = _previous_row_count(output_path)
+ if previous and merged:
+ drop = (previous - len(merged)) / previous
+ if drop > MAX_ROW_DROP_RATIO:
+ failures.append(
+ f"merged total collapsed from {previous} to {len(merged)} "
+ f"({drop:.0%} drop, limit {MAX_ROW_DROP_RATIO:.0%})"
+ )
+ return failures
+
+
def _scrape_rightmove(
outcodes: list[str],
pc_index: PostcodeSpatialIndex,
@@ -335,6 +402,12 @@ def _scrape_rightmove(
try:
outcode_id = resolve_outcode_id(client, outcode)
+ except BlockedError:
+ # Our egress IP is refused, not this outcode. Grinding through
+ # the remaining outcodes would just collect 403s, so abandon the
+ # source and let _run_sources mark it abandoned, which fails the
+ # run however many outcodes we got through first.
+ raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
shutdown.sleep(DELAY_BETWEEN_OUTCODES)
@@ -368,6 +441,8 @@ def _scrape_rightmove(
max_properties_per_source,
)
log.info("Rightmove %s: +%d", outcode, added)
+ except BlockedError:
+ raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
@@ -396,6 +471,8 @@ def _scrape_rightmove(
max_properties_per_source,
)
log.info("Rightmove %s new-homes: +%d", outcode, added_new)
+ except BlockedError:
+ raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
@@ -731,6 +808,7 @@ def _run_sources(
run_zoopla: Callable[[], None] | None,
background_runners: list[tuple[str, Callable[[], None]]],
errors: list[str],
+ abandoned: set[str],
) -> None:
"""Run the HTTP-based providers concurrently while Zoopla runs inline.
@@ -740,17 +818,28 @@ def _run_sources(
writes only its own ``results[source]`` list and appends to the shared
``errors`` list (atomic under the GIL), so there is no cross-source data
race. One source raising never kills the others: each failure is recorded
- and the remaining sources still finish."""
+ and the remaining sources still finish.
+
+ A source that hit an egress block is named in ``abandoned`` as well as
+ ``errors``: it stopped partway, so whatever it did collect covers only the
+ outcodes it reached, and _sanity_failures must reject the run even though
+ the source is not empty."""
with ThreadPoolExecutor(max_workers=max(1, len(background_runners))) as pool:
futures = {pool.submit(fn): name for name, fn in background_runners}
if run_zoopla is not None:
try:
run_zoopla()
+ except BlockedError as exc:
+ _record_error(errors, "zoopla", "*", exc)
+ abandoned.add("zoopla")
except Exception as exc: # noqa: BLE001 - one source must not kill the run
_record_error(errors, "zoopla", "*", exc)
for future, name in futures.items():
try:
future.result()
+ except BlockedError as exc:
+ _record_error(errors, name, "*", exc)
+ abandoned.add(name)
except Exception as exc: # noqa: BLE001 - one source must not kill the run
_record_error(errors, name, "*", exc)
@@ -775,6 +864,10 @@ def run_scrape(
output_base.mkdir(parents=True, exist_ok=True)
errors: list[str] = []
+ # Sources that stopped partway because their egress was blocked. Tracked
+ # separately from `errors`: a partial source is a reason to reject the run,
+ # not just something to log.
+ abandoned: set[str] = set()
results = {source: [] for source in SOURCE_ORDER}
started_at = time.time()
@@ -834,16 +927,28 @@ def run_scrape(
)
try:
- _run_sources(run_zoopla, background_runners, errors)
+ _run_sources(run_zoopla, background_runners, errors, abandoned)
finally:
_save_detail_caches(selected_sources, cache_dir)
merged, source_counts, deduped = _merge_properties(results)
output_path = output_base / "online_listings_buy.parquet"
- if merged:
+ failures = _sanity_failures(
+ results, selected_sources, merged, output_path, abandoned
+ )
+ if failures:
+ # Publishing here is worse than publishing nothing: downstream (the
+ # enrich step, then the server) has no way to tell a gutted dataset from
+ # a real one, so the previous parquet stays and the run exits non-zero.
+ # The asking-price history is left alone too, since it is forward-only
+ # and would bake this run's gaps in permanently.
+ for failure in failures:
+ log.error("Sanity check failed: %s", failure)
+ log.error("Refusing to overwrite %s; keeping the previous data", output_path)
+ else:
# Accrue the per-listing asking-price history before writing: load the
# persistent store, append this run's price moves, dump it, then embed
- # each listing's series in the parquet. Forward-only by nature — the
+ # each listing's series in the parquet. Forward-only by nature: the
# first run seeds one point per listing and reductions appear over time.
history_path = output_base / "price_history" / "listings.json"
history = load_history(history_path)
@@ -853,10 +958,6 @@ def run_scrape(
update_history(history, merged, run_date)
save_history(history_path, history)
write_parquet(merged, output_path, price_history=history)
- else:
- if output_path.exists():
- output_path.unlink()
- log.warning("No London-ish properties to write to %s", output_path)
counts = {
"total": len(merged),
@@ -881,6 +982,10 @@ def run_scrape(
},
"counts": counts,
"path": str(output_path),
+ # `errors` are per-outcode and survivable; `failures` mean the run's
+ # output was rejected and nothing was written. Only the latter is fatal.
"errors": errors,
+ "failures": failures,
+ "written": not failures,
"elapsed_seconds": round(time.time() - started_at, 3),
}
diff --git a/finder/test_block_detection.py b/finder/test_block_detection.py
new file mode 100644
index 0000000..680b048
--- /dev/null
+++ b/finder/test_block_detection.py
@@ -0,0 +1,321 @@
+"""Egress-block detection: 403 storms must rotate the IP, then fail the run.
+
+Regression cover for 2026-07-15, when Rightmove's edge 403'd every typeahead
+call, fetch_with_retry returned None, resolve_outcode_id read that as "no such
+outcode" for all 366 outcodes, and the run published a Rightmove-less parquet
+with errors: [] and exit 0.
+"""
+
+import httpx
+import pytest
+
+import http_client
+import scraper
+from http_client import BlockedError, _EgressBlockTracker
+
+
+class _StubResponse:
+ def __init__(self, status_code: int, payload: dict | None = None):
+ self.status_code = status_code
+ self._payload = payload or {}
+
+ def json(self) -> dict:
+ return self._payload
+
+
+class _StubClient:
+ """Replays a fixed status sequence, recording the URLs it was asked for."""
+
+ def __init__(self, statuses: list[int], payload: dict | None = None):
+ self._statuses = list(statuses)
+ self._payload = payload or {"matches": []}
+ self.calls: list[str] = []
+
+ def get(self, url, params=None, headers=None):
+ self.calls.append(url)
+ status = self._statuses.pop(0) if self._statuses else 200
+ return _StubResponse(status, self._payload)
+
+
+@pytest.fixture(autouse=True)
+def _no_sleeping(monkeypatch):
+ # Backoff delays would otherwise make this suite take minutes.
+ monkeypatch.setattr(http_client.shutdown, "sleep", lambda _s: None)
+ monkeypatch.setattr(http_client.RATE_LIMITER, "acquire", lambda: None)
+
+
+@pytest.fixture(autouse=True)
+def _fresh_tracker(monkeypatch):
+ tracker = _EgressBlockTracker(threshold=3, max_rotations=1)
+ monkeypatch.setattr(http_client, "BLOCK_TRACKER", tracker)
+ return tracker
+
+
+def test_403_is_retried_not_swallowed(monkeypatch):
+ """A transient 403 must not end the call: this is what broke last time."""
+ rotations = []
+ monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: rotations.append(1))
+
+ client = _StubClient([403, 200], payload={"matches": ["ok"]})
+ result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
+
+ assert result == {"matches": ["ok"]}
+ assert len(client.calls) == 2, "the 403 should have been retried"
+ assert rotations == [], "one 403 is not a block; rotating would be disruptive"
+
+
+def test_sustained_403s_rotate_the_egress_ip(monkeypatch):
+ """Threshold consecutive 403s from one host trigger exactly one rotation."""
+ rotations = []
+
+ def _rotate():
+ rotations.append(1)
+ return True
+
+ monkeypatch.setattr(http_client.gluetun, "rotate_ip", _rotate)
+
+ # 3 x 403 hits the threshold and rotates; the retry budget resets and the
+ # next call succeeds on the "new IP".
+ client = _StubClient([403, 403, 403, 200], payload={"matches": ["ok"]})
+ result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
+
+ assert result == {"matches": ["ok"]}
+ assert rotations == [1], "expected exactly one rotation"
+
+
+def test_403s_surviving_rotation_raise_blocked_error(monkeypatch):
+ """Once the rotation budget is spent, a block must raise, never return None.
+
+ Returning None is what let resolve_outcode_id read a site-wide block as
+ "this outcode does not exist" and silently produce zero listings.
+ """
+ monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)
+
+ client = _StubClient([403] * 20)
+ with pytest.raises(BlockedError, match="refusing this egress IP"):
+ http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
+
+
+def test_failed_rotation_is_treated_as_blocked(monkeypatch):
+ """If we cannot rotate away from a blocked IP, we are blocked. Say so."""
+ monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: False)
+
+ client = _StubClient([403] * 10)
+ with pytest.raises(BlockedError):
+ http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
+
+
+def test_success_resets_the_consecutive_count(monkeypatch):
+ """Interleaved successes mean the IP is fine, so 403s must not accumulate."""
+ rotations = []
+ monkeypatch.setattr(
+ http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
+ )
+
+ # 403,200 repeated: never 3 consecutive, so never a rotation.
+ for _ in range(5):
+ client = _StubClient([403, 200], payload={"matches": ["ok"]})
+ http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
+
+ assert rotations == []
+
+
+def test_on_403_false_opts_out_of_block_detection(monkeypatch):
+ monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)
+ client = _StubClient([403])
+ assert (
+ http_client.fetch_with_retry(client, "https://x.example.com/y", on_403=False)
+ is None
+ )
+
+
+def test_403_tracking_is_per_host(monkeypatch):
+ """A block on one host must not be charged against another."""
+ rotations = []
+ monkeypatch.setattr(
+ http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
+ )
+
+ for host in ("a.example.com", "b.example.com", "c.example.com"):
+ client = _StubClient([403, 200], payload={"matches": []})
+ http_client.fetch_with_retry(client, f"https://{host}/typeahead")
+
+ assert rotations == [], "one 403 each across three hosts is not a block"
+
+
+def test_connection_errors_still_retry(monkeypatch):
+ """The rewritten loop must not regress non-403 retry behaviour."""
+
+ class _FlakyClient:
+ def __init__(self):
+ self.calls = 0
+
+ def get(self, url, params=None, headers=None):
+ self.calls += 1
+ if self.calls == 1:
+ raise httpx.ConnectError("boom")
+ return _StubResponse(200, {"matches": ["ok"]})
+
+ client = _FlakyClient()
+ assert http_client.fetch_with_retry(client, "https://x.example.com/y") == {
+ "matches": ["ok"]
+ }
+ assert client.calls == 2
+
+
+# ---------------------------------------------------------------------------
+# Run-level sanity gates
+# ---------------------------------------------------------------------------
+
+
+def test_zero_yield_from_a_selected_source_is_a_failure(tmp_path):
+ results = {"rightmove": [], "onthemarket": [{"a": 1}], "zoopla": []}
+ failures = scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}],
+ output_path=tmp_path / "missing.parquet",
+ abandoned=set(),
+ )
+ assert len(failures) == 1
+ assert "rightmove yielded 0 listings" in failures[0]
+
+
+def test_healthy_run_has_no_failures(tmp_path):
+ results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
+ assert (
+ scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}, {"b": 2}],
+ output_path=tmp_path / "missing.parquet",
+ abandoned=set(),
+ )
+ == []
+ )
+
+
+def test_unselected_source_yielding_zero_is_fine(tmp_path):
+ """zoopla is not scheduled in production; its 0 must not fail the run."""
+ results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
+ assert (
+ scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}],
+ output_path=tmp_path / "missing.parquet",
+ abandoned=set(),
+ )
+ == []
+ )
+
+
+def test_source_abandoned_mid_run_is_a_failure_even_with_listings(tmp_path):
+ """A block that starts mid-run must fail even though the source is non-empty.
+
+ Without this gate a single banked listing defeats the zero-yield check: the
+ source stops at outcode 300 of 366, keeps its 60k listings, OnTheMarket
+ backfills the merged total to within the drop limit, and the partial
+ dataset publishes with exit 0. That is the original incident with one
+ outcode's head start.
+ """
+ import polars as pl
+
+ path = tmp_path / "online_listings_buy.parquet"
+ pl.DataFrame({"x": range(103008)}).write_parquet(path)
+
+ results = {
+ "rightmove": [{"a": 1}] * 60000,
+ "onthemarket": [{"b": 2}] * 83785,
+ "zoopla": [],
+ }
+ failures = scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}] * 99000, # only a 4% dip: both other gates pass
+ output_path=path,
+ abandoned={"rightmove"},
+ )
+ assert len(failures) == 1
+ assert "abandoned mid-run" in failures[0]
+ assert "60000 listings cover only part" in failures[0]
+
+
+def test_the_real_incident_drop_now_trips_the_collapse_gate(tmp_path):
+ """103,008 -> 83,785 is an 18.7% drop, which the original 25% limit missed."""
+ import polars as pl
+
+ path = tmp_path / "online_listings_buy.parquet"
+ pl.DataFrame({"x": range(103008)}).write_parquet(path)
+
+ # Pretend rightmove banked listings so only the collapse gate can fire.
+ results = {
+ "rightmove": [{"a": 1}] * 100,
+ "onthemarket": [{"b": 2}] * 83785,
+ "zoopla": [],
+ }
+ failures = scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"b": 2}] * 83785,
+ output_path=path,
+ abandoned=set(),
+ )
+ assert len(failures) == 1
+ assert "collapsed from 103008 to 83785" in failures[0]
+
+
+def test_row_count_collapse_is_a_failure(tmp_path):
+ """Backstop for degradation that is neither a clean zero nor a raised block.
+
+ Dedup masks a lost source (OnTheMarket backfilled the keys Rightmove used to
+ win), so a partial block can still look plausible per-source.
+ """
+ import polars as pl
+
+ path = tmp_path / "online_listings_buy.parquet"
+ pl.DataFrame({"x": range(1000)}).write_parquet(path)
+
+ results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
+ failures = scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}] * 700, # 30% drop, over the 10% limit
+ output_path=path,
+ abandoned=set(),
+ )
+ assert len(failures) == 1
+ assert "collapsed from 1000 to 700" in failures[0]
+
+
+def test_normal_market_movement_is_not_a_failure(tmp_path):
+ import polars as pl
+
+ path = tmp_path / "online_listings_buy.parquet"
+ pl.DataFrame({"x": range(1000)}).write_parquet(path)
+
+ results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
+ assert (
+ scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}] * 970, # 3% drop, within normal movement
+ output_path=path,
+ abandoned=set(),
+ )
+ == []
+ )
+
+
+def test_first_ever_run_has_no_baseline_to_compare(tmp_path):
+ results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
+ assert (
+ scraper._sanity_failures(
+ results,
+ ["rightmove", "onthemarket"],
+ merged=[{"a": 1}],
+ output_path=tmp_path / "does-not-exist.parquet",
+ abandoned=set(),
+ )
+ == []
+ )
diff --git a/finder/test_scraper_concurrency.py b/finder/test_scraper_concurrency.py
index 140ed60..875e1b9 100644
--- a/finder/test_scraper_concurrency.py
+++ b/finder/test_scraper_concurrency.py
@@ -31,7 +31,7 @@ def test_run_sources_runs_every_source_and_isolates_failures():
order.append("zoo")
errors: list[str] = []
- scraper._run_sources(zoo, [("rightmove", rm), ("onthemarket", otm)], errors)
+ scraper._run_sources(zoo, [("rightmove", rm), ("onthemarket", otm)], errors, set())
assert set(order) == {"rm", "otm", "zoo"}
# The failing source is recorded but did not stop the others.
@@ -45,14 +45,14 @@ def test_run_sources_records_zoopla_failure():
def zoo():
raise ValueError("zoo down")
- scraper._run_sources(zoo, [], errors)
+ scraper._run_sources(zoo, [], errors, set())
assert any("zoopla" in e and "zoo down" in e for e in errors)
def test_run_sources_handles_no_background_runners():
ran = []
errors: list[str] = []
- scraper._run_sources(lambda: ran.append("z"), [], errors)
+ scraper._run_sources(lambda: ran.append("z"), [], errors, set())
assert ran == ["z"]
assert errors == []
@@ -60,7 +60,7 @@ def test_run_sources_handles_no_background_runners():
def test_run_sources_handles_zoopla_absent():
ran = []
errors: list[str] = []
- scraper._run_sources(None, [("rightmove", lambda: ran.append("rm"))], errors)
+ scraper._run_sources(None, [("rightmove", lambda: ran.append("rm"))], errors, set())
assert ran == ["rm"]
assert errors == []
diff --git a/finder/zoopla.py b/finder/zoopla.py
index 0d53d9c..ff457d2 100644
--- a/finder/zoopla.py
+++ b/finder/zoopla.py
@@ -27,14 +27,11 @@ import time
from pathlib import Path
from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
-import httpx
-
+import gluetun
import shutdown
from constants import (
DATA_DIR,
DELAY_BETWEEN_PAGES,
- GLUETUN_API_KEY,
- GLUETUN_CONTROL_URL,
GLUETUN_MAX_ROTATIONS,
GLUETUN_PROXY,
MAX_BEDROOMS,
@@ -472,110 +469,16 @@ def _challenge_timeout_seconds() -> int:
# ---------------------------------------------------------------------------
#
# When Cloudflare Turnstile fires mid-scrape, the cheapest unblocker is to
-# swap the egress IP via Gluetun's HTTP control server. We stop and re-start
-# the VPN, poll until the public IP changes, drop the stale cf_clearance
-# cookies (bound to the previous IP), then reload and re-check the challenge.
-
-
-def _gluetun_base_url() -> str:
- return GLUETUN_CONTROL_URL.rstrip("/")
-
-
-def _gluetun_api_key() -> str | None:
- return GLUETUN_API_KEY
+# swap the egress IP via Gluetun's HTTP control server, then drop the stale
+# cf_clearance cookies (bound to the previous IP) and re-check the challenge.
+# The control-server plumbing itself lives in gluetun.py, shared with
+# http_client.py's egress-block detection.
def _gluetun_max_rotations() -> int:
return max(GLUETUN_MAX_ROTATIONS, 0)
-def _gluetun_client() -> httpx.Client:
- # Talks to the control server directly (not through the VPN proxy).
- headers = {}
- api_key = _gluetun_api_key()
- if api_key:
- headers["X-API-Key"] = api_key
- return httpx.Client(headers=headers)
-
-
-def _gluetun_public_ip(client: httpx.Client) -> str | None:
- try:
- resp = client.get(f"{_gluetun_base_url()}/v1/publicip/ip", timeout=5.0)
- if resp.status_code != 200:
- return None
- data = resp.json()
- except (httpx.HTTPError, ValueError):
- return None
- return data.get("public_ip") or data.get("ip")
-
-
-def _gluetun_set_vpn_status(client: httpx.Client, status: str) -> bool:
- """PUT /v1/vpn/status with {'status': status}. Returns True on 2xx."""
- try:
- resp = client.put(
- f"{_gluetun_base_url()}/v1/vpn/status",
- json={"status": status},
- timeout=15.0,
- )
- except httpx.HTTPError as exc:
- log.warning("Gluetun vpn/status %s failed: %s", status, exc)
- return False
- if resp.status_code == 401:
- log.warning(
- "Gluetun vpn/status %s: 401 Unauthorized. The API key must be "
- "authorised for 'PUT /v1/vpn/status' in Gluetun's auth config.toml",
- status,
- )
- return False
- if resp.status_code >= 400:
- log.warning(
- "Gluetun vpn/status %s returned HTTP %d: %s",
- status, resp.status_code, resp.text[:200],
- )
- return False
- return True
-
-
-def _rotate_gluetun_ip(wait_seconds: int = 45) -> bool:
- """Restart Gluetun's VPN and wait for the public IP to change.
-
- Returns True if a new IP was observed within wait_seconds."""
- with _gluetun_client() as client:
- old_ip = _gluetun_public_ip(client)
- log.info("Requesting Gluetun IP rotation (current IP: %s)", old_ip or "unknown")
-
- stop_attempted = False
- restart_confirmed = False
- try:
- stop_attempted = True
- if not _gluetun_set_vpn_status(client, "stopped"):
- return False
- time.sleep(2)
- restart_confirmed = _gluetun_set_vpn_status(client, "running")
- if not restart_confirmed:
- return False
-
- deadline = time.monotonic() + wait_seconds
- while time.monotonic() < deadline:
- time.sleep(2)
- new_ip = _gluetun_public_ip(client)
- if new_ip and new_ip != old_ip:
- log.info("Gluetun rotated IP: %s -> %s", old_ip or "?", new_ip)
- return True
- finally:
- if stop_attempted and not restart_confirmed:
- log.warning(
- "Gluetun VPN may be stopped after failed rotation; attempting recovery start"
- )
- if not _gluetun_set_vpn_status(client, "running"):
- log.error(
- "Gluetun VPN recovery start failed; manual intervention required"
- )
-
- log.warning("Gluetun IP did not change within %ds", wait_seconds)
- return False
-
-
def _clear_cloudflare_cookies(page) -> None:
"""Drop cf_clearance / __cf_bm which are bound to the previous egress IP."""
try:
@@ -596,7 +499,7 @@ def _rotate_and_retry_challenge(page, max_rotations: int) -> bool:
"Cloudflare Turnstile challenge, rotating Gluetun IP (attempt %d/%d)",
attempt, max_rotations,
)
- if not _rotate_gluetun_ip():
+ if not gluetun.rotate_ip():
continue
_clear_cloudflare_cookies(page)
diff --git a/frontend/src/components/map/Map.tsx b/frontend/src/components/map/Map.tsx
index c7bf6e8..8923b97 100644
--- a/frontend/src/components/map/Map.tsx
+++ b/frontend/src/components/map/Map.tsx
@@ -1,7 +1,7 @@
import { useCallback, useRef, useEffect, useState, useMemo, memo } from 'react';
import type { CSSProperties } from 'react';
import { useTranslation } from 'react-i18next';
-import { Map as MapGL, NavigationControl, ScaleControl } from 'react-map-gl/maplibre';
+import { Map as MapGL, ScaleControl } from 'react-map-gl/maplibre';
import type { MapRef } from 'react-map-gl/maplibre';
import 'maplibre-gl/dist/maplibre-gl.css';
import type {
@@ -350,6 +350,16 @@ export default memo(function Map({
}, [screenshotMode]);
const handleLoad = useCallback(() => {
+ // touchZoomRotate is a single handler for pinch-zoom AND pinch-rotate, so
+ // leaving it enabled for the zoom would also let touch users rotate the map
+ // off north with no compass to reset it. Split them: keep the zoom, drop the
+ // rotate, matching dragRotate/pitchWithRotate being off.
+ const map = mapRef.current?.getMap();
+ map?.touchZoomRotate.disableRotation();
+ // Same split for the keyboard. Its disableRotation docstring claims it kills
+ // panning too, but the handler only zeroes the bearing/pitch deltas, so plain
+ // arrow-key panning survives and Shift+arrow rotate/tilt does not.
+ map?.keyboard.disableRotation();
setMapReady(true);
}, []);
@@ -550,9 +560,6 @@ export default memo(function Map({
zoom={viewState.zoom}
/>