Make worker startup logging visible, bound dataset downloads, persist blocked list

README.md

@@ -21,7 +21,7 @@ commit to git (they're gitignored and volume-mounted from the host):
 
 - **`aragorn_omnicorp`** → `./omnicorp_lmdb/` (`curies.lmdb`, `shared_counts.lmdb`)
 - **`score_paths`** → `./pathfinder_embeddings/` (a directory-style LMDB)
-- **`arax_pathfinder`** → `./arax_pathfinder_dbs/` (`curie_ngd_v1.0_<tier-version>.sqlite`, `tier0-info-for-overlay_v1.0_<tier-version>.sqlite`)
+- **`arax_pathfinder`** → `./arax_pathfinder_dbs/` (`curie_ngd_v1.0_<tier-version>.sqlite`, `tier0-info-for-overlay_v1.0_<tier-version>.sqlite`, `general_concepts.json`)
 
 So a new developer doesn't have to source these by hand, each worker can fetch its dataset on first
 startup. Two download mechanisms are supported, depending on where the dataset lives:
@@ -37,22 +37,51 @@ PATHFINDER_EMBEDDINGS_URL=https://example.org/path/pathfinder_embeddings.tar.gz
 The archive for each dataset should contain the expected files at its top level: `curies.lmdb` and
 `shared_counts.lmdb` for omnicorp, `data.mdb` (and `lock.mdb`) for the embeddings.
 
-**arax_pathfinder's sqlite databases** are served as plain files over HTTPS. The
+**arax_pathfinder's sqlite databases** are served as plain files over HTTPS, no credentials needed. The
 filenames embed a Knowledge Graph version that changes periodically, so only one variable needs updating
 when a new Knowledge Graph ships:
 
 ```dotenv
 ARAX_PATHFINDER_TIER_VERSION=tier0-20260621
 ```
 
-This requires an SSH key with access to that host, mounted read-only into the container
-(`~/.ssh:/home/nru/.ssh:ro` in compose.yml).
+The ARAX blocked-concept list (`general_concepts.json`, fetched from GitHub) lands in this same
+directory, so it is downloaded once and then persists with the databases rather than being re-fetched
+by every new container. It is only fetched when absent — delete it from the volume to pick up an
+updated upstream list.
 
 On startup, each worker checks whether its files already exist in the volume-mounted directory. If
-they're missing and a source is configured (URL or scp path), it fetches them into that directory —
-which lives on the host, so the data persists across restarts and is only downloaded once. If the
-files are already present, or no source is configured, the download is skipped (production mounts
-this data out of band, so it's unaffected).
+they're missing and a URL is configured, it fetches them into that directory — which lives on the
+host, so the data persists across restarts and is only downloaded once. If the files are already
+present, or no source is configured, the download is skipped (production mounts this data out of
+band, so it's unaffected).
+
+Downloads are bounded by `DATASET_DOWNLOAD_TIMEOUT_SEC` (default 60), which applies per socket
+operation rather than to the whole transfer — a large file downloads for as long as it needs, but a
+connection that opens and then stalls fails loudly instead of hanging worker startup.
+
+#### Deploying these workers
+
+The presence check is an exact match on the configured directory **and** the tier-versioned filenames.
+A deployment that mounts the data somewhere else, or whose `ARAX_PATHFINDER_TIER_VERSION` doesn't match
+the filenames on the volume, will not use the mounted copies — it will decide the dataset is missing and
+download it again, into whatever path the settings do point at. If that path isn't the mount, the files
+land on the container's writable layer and the pod is eventually evicted for exceeding its ephemeral
+storage. So when deploying, set the directory explicitly and confirm it resolves to your mount:
+
+```dotenv
+ARAX_PATHFINDER_DBS_DIR=/data/arax_pathfinder_dbs   # default is relative: resolved against /app
+```
+
+```console
+$ kubectl exec deploy/arax-pathfinder -- python -c \
+    "from shepherd_utils.data_download import arax_pathfinder_sqlite_paths as p; print(*p(), sep='\n')"
+```
+
+Because `general_concepts.json` now shares that directory, the volume needs to be writable for the
+first startup that fetches it — or the file can be preloaded alongside the sqlite databases, after
+which the worker only ever reads it. A read-only mount with no preloaded copy fails at startup with a
+permission error rather than silently continuing.
 
 ### Worker
 

compose.yml

@@ -359,9 +359,14 @@ services:
       - ./logs:/app/logs
       - ./.env:/app/.env
       # First run? The worker downloads its two sqlite dbs from
-      # kg2webhost.rtx.ai on startup, no credentials needed. Set
+      # kg2webhost.rtx.ai, plus the ARAX blocked-concept list from GitHub, on
+      # startup -- no credentials needed. All three live here, so they persist
+      # across restarts instead of being re-fetched by each new container. Set
       # ARAX_PATHFINDER_TIER_VERSION in your .env if you need a tier other than
       # the default (see README "Worker data (LMDB / sqlite) downloads").
+      # Deploying elsewhere? ARAX_PATHFINDER_DBS_DIR must point at the mount --
+      # it defaults to a path relative to /app, and a mismatch silently
+      # re-downloads everything instead of using the mounted copies.
       - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs
 
   arax_rank:

shepherd_utils/config.py

@@ -122,6 +122,19 @@ class Settings(BaseSettings):
     omnicorp_lmdb_url: str = ""
     pathfinder_embeddings_url: str = ""
 
+    # Socket timeout (seconds) for those startup dataset downloads. This bounds
+    # each individual blocking socket operation -- the connect and every
+    # subsequent read -- not the transfer as a whole, so a genuinely large file
+    # still downloads for as long as it needs provided it keeps making progress.
+    # Without it, urllib inherits Python's default of no timeout: a connection
+    # that opens and then stalls (an egress proxy that swallows the request, a
+    # server that accepts and never responds) hangs the worker's startup
+    # forever, before it ever reaches the poll loop. That is silent -- the
+    # worker never registers a heartbeat and never picks up a task, but it also
+    # never crashes. With a timeout the download fails loudly instead. Set to 0
+    # to restore the unbounded behavior.
+    dataset_download_timeout_sec: float = 60.0
+
     otel_enabled: bool = True
     jaeger_host: str = "http://jaeger"
     jaeger_port: int = 4317

shepherd_utils/data_download.py

@@ -15,7 +15,8 @@
   ``aragorn_omnicorp`` and ``score_paths`` below).
 * **Per-file** -- individual files fetched directly, no archive/extract step
   (``arax_pathfinder`` below, whose two sqlite databases are served as plain
-  files rather than bundled into one archive).
+  files rather than bundled into one archive, plus the ARAX blocked-concept
+  list that shares their directory).
 
 When a download source is configured (read via :mod:`shepherd_utils.config`),
 each worker calls its matching ``ensure_*`` helper at startup:
@@ -50,14 +51,25 @@ def _missing_files(target_dir: str, required_files: List[str]) -> List[str]:
 
 
 def _download(url: str, dest_path: str, logger: logging.Logger) -> None:
-    """Stream ``url`` to ``dest_path``, logging progress periodically."""
+    """Stream ``url`` to ``dest_path``, logging progress periodically.
+
+    Bounded by ``dataset_download_timeout_sec`` (see config). The timeout is
+    per socket operation rather than for the whole transfer, so a multi-GB file
+    downloads for as long as it needs while a stalled connection raises instead
+    of hanging worker startup indefinitely.
+    """
     logger.info(f"Downloading dataset from {url} ...")
+    timeout = float(settings.dataset_download_timeout_sec)
+    # urlopen treats timeout=None as "no timeout"; 0 is the documented opt-out.
+    kwargs = {"timeout": timeout} if timeout > 0 else {}
+    # Hoisted so the timeout handler below can report how far the transfer got
+    # even when it never made it past opening the connection.
+    read = 0
     try:
         # nosec B310: the URL is operator-configured (an env var), not user input.
-        with urllib.request.urlopen(url) as resp:  # noqa: S310
+        with urllib.request.urlopen(url, **kwargs) as resp:  # noqa: S310
             header = resp.headers.get("Content-Length")
             total = int(header) if header and header.isdigit() else None
-            read = 0
             step = 50 * 1024 * 1024  # log roughly every 50 MB
             next_log = step
             with open(dest_path, "wb") as out:
@@ -82,10 +94,22 @@ def _download(url: str, dest_path: str, logger: logging.Logger) -> None:
             f"URL is correct and reachable."
         ) from e
     except urllib.error.URLError as e:
