From 7351993feeefceebf375667ad1ced285c1b6d7ca Mon Sep 17 00:00:00 2001 From: Jens Date: Sun, 23 Aug 2026 00:13:38 +0200 Subject: [PATCH] stop paging when the provider stops making progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An ArcGIS layer without supportsPagination accepts resultOffset and ignores it, answering every page with the first one. The VHA profile reader advanced its offset by the page length and stopped at the announced count, so for a count that is a multiple of the page size it collected N copies of page one — and its completeness check, len(features) == candidate_count, passed. Four announced records became four stored records, two of them duplicates, filed under an official provenance. That is the substitution bounded acquisition exists to prevent, arriving through the front door. The reader now refuses a record it already collected. It fails rather than silently dropping the duplicate: a provider that cannot page is a provider whose count proves nothing, so a smaller-but-clean result would still be unverifiable. Its watercourse-name loop was worse — a bare `while True` that ended only when the provider stopped setting exceededTransferLimit, with names deduplicated by code so a stuck provider produced no visible change while the requests continued. It now refuses a repeated page body, and both loops have the page budget the sibling readers already had. Those siblings turned out to be fine. GRB and official vector already refuse a repeated page URL, bound the page count, and deduplicate on feature identity — but none of it had a test, so none of it was known to work. Exercised now, including the case where distinct URLs defeat the loop check and the budget is the only backstop. A duplicate across two genuinely different pages is kept once rather than failing, because a cursor over a changing table produces that legitimately. Also: _bash_path fell back to the raw path whenever wslpath failed, except on timeout, which propagated and reddened the suite when starting WSL took more than ten seconds under load. Co-Authored-By: Claude Opus 5 --- .env.example | 1 + backend/app/core/config.py | 6 + .../bathymetry_profile_acquisition_service.py | 69 +++++++++- .../test_bathymetry_pagination_integrity.py | 128 ++++++++++++++++++ backend/tests/test_paged_acquisition_stops.py | 120 ++++++++++++++++ ...test_sprint159_split_promotion_workflow.py | 20 ++- docs/KNOWN_LIMITATIONS.md | 10 ++ 7 files changed, 346 insertions(+), 8 deletions(-) create mode 100644 backend/tests/test_bathymetry_pagination_integrity.py create mode 100644 backend/tests/test_paged_acquisition_stops.py diff --git a/.env.example b/.env.example index 7f3e31af..17f5cae1 100644 --- a/.env.example +++ b/.env.example @@ -83,6 +83,7 @@ BATHYMETRY_PROFILES_ENABLED=true BATHYMETRY_PROFILES_LAYER_URL=https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/0 BATHYMETRY_WATERCOURSE_LAYER_URL=https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/1 BATHYMETRY_PROFILES_PAGE_SIZE=1000 +BATHYMETRY_PROFILES_MAX_PAGES=200 BATHYMETRY_PROFILES_MAX_FEATURES=50000 BATHYMETRY_PROFILES_TIMEOUT_SECONDS=120 BATHYMETRY_PROFILES_MAX_RESPONSE_MB=32 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index b1829ac6..e530c78d 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -255,6 +255,12 @@ class Settings(BaseSettings): le=250_000, validation_alias="BATHYMETRY_PROFILES_MAX_FEATURES", ) + bathymetry_profiles_max_pages: int = Field( + default=200, + ge=1, + le=5_000, + validation_alias="BATHYMETRY_PROFILES_MAX_PAGES", + ) bathymetry_profiles_timeout_seconds: int = Field( default=120, ge=1, diff --git a/backend/app/services/bathymetry_profile_acquisition_service.py b/backend/app/services/bathymetry_profile_acquisition_service.py index 97e4eb78..8bc0e0fe 100644 --- a/backend/app/services/bathymetry_profile_acquisition_service.py +++ b/backend/app/services/bathymetry_profile_acquisition_service.py @@ -276,6 +276,45 @@ class BathymetryProfileAcquisitionService: "spatialRel": "esriSpatialRelIntersects", } + @staticmethod + def _unseen_records( + page_features: list[Any], + seen_object_ids: set[str], + ) -> list[dict[str, Any]]: + """Every page must bring records the earlier pages did not. + + An ArcGIS layer without ``supportsPagination`` accepts ``resultOffset`` + and ignores it, answering every page with the first one. Advancing the + offset by the page length still reaches the announced count, so the + completeness check below passed while the dataset held N copies of page + one — a silent substitution of the source data, which is the one thing + bounded acquisition exists to prevent. + """ + + fresh: list[dict[str, Any]] = [] + for item in page_features: + if not isinstance(item, dict): + continue + attributes = item.get("attributes") + object_id = attributes.get("OBJECTID") if isinstance(attributes, dict) else None + if object_id is None: + raise AppError( + code="BATHYMETRY_PROVIDER_INVALID_RESPONSE", + message="VHA profile record has no OBJECTID, so pagination cannot be verified", + status_code=502, + ) + key = str(object_id) + if key in seen_object_ids: + raise AppError( + code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION", + message="VHA profile pagination repeated a record; the layer is not honouring resultOffset", + details={"object_id": key}, + status_code=502, + ) + seen_object_ids.add(key) + fresh.append(item) + return fresh + @staticmethod def _fetch_profiles( bbox_values: tuple[float, float, float, float], @@ -307,8 +346,16 @@ class BathymetryProfileAcquisitionService: features: list[dict[str, Any]] = [] response_hashes: list[str] = [] request_urls: list[str] = [count_url] + seen_object_ids: set[str] = set() offset = 0 while offset < candidate_count: + if len(request_urls) > settings.bathymetry_profiles_max_pages: + raise AppError( + code="BATHYMETRY_SCOPE_TOO_LARGE", + message="VHA profile pagination exceeded the configured page limit; acquire smaller area partitions", + details={"max_pages": settings.bathymetry_profiles_max_pages}, + status_code=422, + ) page_url = BathymetryProfileAcquisitionService._query_url( base, { @@ -328,11 +375,13 @@ class BathymetryProfileAcquisitionService: message="VHA profile response does not contain a feature list", status_code=502, ) - features.extend(item for item in page_features if isinstance(item, dict)) response_hashes.append(page_sha) request_urls.append(page_url) if not page_features: break + features.extend( + BathymetryProfileAcquisitionService._unseen_records(page_features, seen_object_ids) + ) offset += len(page_features) if len(features) != candidate_count: raise AppError( @@ -363,7 +412,15 @@ class BathymetryProfileAcquisitionService: for start in range(0, len(ordered_codes), 100): chunk = ordered_codes[start : start + 100] offset = 0 + seen_page_hashes: set[str] = set() while True: + if len(seen_page_hashes) >= settings.bathymetry_profiles_max_pages: + raise AppError( + code="BATHYMETRY_SCOPE_TOO_LARGE", + message="VHA watercourse pagination exceeded the configured page limit", + details={"max_pages": settings.bathymetry_profiles_max_pages}, + status_code=422, + ) url = BathymetryProfileAcquisitionService._query_url( base, { @@ -386,6 +443,16 @@ class BathymetryProfileAcquisitionService: message="VHA watercourse response does not contain a feature list", status_code=502, ) + if response_sha in seen_page_hashes: + # The names themselves deduplicate by code, so a stuck + # provider produced no visible change while the loop, which + # ended only on exceededTransferLimit, kept requesting. + raise AppError( + code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION", + message="VHA watercourse pagination returned the same page again", + status_code=502, + ) + seen_page_hashes.add(response_sha) for feature in page_features: attributes = feature.get("attributes") if isinstance(feature, dict) else None if not isinstance(attributes, dict): diff --git a/backend/tests/test_bathymetry_pagination_integrity.py b/backend/tests/test_bathymetry_pagination_integrity.py new file mode 100644 index 00000000..d2b273ff --- /dev/null +++ b/backend/tests/test_bathymetry_pagination_integrity.py @@ -0,0 +1,128 @@ +"""A provider that ignores ``resultOffset`` must not produce duplicated data. + +ArcGIS layers without ``supportsPagination`` accept ``resultOffset`` and ignore +it, answering every page with the first one. The profile reader advanced its +offset by the page length and stopped when it reached the announced count, so +for a count that is a multiple of the page size it collected N copies of page +one, matched the expected total exactly, and stored that as an official +dataset. The watercourse-name reader had no bound at all: it looped for as long +as the provider kept setting ``exceededTransferLimit``. + +The sibling reader for official vector products already refuses a repeated page +and deduplicates on feature identity. These tests hold this reader to the same +rule. +""" + +from __future__ import annotations + +import json +from typing import Any +from urllib.parse import parse_qs, urlparse + +import pytest + +from app.core.config import Settings, get_settings +from app.core.errors import AppError +from app.services.bathymetry_profile_acquisition_service import ( + BathymetryProfileAcquisitionService, +) + +BBOX = (4.30, 51.20, 4.32, 51.22) + + +def _settings(**overrides: Any) -> Settings: + base = get_settings() + return base.model_copy(update={"bathymetry_profiles_page_size": 2, **overrides}) + + +class _Response: + def __init__(self, payload: dict[str, Any]) -> None: + self._body = json.dumps(payload).encode("utf-8") + + def read(self, _limit: int | None = None) -> bytes: + return self._body + + def __enter__(self) -> "_Response": + return self + + def __exit__(self, *_exc: object) -> bool: + return False + + +def _profile(object_id: int) -> dict[str, Any]: + return { + "attributes": {"OBJECTID": object_id, "vhag": 7, "opg_diepte": 1.5}, + "geometry": {"x": 4.31, "y": 51.21}, + } + + +class StuckProvider: + """Answers every page with the same records, as an unpaged layer does.""" + + def __init__(self, *, count: int, page: list[dict[str, Any]]) -> None: + self.count = count + self.page = page + self.requests: list[str] = [] + + def __call__(self, request: Any, **_kwargs: Any) -> _Response: + url = request.full_url if hasattr(request, "full_url") else str(request) + self.requests.append(url) + query = parse_qs(urlparse(url).query) + if query.get("returnCountOnly") == ["true"]: + return _Response({"count": self.count}) + return _Response({"features": list(self.page), "exceededTransferLimit": True}) + + +def test_a_provider_that_ignores_the_offset_is_refused_not_duplicated() -> None: + """Four announced records, two per page, the same two every time. + + Advancing by page length reaches the announced total after two pages, so the + completeness check passed while every record was stored twice. + """ + + provider = StuckProvider(count=4, page=[_profile(1), _profile(2)]) + + with pytest.raises(AppError) as exc_info: + BathymetryProfileAcquisitionService._fetch_profiles(BBOX, _settings(), provider) + + assert exc_info.value.code == "BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION" + assert exc_info.value.status_code == 502 + + +def test_honest_pagination_still_returns_every_record() -> None: + """The guard must not reject a provider that pages correctly.""" + + pages = [[_profile(1), _profile(2)], [_profile(3), _profile(4)]] + + def opener(request: Any, **_kwargs: Any) -> _Response: + url = request.full_url + query = parse_qs(urlparse(url).query) + if query.get("returnCountOnly") == ["true"]: + return _Response({"count": 4}) + offset = int(query.get("resultOffset", ["0"])[0]) + index = offset // 2 + page = pages[index] if index < len(pages) else [] + return _Response({"features": page}) + + features, provenance = BathymetryProfileAcquisitionService._fetch_profiles( + BBOX, _settings(), opener + ) + + assert [item["attributes"]["OBJECTID"] for item in features] == [1, 2, 3, 4] + assert provenance["candidate_count"] == 4 + + +def test_watercourse_names_stop_instead_of_looping_forever() -> None: + """``exceededTransferLimit`` forever is not a reason to request forever.""" + + provider = StuckProvider( + count=0, + page=[{"attributes": {"wlasvl.vhag": 7, "VHAG_TABEL.naam": "Schelde"}}], + ) + + with pytest.raises(AppError) as exc_info: + BathymetryProfileAcquisitionService._fetch_watercourse_names({7}, _settings(), provider) + + assert exc_info.value.code == "BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION" + # Bounded, and bounded early: it must not have hammered the provider first. + assert len(provider.requests) <= 3 diff --git a/backend/tests/test_paged_acquisition_stops.py b/backend/tests/test_paged_acquisition_stops.py new file mode 100644 index 00000000..72fd0ae6 --- /dev/null +++ b/backend/tests/test_paged_acquisition_stops.py @@ -0,0 +1,120 @@ +"""The paged readers' loop protections, exercised rather than assumed. + +GRB and official vector both refuse a repeated page URL and bound the page +count, and GRB deduplicates on feature identity. None of that had a test, so +none of it was known to work — the same category as the redirect handler that +turned out to be dead code while looking like protection. + +A provider that answers every page with a "next" link pointing back at itself +is not hypothetical: it is what a misconfigured cursor or a caching proxy in +front of an OGC endpoint produces. +""" + +from __future__ import annotations + +import pytest +from shapely.geometry import Polygon + +from app.core.config import Settings +from app.core.errors import AppError +from app.services.grb_acquisition_service import GrbAcquisitionService +from tests.test_sprint239_bounded_grb_acquisition import JsonResponse, polygon_feature + +COLLECTION_ITEMS = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items" +SCOPE = Polygon([(5.15, 51.18), (5.17, 51.18), (5.17, 51.20), (5.15, 51.20), (5.15, 51.18)]) + + +def _building(feature_id: str) -> dict: + return polygon_feature( + feature_id, + [(5.155, 51.185), (5.175, 51.185), (5.175, 51.195), (5.155, 51.195), (5.155, 51.185)], + ) + + +def _fetch(opener): + return GrbAcquisitionService._fetch_features( + GrbAcquisitionService._product("buildings"), + SCOPE, + SCOPE.bounds, + "bounded_selection", + Settings(_env_file=None), + opener, + ) + + +def test_a_next_link_pointing_at_itself_is_refused() -> None: + requests: list[str] = [] + + def opener(request, timeout): # noqa: ARG001 + requests.append(request.full_url) + return JsonResponse( + { + "type": "FeatureCollection", + "features": [_building("GBG.1")], + "links": [{"rel": "next", "href": f"{COLLECTION_ITEMS}?cursor=stuck"}], + } + ) + + with pytest.raises(AppError) as exc_info: + _fetch(opener) + + assert exc_info.value.code == "GRB_PROVIDER_PAGINATION_LOOP" + # Refused on the second sighting, not after exhausting the page budget. + assert len(requests) == 2 + + +def test_an_endless_chain_of_fresh_pages_stops_at_the_page_limit() -> None: + """Distinct URLs defeat the loop check, so the page budget is the backstop.""" + + settings = Settings(_env_file=None) + requests: list[str] = [] + + def opener(request, timeout): # noqa: ARG001 + requests.append(request.full_url) + cursor = len(requests) + return JsonResponse( + { + "type": "FeatureCollection", + "features": [_building(f"GBG.{cursor}")], + "links": [{"rel": "next", "href": f"{COLLECTION_ITEMS}?cursor=p{cursor}"}], + } + ) + + with pytest.raises(AppError) as exc_info: + _fetch(opener) + + assert exc_info.value.code == "GRB_SELECTION_TOO_LARGE" + assert exc_info.value.status_code == 422 + assert len(requests) == settings.grb_max_pages + + +def test_a_repeated_feature_across_pages_is_counted_once() -> None: + """Two pages, distinct URLs, overlapping content. + + Unlike a repeated URL this is not necessarily provider misbehaviour — a + cursor over a changing table can hand back a record twice — so the reader + keeps it once rather than failing the acquisition. + """ + + def opener(request, timeout): # noqa: ARG001 + if "cursor=next" in request.full_url: + return JsonResponse( + { + "type": "FeatureCollection", + "features": [_building("GBG.1"), _building("GBG.2")], + "links": [], + } + ) + return JsonResponse( + { + "type": "FeatureCollection", + "features": [_building("GBG.1")], + "links": [{"rel": "next", "href": f"{COLLECTION_ITEMS}?cursor=next"}], + } + ) + + features, transfer = _fetch(opener) + + assert {feature["id"] for feature in features} == {"GBG:GBG.1", "GBG:GBG.2"} + assert transfer["candidate_feature_count"] == 3 + assert transfer["feature_count"] == 2 diff --git a/backend/tests/test_sprint159_split_promotion_workflow.py b/backend/tests/test_sprint159_split_promotion_workflow.py index b4b2f269..1d7b3d4a 100644 --- a/backend/tests/test_sprint159_split_promotion_workflow.py +++ b/backend/tests/test_sprint159_split_promotion_workflow.py @@ -15,13 +15,19 @@ def _bash_path(path: Path) -> str: if os.name != "nt": return raw_path - result = subprocess.run( - ["bash", "-lc", f"wslpath -a {shlex.quote(raw_path)}"], - capture_output=True, - text=True, - timeout=10, - check=False, - ) + try: + result = subprocess.run( + ["bash", "-lc", f"wslpath -a {shlex.quote(raw_path)}"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except (subprocess.TimeoutExpired, OSError): + # Starting WSL can exceed ten seconds while the rest of the suite is + # running. The fallback below is what this helper already does whenever + # the conversion does not work, so a slow shell must not red the suite. + return raw_path if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip() return raw_path diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index 39bb28ae..d6387576 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -71,6 +71,16 @@ runtime source of truth. that legitimately moves to a new host therefore fails closed until the operator updates the configured URL, which is the intended trade: bytes from an unexpected host must never be persisted under an official provenance. +- A paged reader stops when the provider stops making progress: a repeated page + URL, a repeated page body, or a record already collected all fail the + acquisition, and every reader has a page budget. An ArcGIS layer that lacks + `supportsPagination` accepts `resultOffset` and ignores it, which otherwise + yields a dataset holding several copies of page one while matching the + announced record count exactly. Failing is deliberate: a provider that cannot + page is a provider whose count proves nothing, so its data cannot carry an + official provenance. A duplicate record arriving across two genuinely + different pages is kept once instead, since a cursor over a changing table + can produce that legitimately. ## Historical analysis