feat(process-storms): fail-fast AORC pre-flight before the scan

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)}"
+    )

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),

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)