+        # A connect timeout arrives here, wrapped, with e.reason set to the
+        # underlying socket.timeout.
         raise RuntimeError(
             f"download failed for {url}: {e.reason}. Confirm the URL is correct "
             f"and reachable from inside the container."
         ) from e
+    except TimeoutError as e:
+        # A stall part-way through the body raises straight out of resp.read()
+        # rather than being wrapped in URLError, so it needs its own handler --
+        # otherwise this surfaces as a bare TimeoutError with no context.
+        raise RuntimeError(
+            f"download failed for {url}: no data for {timeout:.0f}s "
+            f"({read / 1e6:.0f} MB transferred). The host is reachable but the "
+            f"transfer stalled -- check for an egress proxy or a rate limit, or "
+            f"raise DATASET_DOWNLOAD_TIMEOUT_SEC."
+        ) from e
     logger.info(f"Download complete: {read / 1e6:.0f} MB")
 
 
@@ -123,7 +147,7 @@ def ensure_lmdb_dataset(
     Idempotent: once the files are present this returns immediately, so it's safe
     to call unconditionally on every worker startup.
     """
-    logger = logger or logging.getLogger(__name__)
+    logger = logger or logging.getLogger("shepherd.data_download")
 
     missing = _missing_files(target_dir, required_files)
     if not missing:
@@ -191,7 +215,7 @@ def ensure_http_files_dataset(
     Idempotent: once a file is present it's left alone, so it's safe to call
     unconditionally on every worker startup.
     """
-    logger = logger or logging.getLogger(__name__)
+    logger = logger or logging.getLogger("shepherd.data_download")
     required_files = list(file_sources.keys())
 
     missing = _missing_files(target_dir, required_files)
@@ -344,3 +368,44 @@ def ensure_arax_pathfinder_dbs(logger: Optional[logging.Logger] = None) -> None:
         },
         logger=logger,
     )
