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