Coverage for src/local_deep_research/journal_quality/downloader.py: 86%
226 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Bulk-fetch and lazy-load journal-quality datasets.
3This module is now a thin orchestration layer over the
4`data_sources` package. Each academic dataset is a `DataSource`
5subclass; this module just iterates `ALL_SOURCES` to drive the bulk
6download flow (the dashboard "Download Data" button) and to compute the
7status payload returned by the `/metrics/api/journal-data/status`
8endpoint.
10Public API (kept stable for existing callers and tests):
11- `JOURNAL_DATA_VERSION`
12- `REQUIRED_SNAPSHOT_FILENAME`
13- `get_journal_data_status()`
14- `download_journal_data(force=False)`
15- `ensure_journal_data(auto_download=True)`
17Test-patch surface (kept stable for tests that mock by string path):
18- `_get_data_dir`
19- `_fetch_openalex_sources` (re-export shim)
20- `_fetch_doaj_journals` (re-export shim)
22Data source registry lives in `.data_sources` and is the single source
23of truth for source metadata, fetch logic, and download policy.
24"""
26from __future__ import annotations
28import json
29import os
30import time
31from pathlib import Path
32from typing import Optional
34from loguru import logger
36from .data_sources import ALL_SOURCES, get_source
38JOURNAL_DATA_VERSION = "v4"
40#: The one REQUIRED snapshot, and therefore the predicate that decides
41#: whether a journal-DB build needs the network at all: with this file in
42#: the data directory ``ensure_journal_data`` reports "available" without
43#: a single request and ``JournalQualityDB._build_or_raise`` compiles
44#: straight from disk. Named here rather than repeated as a literal so
45#: the callers that PREDICT a download — ``web/routers/metrics.py``'s
46#: ``_journal_ref_db_read_would_download`` — cannot drift from the caller
47#: that performs one.
48REQUIRED_SNAPSHOT_FILENAME = "openalex_sources.json.gz"
50_SENTINEL = ".downloading"
53# If the sentinel is older than this, assume the previous download
54# crashed mid-way (thread died before the `finally` cleanup could run)
55# and reclaim it. The expected wall-clock is ~7 minutes, so 20 minutes
56# is a generous safety margin that still catches stuck sentinels.
57_SENTINEL_STALE_SECS = 20 * 60
59# Shared progress state for the dashboard's status endpoint.
60# A module-level dict is deliberate: there's only ever one concurrent
61# download (enforced by the O_EXCL sentinel), and the status endpoint
62# reads it on a best-effort basis. Structure:
63# state: "idle" | "running" | "error" | "success"
64# started_at: epoch seconds, or None
65# finished_at: epoch seconds, or None
66# sources: {src_key: {name, state, detail, count}}
67# db_build: {state, detail}
68# error_msg: str or None
69#
70# Per-source entries track independent parallel downloads. Writes are
71# atomic at the per-key level in CPython, so workers can update their
72# own sub-dict without a lock; the main thread composes the overall
73# `state` / `error_msg` after joining.
74_download_state: dict = {
75 "state": "idle",
76 "started_at": None,
77 "finished_at": None,
78 "sources": {},
79 "db_build": {"state": "pending", "detail": ""},
80 "error_msg": None,
81 # Per-source final counts from the most recently COMPLETED download.
82 # Set to a dict only on the success path; explicitly invalidated to
83 # None on every other return in `download_journal_data` (up-to-date,
84 # disk-space, sentinel-race, required-failure, DB-build-failure) so
85 # callers cannot read stale counts from a prior successful run.
86 # Callers rendering a user-facing summary should prefer this
87 # structured field over parsing the `(success, message)` tuple's
88 # string.
89 "counts": None,
90}
93def get_download_state() -> dict:
94 """Return a copy of the current download progress state.
96 A shallow dict copy is not enough — callers (the status endpoint)
97 serialize this to JSON and would otherwise race with live updates.
98 We copy the nested dicts too.
99 """
100 counts = _download_state["counts"]
101 return {
102 "state": _download_state["state"],
103 "started_at": _download_state["started_at"],
104 "finished_at": _download_state["finished_at"],
105 "db_build": dict(_download_state["db_build"]),
106 "error_msg": _download_state["error_msg"],
107 "sources": {k: dict(v) for k, v in _download_state["sources"].items()},
108 "counts": dict(counts) if counts is not None else None,
109 }
112def _set_source_state(key: str, **updates) -> None:
113 """Update a single source's state entry. Safe to call from worker
114 threads because the only writer of ``sources[key]`` is that key's
115 own worker (CPython guarantees dict-item writes are atomic).
116 """
117 entry = _download_state["sources"].setdefault(
118 key, {"name": key, "state": "pending", "detail": "", "count": 0}
119 )
120 entry.update(updates)
121 logger.info(
122 f"journal-data progress: {entry.get('name', key)} "
123 f"{entry.get('state', '?')} {entry.get('detail', '')}".rstrip()
124 )
127def _get_data_dir() -> Path:
128 """Get the journal data directory (user-writable).
130 Kept as a module-level function so tests can patch it via
131 `mock.patch("...journal_data_downloader._get_data_dir", ...)`.
132 """
133 from ..config.paths import get_journal_data_directory
135 return get_journal_data_directory()
138def _clear_orphan_sentinel_on_startup() -> None:
139 """Remove any ``.downloading`` sentinel left over from a previous
140 process that got killed mid-download.
142 The sentinel is created inside ``download_journal_data`` and cleaned
143 up in its ``finally`` block. If the process is SIGKILLed (or crashes
144 hard enough that the ``finally`` doesn't run) the file sits on disk
145 forever. Every subsequent call then sees "Download already in
146 progress" and bows out, even though nothing is actually downloading.
148 Called once at import time. A fresh process can't possibly own an
149 in-progress download, so any pre-existing sentinel is by definition
150 an orphan. The 20-minute ``_SENTINEL_STALE_SECS`` recovery path
151 still exists for the "process still alive but hung" case.
153 Swallows all exceptions — a misread on startup must not break the
154 module. Tests that run concurrently in the same process
155 (test_concurrent_download_blocked) set the sentinel deliberately
156 *after* import, so this runs once and doesn't interfere.
157 """
158 try:
159 sentinel = _get_data_dir() / _SENTINEL
160 if sentinel.exists(): 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 sentinel.unlink()
162 logger.warning(
163 f"Cleared orphan {_SENTINEL} sentinel from a previous run "
164 f"(the old process was killed mid-download; a fresh process "
165 f"cannot own an in-progress download)."
166 )
167 except Exception:
168 logger.exception(
169 "Could not clear orphan sentinel on startup; "
170 "new downloads may be blocked until the 20-minute stale timer."
171 )
174_clear_orphan_sentinel_on_startup()
177# ---------------------------------------------------------------------------
178# Test-patch shims for `_fetch_openalex_sources` and `_fetch_doaj_journals`
179#
180# Tests in `tests/journal_quality/test_downloader.py` and
181# `tests/journal_quality/test_downloader_exception_sanitization.py` patch
182# these names by string path. The bodies have moved into `OpenAlexSource.fetch`
183# and `DOAJSource.fetch`, but we expose module-level wrappers so the
184# existing patches keep intercepting calls. The bulk download loop below
185# routes both sources through these wrappers (not directly through
186# `.fetch()`) so that `mock.patch` substitutions take effect.
187# ---------------------------------------------------------------------------
190def _fetch_openalex_sources(data_dir: Path, progress_cb=None) -> int:
191 return get_source("openalex").fetch(data_dir, progress_cb=progress_cb)
194def _fetch_doaj_journals(data_dir: Path, progress_cb=None) -> int:
195 return get_source("doaj").fetch(data_dir, progress_cb=progress_cb)
198def _fetch_predatory(data_dir: Path, progress_cb=None) -> int:
199 return get_source("predatory").fetch(data_dir, progress_cb=progress_cb)
202def _fetch_jabref_abbreviations(data_dir: Path, progress_cb=None) -> int:
203 return get_source("jabref").fetch(data_dir, progress_cb=progress_cb)
206def _fetch_institutions(data_dir: Path, progress_cb=None) -> int:
207 return get_source("institutions").fetch(data_dir, progress_cb=progress_cb)
210# Map source key → module-level shim *name*, resolved at call time so
211# `mock.patch` substitutions on the module attribute take effect (capturing
212# function objects in a dict at import time would defeat patching).
213_FETCH_SHIM_NAMES = {
214 "openalex": "_fetch_openalex_sources",
215 "doaj": "_fetch_doaj_journals",
216 "predatory": "_fetch_predatory",
217 "jabref": "_fetch_jabref_abbreviations",
218 "institutions": "_fetch_institutions",
219}
222# ---------------------------------------------------------------------------
223# Status
224# ---------------------------------------------------------------------------
227def get_journal_data_status() -> dict:
228 """Return the status payload for the dashboard data sources banner.
230 Shape (kept stable for existing JS consumers and tests):
231 {
232 "available": bool, # OpenAlex source OR compiled DB present
233 "version": Optional[str],
234 "latest_version": str,
235 "needs_update": bool,
236 "files": dict[str, bool], # legacy: filename → present
237 "sources": list[dict], # per-source detail for the banner
238 "data_dir": str,
239 }
240 """
241 # Lazy import to avoid any downloader <-> db module-level cycle,
242 # matching the existing db.py -> downloader.py lazy imports for this
243 # same pair (`_warn_on_stale_data_version`, `_build_or_raise`).
244 from .db import DB_FILENAME, journal_db_file_is_valid
246 data_dir = _get_data_dir()
247 version_file = data_dir / "version.json"
249 installed_version: Optional[str] = None
250 if version_file.exists():
251 try:
252 with open(version_file, encoding="utf-8") as f:
253 info = json.load(f)
254 installed_version = info.get("version")
255 except (json.JSONDecodeError, OSError):
256 pass
258 files = {src.filename: src.is_present(data_dir) for src in ALL_SOURCES}
259 sources = [src.status_dict(data_dir) for src in ALL_SOURCES]
261 # Available if the (required) OpenAlex source exists OR the compiled
262 # reference DB exists AND is usable. The compiled DB is useful even
263 # when the source JSON has been deleted, since the dashboard can
264 # still query it — but a file that merely EXISTS while being
265 # unreadable (stale schema, corrupt) must not count: every actual
266 # read would rebuild it (or, under a forbidding egress scope,
267 # refuse) instead of serving it, so reporting "available" here would
268 # hide the install banner behind a file that answers with silent
269 # zeros. `journal_reference.db` is the pre-rename legacy filename
270 # the downloader below deletes after a fresh download; an install
271 # that hasn't gone through one yet is checked the same way. The
272 # `exists()` guard keeps an absent file (the usual case for the
273 # legacy name, and for both on first run) a quiet "not present":
274 # the validator logs an exception traceback on any open failure.
275 has_source = files.get(REQUIRED_SNAPSHOT_FILENAME, False)
276 has_db = any(
277 path.exists() and journal_db_file_is_valid(path)
278 for path in (data_dir / DB_FILENAME, data_dir / "journal_reference.db")
279 )
281 return {
282 "available": has_source or has_db,
283 "version": installed_version,
284 "latest_version": JOURNAL_DATA_VERSION,
285 # `needs_update` is True both when no version is installed at all
286 # (first run — show the download CTA) and when the installed
287 # version is older than the bundled latest. The previous `and`
288 # spelling silently hid the first-run case from the dashboard.
289 "needs_update": (
290 installed_version is None
291 or installed_version != JOURNAL_DATA_VERSION
292 ),
293 "files": files,
294 "sources": sources,
295 "data_dir": str(data_dir),
296 # Live progress for the dashboard's status indicator. The client
297 # polls this endpoint while a download is in flight; see
298 # downloadJournalData() in journal_quality.html.
299 "download_progress": get_download_state(),
300 }
303# ---------------------------------------------------------------------------
304# Bulk download (dashboard "Download Data" button)
305# ---------------------------------------------------------------------------
308def download_journal_data(force: bool = False) -> tuple[bool, str]:
309 """Fetch every registered data source into the user data directory.
311 Iterates `ALL_SOURCES` in order. Sources marked `required=True`
312 (OpenAlex) abort the batch on failure; `required=False` sources are
313 best-effort and continue on error.
315 Args:
316 force: Re-fetch even if data exists and is current version.
318 Returns:
319 (success, message) tuple. Message format is
320 `"Fetched <N1> <label1> + <N2> <label2> + ... in <S>s"` so the
321 existing test substring assertions ("100 OpenAlex", "50 DOAJ")
322 continue to match.
323 """
324 data_dir = _get_data_dir()
325 sentinel = data_dir / _SENTINEL
327 if not force:
328 status = get_journal_data_status()
329 if status["available"] and not status["needs_update"]:
330 # No fresh fetch ran, so invalidate any counts cached from a
331 # previous in-process download. Callers keying a "what just
332 # happened" summary off `counts` must see None here.
333 _download_state["counts"] = None
334 return True, "Journal data is already up to date"
336 # Disk-space pre-check. The five data sources uncompress to ~1 GB
337 # intermediate, plus the compiled DB. Fail fast with a clear message
338 # rather than crashing mid-download and leaving a corrupt tmp file.
339 import shutil as _shutil
341 from ..constants import JOURNAL_QUALITY_MIN_FREE_DISK_BYTES
343 try:
344 free_bytes = _shutil.disk_usage(str(data_dir)).free
345 except OSError:
346 logger.warning(
347 f"Could not check free disk space for {data_dir}; proceeding."
348 )
349 free_bytes = None
350 if (
351 free_bytes is not None
352 and free_bytes < JOURNAL_QUALITY_MIN_FREE_DISK_BYTES
353 ):
354 # No fetch ran → invalidate any stale counts from a prior call
355 # so `get_download_state()["counts"]` cannot leak them.
356 _download_state["counts"] = None
357 return False, (
358 f"Insufficient disk space: "
359 f"{free_bytes / (1024**3):.1f} GB available, "
360 f"{JOURNAL_QUALITY_MIN_FREE_DISK_BYTES / (1024**3):.0f} GB required."
361 )
363 # Atomic sentinel creation (O_CREAT | O_EXCL). Replaces the previous
364 # exists()+touch() TOCTOU race so two concurrent download triggers
365 # (dashboard click + scheduler) cannot both proceed.
366 #
367 # Stale-sentinel recovery. Two triggers are checked each call:
368 #
369 # 1. PID-based: the sentinel holds the PID of the process that
370 # created it. If that PID is not alive, the owner process
371 # crashed or was killed and the sentinel is orphan.
372 # 2. Age-based (fallback): if the sentinel is older than
373 # _SENTINEL_STALE_SECS we reclaim it even if the PID check is
374 # inconclusive (exotic environments, unreadable sentinel).
375 #
376 # The startup hook in _clear_orphan_sentinel_on_startup handles the
377 # common case of "server was restarted mid-download"; these two
378 # runtime checks cover the case where the server is still running
379 # but the download worker thread itself crashed out of the sentinel.
381 def _sentinel_owner_alive() -> bool:
382 try:
383 owner_pid = int(sentinel.read_text(encoding="utf-8").strip())
384 except (OSError, ValueError):
385 return False # unreadable / malformed → treat as orphan
386 if owner_pid == os.getpid(): 386 ↛ 391line 386 didn't jump to line 391 because the condition on line 386 was always true
387 # Same process. Something is wrong (we should never race
388 # with ourselves — the module-level lock guards that) but
389 # err on the side of "alive" to avoid self-nuking.
390 return True
391 try:
392 os.kill(owner_pid, 0) # signal 0 = liveness probe
393 except ProcessLookupError:
394 return False
395 except PermissionError:
396 # Process exists, owned by another user — treat as alive.
397 return True
398 except OSError:
399 return False
400 return True
402 def _try_claim_sentinel() -> bool:
403 """Create the sentinel + stamp our PID. Returns True on success."""
404 try:
405 with sentinel.open("x", encoding="utf-8") as f:
406 f.write(str(os.getpid()))
407 return True
408 except FileExistsError:
409 return False
411 if not _try_claim_sentinel():
412 try:
413 age = time.time() - sentinel.stat().st_mtime
414 except OSError:
415 age = 0
416 orphan = not _sentinel_owner_alive()
417 if orphan or age > _SENTINEL_STALE_SECS:
418 reason = (
419 "owner process not alive"
420 if orphan
421 else f"age {age:.0f}s > {_SENTINEL_STALE_SECS}s"
422 )
423 logger.warning(
424 f"Reclaiming stale .downloading sentinel ({reason}); "
425 "previous download likely crashed without cleanup."
426 )
427 sentinel.unlink(missing_ok=True)
428 if not _try_claim_sentinel(): 428 ↛ 431line 428 didn't jump to line 431 because the condition on line 428 was never true
429 # Lost a race with another caller reclaiming the same
430 # stale sentinel — bow out cleanly.
431 _download_state["counts"] = None
432 return False, "Download already in progress"
433 else:
434 _download_state["counts"] = None
435 return False, "Download already in progress"
436 try:
437 start = time.time()
438 counts: dict[str, int] = {}
439 parts: list[str] = []
441 # Reset per-source state to a clean "pending" row for each
442 # known source. The dashboard renders one row per entry.
443 _download_state["state"] = "running"
444 _download_state["started_at"] = start
445 _download_state["finished_at"] = None
446 _download_state["error_msg"] = None
447 # Invalidate counts from any prior run — callers inspecting the
448 # structured summary must not see stale data if this download
449 # fails or is still running.
450 _download_state["counts"] = None
451 _download_state["db_build"] = {"state": "pending", "detail": ""}
452 _download_state["sources"] = {
453 src.key: {
454 "name": src.name,
455 "state": "pending",
456 "detail": "",
457 "percent": 0,
458 "count": 0,
459 "required": src.required,
460 }
461 for src in ALL_SOURCES
462 }
464 def _fetch_one(src):
465 """Worker: run one source's fetch, mirror state as it goes."""
466 _set_source_state(
467 src.key, state="running", detail="downloading", percent=5
468 )
470 # Per-partition callback: the chunked sources (openalex
471 # sources + institutions) call this on every partition so
472 # the dashboard's bar moves smoothly. One-shot sources
473 # don't call it and stay at the initial 5% → final 100%.
474 def _on_progress(done, total, detail):
475 pct = int(5 + (done / total) * 90) if total > 0 else 5
476 _set_source_state(
477 src.key,
478 state="running",
479 detail=detail,
480 percent=max(5, min(95, pct)),
481 )
483 try:
484 shim_name = _FETCH_SHIM_NAMES.get(src.key)
485 if shim_name:
486 # bearer:disable python_lang_code_injection
487 # _FETCH_SHIM_NAMES is a hardcoded dict (line 188); the
488 # late-bound globals() lookup is needed for mock.patch
489 # compatibility in tests. No user input reaches the key.
490 n = globals()[shim_name](data_dir, progress_cb=_on_progress)
491 else:
492 n = src.fetch(data_dir, progress_cb=_on_progress)
493 _set_source_state(
494 src.key,
495 state="success",
496 detail=f"{n} {src.count_label}",
497 percent=100,
498 count=n,
499 )
500 return (src, n, None)
501 except Exception as exc:
502 logger.exception(
503 f"{src.name} fetch failed "
504 f"({'required' if src.required else 'optional'})"
505 )
506 _set_source_state(
507 src.key,
508 state="error",
509 detail=exc.__class__.__name__,
510 percent=100,
511 )
512 return (src, 0, exc)
514 # Parallel fetch — every source streams from a different host
515 # (openalex S3, DOAJ CSV, raw.githubusercontent.com, api.openalex.org
516 # for institutions). No single-host contention; the wall-clock is
517 # dominated by the slowest source (OpenAlex snapshot ~30-60 s).
518 from concurrent.futures import ThreadPoolExecutor
520 with ThreadPoolExecutor(
521 max_workers=max(1, len(ALL_SOURCES)),
522 thread_name_prefix="journal-dl",
523 ) as pool:
524 results = list(pool.map(_fetch_one, ALL_SOURCES))
526 required_failure = None
527 for src, n, exc in results:
528 counts[src.key] = n
529 parts.append(f"{n} {src.count_label}")
530 if exc is not None and src.required:
531 required_failure = (src, exc)
533 if required_failure is not None:
534 src, _exc = required_failure
535 msg = f"Failed to fetch {src.name}. Check your network connection."
536 _download_state["state"] = "error"
537 _download_state["error_msg"] = msg
538 _download_state["finished_at"] = time.time()
539 return False, msg
541 # Write version marker. Per-source key names are preserved for
542 # any external consumer that might read them, even though no
543 # production code does today.
544 version_file = data_dir / "version.json"
545 with open(version_file, "w", encoding="utf-8") as f:
546 json.dump(
547 {
548 "version": JOURNAL_DATA_VERSION,
549 "downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
550 "openalex_count": counts.get("openalex", 0),
551 "doaj_count": counts.get("doaj", 0),
552 "jabref_count": counts.get("jabref", 0),
553 "predatory_count": counts.get("predatory", 0),
554 },
555 f,
556 )
558 # Rebuild journal_quality.db synchronously from the freshly
559 # downloaded gz files. The build is the ONLY writer of this
560 # file; everything else opens it read-only via mode=ro.
561 # Also clean up any leftover legacy journal_reference.db files
562 # from before the rename so existing installs don't carry junk.
563 legacy_db = data_dir / "journal_reference.db"
564 if legacy_db.exists(): 564 ↛ 565line 564 didn't jump to line 565 because the condition on line 564 was never true
565 try:
566 import os as _os
568 # bearer:disable python_lang_file_permissions
569 _os.chmod(legacy_db, 0o644)
570 legacy_db.unlink()
571 logger.info(
572 "Removed legacy journal_reference.db "
573 "(replaced by journal_quality.db)"
574 )
575 except OSError:
576 logger.exception("Could not remove legacy DB")
578 _download_state["db_build"] = {
579 "state": "running",
580 "detail": "parsing bundled data",
581 }
582 db_build_error: Optional[str] = None
583 try:
584 from .db import DB_FILENAME, build_db
586 new_db_file = data_dir / DB_FILENAME
587 if new_db_file.exists(): 587 ↛ 588line 587 didn't jump to line 588 because the condition on line 587 was never true
588 import os as _os
590 # bearer:disable python_lang_file_permissions
591 _os.chmod(new_db_file, 0o644)
592 new_db_file.unlink()
593 build_db(data_dir=data_dir, output_path=new_db_file)
594 except Exception as exc:
595 logger.exception(
596 "Failed to rebuild journal_quality.db; "
597 "the runtime accessor will lazy-build on next access"
598 )
599 # SchemaDriftError messages are developer-authored literals
600 # (no SQL, paths, or stack fragments) so they're safe to
601 # surface — operators need to see *which* field drifted to
602 # act on it. For any other exception, fall back to the
603 # class name only, per CodeQL "Information exposure through
604 # an exception" (alerts 7650, 7684). The full trace always
605 # stays in logger.exception above (server-side only).
606 from .data_sources.openalex import SchemaDriftError
608 if isinstance(exc, SchemaDriftError):
609 db_build_error = str(exc)
610 else:
611 db_build_error = exc.__class__.__name__
613 elapsed = time.time() - start
614 if db_build_error:
615 msg = (
616 f"Downloaded data ({' + '.join(parts)}) in {elapsed:.0f}s "
617 f"but DB build failed ({db_build_error}). "
618 f"Lazy-build will retry on next access."
619 )
620 _download_state["db_build"] = {
621 "state": "error",
622 "detail": db_build_error,
623 }
624 _download_state["state"] = "error"
625 _download_state["error_msg"] = msg
626 _download_state["finished_at"] = time.time()
627 return False, msg
629 success_msg = f"Fetched {' + '.join(parts)} in {elapsed:.0f}s"
630 _download_state["db_build"] = {
631 "state": "success",
632 "detail": "ready",
633 }
634 _download_state["state"] = "success"
635 _download_state["error_msg"] = None
636 _download_state["finished_at"] = time.time()
637 # Publish structured counts for callers that want to render a
638 # user-facing summary without echoing `success_msg`. All values
639 # are ints populated from source `.fetch()` returns above.
640 _download_state["counts"] = dict(counts)
641 return True, success_msg
643 finally:
644 sentinel.unlink(missing_ok=True)
647_ensure_cache: Optional[tuple[float, tuple[Optional[Path], bool]]] = None
648_ENSURE_CACHE_TTL = 30.0 # seconds
651def ensure_journal_data(
652 auto_download: bool = True,
653) -> tuple[Optional[Path], bool]:
654 """Ensure journal data is available, optionally triggering a bulk fetch.
656 Returns:
657 (data_dir, is_available) — data_dir is None if unavailable.
659 Thundering-herd guard: when a search runs, every search engine's
660 reputation-filter worker (~30 threads) calls this concurrently.
661 Without the cache, 29 of them race to create the sentinel and
662 each logs a WARNING. One call does the real work; the rest get
663 the cached answer for 30 seconds. The success path (data files
664 present) is already fast — we only cache the negative / race
665 result, which is the noisy path.
667 A caller that is not allowed to download never writes the cache:
668 it attempted nothing worth memoising, and the entry carries no
669 record of which question it answered, so it would go on to answer
670 callers that ARE allowed to download.
671 """
672 global _ensure_cache
674 user_dir = _get_data_dir()
675 if (user_dir / REQUIRED_SNAPSHOT_FILENAME).exists():
676 # Positive path is cheap (one stat call) — no need to cache.
677 return user_dir, True
679 now = time.time()
680 if _ensure_cache is not None:
681 ts, cached = _ensure_cache
682 if now - ts < _ENSURE_CACHE_TTL: 682 ↛ 685line 682 didn't jump to line 685 because the condition on line 682 was always true
683 return cached
685 if not auto_download:
686 # Refusals (test mock mode, or a lazy build whose egress scope
687 # forbids the public fetch) deliberately do NOT populate the
688 # cache. The cached tuple records no trace of which question it
689 # answered, so caching this one would answer the next 30 seconds
690 # of ensure_journal_data(auto_download=True) calls — another
691 # user's search, whose scope permits the fetch — with a stale
692 # "unavailable" and silently suppress a legitimate download.
693 # Nothing is lost: this branch performed a single stat, not the
694 # download/sentinel race the cache exists to absorb.
695 return None, False
697 logger.info("Journal data not found — fetching from upstream sources...")
698 success, message = download_journal_data()
699 if success:
700 logger.info(message)
701 _ensure_cache = (now, (user_dir, True))
702 return user_dir, True
703 logger.warning(f"Journal data fetch failed: {message}")
705 result: tuple[Optional[Path], bool] = (None, False)
706 _ensure_cache = (now, result)
707 return result