diff --git a/src/actions/aorc_preflight.py b/src/actions/aorc_preflight.py new file mode 100644 index 0000000..efb57e7 --- /dev/null +++ b/src/actions/aorc_preflight.py @@ -0,0 +1,130 @@ +"""Pre-flight: verify every year in the payload's date range is mirrored. + +Before kicking off process-storms (which can take 10+ minutes), HEAD each +year's ``.zmetadata`` in the AORC cache. If any year is missing, raise +with a clear "mirror these years first" message — failing in seconds +instead of mid-scan with a cryptic xarray/zarr ``KeyError: '.zmetadata'``. + +This is the regression guard for the 2026-05-27 incident where the +2025.zarr key was never mirrored, the catalog runs got 30+ minutes into +process-storms, the cumsum path opened (2024, 2025) for cross-year +storms, and the whole run crashed when zarr couldn't find ``.zmetadata``. + +No-op when ``AORC_S3_BASE_URL`` isn't set (local dev). Logs warnings but +doesn't raise when ``AORC_S3_KEY`` / ``AORC_S3_SECRET`` are absent — we +have no way to probe a private bucket without them, and the actual scan +will fail later with a more specific error. +""" + +from __future__ import annotations + +import datetime +import logging +import os +from typing import Iterable +from urllib.parse import urlparse + +log = logging.getLogger(__name__) + + +def required_years( + start_date: str, end_date: str | None, storm_duration_hours: int = 0 +) -> list[int]: + """Years whose AORC zarrs the scan needs. + + ``start.year`` through ``end.year`` inclusive; one extra year tacked on + when ``end_date`` + ``storm_duration`` crosses a year boundary (the + cumsum path opens ``(year, year+1)`` for late-year storm windows). + """ + start = datetime.date.fromisoformat(start_date) + end = datetime.date.fromisoformat(end_date) if end_date else start + last = end + datetime.timedelta(hours=storm_duration_hours) + return list(range(start.year, last.year + 1)) + + +def verify_aorc_cache_years( + years: Iterable[int], aorc_base_url: str | None = None +) -> list[int]: + """Return the subset of ``years`` whose ``.zmetadata`` is missing in the + AORC cache. + + Empty list means the cache covers every requested year. Returns the full + list of ``years`` when the cache isn't probeable (no AORC_S3_BASE_URL). + """ + base = aorc_base_url or os.environ.get("AORC_S3_BASE_URL") + if not base: + log.info("AORC pre-flight: AORC_S3_BASE_URL not set — skipping cache check") + return [] + + parsed = urlparse(base) + if parsed.scheme != "s3": + log.warning( + "AORC pre-flight: AORC_S3_BASE_URL=%s is not an s3:// URL, skipping", + base, + ) + return [] + bucket = parsed.netloc + prefix = parsed.path.lstrip("/").rstrip("/") + + try: + import boto3 + except ImportError: + log.warning("AORC pre-flight: boto3 unavailable — skipping cache check") + return [] + + client_kwargs: dict = {"region_name": "us-east-1"} + endpoint = os.environ.get("AORC_S3_ENDPOINT") + if endpoint: + client_kwargs["endpoint_url"] = endpoint + key = os.environ.get("AORC_S3_KEY") + secret = os.environ.get("AORC_S3_SECRET") + if key and secret: + client_kwargs["aws_access_key_id"] = key + client_kwargs["aws_secret_access_key"] = secret + + s3 = boto3.client("s3", **client_kwargs) + + missing: list[int] = [] + for y in years: + key_path = f"{prefix}/{y}.zarr/.zmetadata" if prefix else f"{y}.zarr/.zmetadata" + try: + s3.head_object(Bucket=bucket, Key=key_path) + except Exception as e: + # Treat any failure (NoSuchKey, AccessDenied, transient 5xx) as + # "year not available". The downstream scan will produce the + # more specific error if we're wrong about access. + log.info( + "AORC pre-flight: %s missing from s3://%s/%s (%s)", + y, + bucket, + prefix, + type(e).__name__, + ) + missing.append(y) + return missing + + +def assert_years_available( + start_date: str, end_date: str | None, storm_duration_hours: int = 0 +) -> None: + """Raise RuntimeError if any year needed for ``start..end`` is unmirrored. + + Logs a clear remediation hint pointing at ``./run.py mirror``. + """ + years = required_years(start_date, end_date, storm_duration_hours) + if not years: + return + missing = verify_aorc_cache_years(years) + if not missing: + log.info( + "AORC pre-flight: all %d year(s) %d-%d present in cache", + len(years), + years[0], + years[-1], + ) + return + raise RuntimeError( + f"AORC pre-flight: {len(missing)} year(s) missing from the private cache: " + f"{missing}. Mirror them before launching this run: " + f"./run.py mirror --year-start {min(missing)} --year-end {max(missing)}" + ) diff --git a/src/actions/process_storms.py b/src/actions/process_storms.py index dcf3ef4..554da39 100644 --- a/src/actions/process_storms.py +++ b/src/actions/process_storms.py @@ -12,6 +12,7 @@ from stormhub.met.storm_catalog import StormCatalog, new_catalog, new_collection from worker_sizing import resolve_num_workers +from actions import aorc_preflight log = logging.getLogger(__name__) @@ -80,6 +81,14 @@ def process_storms(ctx: dict[str, Any], action: Any) -> None: ) if collection is None: + # Fail fast: HEAD each required AORC year before the multi-hour scan, + # instead of dying mid-scan on a missing year. + aorc_preflight.assert_years_available( + start_date=attrs["start_date"], + end_date=end_date, + storm_duration_hours=storm_params["storm_duration"], + ) + catalog = new_catalog( catalog_id, str(config_path), diff --git a/test/test_aorc_preflight.py b/test/test_aorc_preflight.py new file mode 100644 index 0000000..808b105 --- /dev/null +++ b/test/test_aorc_preflight.py @@ -0,0 +1,37 @@ +"""Unit tests for aorc_preflight — fail-fast AORC year check.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +from actions import aorc_preflight # noqa: E402 + + +def test_required_years_basic(): + assert aorc_preflight.required_years("1979-02-01", "1981-06-01") == [1979, 1980, 1981] + + +def test_required_years_single_day(): + assert aorc_preflight.required_years("2001-05-05", None) == [2001] + + +def test_required_years_crosses_boundary_with_duration(): + # A 72-hr window starting 2020-12-31 spills into 2021. + assert aorc_preflight.required_years("2020-12-31", "2020-12-31", 72) == [2020, 2021] + + +def test_assert_skips_when_no_base_url(monkeypatch): + monkeypatch.delenv("AORC_S3_BASE_URL", raising=False) + # No base URL -> not probeable -> must not raise. + aorc_preflight.assert_years_available("2001-01-01", "2001-12-31", 72) + + +def test_assert_raises_on_missing_year(monkeypatch): + monkeypatch.setattr(aorc_preflight, "verify_aorc_cache_years", lambda years: [2025]) + with pytest.raises(RuntimeError, match="2025"): + aorc_preflight.assert_years_available("2025-01-01", "2025-12-31", 72)