diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 39d66ff..a04650c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,7 +15,7 @@ Thank you for your interest in improving paperscout. This document describes how When adding blocking I/O or new data sources: 1. Use **`run_blocking_io()`** (see `src/paperscout/concurrency.py`) only for pure blocking I/O with **no shared in-process mutable state** (e.g. psycopg2 via the connection pool). -2. **Never access `ISOProber._stats`, `WG21Index.papers`, or other source internals from a thread** — keep probes and index refresh on the asyncio event loop. +2. **Never access `ISOProber._stats`, `WG21Index._papers`, or other source internals from a thread** — keep probes and index refresh on the asyncio event loop. Use `paper_count()` or `get_papers_snapshot()` for cross-thread reads. 3. See **[docs/architecture.md](docs/architecture.md)** for the full model and **[docs/probe-operations.md](docs/probe-operations.md)** for probe tuning. ## Workflow diff --git a/docs/architecture.md b/docs/architecture.md index 5c902fd..4c4267b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -17,7 +17,7 @@ These components share one thread and the main event loop. They may await I/O bu | Thread | Role | |--------|------| -| **Health server** (`health.py`) | Serves `GET /health`; reads `len(index.papers)` via a callback and scheduler fields from `Scheduler.health_snapshot()` (immutable snapshot, lock-protected publish). | +| **Health server** (`health.py`) | Serves `GET /health`; reads paper count via `index.paper_count()` callback and scheduler fields from `Scheduler.health_snapshot()` (immutable snapshot, lock-protected publish). | | **MessageQueue sender** (`scout.py`) | Drains Slack post queue with rate limiting. | | **`run_blocking_io` / `asyncio.to_thread`** | Runs blocking psycopg2 calls (e.g. `UserWatchlist.matches_for_users`) off the loop. | @@ -26,8 +26,8 @@ These components share one thread and the main event loop. They may await I/O bu When adding or changing code: 1. **Use `run_blocking_io()` (or `asyncio.to_thread`) only for pure blocking I/O** with no shared in-process mutable state. The function should use its own DB connection from the pool. -2. **Never access `ISOProber._stats`, `WG21Index.papers`, `WG21Index._max_rev`, or other source internals from a thread.** Read them only on the event loop, or use lock-protected snapshots (`snapshot_stats()`). -3. **`WG21Index.papers` is replaced wholesale on every `refresh()`** — do not mutate the dict in place. Assign a new dict from `_parse_and_index()`. +2. **Never access `ISOProber._stats`, `WG21Index._papers`, `WG21Index._max_rev`, or other source internals from a thread.** Read them only on the event loop, or use snapshot accessors (`paper_count()`, `get_papers_snapshot()`, `snapshot_stats()`). +3. **`WG21Index._papers` is replaced wholesale on every `refresh()`** — do not mutate the dict in place. Publication is an atomic reference swap in `_commit_index()`. 4. **New HTTP data sources** should follow the async pattern (`httpx` + coroutines on the loop), like `WG21Index` and `ISOProber`. The optional open-std.org scraper in `sources.py` is a future extension point: if integrated, either keep it async on the loop or isolate it in a thread with no shared mutable state. ## Related docs diff --git a/src/paperscout/__main__.py b/src/paperscout/__main__.py index ab1a360..6f526ed 100644 --- a/src/paperscout/__main__.py +++ b/src/paperscout/__main__.py @@ -250,7 +250,7 @@ async def _async_main() -> None: mq.start() def paper_count_fn() -> int: - return len(index.papers) + return index.paper_count() def _on_poll_result(result: PollResult) -> None: notify_channel(app, result, mq) diff --git a/src/paperscout/concurrency.py b/src/paperscout/concurrency.py index 8ef6466..086f990 100644 --- a/src/paperscout/concurrency.py +++ b/src/paperscout/concurrency.py @@ -14,7 +14,7 @@ async def run_blocking_io(fn: Callable[..., _T], /, *args, **kwargs) -> _T: Use only for pure blocking I/O (e.g. psycopg2 queries) that does **not** touch shared in-process mutable state such as ``ISOProber._stats``, - ``WG21Index.papers``, or other source internals. Each call should use its + ``WG21Index._papers``, or other source internals. Each call should use its own database connection from the pool. See ``docs/architecture.md`` for the full concurrency model. diff --git a/src/paperscout/monitor.py b/src/paperscout/monitor.py index a63631c..a9e0de5 100644 --- a/src/paperscout/monitor.py +++ b/src/paperscout/monitor.py @@ -371,12 +371,12 @@ async def seed(self) -> SeedResult: wg21 = self._wg21_index() paper_count = ( - len(wg21.papers) + wg21.paper_count() if wg21 is not None else len(self._snapshots.get(SOURCE_WG21_INDEX, {})) ) if self.cfg.enable_bulk_wg21 and wg21 is not None: - log.info("SEED wg21.link loaded papers=%d", len(wg21.papers)) + log.info("SEED wg21.link loaded papers=%d", wg21.paper_count()) if self.cfg.enable_iso_probe: log.info("SEED isocpp.org probe existing=%d", len(hits)) @@ -496,7 +496,7 @@ async def poll_once(self) -> PollResult: # Safe to run off the event loop: matches_for_users only performs blocking # PostgreSQL I/O via psycopg2 on its own pool connection — it does not - # touch ISOProber._stats, WG21Index.papers, or other shared source state. + # touch ISOProber._stats, WG21Index._papers, or other shared source state. per_user_matches = await run_blocking_io( self.user_watchlist.matches_for_users, diff.new_papers, diff --git a/src/paperscout/sources.py b/src/paperscout/sources.py index 398af92..bb387c6 100644 --- a/src/paperscout/sources.py +++ b/src/paperscout/sources.py @@ -52,7 +52,7 @@ class WG21Index: Likewise, do not access from the MessageQueue sender thread or any other sync worker — ``refresh()`` mutates internal state and would - race with cross-thread reads on ``papers`` / ``_max_rev``. + race with cross-thread reads on ``_papers`` / ``_max_rev``. """ source_id: str = SOURCE_WG21_INDEX @@ -61,20 +61,20 @@ def __init__(self, pool, cfg: Settings | None = None): self._cfg = cfg or settings self._cache = PaperCache(pool, ttl_hours=self._cfg.cache_ttl_hours) # Replaced wholesale on every refresh(); never mutate in place. - self.papers: dict[str, Paper] = {} + self._papers: dict[str, Paper] = {} self._max_rev: dict[int, int] = {} # P-number -> highest revision self._max_p: int = 0 # absolute highest P-number self._sorted_p_nums: list[int] = [] # sorted unique P-numbers, for gap analysis async def refresh(self) -> IndexRefreshResult: - """Load index from cache or network; populate ``self.papers``.""" + """Load index from cache or network; populate ``_papers``.""" cached = self._cache.read_if_fresh() if cached: papers, max_rev, max_p = self._parse_raw(cached) if papers: self._commit_index(papers, max_rev, max_p) log.info("Loaded %d entries from cache", len(papers)) - return IndexRefreshResult(self.papers, stale=False) + return IndexRefreshResult(dict(self._papers), stale=False) raw, category = await self._download() if raw: @@ -83,7 +83,7 @@ async def refresh(self) -> IndexRefreshResult: self._cache.write(raw) self._commit_index(papers, max_rev, max_p) log.info("Downloaded and cached %d entries", len(papers)) - return IndexRefreshResult(self.papers, stale=False) + return IndexRefreshResult(dict(self._papers), stale=False) log.warning( "INDEX-FETCH failure_category=%s url=%s payload produced zero parseable papers", FailureCategory.CONFIGURATION.value, @@ -100,7 +100,7 @@ async def refresh(self) -> IndexRefreshResult: "INDEX-STALE-FALLBACK entries=%d (download failed; using persisted cache)", len(papers), ) - return IndexRefreshResult(self.papers, stale=True) + return IndexRefreshResult(dict(self._papers), stale=True) log.error("No index data available") self._raise_on_download_failure(category or FailureCategory.CONFIGURATION) @@ -192,10 +192,17 @@ def _parse_raw(self, raw: dict) -> tuple[dict[str, Paper], dict[int, int], int]: return papers, max_rev, max_p def _commit_index(self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int) -> None: - self.papers = papers - self._max_rev = max_rev + """Atomically publish a new index snapshot. + + Assigns a new dict reference so readers see either the old or new map, + never a partially-built one. Callers must read via ``paper_count()``, + ``get_papers_snapshot()``, or ``get_known_paper_ids()`` — not the live + ``_papers`` dict, especially from other threads. + """ + self._papers = dict(papers) + self._max_rev = dict(max_rev) self._max_p = max_p - self._sorted_p_nums = sorted(max_rev.keys()) + self._sorted_p_nums = sorted(self._max_rev.keys()) def _parse_and_index(self, raw: dict) -> dict[str, Paper]: papers, max_rev, max_p = self._parse_raw(raw) @@ -238,11 +245,19 @@ def known_p_numbers(self) -> set[int]: def get_known_paper_ids(self) -> frozenset[str]: """Immutable snapshot of all paper ids currently in the index.""" - return frozenset(self.papers) + return frozenset(self._papers) def get_papers_snapshot(self) -> Mapping[str, Paper]: - """Read-only mapping copy of ``papers`` (not the live dict object).""" - return MappingProxyType(dict(self.papers)) + """Read-only mapping copy of the index (not the live dict object).""" + return MappingProxyType(dict(self._papers)) + + def paper_count(self) -> int: + """Number of papers in the current index snapshot.""" + return len(self._papers) + + def _replace_papers_for_test(self, papers: Mapping[str, Paper]) -> None: + """Test-only atomic swap of the papers map (does not recompute ``_max_rev``).""" + self._papers = dict(papers) async def fetch(self) -> IndexRefreshResult: """DataSource: load index and return papers plus stale signal.""" diff --git a/tests/test_callback_protocols.py b/tests/test_callback_protocols.py index 7a9e3bc..5ba8d0f 100644 --- a/tests/test_callback_protocols.py +++ b/tests/test_callback_protocols.py @@ -65,7 +65,7 @@ def on_ops_alert(message: str) -> None: assert isinstance(scheduler.notify_callback, NotifyCallback) assert isinstance(scheduler.ops_alert_fn, OpsAlertFn) - wg21.papers = {} + wg21._papers = {} await scheduler.poll_once() assert notified == [] diff --git a/tests/test_monitor.py b/tests/test_monitor.py index b7c57ad..98039dd 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -234,10 +234,11 @@ def test_rejects_invalid_per_user_matches_value(self): def _make_mock_wg21() -> MagicMock: mock = MagicMock(spec=WG21Index) mock.source_id = SOURCE_WG21_INDEX - mock.papers = {} + mock._papers = {} + mock.paper_count = lambda: len(mock._papers) async def _fetch(): - return IndexRefreshResult(dict(mock.papers), stale=False) + return IndexRefreshResult(dict(mock._papers), stale=False) mock.fetch = AsyncMock(side_effect=_fetch) mock.diff = lambda previous, current: diff_snapshots(previous or {}, current) @@ -320,7 +321,7 @@ async def test_poll_once_detects_new_papers(self, fake_pool): await scheduler.poll_once() new_paper = Paper(id="P9999R0", title="New", author="Author", date="2024-01-01") - index.papers = {"P9999R0": new_paper} + index._papers = {"P9999R0": new_paper} _set_iso_cycle(prober, _empty_cycle()) result = await scheduler.poll_once() assert len(result.diff.new_papers) == 1 @@ -331,7 +332,7 @@ async def test_poll_once_surfaces_only_recent_probe_hits(self, fake_pool): recent = _recent_hit() old = _old_hit() - index.papers = {} + index._papers = {} _set_iso_cycle(prober, _success_cycle([recent, old])) result = await scheduler.poll_once() assert len(result.probe_hits) == 1 @@ -350,7 +351,7 @@ async def test_poll_once_detects_dp_transition(self, fake_pool): author="Author", date="2025-01-01", ) - index.papers = {"P9999R0": new_paper} + index._papers = {"P9999R0": new_paper} _set_iso_cycle(prober, _empty_cycle()) result = await scheduler.poll_once() @@ -365,7 +366,7 @@ async def test_poll_once_dp_skip_non_p_papers(self, fake_pool): await scheduler.poll_once() n_paper = Paper(id="N4950", title="Working Draft", author="Ed", date="2025-01-01") - index.papers = {"N4950": n_paper} + index._papers = {"N4950": n_paper} _set_iso_cycle(prober, _empty_cycle()) result = await scheduler.poll_once() @@ -376,7 +377,7 @@ async def test_poll_once_no_dp_transition_when_no_draft(self, fake_pool): await scheduler.poll_once() new_paper = Paper(id="P8888R0", title="Entirely New", author="X", date="2025-01-01") - index.papers = {"P8888R0": new_paper} + index._papers = {"P8888R0": new_paper} _set_iso_cycle(prober, _empty_cycle()) result = await scheduler.poll_once() @@ -391,7 +392,7 @@ async def test_poll_once_dp_transition_logged(self, fake_pool, caplog): draft_url = "https://isocpp.org/files/papers/D7777R0.pdf" state.mark_discovered(draft_url) new_paper = Paper(id="P7777R0", title="X", author="Y", date="2025-01-01") - index.papers = {"P7777R0": new_paper} + index._papers = {"P7777R0": new_paper} _set_iso_cycle(prober, _empty_cycle()) with caplog.at_level(logging.INFO): @@ -416,7 +417,7 @@ async def test_poll_once_logs_updated_papers(self, fake_pool, caplog): old_paper = Paper(id="P9999R0", title="Old Title", author="A", date="2024-01-01") scheduler._snapshots[SOURCE_WG21_INDEX] = {"P9999R0": old_paper} updated_paper = Paper(id="P9999R0", title="New Title", author="A", date="2024-01-01") - index.papers = {"P9999R0": updated_paper} + index._papers = {"P9999R0": updated_paper} _set_iso_cycle(prober, _empty_cycle()) with caplog.at_level(logging.DEBUG): await scheduler.poll_once() @@ -428,7 +429,7 @@ async def test_poll_old_hits_logged(self, fake_pool, caplog): scheduler, index, prober, _, _ = _make_scheduler(fake_pool) await scheduler.poll_once() old = _old_hit() - index.papers = {} + index._papers = {} _set_iso_cycle(prober, _success_cycle([old])) with caplog.at_level(logging.INFO): result = await scheduler.poll_once() @@ -440,7 +441,7 @@ async def test_poll_once_populates_per_user_matches(self, fake_pool): await scheduler.poll_once() new_paper = Paper(id="P9999R0", title="Senders", author="Eric Niebler", date="2024-01-01") - index.papers = {"P9999R0": new_paper} + index._papers = {"P9999R0": new_paper} _set_iso_cycle(prober, _empty_cycle()) user_watchlist.matches_for_users.return_value = { @@ -456,7 +457,7 @@ async def test_poll_once_per_user_probe_hit(self, fake_pool): hit = _recent_hit(front_text="written by eric niebler") _set_iso_cycle(prober, _success_cycle([hit])) - index.papers = {} + index._papers = {} user_watchlist.matches_for_users.return_value = { "U123": PerUserMatches(papers=[], probe_hits=[(hit, MatchReason.AUTHOR)]) @@ -543,7 +544,7 @@ async def test_seed_does_not_log_index_new_on_baseline(self, fake_pool, caplog): scheduler, wg21, _, _, _ = _make_scheduler(fake_pool) paper = Paper(id="P9999R0", title="Baseline", author="Author", date="2024-01-01") - wg21.papers = {"P9999R0": paper} + wg21._papers = {"P9999R0": paper} with caplog.at_level(logging.INFO): await scheduler.seed() @@ -702,7 +703,7 @@ async def test_failed_probe_cycle_does_not_advance_last_successful_poll_normal_p scheduler, index, prober, _, _ = _make_scheduler(fake_pool) await scheduler.poll_once() before = scheduler._last_successful_poll - index.papers = {} + index._papers = {} _set_iso_cycle(prober, _failed_cycle("network down")) await scheduler.poll_once() assert scheduler._last_successful_poll == before @@ -832,7 +833,7 @@ async def test_poll_once_propagates_index_refresh_error(self, fake_pool): with pytest.raises(IndexRefreshError) as exc_info: await scheduler.poll_once() assert exc_info.value.category is FailureCategory.TIMEOUT - assert wg21.papers == {} + assert wg21.paper_count() == 0 async def test_poll_once_uses_stale_index(self, fake_pool, caplog): scheduler, wg21, prober, _, _ = _make_scheduler_with_real_wg21(fake_pool) @@ -846,7 +847,7 @@ async def test_poll_once_uses_stale_index(self, fake_pool, caplog): mock_cls.return_value.__aexit__ = AsyncMock(return_value=False) with caplog.at_level(logging.WARNING, logger="paperscout.monitor"): await scheduler.poll_once() - assert "P2300R10" in wg21.papers + assert "P2300R10" in wg21.get_known_paper_ids() assert "INDEX-STALE" in caplog.text assert scheduler._last_successful_poll is None prober.fetch.assert_called_once() diff --git a/tests/test_sources.py b/tests/test_sources.py index 4522ef1..ca9c454 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -308,7 +308,7 @@ async def test_refresh_rejected_parse_preserves_index_metadata(self, fake_pool): index = WG21Index(fake_pool) index._parse_and_index(SAMPLE_INDEX_DATA) original_max_p = index._max_p - original_papers = dict(index.papers) + original_papers = dict(index.get_papers_snapshot()) index._cache.write({"bad": "not-a-dict"}) with patch.object( index, "_download", AsyncMock(return_value=(None, FailureCategory.NETWORK)) @@ -316,7 +316,7 @@ async def test_refresh_rejected_parse_preserves_index_metadata(self, fake_pool): with pytest.raises(IndexRefreshError): await index.refresh() assert index._max_p == original_max_p - assert index.papers == original_papers + assert dict(index.get_papers_snapshot()) == original_papers async def test_download_json_parse_error_emits_configuration(self, fake_pool, caplog): index = WG21Index(fake_pool) @@ -419,11 +419,10 @@ def test_get_papers_snapshot_and_known_ids_are_independent_snapshots(self, fake_ from types import MappingProxyType index = WG21Index(fake_pool) - index.papers = index._parse_and_index({"P1000R0": {"title": "T"}}) + index._parse_and_index({"P1000R0": {"title": "T"}}) snap = index.get_papers_snapshot() - assert snap is not index.papers assert isinstance(snap, MappingProxyType) - assert dict(snap) == dict(index.papers) + assert dict(snap) == dict(index.get_papers_snapshot()) with pytest.raises(TypeError): snap["X"] = Paper(id="X") @@ -432,7 +431,7 @@ def test_get_papers_snapshot_and_known_ids_are_independent_snapshots(self, fake_ with pytest.raises(AttributeError): frozen.add("X") - index.papers = index._parse_and_index({}) + index._parse_and_index({}) assert "P1000R0" in snap assert snap["P1000R0"].id == "P1000R0" @@ -671,9 +670,8 @@ def test_hot_recent_paper_by_date(self, fake_pool): frontier_window_below=0, ) recent_date = (date.today() - timedelta(days=30)).isoformat() - # _parse_and_index updates _max_rev/_sorted_p_nums but not self.papers; - # assign both so that the date-based hot filter can find the paper. - index.papers = index._parse_and_index( + # _parse_and_index updates _max_rev/_sorted_p_nums and commits _papers. + index._parse_and_index( { "P5000R2": {"title": "T", "date": recent_date, "type": "paper"}, } @@ -801,7 +799,7 @@ def test_hot_paper_skipped_when_no_date(self, fake_pool): fake_pool, hot_lookback_months=6, frontier_window_above=0, frontier_window_below=0 ) # Paper with no date should be silently skipped (the `continue` branch) - index.papers = index._parse_and_index({"P6000R0": {"title": "T", "type": "paper"}}) + index._parse_and_index({"P6000R0": {"title": "T", "type": "paper"}}) frontier = index.effective_frontier() hot_known, _ = prober._hot_numbers(frontier) assert 6000 not in hot_known @@ -810,9 +808,7 @@ def test_hot_paper_skipped_when_bad_date(self, fake_pool): prober, index, _ = self._make_prober( fake_pool, hot_lookback_months=6, frontier_window_above=0, frontier_window_below=0 ) - index.papers = index._parse_and_index( - {"P6001R0": {"title": "T", "date": "not-a-date", "type": "paper"}} - ) + index._parse_and_index({"P6001R0": {"title": "T", "date": "not-a-date", "type": "paper"}}) frontier = index.effective_frontier() hot_known, _ = prober._hot_numbers(frontier) assert 6001 not in hot_known @@ -973,7 +969,7 @@ async def test_skips_already_discovered(self, fake_pool): async def test_skips_already_in_index(self, fake_pool): prober, index, _ = self._make_prober(fake_pool) - index.papers = {"D2300R11": Paper(id="D2300R11")} + index._replace_papers_for_test({"D2300R11": Paper(id="D2300R11")}) url = "https://isocpp.org/files/papers/D2300R11.pdf" sem = asyncio.Semaphore(5) result = await prober._probe_one(AsyncMock(), sem, url, "D", 2300, 11, ".pdf", "hot") @@ -1072,7 +1068,7 @@ async def test_stats_skipped_discovered(self, fake_pool): async def test_stats_skipped_in_index(self, fake_pool): prober, index, _ = self._make_prober(fake_pool) - index.papers = {"D9998R0": Paper(id="D9998R0")} + index._replace_papers_for_test({"D9998R0": Paper(id="D9998R0")}) url = "https://isocpp.org/files/papers/D9998R0.pdf" sem = asyncio.Semaphore(5) await prober._probe_one(AsyncMock(), sem, url, "D", 9998, 0, ".pdf", "hot")