stac: retry transient child failures once; exit on root failure (#73)

Two related changes that sharpen the startup-resilience story:

1. Retry pass for transient child failures. After the main pool drains,
   children that timed out get one more attempt in a fresh pool using a
   longer STAC_CHILD_RETRY_TIMEOUT (default 8s, env-overridable). Rescues
   the tail-latency case (~2 of 81 collections observed on 2026-04-18 dev
   rollout) without needing to kill the pod. A retry that also fails
   leaves the original error in STAC_LOAD_ERRORS.

2. Exit on root-catalog failure. When the root fetch itself fails at
   startup, sys.exit(1) rather than starting uvicorn with an empty
   catalog. Lets Kubernetes restart the pod so the next attempt can
   catch better S3 conditions, instead of presenting a useless server.
   Partial catalogs (some children failed) still serve — that's the
   point of the resilience design; only total failure triggers exit.

3. Default STAC_FETCH_CONCURRENCY bumped from 8 to 16 so the main pool
   plus retry pass both fit within the readiness-probe budget.

Tests: 4 new (retry rescue, retry failure is final, retry uses configured
timeout, new concurrency default). Full suite: 95 passed.

Co-authored-by: Carl Boettiger <cboettig@berkeley.edu>

Author: boettiger-lab-llm-agent[bot]

server.py

@@ -11,7 +11,7 @@
 from mcp.shared.session import BaseSession
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.responses import JSONResponse
-from stac import STAC_DATASETS, STAC_CATALOG_URL, list_datasets as _stac_list, get_dataset as _stac_get, get_collection as _stac_get_collection
+from stac import STAC_DATASETS, STAC_LOAD_ERRORS, STAC_CATALOG_URL, list_datasets as _stac_list, get_dataset as _stac_get, get_collection as _stac_get_collection
 
 # Workaround for https://github.com/boettiger-lab/mcp-data-server/issues/5
 # send_notification crashes with ClosedResourceError when the client disconnects
@@ -276,6 +276,19 @@ async def dispatch(self, request, call_next):
 # 10. SERVER START
 # -------------------------------------------------------------------------
 if __name__ == "__main__":
+    # If the STAC root catalog was unreachable at startup, serving would give
+    # clients a useless empty catalog. Exit non-zero so Kubernetes restarts the
+    # pod and gets a fresh attempt against whatever S3 looks like now. Child
+    # failures (partial catalog) are fine — the resilience design serves what
+    # loaded and records the rest in STAC_LOAD_ERRORS for list_datasets's footer.
+    if "__root__" in STAC_LOAD_ERRORS:
+        print(
+            "💀 STAC root catalog unreachable at startup — exiting so k8s can "
+            f"restart and retry. Reason: {STAC_LOAD_ERRORS['__root__']}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
     app = mcp.streamable_http_app()
     app.router.redirect_slashes = False
     mount_tiles(app)

stac.py

@@ -30,7 +30,13 @@
 _STAC_CHILD_TIMEOUT = int(
     os.environ.get("STAC_CHILD_TIMEOUT", os.environ.get("STAC_TIMEOUT", "5"))
 )
-_STAC_FETCH_CONCURRENCY = int(os.environ.get("STAC_FETCH_CONCURRENCY", "8"))
+_STAC_FETCH_CONCURRENCY = int(os.environ.get("STAC_FETCH_CONCURRENCY", "16"))
+
+# Child fetches that time out on the first pass are retried once, each with a
+# longer per-child timeout. Rescues tail-latency failures without leaving
+# otherwise-healthy collections out of the catalog. A persistent failure
+# (both attempts time out) stays in STAC_LOAD_ERRORS.
+_STAC_CHILD_RETRY_TIMEOUT = int(os.environ.get("STAC_CHILD_RETRY_TIMEOUT", "8"))
 
 # Legacy alias retained so external callers (if any) that read the old name still work.
 _STAC_TIMEOUT = _STAC_CHILD_TIMEOUT
@@ -307,7 +313,7 @@ def _collection_to_dict(col, sub_children=None) -> dict:
     return {k: v for k, v in result.items() if v is not None}
 
 
-def _fetch_parent(href: str, title: str | None, token: str | None):
+def _fetch_parent(href: str, title: str | None, token: str | None, timeout: int = None):
     """Thread-worker: fetch one top-level child Collection.
 
     Returns a 3-tuple (col, subchild_hrefs, error):
@@ -322,8 +328,13 @@ def _fetch_parent(href: str, title: str | None, token: str | None):
 
     This function must NEVER raise — all exceptions are caught and translated
     to the error dict.
+
+    `timeout` overrides the default child timeout; used by the retry pass.
     """
-    child_io = _TimeoutStacIO(token=token, timeout=_STAC_CHILD_TIMEOUT)
+    child_io = _TimeoutStacIO(
+        token=token,
+        timeout=timeout if timeout is not None else _STAC_CHILD_TIMEOUT,
+    )
     try:
         col = pystac.Collection.from_file(href, stac_io=child_io)
         subchild_hrefs = [l.href for l in (col.links or []) if l.rel == "child"]
@@ -334,16 +345,21 @@ def _fetch_parent(href: str, title: str | None, token: str | None):
         return None, [], {ident: reason}
 
 
-def _fetch_subchild(href: str, parent_id: str, token: str | None):
+def _fetch_subchild(href: str, parent_id: str, token: str | None, timeout: int = None):
     """Thread-worker: fetch one sub-child Collection (a leaf of a parent).
 
     Returns a 2-tuple (col, error):
     - col: the parsed pystac.Collection on success, else None
     - error: None on success, or {identifier: reason} on failure
 
     Never raises — all exceptions caught.
+
+    `timeout` overrides the default child timeout; used by the retry pass.
     """
-    child_io = _TimeoutStacIO(token=token, timeout=_STAC_CHILD_TIMEOUT)
+    child_io = _TimeoutStacIO(
+        token=token,
+        timeout=timeout if timeout is not None else _STAC_CHILD_TIMEOUT,
+    )
     try:
         col = pystac.Collection.from_file(href, stac_io=child_io)
         return col, None
@@ -402,19 +418,27 @@ def fetch_stac_catalog(catalog_url: str = None, catalog_token: str = None) -> di
     subchild_cols_by_parent: dict = {}
     errors: dict = {}
 
+    # Failed items recorded during the initial pool drain so we can retry them once
+    # below with a longer per-child timeout. Rescues tail-latency failures without
+    # leaving healthy-but-slow collections out of the catalog.
+    failed_parents: list[tuple[str, str | None]] = []      # (href, title)
+    failed_subchildren: list[tuple[str, str]] = []         # (href, parent_col_id)
+
     with ThreadPoolExecutor(max_workers=_STAC_FETCH_CONCURRENCY) as pool:
-        # future -> ("parent", href) OR ("subchild", parent_col_id)
+        # future -> ("parent", href, title) OR ("subchild", href, parent_col_id)
         pending: dict = {}
 
         for href, title in parent_links:
             fut = pool.submit(_fetch_parent, href, title, catalog_token)
-            pending[fut] = ("parent", href)
+            pending[fut] = ("parent", href, title)
 
         while pending:
             done, _ = wait(pending.keys(), return_when=FIRST_COMPLETED)
             for fut in done:
-                kind, ctx = pending.pop(fut)
+                entry = pending.pop(fut)
+                kind = entry[0]
                 if kind == "parent":
+                    _, href, title = entry
                     col, subchild_hrefs, error = fut.result()
                     if col is not None:
                         parent_cols[col.id] = col
@@ -424,14 +448,67 @@ def fetch_stac_catalog(catalog_url: str = None, catalog_token: str = None) -> di
                             sub_fut = pool.submit(
                                 _fetch_subchild, sub_href, col.id, catalog_token,
                             )
-                            pending[sub_fut] = ("subchild", col.id)
+                            pending[sub_fut] = ("subchild", sub_href, col.id)
+                    if error:
+                        errors.update(error)
+                        failed_parents.append((href, title))
+                else:  # subchild
+                    _, sub_href, parent_col_id = entry
+                    col, error = fut.result()
+                    if col is not None:
+                        subchild_cols_by_parent.setdefault(parent_col_id, {})[col.id] = col
+                    if error:
+                        errors.update(error)
+                        failed_subchildren.append((sub_href, parent_col_id))
+
+    # --- Phase 2.5: retry failed children once with a longer timeout ---
+    # Rescues tail-latency failures (borderline-slow S3 responses). A retry that
+    # also fails leaves the original error in STAC_LOAD_ERRORS. If a previously-failed
+    # parent succeeds on retry, its sub-children are NOT re-fetched here — they
+    # weren't attempted the first time because we didn't have the parent's JSON;
+    # a cache-miss on any specific sub-child ID will trigger a fresh walk later.
+    if failed_parents or failed_subchildren:
+        with ThreadPoolExecutor(max_workers=_STAC_FETCH_CONCURRENCY) as retry_pool:
+            retry_futures: dict = {}
+            for href, title in failed_parents:
+                fut = retry_pool.submit(
+                    _fetch_parent, href, title, catalog_token, _STAC_CHILD_RETRY_TIMEOUT,
+                )
+                retry_futures[fut] = ("parent", href, title)
+            for sub_href, parent_col_id in failed_subchildren:
+                fut = retry_pool.submit(
+                    _fetch_subchild, sub_href, parent_col_id, catalog_token, _STAC_CHILD_RETRY_TIMEOUT,
+                )
+                retry_futures[fut] = ("subchild", sub_href, parent_col_id)
+            for fut in retry_futures:
+                entry = retry_futures[fut]
+                kind = entry[0]
+                if kind == "parent":
+                    _, href, title = entry
+                    col, _subchild_hrefs, error = fut.result()
+                    if col is not None:
+                        parent_cols[col.id] = col
+                        # Clear the first-pass error for this parent since retry succeeded
+                        errors.pop(_child_identifier(href, title_hint=title), None)
+                        print(
+                            f"🔁 Retry succeeded for parent: {col.id}",
+                            file=sys.stderr,
+                        )
+                    # If retry also failed, `error` carries the same identifier key as the
+                    # first-pass error; updating with it leaves the STAC_LOAD_ERRORS entry
+                    # pointing at the most-recent (retry) reason.
                     if error:
                         errors.update(error)
                 else:  # subchild
-                    parent_col_id = ctx
+                    _, sub_href, parent_col_id = entry
                     col, error = fut.result()
                     if col is not None:
                         subchild_cols_by_parent.setdefault(parent_col_id, {})[col.id] = col
+                        errors.pop(_child_identifier(sub_href, title_hint=None), None)
+                        print(
+                            f"🔁 Retry succeeded for sub-child: {col.id}",
+                            file=sys.stderr,
+                        )
                     if error:
                         errors.update(error)
 

tests/test_stac.py

@@ -541,12 +541,13 @@ def test_child_timeout_default_is_5(self, monkeypatch):
         importlib.reload(stac)
         assert stac._STAC_CHILD_TIMEOUT == 5
 
-    def test_fetch_concurrency_default_is_8(self, monkeypatch):
-        """With no env var set, STAC_FETCH_CONCURRENCY defaults to 8."""
+    def test_fetch_concurrency_default_is_16(self, monkeypatch):
+        """Default concurrency is 16 — sized so the main pool plus the retry pass
+        both fit in the readiness-probe budget."""
         monkeypatch.delenv("STAC_FETCH_CONCURRENCY", raising=False)
         import stac
         importlib.reload(stac)
-        assert stac._STAC_FETCH_CONCURRENCY == 8
+        assert stac._STAC_FETCH_CONCURRENCY == 16
 
     def test_stac_timeout_back_compat_applies_to_both(self, monkeypatch):
         """If only STAC_TIMEOUT is set, both root and child timeouts adopt its value."""
@@ -992,3 +993,107 @@ def test_cache_miss_refetch_with_all_children_failing_preserves_previous_cache(s
         assert len(stac.STAC_LOAD_ERRORS) == 1
         assert any("public-new" in k for k in stac.STAC_LOAD_ERRORS.keys())
 
+    def test_child_retry_timeout_default_is_8(self, monkeypatch):
+        """Retry pass uses a longer per-child timeout (default 8s) to rescue borderline slow children."""
+        monkeypatch.delenv("STAC_CHILD_RETRY_TIMEOUT", raising=False)
+        import stac
+        importlib.reload(stac)
+        assert stac._STAC_CHILD_RETRY_TIMEOUT == 8
+
+    def test_retry_rescues_transient_parent_failure(self):
+        """A parent that fails the first pass but succeeds on retry ends up in the catalog
+        and out of STAC_LOAD_ERRORS."""
+        from unittest.mock import patch
+        import requests
+        import stac
+
+        self._reset_module_state(stac)
+
+        cat = self._make_root_catalog([
+            "https://example.com/public-flaky/stac-collection.json",
+        ])
+        col = self._make_leaf_collection("flaky")
+
+        # First call raises Timeout; subsequent calls return the Collection.
+        call_count = {"n": 0}
+        def flaky_from_file(href, *args, **kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                raise requests.exceptions.Timeout("first pass slow")
+            return col
+
+        with patch("stac.pystac.Catalog.from_file", return_value=cat), \
+             patch("stac.pystac.Collection.from_file", side_effect=flaky_from_file):
+            result = stac.fetch_stac_catalog()
+
+        # Retry saved it
+        assert "flaky" in result
+        assert call_count["n"] == 2  # one failure + one retry
+        # Nothing in the error dict — retry cleared the first-pass failure
+        assert stac.STAC_LOAD_ERRORS == {}
+
+    def test_retry_failure_is_final(self):
+        """If both the initial attempt and the retry fail, the error stays in STAC_LOAD_ERRORS
+        and the collection is missing from the catalog."""
+        from unittest.mock import patch
+        import requests
+        import stac
+
+        self._reset_module_state(stac)
+
+        cat = self._make_root_catalog([
+            "https://example.com/public-dead/stac-collection.json",
+        ])
+
+        with patch("stac.pystac.Catalog.from_file", return_value=cat), \
+             patch(
+                 "stac.pystac.Collection.from_file",
+                 side_effect=requests.exceptions.Timeout("always slow"),
+             ):
+            result = stac.fetch_stac_catalog()
+
+        assert result == {}
+        assert len(stac.STAC_LOAD_ERRORS) == 1
+        assert any("public-dead" in k for k in stac.STAC_LOAD_ERRORS.keys())
+
+    def test_retry_uses_child_retry_timeout(self, monkeypatch):
+        """The retry pass constructs _TimeoutStacIO with _STAC_CHILD_RETRY_TIMEOUT, not _STAC_CHILD_TIMEOUT."""
+        from unittest.mock import patch, MagicMock
+        import requests
+        import stac
+
+        monkeypatch.setenv("STAC_CHILD_TIMEOUT", "5")
+        monkeypatch.setenv("STAC_CHILD_RETRY_TIMEOUT", "8")
+        importlib.reload(stac)
+        self._reset_module_state(stac)
+
+        cat = self._make_root_catalog([
+            "https://example.com/public-flaky/stac-collection.json",
+        ])
+        col = self._make_leaf_collection("flaky")
+
+        seen_timeouts = []
+        orig_TimeoutStacIO = stac._TimeoutStacIO
+
+        def capturing_stac_io(*args, **kwargs):
+            seen_timeouts.append(kwargs.get("timeout"))
+            return orig_TimeoutStacIO(*args, **kwargs)
+
+        call_count = {"n": 0}
+        def flaky_from_file(href, *args, **kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                raise requests.exceptions.Timeout("slow")
+            return col
+
+        with patch("stac._TimeoutStacIO", side_effect=capturing_stac_io), \
+             patch("stac.pystac.Catalog.from_file", return_value=cat), \
+             patch("stac.pystac.Collection.from_file", side_effect=flaky_from_file):
+            stac.fetch_stac_catalog()
+
+        # Filter out the root_io construction (timeout=_STAC_ROOT_TIMEOUT=15)
+        # and look only at the child-fetch constructions.
+        child_timeouts = [t for t in seen_timeouts if t != stac._STAC_ROOT_TIMEOUT]
+        assert 5 in child_timeouts, f"expected first-pass child timeout (5s) in {child_timeouts}"
+        assert 8 in child_timeouts, f"expected retry child timeout (8s) in {child_timeouts}"
+