From 721173aea3366e9ccf5a32bf8b54f8dd117c16c2 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 8 Jun 2026 13:19:52 +0800 Subject: [PATCH] Isolate observation collector from web API --- deploy.sh | 3 + docker-compose.yml | 56 ++++++++- src/database/db_manager.py | 98 ++++++++++------ tests/test_deployment_runtime_config.py | 46 ++++++-- tests/test_observation_collector.py | 32 ++++++ tests/test_web_observability.py | 41 ++++++- web/app_factory.py | 2 +- web/observation_collector_worker.py | 47 ++++++++ web/services/city_api.py | 144 +++++++++++++++--------- 9 files changed, 369 insertions(+), 100 deletions(-) create mode 100644 web/observation_collector_worker.py diff --git a/deploy.sh b/deploy.sh index 8dc12d1d..6669ae10 100644 --- a/deploy.sh +++ b/deploy.sh @@ -220,6 +220,9 @@ if [ "$FAILED_BACKEND" = "1" ]; then exit 1 fi +echo "Updating observation collector..." +compose_up_retry "observation collector" -d --no-deps polyweather_collector + echo "Updating frontend..." compose_up_retry "frontend" -d --no-deps polyweather_frontend diff --git a/docker-compose.yml b/docker-compose.yml index bfcdae95..50a7cc91 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,6 +29,7 @@ services: - .env environment: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: '' POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: bot @@ -99,15 +100,21 @@ services: env_file: *id001 environment: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: '' + POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1} + POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1} + POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-3000} POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180} POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60} - POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600} POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-5} POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: web + UVICORN_WORKERS: ${UVICORN_WORKERS:-2} healthcheck: interval: 30s retries: 3 @@ -129,6 +136,53 @@ services: volumes: - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data + polyweather_collector: + command: python -m web.observation_collector_worker + container_name: polyweather_collector + depends_on: + polyweather_redis: + condition: service_healthy + polyweather_web: + condition: service_started + logging: + driver: "json-file" + options: + max-size: "50m" + max-file: "3" + cpus: ${POLYWEATHER_COLLECTOR_CPUS:-0.75} + env_file: *id001 + environment: + POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ${POLYWEATHER_COLLECTOR_PATCH_ENDPOINT:-http://polyweather_web:8000/api/internal/collector-patch} + POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60} + POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180} + POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true' + POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600} + POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-15} + POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300} + POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30} + POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' + POLYWEATHER_SERVICE_ROLE: collector + healthcheck: + interval: 60s + retries: 3 + test: + - CMD + - python + - -c + - import sqlite3; c=sqlite3.connect('/var/lib/polyweather/polyweather.db'); + c.execute('SELECT 1'); c.close() + timeout: 10s + image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} + mem_limit: ${POLYWEATHER_COLLECTOR_MEM_LIMIT:-768m} + memswap_limit: ${POLYWEATHER_COLLECTOR_MEMSWAP_LIMIT:-1g} + pids_limit: 256 + restart: unless-stopped + user: ${UID:-1000}:${GID:-1000} + volumes: + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data x-polyweather-base: env_file: *id001 image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} diff --git a/src/database/db_manager.py b/src/database/db_manager.py index 7bba6bfb..5f3fb90a 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -39,6 +39,10 @@ class DBManager: conn.execute("PRAGMA busy_timeout=5000") return conn + @staticmethod + def _is_sqlite_locked_error(exc: sqlite3.OperationalError) -> bool: + return "database is locked" in str(exc).lower() + def _init_cache_key(self) -> str: return os.path.abspath(self.db_path) @@ -3015,19 +3019,30 @@ class DBManager: pressure_hpa: Optional[float] = None, obs_time: str, ) -> None: - with self._get_connection() as conn: - conn.execute( - """ - INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time) - VALUES (?, ?, ?, ?, ?, ?) - """, - (str(icao).strip().upper(), str(city).strip().lower(), - temp_c, wind_kt, pressure_hpa, str(obs_time)), - ) - conn.execute( - "DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')" - ) - conn.commit() + safe_icao = str(icao).strip().upper() + safe_city = str(city).strip().lower() + try: + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time) + VALUES (?, ?, ?, ?, ?, ?) + """, + (safe_icao, safe_city, temp_c, wind_kt, pressure_hpa, str(obs_time)), + ) + conn.execute( + "DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')" + ) + conn.commit() + except sqlite3.OperationalError as exc: + if self._is_sqlite_locked_error(exc): + logger.warning( + "airport obs log skipped because sqlite is locked icao={} city={}", + safe_icao, + safe_city, + ) + return + raise def get_airport_obs_recent( self, icao: str, minutes: int = 30 @@ -3068,30 +3083,41 @@ class DBManager: safe_otime = str(otime_utc or "").strip() if not safe_icao or not safe_runway or not safe_otime: return - with self._get_connection() as conn: - existing = conn.execute( - "SELECT id FROM runway_obs_log WHERE icao=? AND runway=? AND otime_utc=? LIMIT 1", - (safe_icao, safe_runway, safe_otime), - ).fetchone() - if existing: + try: + with self._get_connection() as conn: + existing = conn.execute( + "SELECT id FROM runway_obs_log WHERE icao=? AND runway=? AND otime_utc=? LIMIT 1", + (safe_icao, safe_runway, safe_otime), + ).fetchone() + if existing: + return + conn.execute( + """ + INSERT INTO runway_obs_log ( + icao, city, runway, + tdz_temp, mid_temp, end_temp, target_runway_max, + wind_dir, wind_speed, rvr, mor, humidity, + otime_utc + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + safe_icao, safe_city, safe_runway, + tdz_temp, mid_temp, end_temp, target_runway_max, + wind_dir, wind_speed, rvr, mor, humidity, + safe_otime, + ), + ) + conn.commit() + except sqlite3.OperationalError as exc: + if self._is_sqlite_locked_error(exc): + logger.warning( + "runway obs log skipped because sqlite is locked icao={} city={} runway={}", + safe_icao, + safe_city, + safe_runway, + ) return - conn.execute( - """ - INSERT INTO runway_obs_log ( - icao, city, runway, - tdz_temp, mid_temp, end_temp, target_runway_max, - wind_dir, wind_speed, rvr, mor, humidity, - otime_utc - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - safe_icao, safe_city, safe_runway, - tdz_temp, mid_temp, end_temp, target_runway_max, - wind_dir, wind_speed, rvr, mor, humidity, - safe_otime, - ), - ) - conn.commit() + raise def get_runway_obs_recent( self, icao: str, minutes: int = 60 diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index d70bc4ab..a8d42b2d 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -52,7 +52,7 @@ def test_scan_terminal_prewarm_only_runs_for_web_service(monkeypatch): assert app_factory._scan_terminal_prewarm_enabled() is True -def test_observation_collector_only_runs_for_web_service(monkeypatch): +def test_observation_collector_only_runs_for_collector_service(monkeypatch): from web import app_factory monkeypatch.delenv("POLYWEATHER_SERVICE_ROLE", raising=False) @@ -63,22 +63,53 @@ def test_observation_collector_only_runs_for_web_service(monkeypatch): assert app_factory._observation_collector_enabled() is False monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "web") + assert app_factory._observation_collector_enabled() is False + + monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "collector") assert app_factory._observation_collector_enabled() is True monkeypatch.setenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", "false") assert app_factory._observation_collector_enabled() is False -def test_docker_compose_isolates_scan_terminal_prewarm_to_web_service(): +def test_docker_compose_isolates_collector_from_web_and_bot_services(): compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") + bot_block = compose.split(" polyweather:", 1)[1].split( + "\n polyweather_frontend:", + 1, + )[0] + web_block = compose.split(" polyweather_web:", 1)[1].split( + "\n polyweather_collector:", + 1, + )[0] + collector_block = compose.split(" polyweather_collector:", 1)[1].split( + "\nx-polyweather-base:", + 1, + )[0] assert "POLYWEATHER_SERVICE_ROLE: web" in compose assert "POLYWEATHER_SERVICE_ROLE: bot" in compose - assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in compose - assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in compose - assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true}" in compose - assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in compose - assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in compose + assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block + assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block + assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in web_block + assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in collector_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in bot_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in web_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block + assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1}" in web_block + assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1}" in web_block + assert "POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-3000}" in web_block + assert "UVICORN_WORKERS: ${UVICORN_WORKERS:-2}" in web_block + assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in bot_block + assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in web_block + assert ( + "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: " + "${POLYWEATHER_COLLECTOR_PATCH_ENDPOINT:-http://polyweather_web:8000/api/internal/collector-patch}" + in collector_block + ) + assert "command: python -m web.observation_collector_worker" in collector_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in collector_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in collector_block def test_scan_terminal_backend_timeout_returns_before_next_proxy_abort(): @@ -152,6 +183,7 @@ def test_deploy_script_retries_compose_recreate_races(): assert "compose_up_retry()" in script assert "removal of container .* is already in progress" in script assert 'compose_up_retry "backend services" -d --no-deps polyweather_web polyweather' in script + assert 'compose_up_retry "observation collector" -d --no-deps polyweather_collector' in script assert 'compose_up_retry "frontend" -d --no-deps polyweather_frontend' in script diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index c7f61801..3e3e72cb 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -1,4 +1,5 @@ from concurrent.futures import ThreadPoolExecutor +import sqlite3 import time @@ -130,3 +131,34 @@ def test_observation_collector_run_due_once_refreshes_panel_cache(): assert collector.run_due_once(now_ts=1180.0) == 1 assert calls == [("qingdao", False), ("qingdao", False)] assert refreshed == ["qingdao", "qingdao"] + + +def test_observation_collector_worker_entrypoint_exists(): + from web import observation_collector_worker + + assert callable(observation_collector_worker.main) + + +def test_ephemeral_observation_log_writes_skip_sqlite_lock(monkeypatch, tmp_path): + from src.database.db_manager import DBManager + + db = DBManager(str(tmp_path / "polyweather.db")) + + def locked_connection(): + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(db, "_get_connection", locked_connection) + + db.append_airport_obs( + icao="ZSQD", + city="qingdao", + temp_c=24.0, + obs_time="2026-06-08T04:00:00Z", + ) + db.append_runway_obs( + icao="ZSQD", + city="qingdao", + runway="17/35", + target_runway_max=24.0, + otime_utc="2026-06-08T04:00:00Z", + ) diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 4516d3c3..e78c5321 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -969,7 +969,45 @@ def test_chart_detail_payload_uses_threadpool_and_reuses_short_cache(monkeypatch def test_city_detail_batch_partial_timeout_default_stays_below_proxy_budget(monkeypatch): monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", raising=False) - assert city_api._city_detail_batch_partial_timeout_seconds() == 6.0 + assert city_api._city_detail_batch_partial_timeout_seconds() == 3.0 + + +def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeypatch): + import asyncio + + build_calls = 0 + + async def build_batch_item(city, **kwargs): + nonlocal build_calls + build_calls += 1 + return city, {"city": city} + + monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1") + monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE", None) + monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE", 0) + monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None) + monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) + monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item) + + semaphore = city_api._city_detail_batch_build_semaphore() + assert semaphore.acquire(blocking=False) is True + try: + payload = asyncio.run( + city_api.get_city_detail_batch_payload( + object(), + cities="Paris,Shanghai", + resolution="10m", + limit=2, + ) + ) + finally: + semaphore.release() + + assert payload["partial"] is True + assert payload["busy"] is True + assert payload["details"] == {} + assert payload["missing"] == ["paris", "shanghai"] + assert build_calls == 0 def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch): @@ -1039,6 +1077,7 @@ def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monke } monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "20") + monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "2") monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None) monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item) diff --git a/web/app_factory.py b/web/app_factory.py index 48764ad3..a949668c 100644 --- a/web/app_factory.py +++ b/web/app_factory.py @@ -41,7 +41,7 @@ def _observation_collector_enabled() -> bool: enabled = str( os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED") or "true" ).strip().lower() in {"1", "true", "yes", "on"} - return enabled and _service_role() in {"web", "api", "backend"} + return enabled and _service_role() in {"collector"} def create_app() -> FastAPI: diff --git a/web/observation_collector_worker.py b/web/observation_collector_worker.py new file mode 100644 index 00000000..8b84ed8e --- /dev/null +++ b/web/observation_collector_worker.py @@ -0,0 +1,47 @@ +"""Standalone observation collector process. + +The FastAPI web worker must stay responsive for health checks and user-facing +API routes. High-frequency source polling runs here instead of inside the web +request process. +""" + +from __future__ import annotations + +import signal +import threading +from types import FrameType +from typing import Optional + +from loguru import logger + +from web.core import _weather +from web.observation_collector_service import start_observation_collector_loop +from web.services.city_runtime import _refresh_city_panel_cache + +_STOP_EVENT = threading.Event() + + +def _handle_stop_signal(signum: int, _frame: Optional[FrameType]) -> None: + logger.info("observation collector worker stopping signal={}", signum) + _STOP_EVENT.set() + + +def main() -> None: + signal.signal(signal.SIGINT, _handle_stop_signal) + signal.signal(signal.SIGTERM, _handle_stop_signal) + + thread = start_observation_collector_loop( + weather=_weather, + cache_refresher=lambda city: _refresh_city_panel_cache(city, force_refresh=False), + ) + if thread is None: + logger.warning("observation collector worker started with collector disabled") + else: + logger.info("observation collector worker started thread={}", thread.name) + + while not _STOP_EVENT.wait(3600): + pass + + +if __name__ == "__main__": + main() diff --git a/web/services/city_api.py b/web/services/city_api.py index 38a25143..702c5dd0 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -46,6 +46,9 @@ _CITY_DETAIL_BATCH_RESPONSE_CACHE: Dict[CityDetailBatchResponseCacheKey, Dict[st _CITY_DETAIL_BATCH_RESPONSE_CACHE_TS: Dict[CityDetailBatchResponseCacheKey, float] = {} _CITY_DETAIL_BATCH_RESPONSE_INFLIGHT: Dict[CityDetailBatchResponseCacheKey, "asyncio.Task[Dict[str, Any]]"] = {} _CITY_DETAIL_BATCH_RESPONSE_LOCK = asyncio.Lock() +_CITY_DETAIL_BATCH_BUILD_SEMAPHORE: Optional[threading.BoundedSemaphore] = None +_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE = 0 +_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_LOCK = threading.Lock() def _city_detail_payload_cache_ttl() -> float: @@ -933,20 +936,38 @@ async def _build_city_detail_batch_item_async( def _city_detail_batch_concurrency() -> int: try: - value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "3") or "3") + value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "1") or "1") except ValueError: - value = 3 - return max(1, min(6, value)) + value = 1 + return max(1, min(4, value)) + + +def _city_detail_batch_global_concurrency() -> int: + try: + value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1") or "1") + except ValueError: + value = 1 + return max(1, min(4, value)) + + +def _city_detail_batch_build_semaphore() -> threading.BoundedSemaphore: + global _CITY_DETAIL_BATCH_BUILD_SEMAPHORE, _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE + size = _city_detail_batch_global_concurrency() + with _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_LOCK: + if _CITY_DETAIL_BATCH_BUILD_SEMAPHORE is None or _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE != size: + _CITY_DETAIL_BATCH_BUILD_SEMAPHORE = threading.BoundedSemaphore(size) + _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE = size + return _CITY_DETAIL_BATCH_BUILD_SEMAPHORE def _city_detail_batch_partial_timeout_seconds() -> Optional[float]: try: timeout_ms = int( - os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "6000") - or "6000" + os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "3000") + or "3000" ) except ValueError: - timeout_ms = 6000 + timeout_ms = 3000 if timeout_ms <= 0: return None return max(0.001, min(60.0, timeout_ms / 1000.0)) @@ -991,58 +1012,73 @@ async def get_city_detail_batch_payload( detail_scope = _normalize_city_detail_scope(scope) async def _build_uncached_payload() -> Dict[str, Any]: - semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) + build_semaphore = _city_detail_batch_build_semaphore() + if not build_semaphore.acquire(blocking=False): + return { + "cities": city_names, + "details": {}, + "errors": {}, + "missing": list(city_names), + "partial": True, + "busy": True, + "stale_reason": "city detail batch builder is busy", + } - async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]: - async with semaphore: - return await _build_city_detail_batch_item_async( - city, - force_refresh=force_refresh, - market_slug=market_slug, - target_date=target_date, - resolution=resolution, - detail_scope=detail_scope, - timing_recorder=timer, - ) + try: + semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) - task_by_city = { - city: asyncio.create_task(_build_with_limit(city)) - for city in city_names - } - task_city_lookup = {task: city for city, task in task_by_city.items()} - done, pending = await timer.measure_async( - "build_details", - lambda: asyncio.wait( - task_by_city.values(), - timeout=_city_detail_batch_partial_timeout_seconds(), - ), - ) - details: Dict[str, Any] = {} - errors: Dict[str, str] = {} - missing: List[str] = [] - for task in done: - city = task_city_lookup[task] - try: - result_city, payload = task.result() - except Exception as exc: - errors[city] = str(exc) - continue - details[result_city] = payload + async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]: + async with semaphore: + return await _build_city_detail_batch_item_async( + city, + force_refresh=force_refresh, + market_slug=market_slug, + target_date=target_date, + resolution=resolution, + detail_scope=detail_scope, + timing_recorder=timer, + ) - for task in pending: - city = task_city_lookup[task] - missing.append(city) - task.cancel() + task_by_city = { + city: asyncio.create_task(_build_with_limit(city)) + for city in city_names + } + task_city_lookup = {task: city for city, task in task_by_city.items()} + done, pending = await timer.measure_async( + "build_details", + lambda: asyncio.wait( + task_by_city.values(), + timeout=_city_detail_batch_partial_timeout_seconds(), + ), + ) + details: Dict[str, Any] = {} + errors: Dict[str, str] = {} + missing: List[str] = [] + for task in done: + city = task_city_lookup[task] + try: + result_city, payload = task.result() + except Exception as exc: + errors[city] = str(exc) + continue + details[result_city] = payload - missing_set = set(missing) - missing = [city for city in city_names if city in missing_set] - return { - "cities": city_names, - "details": details, - "errors": errors, - "missing": missing, - "partial": bool(missing or errors), - } + for task in pending: + city = task_city_lookup[task] + missing.append(city) + task.cancel() + + missing_set = set(missing) + missing = [city for city in city_names if city in missing_set] + return { + "cities": city_names, + "details": details, + "errors": errors, + "missing": missing, + "partial": bool(missing or errors), + } + finally: + build_semaphore.release() cache_ttl = _city_detail_batch_response_cache_ttl() cache_key = _city_detail_batch_response_cache_key(