From 41615b80ec637d8bd27da3066c8c6da261fd45dd Mon Sep 17 00:00:00 2001 From: prbatero <42007693+prbatero@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:30:41 -0400 Subject: [PATCH] perf(project-details): optimize loading and conditional caching Move project-detail orchestration into a keyed processor, preserve legacy storage-key joins, batch related reads, add a bounded process-local single-flight cache with ETag and refresh semantics, and cover the API contract and stats path. --- api/hastefuncapi/README.md | 13 +- api/hastefuncapi/function_app.py | 212 ++++---- .../tests/test_generate_project_stats.py | 86 ++++ .../tests/test_model_artifact_route.py | 66 +++ .../tests/test_project_details_route.py | 183 +++++++ docker/nginx.conf | 5 +- docs/api-overview.md | 20 + docs/api/hastefuncapi.md | 9 +- docs/configuration.md | 19 + .../core/processors/project_details.py | 213 ++++++++ .../src/hastegeo/core/utils/async_cache.py | 103 ++++ .../core/processors/test_project_details.py | 463 ++++++++++++++++++ hastelib/tests/core/utils/test_async_cache.py | 180 +++++++ .../tools/bench_api_http.py | 27 +- 14 files changed, 1467 insertions(+), 132 deletions(-) create mode 100644 api/hastefuncapi/tests/test_generate_project_stats.py create mode 100644 api/hastefuncapi/tests/test_model_artifact_route.py create mode 100644 api/hastefuncapi/tests/test_project_details_route.py create mode 100644 hastelib/src/hastegeo/core/processors/project_details.py create mode 100644 hastelib/src/hastegeo/core/utils/async_cache.py create mode 100644 hastelib/tests/core/processors/test_project_details.py create mode 100644 hastelib/tests/core/utils/test_async_cache.py diff --git a/api/hastefuncapi/README.md b/api/hastefuncapi/README.md index 909efe18..cfd5bac2 100644 --- a/api/hastefuncapi/README.md +++ b/api/hastefuncapi/README.md @@ -21,11 +21,22 @@ All functions are defined in `function_app.py` as a single Azure Functions app. |--------|-------|-------------| | GET | `GetDashboardData` | Aggregated dashboard stats: project summaries, layer info, model status, and system-wide metrics. | | GET | `GetProjects` | All projects with aggregated layer and model counts. | -| GET | `GetProjectDetails` | Full project details including image layers, models, and processing status. Requires `projectId`. | +| GET | `GetProjectDetails` | Project, layer, validation, and optional model details. Supports `ETag`/`If-None-Match`; requires `projectId`. | | PUT | `PutProject` | Create or update a project. Auto-generates `projectId` and `creationDate` if not provided. | | DELETE | `DeleteProject` | Delete a project by `projectId`. | | GET | `GenerateProjectStats` | Regenerates project stats from raw data — useful if stats fall out of sync. | +#### Project Detail Caching + +`GetProjectDetails` returns `ETag`, `Cache-Control`, and `X-Haste-Cache` headers. +Send `If-None-Match` to receive an empty `304` when the fresh cached representation is +unchanged. Send `Cache-Control: no-cache` after a mutation to force storage refresh and +then compare the ETag. + +The response cache is bounded and process-local. It deduplicates concurrent requests but +does not provide coherence across scaled-out Function workers. Performance headers named +`X-Haste-Data-Layer-*` count logical calls, not Azure Storage REST transactions. + ### Image Layers | Method | Route | Description | diff --git a/api/hastefuncapi/function_app.py b/api/hastefuncapi/function_app.py index 466f741f..39fa3bdc 100644 --- a/api/hastefuncapi/function_app.py +++ b/api/hastefuncapi/function_app.py @@ -4,6 +4,7 @@ import asyncio import base64 import binascii +import hashlib import json import os import re @@ -44,6 +45,7 @@ from hastegeo.core.processors.imagery import ImageryPreProcessor from hastegeo.core.processors.inference import InferencePreprocessor from hastegeo.core.processors.metadata import MetadataProcessor +from hastegeo.core.processors.project_details import ProjectDetailsProcessor from hastegeo.core.processors.publishing import ( PublishingDependencyError, PublishingDisabledError, @@ -74,6 +76,10 @@ PublishingSourceResolver, ) from hastegeo.core.utils import perf +from hastegeo.core.utils.async_cache import ( + AsyncTTLCache, + configured_cache_value, +) from hastegeo.core.utils.blob import ( download_blob_to_tempfile, parse_byte_range, @@ -106,6 +112,17 @@ ) app = func.FunctionApp() +_PROJECT_DETAILS_CACHE_SECONDS = configured_cache_value( + "HASTE_PROJECTDETAILS_CACHE_SECONDS", 15, 0, 300 +) +_PROJECT_DETAILS_CACHE_ENTRIES = configured_cache_value( + "HASTE_PROJECTDETAILS_CACHE_ENTRIES", 64, 1, 512 +) +_project_details_cache = AsyncTTLCache( + ttl_seconds=_PROJECT_DETAILS_CACHE_SECONDS, + max_entries=_PROJECT_DETAILS_CACHE_ENTRIES, +) + # Development mode check - when running locally with Docker/Azurite # Set DEVELOPMENT_MODE=true to disable function key authentication DEVELOPMENT_MODE = ( @@ -142,6 +159,25 @@ def _require_guid_param(req: func.HttpRequest, name: str) -> str: return value +def _etag_matches(if_none_match: str | None, etag: str) -> bool: + if not if_none_match: + return False + for candidate in if_none_match.split(","): + candidate = candidate.strip() + if candidate == "*" or candidate.removeprefix("W/") == etag: + return True + return False + + +def _cache_refresh_requested(cache_control: str | None) -> bool: + if not cache_control: + return False + directives = { + directive.strip().lower() for directive in cache_control.split(",") + } + return "no-cache" in directives or "max-age=0" in directives + + def _require_short_int_id_param(req: func.HttpRequest, name: str) -> str: """Return a request parameter validated as a short integer id, or raise ValueError. @@ -743,111 +779,30 @@ async def GetProjectDetails(req: func.HttpRequest) -> func.HttpResponse: _perf = perf.begin(_perf_on) _perf_wall = time.perf_counter() - project = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().PROJECT.value, - partition_key=project_id, - ).load, - project_id, - ) - image_layers = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().IMAGELAYER.value, - partition_key=project_id, - ).load_all_from_partition - ) - if include_models: - models = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().MODEL.value, - partition_key=project_id, - ).load_all_from_partition - ) - for image_layer in image_layers: - image_layer_id = image_layer["imageLayerId"] - if include_models: - match_models = [ - model - for model in models - if model["imageLayerId"] == image_layer_id - ] - match_models.sort( - key=lambda x: x["creationDate"], reverse=True - ) - for model in match_models: - try: - artifacts = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().MODEL_ARTIFACTS.value, - partition_key=project_id, - ).load, - model["modelId"], - ) - model["artifacts"] = artifacts - except FileNotFoundError: - model["artifacts"] = None - try: - if not model.get("labelsUrl"): - # Older models may not have labelsUrl - model["labelsUrl"] = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().TRAIN_LABELS.value, - partition_key=project_id, - ).export, - key=model["modelId"], - data_format="geojson", - ) - except FileNotFoundError: - # This is a noop until the export method is properly - # implemented - model["labelsUrl"] = None - image_layer["models"] = match_models - image_layer["modelCount"] = len(match_models) - label_projects = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().LABELS.value, - partition_key=project_id, - ).load_all_from_partition - ) - match_label_projects = next( - ( - label_project - for label_project in label_projects - if label_project["imageLayerId"] == image_layer_id - ), - None, - ) - if match_label_projects is not None: - if ( - "labels" in match_label_projects - and match_label_projects["labels"] is not None - ): - image_layer["labelProjectCount"] = len( - match_label_projects["labels"] - ) - else: - image_layer["labelProjectCount"] = 0 - if not image_layer.get("labelsUrl"): - # Older image layers will not have the generated geoJSON - image_layer["labelsUrl"] = None - try: - validation_data = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().VALIDATION.value, - partition_key=project_id, - ).load, - image_layer_id, - ) - labels = validation_data.get("labels") or {} - image_layer["validationLabelCount"] = len(labels) - except FileNotFoundError: - image_layer["validationLabelCount"] = 0 - project["imageLayer"] = image_layers - project["imageLayerCount"] = len(image_layers) - project["imageLayer"].sort( - key=lambda x: x["creationDate"], reverse=True - ) - _payload = json.dumps(project) + async def _load_response(): + project = await ProjectDetailsProcessor( + project_id=project_id, config=config + ).load(include_models=include_models) + payload = json.dumps(project) + return { + "payload": payload, + "etag": '"' + + hashlib.sha256(payload.encode()).hexdigest()[:32] + + '"', + "layer_count": len(project["imageLayer"]), + } + + cache_key = (project_id, include_models) + ( + cached_response, + cache_hit, + ) = await _project_details_cache.get_or_create( + cache_key, + _load_response, + refresh=_cache_refresh_requested(req.headers.get("Cache-Control")), + ) + _payload = cached_response["payload"] + etag = cached_response["etag"] _perf_headers = perf.headers(_perf, _perf_wall) perf.log_summary( logger, @@ -856,17 +811,31 @@ async def GetProjectDetails(req: func.HttpRequest) -> func.HttpResponse: _perf_wall, project_id=project_id, include_models=include_models, - layers=len(image_layers), + layers=cached_response["layer_count"], payload_bytes=len(_payload), + cache_hit=cache_hit, ) + cache_headers = { + "Cache-Control": ( + f"private, max-age={_PROJECT_DETAILS_CACHE_SECONDS}" + ), + "ETag": etag, + "X-Haste-Cache": "HIT" if cache_hit else "MISS", + } + if _perf_headers: + cache_headers.update(_perf_headers) + if _etag_matches(req.headers.get("If-None-Match"), etag): + return func.HttpResponse(status_code=304, headers=cache_headers) return func.HttpResponse( - _payload, status_code=200, headers=_perf_headers or None + _payload, status_code=200, headers=cache_headers ) except FileNotFoundError as e: + perf.end() logger.error(f"Project not found: {e}\n{traceback.format_exc()}") return func.HttpResponse("Project not found.", status_code=404) except Exception as e: + perf.end() logger.error( f"Error loading project details: {e}\n{traceback.format_exc()}" ) @@ -2850,6 +2819,19 @@ async def GenerateProjectStats(req: func.HttpRequest) -> func.HttpResponse: partition_key=project_id, ).load_all_from_partition ) + # PERF (Phase 1, B5): load LABELS once per project and index by + # imageLayerId, instead of a full partition scan per image layer. + label_projects = await asyncio.to_thread( + MetadataProcessor( + data_type=config.get_metadata_types().LABELS.value, + partition_key=project_id, + ).load_all_from_partition + ) + labels_by_layer = {} + for label_project in label_projects: + lid = label_project.get("imageLayerId") + if lid is not None and lid not in labels_by_layer: + labels_by_layer[lid] = label_project for image_layer in image_layers: image_layer_id = image_layer["imageLayerId"] match_models = [ @@ -2859,20 +2841,8 @@ async def GenerateProjectStats(req: func.HttpRequest) -> func.HttpResponse: ] image_layer["models"] = match_models image_layer["modelCount"] = len(match_models) - label_projects = await asyncio.to_thread( - MetadataProcessor( - data_type=config.get_metadata_types().LABELS.value, - partition_key=project_id, - ).load_all_from_partition - ) - match_label_projects = next( - ( - label_project - for label_project in label_projects - if label_project["imageLayerId"] == image_layer_id - ), - None, - ) + image_layer["labelProjectCount"] = 0 + match_label_projects = labels_by_layer.get(image_layer_id) if match_label_projects is not None: if ( "labels" in match_label_projects diff --git a/api/hastefuncapi/tests/test_generate_project_stats.py b/api/hastefuncapi/tests/test_generate_project_stats.py new file mode 100644 index 00000000..e0d10379 --- /dev/null +++ b/api/hastefuncapi/tests/test_generate_project_stats.py @@ -0,0 +1,86 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import io +import json +import os +import unittest +from contextlib import redirect_stderr +from unittest.mock import Mock, patch + +import azure.functions as func + +os.environ.setdefault("DEVELOPMENT_MODE", "true") +os.environ.setdefault("METADATA_STORAGE_TYPE", "local") +os.environ.setdefault("ARTIFACT_STORAGE_TYPE", "local") +os.environ.setdefault("DATA_PATH", "/tmp/haste-project-stats-api-tests") +os.environ.setdefault("TEMP_DATA_PATH", "/tmp/haste-project-stats-api-tests") + +with redirect_stderr(io.StringIO()): + from api.hastefuncapi import function_app + + +class TestGenerateProjectStats(unittest.IsolatedAsyncioTestCase): + async def test_unlabeled_layer_counts_zero_and_labels_load_once( + self, + ) -> None: + types = function_app.config.get_metadata_types() + processors = { + (types.PROJECT.value, None): Mock(), + (types.IMAGELAYER.value, "project-1"): Mock(), + (types.MODEL.value, "project-1"): Mock(), + (types.LABELS.value, "project-1"): Mock(), + } + processors[(types.PROJECT.value, None)].load_all.return_value = [ + { + "projectId": "project-1", + "name": "Project", + "description": "Description", + "creationDate": "2026-01-01T00:00:00Z", + "affectedCountries": [], + } + ] + processors[ + (types.IMAGELAYER.value, "project-1") + ].load_all_from_partition.return_value = [ + {"imageLayerId": "labeled"}, + {"imageLayerId": "unlabeled"}, + ] + processors[ + (types.MODEL.value, "project-1") + ].load_all_from_partition.return_value = [] + labels = processors[(types.LABELS.value, "project-1")] + labels.load_all_from_partition.return_value = [ + {"imageLayerId": "labeled", "labels": [{"id": "label-1"}]} + ] + + def processor_factory(*, data_type, partition_key=None): + return processors[(data_type, partition_key)] + + request = func.HttpRequest( + method="GET", + url="http://localhost/api/GenerateProjectStats", + headers={}, + params={}, + route_params={}, + body=b"", + ) + with patch.object( + function_app, "MetadataProcessor", side_effect=processor_factory + ): + response = await function_app.GenerateProjectStats(request) + + self.assertEqual(response.status_code, 200) + project = json.loads(response.get_body())["projects"][0] + self.assertEqual(project["labelsCount"], 1) + self.assertEqual( + project["imageLayerStats"], + [ + {"imageLayerId": "labeled", "labelsCount": 1}, + {"imageLayerId": "unlabeled", "labelsCount": 0}, + ], + ) + labels.load_all_from_partition.assert_called_once_with() + + +if __name__ == "__main__": + unittest.main() diff --git a/api/hastefuncapi/tests/test_model_artifact_route.py b/api/hastefuncapi/tests/test_model_artifact_route.py new file mode 100644 index 00000000..d0a82ff4 --- /dev/null +++ b/api/hastefuncapi/tests/test_model_artifact_route.py @@ -0,0 +1,66 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import io +import os +import unittest +from contextlib import redirect_stderr +from unittest.mock import AsyncMock, Mock, patch + +import azure.functions as func +from hastegeo.core.utils.blob import BlobRange + +os.environ.setdefault("DEVELOPMENT_MODE", "true") +os.environ.setdefault("METADATA_STORAGE_TYPE", "local") +os.environ.setdefault("ARTIFACT_STORAGE_TYPE", "local") +os.environ.setdefault("DATA_PATH", "/tmp/haste-model-artifact-api-tests") +os.environ.setdefault("TEMP_DATA_PATH", "/tmp/haste-model-artifact-api-tests") + +with redirect_stderr(io.StringIO()): + from api.hastefuncapi import function_app + +PROJECT_ID = "123e4567-e89b-12d3-a456-426614174000" + + +class TestGetModelArtifact(unittest.IsolatedAsyncioTestCase): + async def test_gpkg_response_is_a_named_download(self) -> None: + request = func.HttpRequest( + method="GET", + url="http://localhost/api/GetModelArtifact", + headers={}, + params={ + "projectId": PROJECT_ID, + "modelId": "42", + "kind": "gpkg", + }, + route_params={}, + body=b"", + ) + processor = Mock() + processor.load.return_value = { + "gpkgUrl": "https://account.test/model.gpkg" + } + blob = BlobRange( + data=b"gpkg", + total_size=4, + content_type="application/octet-stream", + etag='"etag"', + ) + + with patch.object( + function_app, "MetadataProcessor", return_value=processor + ), patch.object( + function_app, + "read_blob_range", + new=AsyncMock(return_value=blob), + ): + response = await function_app.GetModelArtifact(request) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.headers["Content-Disposition"], + 'attachment; filename="building_predictions_42.gpkg"', + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/hastefuncapi/tests/test_project_details_route.py b/api/hastefuncapi/tests/test_project_details_route.py new file mode 100644 index 00000000..1d2c75c6 --- /dev/null +++ b/api/hastefuncapi/tests/test_project_details_route.py @@ -0,0 +1,183 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import io +import json +import os +import unittest +from contextlib import redirect_stderr +from unittest.mock import AsyncMock, patch + +import azure.functions as func +from hastegeo.core.utils.async_cache import AsyncTTLCache + +os.environ.setdefault("DEVELOPMENT_MODE", "true") +os.environ.setdefault("METADATA_STORAGE_TYPE", "local") +os.environ.setdefault("ARTIFACT_STORAGE_TYPE", "local") +os.environ.setdefault("DATA_PATH", "/tmp/haste-project-details-api-tests") +os.environ.setdefault("TEMP_DATA_PATH", "/tmp/haste-project-details-api-tests") + +with redirect_stderr(io.StringIO()): + from api.hastefuncapi import function_app + +PROJECT_ID = "123e4567-e89b-12d3-a456-426614174000" + + +def make_request( + *, include_models: str | None = None, headers: dict | None = None +) -> func.HttpRequest: + params = {"projectId": PROJECT_ID} + if include_models is not None: + params["includeModels"] = include_models + return func.HttpRequest( + method="GET", + url="http://localhost/api/GetProjectDetails", + headers=headers or {}, + params=params, + route_params={}, + body=b"", + ) + + +class TestGetProjectDetails(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.project = { + "projectId": PROJECT_ID, + "name": "Project", + "imageLayer": [], + "imageLayerCount": 0, + } + self.loader = AsyncMock() + self.loader.load.return_value = self.project + patcher = patch.object( + function_app, + "ProjectDetailsProcessor", + return_value=self.loader, + ) + self.addCleanup(patcher.stop) + self.processor_class = patcher.start() + cache_patcher = patch.object( + function_app, + "_project_details_cache", + AsyncTTLCache(ttl_seconds=15, max_entries=8), + ) + self.addCleanup(cache_patcher.stop) + cache_patcher.start() + + async def test_returns_project_and_cache_headers(self) -> None: + response = await function_app.GetProjectDetails( + make_request(include_models="True") + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(json.loads(response.get_body()), self.project) + self.assertEqual( + response.headers["Cache-Control"], "private, max-age=15" + ) + self.assertEqual(response.headers["X-Haste-Cache"], "MISS") + self.assertTrue(response.headers["ETag"].startswith('"')) + self.processor_class.assert_called_once_with( + project_id=PROJECT_ID, config=function_app.config + ) + self.loader.load.assert_awaited_once_with(include_models=True) + + async def test_include_models_defaults_to_false(self) -> None: + response = await function_app.GetProjectDetails(make_request()) + + self.assertEqual(response.status_code, 200) + self.loader.load.assert_awaited_once_with(include_models=False) + + async def test_matching_etag_returns_empty_304(self) -> None: + first = await function_app.GetProjectDetails(make_request()) + etag = first.headers["ETag"] + self.loader.reset_mock() + + response = await function_app.GetProjectDetails( + make_request(headers={"If-None-Match": etag}) + ) + + self.assertEqual(response.status_code, 304) + self.assertEqual(response.get_body(), b"") + self.assertEqual(response.headers["ETag"], etag) + self.assertEqual(response.headers["X-Haste-Cache"], "HIT") + self.loader.load.assert_not_awaited() + + async def test_weak_etag_in_a_list_matches(self) -> None: + first = await function_app.GetProjectDetails(make_request()) + etag = first.headers["ETag"] + + response = await function_app.GetProjectDetails( + make_request(headers={"If-None-Match": f'"different", W/{etag}'}) + ) + + self.assertEqual(response.status_code, 304) + + async def test_include_models_uses_a_separate_cache_entry(self) -> None: + await function_app.GetProjectDetails(make_request()) + await function_app.GetProjectDetails( + make_request(include_models="true") + ) + + self.assertEqual(self.loader.load.await_count, 2) + self.loader.load.assert_any_await(include_models=False) + self.loader.load.assert_any_await(include_models=True) + + async def test_no_cache_request_refreshes_cached_response(self) -> None: + await function_app.GetProjectDetails(make_request()) + self.loader.load.return_value = dict(self.project, name="Updated") + + response = await function_app.GetProjectDetails( + make_request(headers={"Cache-Control": "no-cache"}) + ) + cached = await function_app.GetProjectDetails(make_request()) + + self.assertEqual(json.loads(response.get_body())["name"], "Updated") + self.assertEqual(json.loads(cached.get_body())["name"], "Updated") + self.assertEqual(self.loader.load.await_count, 2) + + def test_cache_refresh_directives(self) -> None: + self.assertTrue(function_app._cache_refresh_requested("no-cache")) + self.assertTrue( + function_app._cache_refresh_requested("public, max-age=0") + ) + self.assertFalse(function_app._cache_refresh_requested("max-age=15")) + + async def test_missing_project_returns_404(self) -> None: + self.loader.load.side_effect = FileNotFoundError("missing") + + first = await function_app.GetProjectDetails(make_request()) + second = await function_app.GetProjectDetails(make_request()) + + self.assertEqual(first.status_code, 404) + self.assertEqual(second.status_code, 404) + self.assertEqual(self.loader.load.await_count, 2) + + async def test_unexpected_error_returns_500(self) -> None: + self.loader.load.side_effect = RuntimeError("storage unavailable") + + response = await function_app.GetProjectDetails(make_request()) + + self.assertEqual(response.status_code, 500) + + async def test_invalid_project_id_returns_400_without_loading( + self, + ) -> None: + request = func.HttpRequest( + method="GET", + url="http://localhost/api/GetProjectDetails", + headers={}, + params={"projectId": "not-a-guid"}, + route_params={}, + body=b"", + ) + + response = await function_app.GetProjectDetails(request) + + self.assertEqual(response.status_code, 400) + self.processor_class.assert_not_called() + + def test_non_matching_etag_is_rejected(self) -> None: + self.assertFalse(function_app._etag_matches('"other"', '"expected"')) + + +if __name__ == "__main__": + unittest.main() diff --git a/docker/nginx.conf b/docker/nginx.conf index 1f159c50..5ab747cb 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -61,7 +61,7 @@ http { if ($request_method = 'OPTIONS') { add_header 'Access-Control-Allow-Origin' '*' always; add_header 'Access-Control-Allow-Methods' 'GET, POST, PUT, DELETE, OPTIONS' always; - add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,Keep-Alive,Origin,User-Agent,X-Requested-With' always; + add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,If-None-Match,Keep-Alive,Origin,User-Agent,X-Requested-With' always; add_header 'Access-Control-Max-Age' 1728000 always; add_header 'Content-Type' 'text/plain; charset=utf-8' always; add_header 'Content-Length' 0 always; @@ -71,7 +71,8 @@ http { # Add CORS headers to all responses add_header 'Access-Control-Allow-Origin' '*' always; add_header 'Access-Control-Allow-Methods' 'GET, POST, PUT, DELETE, OPTIONS' always; - add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,Keep-Alive,Origin,User-Agent,X-Requested-With' always; + add_header 'Access-Control-Allow-Headers' 'Accept,Authorization,Cache-Control,Content-Type,DNT,If-Modified-Since,If-None-Match,Keep-Alive,Origin,User-Agent,X-Requested-With' always; + add_header 'Access-Control-Expose-Headers' 'ETag,Server-Timing,X-Haste-Cache,X-Haste-Data-Layer-Calls,X-Haste-Data-Layer-Ms' always; # Proxy to Azure Functions - preserve full URI proxy_pass http://functions_backend$request_uri; diff --git a/docs/api-overview.md b/docs/api-overview.md index 03dde3c5..105f3187 100644 --- a/docs/api-overview.md +++ b/docs/api-overview.md @@ -5,6 +5,15 @@ poison-queue handler) for managing disaster assessment projects, processing sate imagery, and running AI models for damage assessment. A separate TiTiler-based tile server handles geospatial imagery visualization. +## Contents + +- [Architecture](#architecture) +- [Authentication](#authentication) +- [Base URLs](#base-urls) +- [Response Formats](#response-formats) +- [Rate Limits](#rate-limits) +- [HTTP Status Codes](#http-status-codes) + ## Architecture The API layer consists of three Azure Functions apps: @@ -61,6 +70,17 @@ Some write endpoints (e.g. the Model Catalog) return a small status wrapper: { "success": true, "message": "…", "catalogModel": {} } ``` +### Conditional Project Details + +`GetProjectDetails` returns a bounded process-local cached representation with `ETag`, +`Cache-Control`, and `X-Haste-Cache` headers. Clients send `If-None-Match` for normal +polls; a matching fresh representation returns `304` without storage work or a body. +Clients send `Cache-Control: no-cache` after mutations to force storage refresh. + +The cache is per Functions worker and is not a distributed consistency mechanism. The +`X-Haste-Data-Layer-Calls` and `X-Haste-Data-Layer-Ms` headers describe logical library +operations; Azure Storage metrics remain the source for REST transaction counts. + ## Rate Limits API throughput is bounded by the underlying platform: diff --git a/docs/api/hastefuncapi.md b/docs/api/hastefuncapi.md index 2c207c75..70370881 100644 --- a/docs/api/hastefuncapi.md +++ b/docs/api/hastefuncapi.md @@ -2,6 +2,13 @@ Azure Functions backend for the HASTE application. Provides REST endpoints for managing projects, image layers, ML model training/inference, labeling, user management, and geospatial data access. +## Contents + +- [Overview](#overview) +- [Endpoints](#endpoints) +- [Development Setup](#development-setup) +- [Auto-generated API Docs](#auto-generated-api-docs) + --- ## Overview @@ -21,7 +28,7 @@ All functions are defined in `function_app.py` as a single Azure Functions app. |--------|-------|-------------| | GET | `GetDashboardData` | Aggregated dashboard stats: project summaries, layer info, model status, and system-wide metrics. | | GET | `GetProjects` | All projects with aggregated layer and model counts. | -| GET | `GetProjectDetails` | Full project details including image layers, models, and processing status. Requires `projectId`. | +| GET | `GetProjectDetails` | Project, layer, validation, and optional model details. Supports `ETag`/`If-None-Match`; requires `projectId`. | | PUT | `PutProject` | Create or update a project. Auto-generates `projectId` and `creationDate` if not provided. | | DELETE | `DeleteProject` | Delete a project by `projectId`. | | GET | `GenerateProjectStats` | Regenerates project stats from raw data — useful if stats fall out of sync. | diff --git a/docs/configuration.md b/docs/configuration.md index 4a2f9a21..4fd874a5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -12,6 +12,7 @@ This guide documents each configuration mode. For the end-to-end workflow, see ## Contents - [Core settings](#core-settings) +- [Project detail performance](#project-detail-performance) - [Batch (create vs. bring-your-own)](#batch-create-vs-bring-your-own) - [Batch image tags and pool immutability](#batch-image-tags-and-pool-immutability) - [hastegeo wheel pinning](#hastegeo-wheel-pinning) @@ -38,6 +39,24 @@ This guide documents each configuration mode. For the end-to-end workflow, see Resource names are `HASTE_RESOURCE_PREFIX` + `HASTE_RANDOM_SUFFIX` based (not azd's `resourceToken`) so `what-if` stays clean against existing deployments. +## Project detail performance + +The API and queue workers use the following optional runtime settings. Defaults are +validated at process startup, so invalid values fail fast instead of creating an +unbounded executor or cache. + +| Function App setting | Default | Allowed | Purpose | +|---|---:|---:|---| +| `HASTE_BLOB_DOWNLOAD_WORKERS` | 16 | 1–64 | Process-wide blocking Blob I/O budget. | +| `HASTE_METADATA_LOAD_WORKERS` | 8 | 1–64 | Per-map limit within the process budget. | +| `HASTE_ARTIFACT_DOWNLOAD_WORKERS` | 8 | 1–64 | Per-artifact limit within the process budget. | +| `HASTE_PROJECTDETAILS_CACHE_SECONDS` | 15 | 0–300 | Process-local response freshness and HTTP `max-age`. | +| `HASTE_PROJECTDETAILS_CACHE_ENTRIES` | 64 | 1–512 | Maximum cached project response variants per worker. | + +These controls have code defaults and are not yet exposed as Bicep parameters. Treat +IaC parameterization of non-default values as a separate deployment change. Load-test +worker-count overrides with concurrent requests before production rollout. + ## Batch (create vs. bring-your-own) HASTE runs GPU workloads on Azure Batch. Both the Batch **account** and the GPU diff --git a/hastelib/src/hastegeo/core/processors/project_details.py b/hastelib/src/hastegeo/core/processors/project_details.py new file mode 100644 index 00000000..bb9cfd99 --- /dev/null +++ b/hastelib/src/hastegeo/core/processors/project_details.py @@ -0,0 +1,213 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +"""Project-details loading and response assembly.""" + +import asyncio +from collections import defaultdict +from collections.abc import Callable, Mapping, Sequence +from typing import Any + +from ..config import Config +from .metadata import MetadataProcessor + + +class ProjectDetailsProcessor: + """Load and assemble one project's API detail response.""" + + def __init__( + self, + project_id: str, + config: Config | None = None, + processor_factory: Callable[ + ..., MetadataProcessor + ] = MetadataProcessor, + ) -> None: + self.project_id = project_id + self.config = config or Config() + self.processor_factory = processor_factory + + def _processor(self, data_type: str) -> MetadataProcessor: + return self.processor_factory( + data_type=data_type, + partition_key=self.project_id, + config=self.config, + ) + + def _load(self, data_type: str, key: str) -> dict[str, Any]: + return self._processor(data_type).load(key) + + def _load_partition(self, data_type: str) -> list[dict[str, Any]]: + return self._processor(data_type).load_all_from_partition() + + def _load_map( + self, data_type: str, keys: Sequence[str] + ) -> dict[str, dict[str, Any] | None]: + return self._processor(data_type).load_map(keys) + + def _load_train_label_urls( + self, models: Sequence[Mapping[str, Any]] + ) -> dict[str, str | None]: + model_ids = [ + model["modelId"] for model in models if not model.get("labelsUrl") + ] + if not model_ids: + return {} + + processor = self._processor( + self.config.get_metadata_types().TRAIN_LABELS.value + ) + try: + existing_keys = set(processor.list_keys(data_format="geojson")) + except NotImplementedError: + urls = {} + for model_id in model_ids: + try: + urls[model_id] = processor.export( + model_id, data_format="geojson" + ) + except (FileNotFoundError, NotImplementedError): + urls[model_id] = None + return urls + + urls = {} + for model_id in model_ids: + if model_id not in existing_keys: + urls[model_id] = None + continue + try: + urls[model_id] = processor.build_url( + model_id, data_format="geojson" + ) + except NotImplementedError: + urls[model_id] = None + return urls + + async def load(self, include_models: bool) -> dict[str, Any]: + """Load project metadata concurrently and assemble the response.""" + types = self.config.get_metadata_types() + project = await asyncio.to_thread( + self._load, types.PROJECT.value, self.project_id + ) + + partition_types = [types.IMAGELAYER.value, types.LABELS.value] + if include_models: + partition_types.append(types.MODEL.value) + partition_results = await asyncio.gather( + *( + asyncio.to_thread(self._load_partition, data_type) + for data_type in partition_types + ) + ) + image_layers = partition_results[0] + label_projects = partition_results[1] + models = partition_results[2] if include_models else [] + + image_layer_ids = [ + image_layer["imageLayerId"] for image_layer in image_layers + ] + validation_task = asyncio.to_thread( + self._load_map, types.VALIDATION.value, image_layer_ids + ) + + if include_models: + model_ids = [model["modelId"] for model in models] + ( + validation_by_layer, + artifacts_by_model, + train_label_urls_by_model, + ) = await asyncio.gather( + validation_task, + asyncio.to_thread( + self._load_map, types.MODEL_ARTIFACTS.value, model_ids + ), + asyncio.to_thread(self._load_train_label_urls, models), + ) + else: + validation_by_layer = await validation_task + artifacts_by_model = {} + train_label_urls_by_model = {} + + return assemble_project_details( + project=project, + image_layers=image_layers, + models=models, + label_projects=label_projects, + artifacts_by_model=artifacts_by_model, + validation_by_layer=validation_by_layer, + train_label_urls_by_model=train_label_urls_by_model, + include_models=include_models, + ) + + +def assemble_project_details( + project: Mapping[str, Any], + image_layers: Sequence[Mapping[str, Any]], + models: Sequence[Mapping[str, Any]], + label_projects: Sequence[Mapping[str, Any]], + artifacts_by_model: Mapping[str, Mapping[str, Any] | None], + validation_by_layer: Mapping[str, Mapping[str, Any] | None], + train_label_urls_by_model: Mapping[str, str | None], + include_models: bool, +) -> dict[str, Any]: + """Assemble the API response from already-loaded metadata records.""" + models_by_layer: dict[str | None, list[Mapping[str, Any]]] = defaultdict( + list + ) + for model in models: + models_by_layer[model.get("imageLayerId")].append(model) + + labels_by_layer: dict[str, Mapping[str, Any]] = {} + for label_project in label_projects: + image_layer_id = label_project.get("imageLayerId") + if ( + image_layer_id is not None + and image_layer_id not in labels_by_layer + ): + labels_by_layer[image_layer_id] = label_project + + assembled_layers: list[dict[str, Any]] = [] + for stored_image_layer in image_layers: + image_layer = dict(stored_image_layer) + image_layer_id = image_layer["imageLayerId"] + + if include_models: + layer_models = sorted( + models_by_layer.get(image_layer_id, []), + key=lambda model: model["creationDate"], + reverse=True, + ) + assembled_models = [] + for stored_model in layer_models: + model = dict(stored_model) + model_id = model["modelId"] + model["artifacts"] = artifacts_by_model.get(model_id) + if not model.get("labelsUrl"): + model["labelsUrl"] = train_label_urls_by_model.get( + model_id + ) + assembled_models.append(model) + image_layer["models"] = assembled_models + image_layer["modelCount"] = len(assembled_models) + + label_project = labels_by_layer.get(image_layer_id) + if label_project is not None: + labels = label_project.get("labels") + image_layer["labelProjectCount"] = ( + len(labels) if labels is not None else 0 + ) + if not image_layer.get("labelsUrl"): + image_layer["labelsUrl"] = None + + validation = validation_by_layer.get(image_layer_id) + validation_labels = validation.get("labels") if validation else None + image_layer["validationLabelCount"] = len(validation_labels or {}) + assembled_layers.append(image_layer) + + assembled_project = dict(project) + assembled_project["imageLayer"] = sorted( + assembled_layers, + key=lambda image_layer: image_layer["creationDate"], + reverse=True, + ) + assembled_project["imageLayerCount"] = len(assembled_layers) + return assembled_project diff --git a/hastelib/src/hastegeo/core/utils/async_cache.py b/hastelib/src/hastegeo/core/utils/async_cache.py new file mode 100644 index 00000000..72a7db95 --- /dev/null +++ b/hastelib/src/hastegeo/core/utils/async_cache.py @@ -0,0 +1,103 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +"""Small bounded async cache with per-key single-flight loading.""" + +import asyncio +import os +import time +from collections import OrderedDict +from collections.abc import Awaitable, Callable +from typing import Generic, TypeVar + +KeyT = TypeVar("KeyT") +ValueT = TypeVar("ValueT") + + +def configured_cache_value( + name: str, default: int, minimum: int, maximum: int +) -> int: + """Read a bounded integer cache setting from the environment.""" + raw_value = os.environ.get(name, str(default)) + try: + value = int(raw_value) + except ValueError as error: + raise ValueError(f"{name} must be an integer") from error + if not minimum <= value <= maximum: + raise ValueError(f"{name} must be between {minimum} and {maximum}") + return value + + +class AsyncTTLCache(Generic[KeyT, ValueT]): + """Cache successful async loads for a bounded freshness window.""" + + def __init__( + self, + ttl_seconds: float, + max_entries: int, + clock: Callable[[], float] = time.monotonic, + ) -> None: + if ttl_seconds < 0: + raise ValueError("ttl_seconds cannot be negative") + if max_entries < 1: + raise ValueError("max_entries must be positive") + self.ttl_seconds = ttl_seconds + self.max_entries = max_entries + self._clock = clock + self._entries: OrderedDict[KeyT, tuple[float, ValueT]] = OrderedDict() + self._inflight: dict[KeyT, asyncio.Task[ValueT]] = {} + self._lock = asyncio.Lock() + + async def get_or_create( + self, + key: KeyT, + factory: Callable[[], Awaitable[ValueT]], + refresh: bool = False, + ) -> tuple[ValueT, bool]: + """Return ``(value, reused)`` and share one in-flight load per key.""" + async with self._lock: + cached = None if refresh else self._entries.get(key) + if cached is not None: + expires_at, value = cached + if self._clock() < expires_at: + self._entries.move_to_end(key) + return value, True + del self._entries[key] + elif refresh: + self._entries.pop(key, None) + + task = None if refresh else self._inflight.get(key) + reused = task is not None + if task is None: + task = asyncio.create_task(factory()) + self._inflight[key] = task + task.add_done_callback( + lambda completed, cache_key=key: asyncio.create_task( + self._complete(cache_key, completed) + ) + ) + + return await asyncio.shield(task), reused + + async def _complete(self, key: KeyT, task: asyncio.Task[ValueT]) -> None: + async with self._lock: + if self._inflight.get(key) is not task: + return + del self._inflight[key] + if task.cancelled() or task.exception() is not None: + return + self._entries[key] = ( + self._clock() + self.ttl_seconds, + task.result(), + ) + self._entries.move_to_end(key) + while len(self._entries) > self.max_entries: + self._entries.popitem(last=False) + + async def clear(self) -> None: + """Clear cached values and cancel unfinished loads.""" + async with self._lock: + tasks = list(self._inflight.values()) + self._inflight.clear() + self._entries.clear() + for task in tasks: + task.cancel() diff --git a/hastelib/tests/core/processors/test_project_details.py b/hastelib/tests/core/processors/test_project_details.py new file mode 100644 index 00000000..15369522 --- /dev/null +++ b/hastelib/tests/core/processors/test_project_details.py @@ -0,0 +1,463 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import json +import os +import tempfile +import unittest +from unittest.mock import Mock, patch + +from hastegeo.core.config import Config +from hastegeo.core.processors.metadata import MetadataProcessor +from hastegeo.core.processors.project_details import ( + ProjectDetailsProcessor, + assemble_project_details, +) + + +class TestAssembleProjectDetails(unittest.TestCase): + def setUp(self) -> None: + self.project = {"projectId": "project-1", "name": "Project"} + self.layers = [ + { + "imageLayerId": "layer-old", + "creationDate": "2026-01-01T00:00:00Z", + }, + { + "imageLayerId": "layer-new", + "creationDate": "2026-02-01T00:00:00Z", + }, + ] + + def test_joins_records_by_storage_key_when_body_ids_are_missing( + self, + ) -> None: + models = [ + { + "modelId": "model-1", + "imageLayerId": "layer-new", + "creationDate": "2026-02-02T00:00:00Z", + } + ] + artifacts = {"model-1": {"metrics": {"iou": 0.8}}} + validations = {"layer-new": {"labels": {"building-1": {}}}} + + result = assemble_project_details( + self.project, + self.layers, + models, + [{"imageLayerId": "layer-new", "labels": [{"id": "a"}]}], + artifacts, + validations, + {"model-1": "https://example.test/model-1.geojson"}, + include_models=True, + ) + + newest_layer = result["imageLayer"][0] + self.assertEqual(newest_layer["imageLayerId"], "layer-new") + self.assertEqual( + newest_layer["models"][0]["artifacts"], artifacts["model-1"] + ) + self.assertEqual(newest_layer["validationLabelCount"], 1) + self.assertEqual(newest_layer["labelProjectCount"], 1) + self.assertEqual( + newest_layer["models"][0]["labelsUrl"], + "https://example.test/model-1.geojson", + ) + + def test_preserves_existing_labels_url(self) -> None: + models = [ + { + "modelId": "model-1", + "imageLayerId": "layer-new", + "creationDate": "2026-02-02T00:00:00Z", + "labelsUrl": "https://stored.test/labels.geojson", + } + ] + + result = assemble_project_details( + self.project, + self.layers, + models, + [], + {}, + {}, + {"model-1": "https://generated.test/labels.geojson"}, + include_models=True, + ) + + model = result["imageLayer"][0]["models"][0] + self.assertEqual( + model["labelsUrl"], "https://stored.test/labels.geojson" + ) + self.assertIsNone(model["artifacts"]) + + def test_preserves_existing_layer_labels_url(self) -> None: + layer = dict( + self.layers[1], labelsUrl="https://stored.test/layer.geojson" + ) + + result = assemble_project_details( + self.project, + [layer], + [], + [{"imageLayerId": "layer-new", "labels": []}], + {}, + {}, + {}, + include_models=False, + ) + + self.assertEqual( + result["imageLayer"][0]["labelsUrl"], + "https://stored.test/layer.geojson", + ) + + def test_uses_first_label_project_for_duplicate_layer(self) -> None: + result = assemble_project_details( + self.project, + [self.layers[1]], + [], + [ + {"imageLayerId": "layer-new", "labels": [{"id": "first"}]}, + { + "imageLayerId": "layer-new", + "labels": [{"id": "second"}, {"id": "third"}], + }, + {"labels": []}, + ], + {}, + {}, + {}, + include_models=False, + ) + + self.assertEqual(result["imageLayer"][0]["labelProjectCount"], 1) + + def test_missing_related_records_keep_legacy_defaults(self) -> None: + result = assemble_project_details( + self.project, + self.layers, + [], + [], + {}, + {}, + {}, + include_models=True, + ) + + for image_layer in result["imageLayer"]: + self.assertEqual(image_layer["models"], []) + self.assertEqual(image_layer["modelCount"], 0) + self.assertEqual(image_layer["validationLabelCount"], 0) + self.assertNotIn("labelProjectCount", image_layer) + + def test_excluding_models_does_not_add_model_fields(self) -> None: + result = assemble_project_details( + self.project, + self.layers, + [], + [], + {}, + {}, + {}, + include_models=False, + ) + + for image_layer in result["imageLayer"]: + self.assertNotIn("models", image_layer) + self.assertNotIn("modelCount", image_layer) + + def test_inputs_are_not_mutated(self) -> None: + assemble_project_details( + self.project, + self.layers, + [], + [], + {}, + {}, + {}, + include_models=False, + ) + + self.assertNotIn("imageLayer", self.project) + for image_layer in self.layers: + self.assertNotIn("validationLabelCount", image_layer) + + def test_complete_fixture_matches_expected_response_bytes(self) -> None: + result = assemble_project_details( + self.project, + [self.layers[1]], + [ + { + "modelId": "model-1", + "imageLayerId": "layer-new", + "creationDate": "2026-02-02T00:00:00Z", + } + ], + [{"imageLayerId": "layer-new", "labels": [{"id": "a"}]}], + {"model-1": {"metrics": {"iou": 0.8}}}, + {"layer-new": {"labels": {"building-1": {}}}}, + {"model-1": "https://example.test/model-1.geojson"}, + include_models=True, + ) + expected = { + "projectId": "project-1", + "name": "Project", + "imageLayer": [ + { + "imageLayerId": "layer-new", + "creationDate": "2026-02-01T00:00:00Z", + "models": [ + { + "modelId": "model-1", + "imageLayerId": "layer-new", + "creationDate": "2026-02-02T00:00:00Z", + "artifacts": {"metrics": {"iou": 0.8}}, + "labelsUrl": ( + "https://example.test/model-1.geojson" + ), + } + ], + "modelCount": 1, + "labelProjectCount": 1, + "labelsUrl": None, + "validationLabelCount": 1, + } + ], + "imageLayerCount": 1, + } + + self.assertEqual(json.dumps(result), json.dumps(expected)) + + +class TestProjectDetailsProcessor(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.types = Config.get_metadata_types() + self.config = Mock() + self.config.get_metadata_types.return_value = self.types + self.processors = {} + self.created_types = [] + + for data_type in ( + self.types.PROJECT.value, + self.types.IMAGELAYER.value, + self.types.LABELS.value, + self.types.MODEL.value, + self.types.VALIDATION.value, + self.types.MODEL_ARTIFACTS.value, + self.types.TRAIN_LABELS.value, + ): + self.processors[data_type] = Mock() + + self.processors[self.types.PROJECT.value].load.return_value = { + "projectId": "project-1", + "name": "Project", + } + self.processors[ + self.types.IMAGELAYER.value + ].load_all_from_partition.return_value = [ + { + "imageLayerId": "layer-1", + "creationDate": "2026-01-01T00:00:00Z", + } + ] + self.processors[ + self.types.LABELS.value + ].load_all_from_partition.return_value = [] + self.processors[ + self.types.MODEL.value + ].load_all_from_partition.return_value = [ + { + "modelId": "model-1", + "imageLayerId": "layer-1", + "creationDate": "2026-01-02T00:00:00Z", + } + ] + self.processors[self.types.VALIDATION.value].load_map.return_value = { + "layer-1": {"labels": {"building-1": {}}} + } + self.processors[ + self.types.MODEL_ARTIFACTS.value + ].load_map.return_value = {"model-1": {"metrics": {"iou": 0.8}}} + train_labels = self.processors[self.types.TRAIN_LABELS.value] + train_labels.list_keys.return_value = ["model-1"] + train_labels.build_url.return_value = ( + "https://example.test/model-1.geojson" + ) + + def factory(self, *, data_type, partition_key, config): + self.assertEqual(partition_key, "project-1") + self.assertIs(config, self.config) + self.created_types.append(data_type) + return self.processors[data_type] + + async def test_load_uses_keyed_maps_for_related_records(self) -> None: + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=True) + + model = result["imageLayer"][0]["models"][0] + self.assertEqual(model["artifacts"], {"metrics": {"iou": 0.8}}) + self.assertEqual(result["imageLayer"][0]["validationLabelCount"], 1) + self.processors[ + self.types.VALIDATION.value + ].load_map.assert_called_once_with(["layer-1"]) + self.processors[ + self.types.MODEL_ARTIFACTS.value + ].load_map.assert_called_once_with(["model-1"]) + + async def test_load_without_models_skips_model_storage(self) -> None: + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=False) + + self.assertNotIn(self.types.MODEL.value, self.created_types) + self.assertNotIn(self.types.MODEL_ARTIFACTS.value, self.created_types) + self.assertNotIn(self.types.TRAIN_LABELS.value, self.created_types) + self.assertNotIn("models", result["imageLayer"][0]) + + async def test_train_label_listing_falls_back_to_legacy_export( + self, + ) -> None: + train_labels = self.processors[self.types.TRAIN_LABELS.value] + train_labels.list_keys.side_effect = NotImplementedError + train_labels.export.return_value = ( + "https://fallback.test/model-1.geojson" + ) + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=True) + + model = result["imageLayer"][0]["models"][0] + self.assertEqual( + model["labelsUrl"], "https://fallback.test/model-1.geojson" + ) + train_labels.export.assert_called_once_with( + "model-1", data_format="geojson" + ) + + async def test_missing_legacy_train_label_export_returns_none( + self, + ) -> None: + train_labels = self.processors[self.types.TRAIN_LABELS.value] + train_labels.list_keys.side_effect = NotImplementedError + train_labels.export.side_effect = FileNotFoundError + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=True) + + self.assertIsNone(result["imageLayer"][0]["models"][0]["labelsUrl"]) + + async def test_existing_model_url_skips_train_label_storage(self) -> None: + model = self.processors[self.types.MODEL.value] + model.load_all_from_partition.return_value[0][ + "labelsUrl" + ] = "https://stored.test/model.geojson" + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=True) + + self.assertEqual( + result["imageLayer"][0]["models"][0]["labelsUrl"], + "https://stored.test/model.geojson", + ) + self.assertNotIn(self.types.TRAIN_LABELS.value, self.created_types) + + async def test_unsupported_train_label_url_is_returned_as_none( + self, + ) -> None: + train_labels = self.processors[self.types.TRAIN_LABELS.value] + train_labels.build_url.side_effect = NotImplementedError + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + result = await processor.load(include_models=True) + + model = result["imageLayer"][0]["models"][0] + self.assertIsNone(model["labelsUrl"]) + + async def test_project_load_failure_stops_partition_reads(self) -> None: + self.processors[ + self.types.PROJECT.value + ].load.side_effect = FileNotFoundError + processor = ProjectDetailsProcessor( + "project-1", self.config, self.factory + ) + + with self.assertRaises(FileNotFoundError): + await processor.load(include_models=True) + + self.assertEqual(self.created_types, [self.types.PROJECT.value]) + + +class TestProjectDetailsLocalStorage(unittest.IsolatedAsyncioTestCase): + async def test_load_preserves_keyed_legacy_records_end_to_end( + self, + ) -> None: + with tempfile.TemporaryDirectory() as data_path: + with patch.dict( + os.environ, + { + "METADATA_STORAGE_TYPE": "local", + "DATA_PATH": data_path, + }, + ): + config = Config() + types = config.get_metadata_types() + records = ( + ( + types.PROJECT.value, + "project-1", + {"projectId": "project-1"}, + ), + ( + types.IMAGELAYER.value, + "layer-1", + { + "imageLayerId": "layer-1", + "creationDate": "2026-01-01T00:00:00Z", + }, + ), + ( + types.MODEL.value, + "model-1", + { + "modelId": "model-1", + "imageLayerId": "layer-1", + "creationDate": "2026-01-02T00:00:00Z", + }, + ), + (types.MODEL_ARTIFACTS.value, "model-1", {"metrics": {}}), + (types.VALIDATION.value, "layer-1", {"labels": {"a": {}}}), + ) + for data_type, key, value in records: + MetadataProcessor( + data_type=data_type, + partition_key="project-1", + config=config, + ).save(key, value) + + result = await ProjectDetailsProcessor( + "project-1", config=config + ).load(include_models=True) + + image_layer = result["imageLayer"][0] + self.assertEqual( + image_layer["models"][0]["artifacts"], {"metrics": {}} + ) + self.assertEqual(image_layer["validationLabelCount"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/hastelib/tests/core/utils/test_async_cache.py b/hastelib/tests/core/utils/test_async_cache.py new file mode 100644 index 00000000..cf2dab3d --- /dev/null +++ b/hastelib/tests/core/utils/test_async_cache.py @@ -0,0 +1,180 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +import asyncio +import os +import unittest +from unittest.mock import AsyncMock, patch + +from hastegeo.core.utils.async_cache import ( + AsyncTTLCache, + configured_cache_value, +) + + +class TestConfiguredCacheValue(unittest.TestCase): + def test_reads_bounded_integer(self) -> None: + with patch.dict(os.environ, {"TEST_CACHE_VALUE": "12"}): + self.assertEqual( + configured_cache_value("TEST_CACHE_VALUE", 5, 0, 20), 12 + ) + + def test_rejects_invalid_value(self) -> None: + for value in ("invalid", "-1", "21"): + with self.subTest(value=value): + with patch.dict(os.environ, {"TEST_CACHE_VALUE": value}): + with self.assertRaises(ValueError): + configured_cache_value("TEST_CACHE_VALUE", 5, 0, 20) + + +class TestAsyncTTLCache(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.now = 0.0 + self.cache = AsyncTTLCache( + ttl_seconds=10, + max_entries=2, + clock=lambda: self.now, + ) + + async def asyncTearDown(self) -> None: + await self.cache.clear() + + async def test_reuses_value_before_expiry(self) -> None: + factory = AsyncMock(return_value="value") + + first, first_reused = await self.cache.get_or_create("key", factory) + second, second_reused = await self.cache.get_or_create("key", factory) + + self.assertEqual((first, second), ("value", "value")) + self.assertFalse(first_reused) + self.assertTrue(second_reused) + factory.assert_awaited_once_with() + + async def test_reloads_value_after_expiry(self) -> None: + factory = AsyncMock(side_effect=["first", "second"]) + await self.cache.get_or_create("key", factory) + self.now = 10 + + value, reused = await self.cache.get_or_create("key", factory) + + self.assertEqual(value, "second") + self.assertFalse(reused) + self.assertEqual(factory.await_count, 2) + + async def test_refresh_replaces_unexpired_value(self) -> None: + factory = AsyncMock(side_effect=["first", "second"]) + await self.cache.get_or_create("key", factory) + + refreshed, reused = await self.cache.get_or_create( + "key", factory, refresh=True + ) + cached, cached_reused = await self.cache.get_or_create("key", factory) + + self.assertEqual((refreshed, cached), ("second", "second")) + self.assertFalse(reused) + self.assertTrue(cached_reused) + self.assertEqual(factory.await_count, 2) + + async def test_refresh_supersedes_an_older_inflight_load(self) -> None: + old_started = asyncio.Event() + old_release = asyncio.Event() + + async def old_factory() -> str: + old_started.set() + await old_release.wait() + return "old" + + old_request = asyncio.create_task( + self.cache.get_or_create("key", old_factory) + ) + await old_started.wait() + + refreshed, reused = await self.cache.get_or_create( + "key", lambda: asyncio.sleep(0, result="new"), refresh=True + ) + old_release.set() + old_value, _ = await old_request + cached, cached_reused = await self.cache.get_or_create( + "key", lambda: asyncio.sleep(0, result="unexpected") + ) + + self.assertEqual((old_value, refreshed, cached), ("old", "new", "new")) + self.assertFalse(reused) + self.assertTrue(cached_reused) + + async def test_concurrent_requests_share_one_load(self) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def factory() -> str: + started.set() + await release.wait() + return "value" + + first = asyncio.create_task(self.cache.get_or_create("key", factory)) + await started.wait() + second = asyncio.create_task(self.cache.get_or_create("key", factory)) + await asyncio.sleep(0) + release.set() + + self.assertEqual( + await asyncio.gather(first, second), + [("value", False), ("value", True)], + ) + + async def test_failed_load_is_not_cached(self) -> None: + factory = AsyncMock(side_effect=[RuntimeError("failed"), "recovered"]) + + with self.assertRaisesRegex(RuntimeError, "failed"): + await self.cache.get_or_create("key", factory) + value, reused = await self.cache.get_or_create("key", factory) + + self.assertEqual(value, "recovered") + self.assertFalse(reused) + self.assertEqual(factory.await_count, 2) + + async def test_evicts_least_recently_used_entry(self) -> None: + factories = { + key: AsyncMock(return_value=key) for key in ("a", "b", "c") + } + for key in ("a", "b", "a", "c"): + await self.cache.get_or_create(key, factories[key]) + + _, reused = await self.cache.get_or_create("b", factories["b"]) + + self.assertFalse(reused) + self.assertEqual(factories["a"].await_count, 1) + self.assertEqual(factories["b"].await_count, 2) + + async def test_clear_removes_cached_values(self) -> None: + factory = AsyncMock(return_value="value") + await self.cache.get_or_create("key", factory) + + await self.cache.clear() + _, reused = await self.cache.get_or_create("key", factory) + + self.assertFalse(reused) + self.assertEqual(factory.await_count, 2) + + async def test_clear_cancels_inflight_load(self) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def factory() -> str: + started.set() + await release.wait() + return "value" + + request = asyncio.create_task(self.cache.get_or_create("key", factory)) + await started.wait() + + await self.cache.clear() + + with self.assertRaises(asyncio.CancelledError): + await request + await asyncio.sleep(0) + + def test_rejects_invalid_configuration(self) -> None: + with self.assertRaises(ValueError): + AsyncTTLCache(ttl_seconds=-1, max_entries=1) + with self.assertRaises(ValueError): + AsyncTTLCache(ttl_seconds=1, max_entries=0) diff --git a/spec/features/perf-layer-loading/tools/bench_api_http.py b/spec/features/perf-layer-loading/tools/bench_api_http.py index 9c7224fa..ec5a76eb 100644 --- a/spec/features/perf-layer-loading/tools/bench_api_http.py +++ b/spec/features/perf-layer-loading/tools/bench_api_http.py @@ -36,6 +36,11 @@ def main(): ap.add_argument("--repeats", type=int, default=30) ap.add_argument("--warmup", type=int, default=3) ap.add_argument("--code", default=None, help="Azure Functions key, if required") + ap.add_argument( + "--allow-cache", + action="store_true", + help="Measure warm server-cache behavior instead of forcing reloads.", + ) args = ap.parse_args() qs = f"?projectId={args.project_id}&includeModels=True" @@ -46,7 +51,9 @@ def main(): latencies, storage_calls, storage_ms, payloads = [], [], [], [] for i in range(args.warmup + args.repeats): t0 = time.perf_counter() - with urllib.request.urlopen(url) as resp: + headers = {} if args.allow_cache else {"Cache-Control": "no-cache"} + request = urllib.request.Request(url, headers=headers) + with urllib.request.urlopen(request) as resp: body = resp.read() hdrs = resp.headers dt = (time.perf_counter() - t0) * 1000.0 @@ -54,10 +61,16 @@ def main(): continue latencies.append(dt) payloads.append(len(body)) - if hdrs.get("X-Haste-Storage-Calls"): - storage_calls.append(int(hdrs["X-Haste-Storage-Calls"])) - if hdrs.get("X-Haste-Storage-Ms"): - storage_ms.append(float(hdrs["X-Haste-Storage-Ms"])) + calls_header = hdrs.get("X-Haste-Data-Layer-Calls") or hdrs.get( + "X-Haste-Storage-Calls" + ) + timing_header = hdrs.get("X-Haste-Data-Layer-Ms") or hdrs.get( + "X-Haste-Storage-Ms" + ) + if calls_header: + storage_calls.append(int(calls_header)) + if timing_header: + storage_ms.append(float(timing_header)) result = { "project_id": args.project_id, @@ -66,8 +79,8 @@ def main(): "latency_p95_ms": round(_pct(latencies, 95), 1), "latency_max_ms": round(max(latencies), 1), "payload_kb": round(statistics.median(payloads) / 1024, 1), - "server_storage_calls": storage_calls[0] if storage_calls else None, - "server_storage_ms_p50": ( + "server_data_layer_calls": storage_calls[0] if storage_calls else None, + "server_data_layer_ms_p50": ( round(statistics.median(storage_ms), 1) if storage_ms else None ), }