test_thread_safety.py

"""Thread-safety contention tests for lock-protected paths.

Intentional OS-thread stress of lock-protected paths; production uses
event-loop cooperative scheduling + snapshot reads.

NOTE: FakePool is not thread-safe. These tests do not call pool methods
concurrently; they only stress ISOProber._stats_lock and Scheduler._health_lock.
"""

from __future__ import annotations

import threading
import time
from unittest.mock import MagicMock

from paperscout.models import CycleStatus
from paperscout.monitor import Scheduler, _compute_probe_success_rate
from paperscout.protocols import SOURCE_ISO_PROBE, SOURCE_WG21_INDEX
from paperscout.sources import ISOProber, WG21Index
from paperscout.storage import ProbeState, UserWatchlist
from tests.conftest import make_test_settings

_STATS_KEYS = frozenset(ISOProber._STATS_TEMPLATE.keys())
_HEALTH_SNAPSHOT_KEYS = frozenset(
    {
        "last_updated",
        "poll_count",
        "last_successful_poll",
        "last_cycle_status",
        "last_cycle_error",
        "probe_stats",
        "probe_success_rate",
    }
)

THREAD_JOIN_TIMEOUT = 5.0
RESET_ITERATIONS = 1000
SNAPSHOT_ITERATIONS = 1000
BUMPER_ITERATIONS = 1000
PUBLISH_ITERATIONS = 500
READ_ITERATIONS = 500


def _make_prober(fake_pool) -> ISOProber:
    index = WG21Index(fake_pool)
    state = ProbeState(fake_pool)
    wl = MagicMock(spec=UserWatchlist)
    wl.get_all_watched_paper_nums.return_value = set()
    return ISOProber(index, state, user_watchlist=wl, cfg=make_test_settings())


def _make_scheduler(fake_pool) -> Scheduler:
    wg21 = MagicMock()
    wg21.source_id = SOURCE_WG21_INDEX
    iso = MagicMock()
    iso.source_id = SOURCE_ISO_PROBE
    user_watchlist = MagicMock(spec=UserWatchlist)
    user_watchlist.matches_for_users.return_value = {}
    state = ProbeState(fake_pool)
    return Scheduler(
        sources=[wg21, iso],
        user_watchlist=user_watchlist,
        state=state,
        cfg=make_test_settings(),
    )


def _join_threads(threads: list[threading.Thread]) -> None:
    for thread in threads:
        thread.join(timeout=THREAD_JOIN_TIMEOUT)
        assert not thread.is_alive(), (
            f"thread {thread.name} did not finish within {THREAD_JOIN_TIMEOUT}s"
        )


def _assert_valid_probe_stats(stats: dict[str, int]) -> None:
    assert set(stats.keys()) == _STATS_KEYS
    assert all(isinstance(v, int) and v >= 0 for v in stats.values())


def _assert_health_snapshot_consistent(snap: dict) -> None:
    assert set(snap.keys()) == _HEALTH_SNAPSHOT_KEYS
    assert isinstance(snap["poll_count"], int) and snap["poll_count"] >= 0
    assert isinstance(snap["probe_stats"], dict)
    assert all(isinstance(v, int) for v in snap["probe_stats"].values())
    assert snap["probe_success_rate"] == _compute_probe_success_rate(snap["probe_stats"])
    assert snap["probe_success_rate"] is None or isinstance(snap["probe_success_rate"], float)
    assert snap["last_cycle_status"] is None or isinstance(snap["last_cycle_status"], str)
    assert snap["last_cycle_error"] is None or isinstance(snap["last_cycle_error"], str)
    assert snap["last_successful_poll"] is None or isinstance(snap["last_successful_poll"], str)
    last_updated = snap["last_updated"]
    assert last_updated is None or (isinstance(last_updated, str) and len(last_updated) > 0)


