From 672def5e8f1c1434f6f2078b574c38152512d1da Mon Sep 17 00:00:00 2001 From: mac Date: Thu, 9 Jul 2026 23:42:36 +0800 Subject: [PATCH 1/6] Encapsulate WG21Index.papers behind snapshot and count accessors Privatize the index map to _papers, add paper_count() and a test seam, migrate cross-thread readers and tests, and update concurrency docs. --- CONTRIBUTING.md | 2 +- docs/architecture.md | 6 +++--- src/paperscout/__main__.py | 2 +- src/paperscout/concurrency.py | 2 +- src/paperscout/monitor.py | 6 +++--- src/paperscout/sources.py | 35 +++++++++++++++++++++++--------- tests/test_callback_protocols.py | 2 +- tests/test_monitor.py | 33 +++++++++++++++--------------- tests/test_sources.py | 24 ++++++++++------------ 9 files changed, 63 insertions(+), 49 deletions(-) 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..2063a0a 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(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(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(self._papers, stale=True) log.error("No index data available") self._raise_on_download_failure(category or FailureCategory.CONFIGURATION) @@ -192,7 +192,14 @@ 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 + """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 = papers self._max_rev = max_rev self._max_p = max_p self._sorted_p_nums = sorted(max_rev.keys()) @@ -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..f114847 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,7 +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( + index._parse_and_index( {"P6001R0": {"title": "T", "date": "not-a-date", "type": "paper"}} ) frontier = index.effective_frontier() @@ -973,7 +971,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 +1070,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") From d26f8553102f818724aef066ff8aa5c305f1e46d Mon Sep 17 00:00:00 2001 From: mac Date: Thu, 9 Jul 2026 23:45:09 +0800 Subject: [PATCH 2/6] fixed lint error --- tests/test_sources.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_sources.py b/tests/test_sources.py index f114847..ca9c454 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -808,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._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 From 43b1ec96bffc1523875c98a18555815cc4ad0753 Mon Sep 17 00:00:00 2001 From: mac Date: Fri, 10 Jul 2026 00:06:43 +0800 Subject: [PATCH 3/6] addressed ai reviews --- src/paperscout/sources.py | 45 ++++++++++++++++++++++++++++----------- 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/src/paperscout/sources.py b/src/paperscout/sources.py index 2063a0a..5b93b0a 100644 --- a/src/paperscout/sources.py +++ b/src/paperscout/sources.py @@ -74,7 +74,7 @@ async def refresh(self) -> IndexRefreshResult: 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=False) log.error("No index data available") self._raise_on_download_failure(category or FailureCategory.CONFIGURATION) @@ -191,7 +191,9 @@ def _parse_raw(self, raw: dict) -> tuple[dict[str, Paper], dict[int, int], int]: max_rev[paper.number] = paper.revision return papers, max_rev, max_p - def _commit_index(self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int) -> None: + def _commit_index( + self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int + ) -> None: """Atomically publish a new index snapshot. Assigns a new dict reference so readers see either the old or new map, @@ -429,7 +431,9 @@ async def run_cycle(self) -> CycleResult: urls = self._build_probe_list() known_ids = self.index.get_known_paper_ids() hot_count = sum( - 1 for u in urls if u.tier in (Tier.WATCHLIST, Tier.FRONTIER, Tier.RECENT) + 1 + for u in urls + if u.tier in (Tier.WATCHLIST, Tier.FRONTIER, Tier.RECENT) ) cold_count = sum(1 for u in urls if u.tier == Tier.COLD) slice_idx = (self._cycle - 1) % self.cfg.cold_cycle_divisor @@ -553,9 +557,9 @@ def _build_probe_list(self) -> list[ProbeTarget]: extra_p_numbers=self.state.paper_nums_from_discovered_iso_urls(), ) hot_known, hot_unknown = self._hot_numbers(frontier) - return self._build_hot_list(frontier, hot_known, hot_unknown) + self._build_cold_slice( - self._cycle, frontier, hot_known, hot_unknown - ) + return self._build_hot_list( + frontier, hot_known, hot_unknown + ) + self._build_cold_slice(self._cycle, frontier, hot_known, hot_unknown) def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: """Return (known_hot, unknown_hot) P-number sets to probe every cycle.""" @@ -573,9 +577,16 @@ def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: # Recently active papers if self.cfg.hot_lookback_months > 0: - cutoff = date.today() - timedelta(days=int(self.cfg.hot_lookback_months * 30.44)) + cutoff = date.today() - timedelta( + days=int(self.cfg.hot_lookback_months * 30.44) + ) for p in self.index.get_papers_snapshot().values(): - if p.prefix != "P" or p.number is None or not p.date or p.date == "unknown": + if ( + p.prefix != "P" + or p.number is None + or not p.date + or p.date == "unknown" + ): continue try: if date.fromisoformat(p.date[:10]) >= cutoff: @@ -586,7 +597,9 @@ def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: known_p_nums = self.index.known_p_numbers() return hot & known_p_nums, hot - known_p_nums - def _tier_label(self, num: int, watchlist_set: set[int], frontier_range: set[int]) -> Tier: + def _tier_label( + self, num: int, watchlist_set: set[int], frontier_range: set[int] + ) -> Tier: if num in watchlist_set: return Tier.WATCHLIST if num in frontier_range: @@ -770,7 +783,9 @@ async def _probe_one( if last_modified.tzinfo is None: last_modified = last_modified.replace(tzinfo=timezone.utc) threshold = timedelta(hours=self.cfg.alert_modified_hours) - is_recent = (datetime.now(timezone.utc) - last_modified) <= threshold + is_recent = ( + datetime.now(timezone.utc) - last_modified + ) <= threshold except (TypeError, ValueError): # Bad Last-Modified: merge with no-LM so we don't silently drop hits. last_modified = None @@ -780,7 +795,11 @@ async def _probe_one( # file; treat as recent so we don't silently drop it. is_recent = True - lm_display = last_modified.strftime("%Y-%m-%d %H:%M UTC") if last_modified else "no-lm" + lm_display = ( + last_modified.strftime("%Y-%m-%d %H:%M UTC") + if last_modified + else "no-lm" + ) log.info( "HIT tier=%-10s recent=%-5s lm=%-20s %s", tier, From 3792d8e3a6d2ef2517ac8714c1269dc6be2157b5 Mon Sep 17 00:00:00 2001 From: mac Date: Fri, 10 Jul 2026 00:19:02 +0800 Subject: [PATCH 4/6] fixed lint error --- src/paperscout/sources.py | 39 ++++++++++----------------------------- 1 file changed, 10 insertions(+), 29 deletions(-) diff --git a/src/paperscout/sources.py b/src/paperscout/sources.py index 5b93b0a..3d84c3b 100644 --- a/src/paperscout/sources.py +++ b/src/paperscout/sources.py @@ -191,9 +191,7 @@ def _parse_raw(self, raw: dict) -> tuple[dict[str, Paper], dict[int, int], int]: max_rev[paper.number] = paper.revision return papers, max_rev, max_p - def _commit_index( - self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int - ) -> None: + def _commit_index(self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int) -> None: """Atomically publish a new index snapshot. Assigns a new dict reference so readers see either the old or new map, @@ -431,9 +429,7 @@ async def run_cycle(self) -> CycleResult: urls = self._build_probe_list() known_ids = self.index.get_known_paper_ids() hot_count = sum( - 1 - for u in urls - if u.tier in (Tier.WATCHLIST, Tier.FRONTIER, Tier.RECENT) + 1 for u in urls if u.tier in (Tier.WATCHLIST, Tier.FRONTIER, Tier.RECENT) ) cold_count = sum(1 for u in urls if u.tier == Tier.COLD) slice_idx = (self._cycle - 1) % self.cfg.cold_cycle_divisor @@ -557,9 +553,9 @@ def _build_probe_list(self) -> list[ProbeTarget]: extra_p_numbers=self.state.paper_nums_from_discovered_iso_urls(), ) hot_known, hot_unknown = self._hot_numbers(frontier) - return self._build_hot_list( - frontier, hot_known, hot_unknown - ) + self._build_cold_slice(self._cycle, frontier, hot_known, hot_unknown) + return self._build_hot_list(frontier, hot_known, hot_unknown) + self._build_cold_slice( + self._cycle, frontier, hot_known, hot_unknown + ) def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: """Return (known_hot, unknown_hot) P-number sets to probe every cycle.""" @@ -577,16 +573,9 @@ def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: # Recently active papers if self.cfg.hot_lookback_months > 0: - cutoff = date.today() - timedelta( - days=int(self.cfg.hot_lookback_months * 30.44) - ) + cutoff = date.today() - timedelta(days=int(self.cfg.hot_lookback_months * 30.44)) for p in self.index.get_papers_snapshot().values(): - if ( - p.prefix != "P" - or p.number is None - or not p.date - or p.date == "unknown" - ): + if p.prefix != "P" or p.number is None or not p.date or p.date == "unknown": continue try: if date.fromisoformat(p.date[:10]) >= cutoff: @@ -597,9 +586,7 @@ def _hot_numbers(self, frontier: int) -> tuple[set[int], set[int]]: known_p_nums = self.index.known_p_numbers() return hot & known_p_nums, hot - known_p_nums - def _tier_label( - self, num: int, watchlist_set: set[int], frontier_range: set[int] - ) -> Tier: + def _tier_label(self, num: int, watchlist_set: set[int], frontier_range: set[int]) -> Tier: if num in watchlist_set: return Tier.WATCHLIST if num in frontier_range: @@ -783,9 +770,7 @@ async def _probe_one( if last_modified.tzinfo is None: last_modified = last_modified.replace(tzinfo=timezone.utc) threshold = timedelta(hours=self.cfg.alert_modified_hours) - is_recent = ( - datetime.now(timezone.utc) - last_modified - ) <= threshold + is_recent = (datetime.now(timezone.utc) - last_modified) <= threshold except (TypeError, ValueError): # Bad Last-Modified: merge with no-LM so we don't silently drop hits. last_modified = None @@ -795,11 +780,7 @@ async def _probe_one( # file; treat as recent so we don't silently drop it. is_recent = True - lm_display = ( - last_modified.strftime("%Y-%m-%d %H:%M UTC") - if last_modified - else "no-lm" - ) + lm_display = last_modified.strftime("%Y-%m-%d %H:%M UTC") if last_modified else "no-lm" log.info( "HIT tier=%-10s recent=%-5s lm=%-20s %s", tier, From 4b73b0b807e464bb4c917e6069b7e279f8565ae9 Mon Sep 17 00:00:00 2001 From: mac Date: Fri, 10 Jul 2026 00:42:11 +0800 Subject: [PATCH 5/6] fixed stale parameter as true --- src/paperscout/sources.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/paperscout/sources.py b/src/paperscout/sources.py index 3d84c3b..e1fe798 100644 --- a/src/paperscout/sources.py +++ b/src/paperscout/sources.py @@ -100,7 +100,7 @@ async def refresh(self) -> IndexRefreshResult: "INDEX-STALE-FALLBACK entries=%d (download failed; using persisted cache)", len(papers), ) - return IndexRefreshResult(dict(self._papers), stale=False) + return IndexRefreshResult(dict(self._papers), stale=True) log.error("No index data available") self._raise_on_download_failure(category or FailureCategory.CONFIGURATION) From 6c94b153cdbe771e44ad809e61631421b2140c6b Mon Sep 17 00:00:00 2001 From: mac Date: Fri, 10 Jul 2026 01:53:04 +0800 Subject: [PATCH 6/6] Keep the atomic reference swap, but publish owned copies. --- src/paperscout/sources.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/paperscout/sources.py b/src/paperscout/sources.py index e1fe798..bb387c6 100644 --- a/src/paperscout/sources.py +++ b/src/paperscout/sources.py @@ -199,10 +199,10 @@ def _commit_index(self, papers: dict[str, Paper], max_rev: dict[int, int], max_p ``get_papers_snapshot()``, or ``get_known_paper_ids()`` — not the live ``_papers`` dict, especially from other threads. """ - self._papers = papers - self._max_rev = max_rev + 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)