+
+
+# The ARAX blocked-concept list lives alongside the pathfinder sqlite databases
+# so it lands on the same mounted volume. It previously went to the worker's
+# working directory (``/app``), which is the container's writable layer and
+# therefore thrown away on every restart -- meaning each new pod re-fetched it
+# from GitHub during startup, before the poll loop, on a code path whose logs
+# were being discarded. On the volume it is fetched once and persists.
+ARAX_BLOCKED_LIST_FILENAME = "general_concepts.json"
+
+
+def arax_blocked_list_path() -> str:
+    """Return the on-disk path of the ARAX blocked-concept list.
+
+    Single source of truth, in the same spirit as
+    ``arax_pathfinder_sqlite_paths``: ``ensure_arax_blocked_list`` uses it to
+    know where to download, and worker.py uses it to know what to open.
+    """
+    return os.path.join(settings.arax_pathfinder_dbs_dir, ARAX_BLOCKED_LIST_FILENAME)
+
+
+def ensure_arax_blocked_list(logger: Optional[logging.Logger] = None) -> None:
+    """Ensure the ARAX blocked-concept list is present next to the sqlite dbs.
+
+    Fetched via the shared downloader so it lands through a temp file + atomic
+    rename (a direct write let concurrent tasks race on a half-written file)
+    and inherits the download timeout. Idempotent, so it is safe to call at
+    startup and again lazily from a pool child.
+
+    Note the flip side of persisting this on the volume: it is now only fetched
+    when absent, so a refreshed upstream list is not picked up until the file is
+    deleted. Delete it from the volume to force a re-fetch on the next restart.
+    """
+    ensure_http_files_dataset(
+        name="arax_blocked_list",
+        # ``or "."`` so an unset/blank dbs dir degrades to the working directory
+        # rather than handing makedirs an empty path.
+        target_dir=os.path.dirname(arax_blocked_list_path()) or ".",
+        file_sources={ARAX_BLOCKED_LIST_FILENAME: settings.arax_blocked_list_url},
+        logger=logger,
+    )