class TestISOProberStatsContention:
    """Intentional OS-thread stress of ISOProber._stats_lock."""

    def test_concurrent_bump_stat_totals(self, fake_pool):
        """Concurrent _bump_stat() calls produce correct totals."""
        prober = _make_prober(fake_pool)
        n_threads = 32
        bumps_per_thread = 100
        barrier = threading.Barrier(n_threads)

        def worker() -> None:
            barrier.wait()
            for _ in range(bumps_per_thread):
                prober._bump_stat("miss")

        threads = [threading.Thread(target=worker, name=f"bump-{i}") for i in range(n_threads)]
        for thread in threads:
            thread.start()
        _join_threads(threads)
        assert prober.snapshot_stats()["miss"] == n_threads * bumps_per_thread

    def test_snapshot_stats_consistent_under_concurrent_reset(self, fake_pool):
        """snapshot_stats() concurrent with _reset_stats() returns consistent state."""
        prober = _make_prober(fake_pool)
        errors: list[Exception] = []
        errors_lock = threading.Lock()
        n_threads = 5
        barrier = threading.Barrier(n_threads)

        def record_error(exc: Exception) -> None:
            with errors_lock:
                errors.append(exc)

        def resetter() -> None:
            try:
                barrier.wait()
                for _ in range(RESET_ITERATIONS):
                    prober._reset_stats()
            except Exception as exc:
                record_error(exc)

        def snapshotter() -> None:
            try:
                barrier.wait()
                for _ in range(SNAPSHOT_ITERATIONS):
                    _assert_valid_probe_stats(prober.snapshot_stats())
            except Exception as exc:
                record_error(exc)

        def bumper() -> None:
            try:
                barrier.wait()
                for _ in range(BUMPER_ITERATIONS):
                    prober._bump_stat("miss")
            except Exception as exc:
                record_error(exc)

        threads = [
            threading.Thread(target=resetter, name="resetter"),
            threading.Thread(target=snapshotter, name="snapshot-0"),
            threading.Thread(target=snapshotter, name="snapshot-1"),
            threading.Thread(target=bumper, name="bumper-0"),
            threading.Thread(target=bumper, name="bumper-1"),
        ]
        for thread in threads:
            thread.start()
        _join_threads(threads)
        assert not errors, f"thread errors: {errors!r}"


class TestSchedulerHealthContention:
    """Intentional OS-thread stress of Scheduler._health_lock."""

    def test_health_snapshot_consistent_under_concurrent_publish(self, fake_pool):
        """health_snapshot() from non-event-loop threads returns consistent data."""
        scheduler = _make_scheduler(fake_pool)
        scheduler._poll_count = 0
        scheduler._last_probe_stats = {}
        scheduler._last_cycle_status = None
        scheduler._last_successful_poll = None
        scheduler._publish_health_snapshot()

        errors: list[Exception] = []
        errors_lock = threading.Lock()

        def record_error(exc: Exception) -> None:
            with errors_lock:
                errors.append(exc)

        def writer() -> None:
            try:
                # Writer thread is the sole mutator; attribute writes happen-before
                # _publish_health_snapshot() reads on the same thread. _health_lock
                # only guards _health_snapshot assignment vs reader threads.
                for i in range(PUBLISH_ITERATIONS):
                    scheduler._poll_count = i + 1
                    scheduler._last_probe_stats = {
                        "miss": i % 10,
                        "error": i % 3,
                        "hit_recent": i % 5,
                        "hit_old": 0,
                        "hit_no_lm": 0,
                        "skipped_discovered": 0,
                        "skipped_in_index": 0,
                    }
                    scheduler._last_cycle_status = (
                        CycleStatus.SUCCESS if i % 2 == 0 else CycleStatus.EMPTY
                    )
                    scheduler._last_successful_poll = time.time()
                    scheduler._publish_health_snapshot()
            except Exception as exc:
                record_error(exc)

        def reader() -> None:
            try:
                for _ in range(READ_ITERATIONS):
                    _assert_health_snapshot_consistent(scheduler.health_snapshot())
            except Exception as exc:
                record_error(exc)

        writer_thread = threading.Thread(target=writer, name="writer")
        reader_threads = [threading.Thread(target=reader, name=f"reader-{i}") for i in range(6)]
        writer_thread.start()
        for thread in reader_threads:
            thread.start()
        _join_threads([writer_thread, *reader_threads])
        assert not errors, f"thread errors: {errors!r}"