shepherd_utils/db.py

@@ -193,7 +193,7 @@ async def initialize_db() -> None:
     except Exception:
         # A failed upgrade must never keep a worker from starting: the schema
         # additions are performance aids, and the janitor/next boot retries.
-        logging.getLogger(__name__).warning(
+        logging.getLogger("shepherd.db").warning(
             "Failed to apply startup schema upgrades", exc_info=True
         )
 

shepherd_utils/logger.py

@@ -128,10 +128,32 @@ def get_logging_config():
             "default": {"format": "[%(asctime)s: %(levelname)s/%(name)s]: %(message)s"}
         },
         "handlers": handlers,
+        # The output handlers live on root, and root alone. Previously they were
+        # attached to ``shepherd`` and root was left unconfigured, so any record
+        # logged outside that namespace -- a stray ``logging.info``, a
+        # third-party library, a logger named for its module or its Redis stream
+        # -- reached a handler-less root and was dropped by logging's
+        # ``lastResort`` fallback, which only emits WARNING+ and ignores our
+        # formatter. That silently swallowed every worker's entire startup
+        # phase; see ``get_worker_logger``.
+        #
+        # Root's level applies only to records logged directly on root, not to
+        # ones propagated up from a child (those are level-checked at the
+        # originating logger). So WARNING here means third-party libraries are
+        # quiet below WARNING -- httpx logs a line per request at INFO, which
+        # would bury our own output -- while ``shepherd.*`` still emits at DEBUG
+        # via the entry below. Everything then reaches these handlers by
+        # propagation, which also keeps pytest's ``caplog`` working.
+        "root": {
+            "level": "WARNING",
+            "handlers": logger_handlers,
+        },
         "loggers": {
+            # No handlers: records propagate to root's. Only the level is set
+            # here, which is what lets our own logging through at DEBUG while
+            # leaving third-party loggers at root's WARNING.
             "shepherd": {
                 "level": "DEBUG",
-                "handlers": logger_handlers,
             },
             # psycopg's pool retries to keep min_size connections warm and logs
             # a WARNING on every failed attempt. When the DB is down that floods
@@ -150,6 +172,25 @@ def get_logging_config():
     return logging_config
 
 
+def get_worker_logger(name: str) -> logging.Logger:
+    """Return a logger under the configured ``shepherd`` namespace.
+
+    ``setup_logging`` only attaches handlers to ``shepherd`` (and, as a
+    WARNING-level backstop, root), so a logger named for the stream alone
+    (``logging.getLogger("arax.pathfinder")``) inherits no handler at INFO and
+    its records vanish. Every worker's startup phase -- dataset downloads, pool
+    sizing, poll-loop errors -- logged through exactly such a logger, so a
+    worker that hung before reaching ``get_tasks`` produced no output at all
+    and looked identical to a healthy idle one.
+
+    Passing a name that is already namespaced is a no-op, so this is safe to
+    apply to existing ``shepherd.``-prefixed names.
+    """
+    if name == "shepherd" or name.startswith("shepherd."):
+        return logging.getLogger(name)
+    return logging.getLogger(f"shepherd.{name}")
+
+
 def setup_logging():
     """Set up logging."""
     config = get_logging_config()

tests/unit/test_data_download.py

@@ -5,7 +5,9 @@
 """
 
 import logging
+import os
 import tarfile
+import urllib.request  # noqa: F401  (patched by name in the timeout tests)
 
 import pytest
 
@@ -155,3 +157,91 @@ def test_ensure_pathfinder_embeddings_wires_settings(tmp_path, mocker):
 
     assert (target / "data.mdb").exists()
     assert (target / "lock.mdb").exists()
+
+
+# --- download timeout --------------------------------------------------------
+#
+# urllib defaults to no timeout, so a connection that opens and then stalls hung
+# worker startup forever -- before the poll loop, before any heartbeat, and with
+# no crash to point at.
+
+
+def test_download_passes_configured_timeout(tmp_path, mocker):
+    mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 12.5)
+    urlopen = mocker.patch.object(data_download.urllib.request, "urlopen")
+    urlopen.return_value.__enter__.return_value.headers.get.return_value = None
+    urlopen.return_value.__enter__.return_value.read.return_value = b""
+
+    data_download._download("http://example.invalid/x", str(tmp_path / "out"), logger)
+
+    assert urlopen.call_args.kwargs["timeout"] == 12.5
+
+
+def test_download_timeout_zero_restores_unbounded_behavior(tmp_path, mocker):
+    mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 0)
+    urlopen = mocker.patch.object(data_download.urllib.request, "urlopen")
+    urlopen.return_value.__enter__.return_value.headers.get.return_value = None
+    urlopen.return_value.__enter__.return_value.read.return_value = b""
+
+    data_download._download("http://example.invalid/x", str(tmp_path / "out"), logger)
+
+    assert "timeout" not in urlopen.call_args.kwargs
+
+
+def test_download_reports_a_mid_transfer_stall(tmp_path, mocker):
+    """A stall inside resp.read() raises TimeoutError directly rather than being
+    wrapped in URLError, so it needs its own handler to get a useful message."""
+    mocker.patch.object(data_download.settings, "dataset_download_timeout_sec", 30)
+    urlopen = mocker.patch.object(data_download.urllib.request, "urlopen")
+    resp = urlopen.return_value.__enter__.return_value
+    resp.headers.get.return_value = None
+    resp.read.side_effect = [b"partial", TimeoutError("timed out")]
+
+    with pytest.raises(RuntimeError, match="transfer stalled"):
+        data_download._download(
+            "http://example.invalid/x", str(tmp_path / "out"), logger
+        )
+
+
+# --- ARAX blocked list -------------------------------------------------------
+
+
+def test_blocked_list_lives_with_the_pathfinder_sqlite_dbs(tmp_path, mocker):
+    """It used to be written to the working directory -- the container's
+    writable layer -- so every new pod re-fetched it from GitHub at startup."""
+    mocker.patch.object(
+        data_download.settings, "arax_pathfinder_dbs_dir", str(tmp_path / "dbs")
+    )
+
+    path = data_download.arax_blocked_list_path()
+
+    assert path == str(tmp_path / "dbs" / "general_concepts.json")
+    curie_ngd, _ = data_download.arax_pathfinder_sqlite_paths()
+    # Same volume as the sqlite databases, which is the whole point.
+    assert os.path.dirname(path) == os.path.dirname(curie_ngd)
+
+
+def test_ensure_arax_blocked_list_downloads_into_the_volume(tmp_path, mocker):
+    target = tmp_path / "dbs"
+    source = tmp_path / "general_concepts.json"
+    source.write_text('{"curies": ["CHEBI:1"], "synonyms": ["Water"]}')
+    mocker.patch.object(data_download.settings, "arax_pathfinder_dbs_dir", str(target))
+    mocker.patch.object(
+        data_download.settings, "arax_blocked_list_url", source.as_uri()
+    )
+
+    data_download.ensure_arax_blocked_list(logger)
+
+    assert (target / "general_concepts.json").exists()
+
+
+def test_ensure_arax_blocked_list_is_a_noop_once_present(tmp_path, mocker):
+    target = tmp_path / "dbs"
+    target.mkdir()
+    (target / "general_concepts.json").write_text('{"curies": [], "synonyms": []}')
+    mocker.patch.object(data_download.settings, "arax_pathfinder_dbs_dir", str(target))
+    spy = mocker.patch.object(data_download, "_download")
+
+    data_download.ensure_arax_blocked_list(logger)
+
+    spy.assert_not_called()