diff --git a/deploy/docker/ARCHITECTURE.md b/deploy/docker/ARCHITECTURE.md index eb49cdaed..8548fb6fc 100644 --- a/deploy/docker/ARCHITECTURE.md +++ b/deploy/docker/ARCHITECTURE.md @@ -272,21 +272,31 @@ async def janitor(): await asyncio.sleep(interval) async with LOCK: - # Clean cold pool first (less valuable) - for sig in list(COLD_POOL.keys()): - if now - LAST_USED[sig] > cold_ttl: - await COLD_POOL[sig].close() - del COLD_POOL[sig], LAST_USED[sig], USAGE_COUNT[sig] - await track_janitor_event("close_cold", sig, {...}) - - # Clean hot pool (more conservative) - for sig in list(HOT_POOL.keys()): - if now - LAST_USED[sig] > hot_ttl: - await HOT_POOL[sig].close() - del HOT_POOL[sig], LAST_USED[sig], USAGE_COUNT[sig] - await track_janitor_event("close_hot", sig, {...}) + # Both pools, same rule (cold first; hot gets the longer TTL): + for sig in list(pool.keys()): + idle = now - LAST_USED[sig] + if idle <= ttl: + continue # used recently + if pool[sig].active_requests > 0: + if idle <= STALE_CEILING: + continue # genuinely serving a request + log.error("Leaked request counter ... force-closing") + # busy past the ceiling = a hung/leaked request pinned it + _close_in_background(pool[sig]) # close() can hang; NEVER awaited under LOCK + del pool[sig], LAST_USED[sig], USAGE_COUNT[sig] + await track_janitor_event("close_cold"/"close_hot", sig, {...}) ``` +Two safeguards keep the pool un-pinnable (issue #2202): + +- **Stale-lease backstop**: `active_requests` is only trusted while the browser was + touched within `STALE_CEILING` (`pool.stale_lease_s`; `0` = auto + `max(2 × limits.wall_clock_s, 21600)`). A counter stuck past that means a request + hung or leaked, so the browser is force-closed instead of pinned forever. +- **Background close**: `close()` on a wedged browser can hang, so the janitor never + awaits it while holding LOCK. Closes run as fire-and-forget tasks with a 60s cap; + `close_all()` drains them (65s cap) so shutdown is clean. + **Config Signature Generation:** ```python @@ -812,6 +822,7 @@ crawler: pool: idle_ttl_sec: 300 # Base TTL for cold pool (5 min) + stale_lease_s: 0 # Leak-backstop ceiling; 0 = auto: max(2 x limits.wall_clock_s, 21600) rate_limiter: enabled: true diff --git a/deploy/docker/MIGRATION.md b/deploy/docker/MIGRATION.md index bcd2097c7..a11fa9623 100644 --- a/deploy/docker/MIGRATION.md +++ b/deploy/docker/MIGRATION.md @@ -161,7 +161,7 @@ longer published. For an **external** redis, set `REDIS_PASSWORD`. ```yaml limits: max_body_bytes: 10485760 # request body cap (413); 0 = unbounded - wall_clock_s: 0 # per-crawl deadline (504); 0 = none + wall_clock_s: 1800 # per-crawl deadline (504); 0 = none queue: maxsize: 1000 # background job queue (503 when full); 0 = unbounded workers: 4 @@ -170,6 +170,25 @@ limits: To keep the previous behavior exactly, set the caps you don't want to `0`. +`wall_clock_s` defaults to **1800** (it was `0` before). A single request that runs +longer than that now returns 504 - this affects large multi-URL batches and deep +crawls. Raise it, or set `0` to disable the deadline. + +### The pool janitor force-closes leaked browsers + +A pooled browser that reports "busy" but that no request has touched for longer +than a ceiling - `crawler.pool.stale_lease_s`, default `0` = automatic +`max(2 × wall_clock_s, 21600)` seconds (6 h floor) - is treated as pinned by a +leaked/hung request. The janitor force-closes it and logs +`🚨 Leaked request counter ... force-closing` at ERROR level. That line means a +request hung or leaked; it is cleanup working as intended, not a crash. + +The server cannot tell a hung browser from one serving a very long crawl - +streaming crawls in particular have no wall-clock deadline. If your crawls +(streaming or with `wall_clock_s: 0`) can legitimately run longer than 6 h, set +`stale_lease_s` higher than your longest expected crawl, or the janitor may +close a browser mid-crawl once it passes the ceiling. + ### Error responses are generic 5xx responses return `{"error": "Internal server error", "correlation_id": "…"}`. diff --git a/deploy/docker/config.yml b/deploy/docker/config.yml index 614ac24d1..74b2e6ed0 100644 --- a/deploy/docker/config.yml +++ b/deploy/docker/config.yml @@ -39,7 +39,7 @@ limits: max_body_bytes: 10485760 # 10 MiB request body cap (413 if exceeded); 0 = unbounded max_pages: 100 # deep-crawl page budget clamp (defense in depth) max_depth: 5 # deep-crawl depth clamp - wall_clock_s: 0 # per-crawl deadline in seconds (504 on timeout); 0 = no deadline + wall_clock_s: 1800 # per-crawl deadline in seconds (504 on timeout); 0 = no deadline queue: # background job queue for /crawl/job and /llm/job maxsize: 1000 # max queued jobs (503 when full); 0 = unbounded workers: 4 # concurrent background workers @@ -84,6 +84,9 @@ crawler: pool: max_pages: 40 # ← GLOBAL_SEM permits idle_ttl_sec: 300 # ← 30 min janitor cutoff + stale_lease_s: 0 # leak backstop ceiling; 0 = auto: max(2 x limits.wall_clock_s, 21600). + # If you set wall_clock_s: 0 for very long crawls, raise this above + # your longest crawl or the janitor may force-close it mid-flight. browser: kwargs: headless: true diff --git a/deploy/docker/crawler_pool.py b/deploy/docker/crawler_pool.py index 516d9562a..9e25fc681 100644 --- a/deploy/docker/crawler_pool.py +++ b/deploy/docker/crawler_pool.py @@ -22,6 +22,12 @@ BASE_IDLE_TTL = CONFIG.get("crawler", {}).get("pool", {}).get("idle_ttl_sec", 300) DEFAULT_CONFIG_SIG = None # Cached sig for default config +# Leak-backstop ceiling: a "busy" browser untouched past it is force-closed. +# 6h floor: streams have no deadline and nothing refreshes LAST_USED mid-crawl. +def _pos(v): return v if isinstance(v, (int, float)) and not isinstance(v, bool) and v > 0 else 0 +_WALL_CLOCK = _pos(CONFIG.get("limits", {}).get("wall_clock_s", 0)) +STALE_CEILING = _pos(CONFIG.get("crawler", {}).get("pool", {}).get("stale_lease_s", 0)) or max(2 * _WALL_CLOCK, 21600) + def get_pool_snapshot() -> dict: """Return a point-in-time snapshot of pool state for monitoring. @@ -125,9 +131,9 @@ async def release_crawler(crawler: AsyncWebCrawler): obtained via get_crawler() so the janitor knows when it's safe to close idle browsers. """ - async with LOCK: - if hasattr(crawler, 'active_requests'): - crawler.active_requests = max(0, crawler.active_requests - 1) + # No lock and no await: a cancelled request must never skip this decrement. + if hasattr(crawler, 'active_requests'): + crawler.active_requests = max(0, crawler.active_requests - 1) async def init_permanent(cfg: BrowserConfig): """Initialize permanent default browser.""" @@ -145,16 +151,42 @@ async def init_permanent(cfg: BrowserConfig): async def close_all(): """Close all browsers.""" async with LOCK: - tasks = [] + # Through _close_in_background so one wedged browser can't hang shutdown + # while holding LOCK; the drain below is the single bounded wait point. if PERMANENT: - tasks.append(PERMANENT.close()) - tasks.extend([c.close() for c in HOT_POOL.values()]) - tasks.extend([c.close() for c in COLD_POOL.values()]) - await asyncio.gather(*tasks, return_exceptions=True) + _close_in_background(PERMANENT) + for c in list(HOT_POOL.values()) + list(COLD_POOL.values()): + _close_in_background(c) HOT_POOL.clear() COLD_POOL.clear() LAST_USED.clear() USAGE_COUNT.clear() + # Drain all closes (janitor's included) so shutdown doesn't destroy live tasks. + if _CLOSE_TASKS: + with suppress(Exception): + await asyncio.wait_for( + asyncio.gather(*list(_CLOSE_TASKS), return_exceptions=True), timeout=65 + ) + +_CLOSE_TASKS = set() + +def _close_in_background(crawler: AsyncWebCrawler): + """Close a browser without holding the pool LOCK. + + close() on a wedged browser can hang; awaiting it under LOCK would freeze + get_crawler() and every future janitor pass for the whole server. + """ + async def _close(): + try: + # 60s gives Playwright's own graceful-then-SIGKILL cycle room to finish + await asyncio.wait_for(crawler.close(), timeout=60) + except asyncio.TimeoutError: + logger.warning("⚠️ Browser close timed out after 60s - abandoning (its processes may linger)") + except Exception: + pass + task = asyncio.create_task(_close()) + _CLOSE_TASKS.add(task) + task.add_done_callback(_CLOSE_TASKS.discard) async def janitor(): """Adaptive cleanup based on memory pressure.""" @@ -177,12 +209,14 @@ async def janitor(): for sig in list(COLD_POOL.keys()): if now - LAST_USED.get(sig, now) > cold_ttl: crawler = COLD_POOL[sig] - if getattr(crawler, 'active_requests', 0) > 0: - continue # still serving requests, skip idle_time = now - LAST_USED[sig] - logger.info(f"🧹 Closing cold browser (sig={sig[:8]}, idle={idle_time:.0f}s)") - with suppress(Exception): - await crawler.close() + if getattr(crawler, 'active_requests', 0) > 0: + if idle_time <= STALE_CEILING: + continue # still serving requests, skip + logger.error(f"🚨 Leaked request counter (sig={sig[:8]}, active={crawler.active_requests}, idle={idle_time:.0f}s > {STALE_CEILING}s) - force-closing") + else: + logger.info(f"🧹 Closing cold browser (sig={sig[:8]}, idle={idle_time:.0f}s)") + _close_in_background(crawler) # close() can hang; never await it under LOCK COLD_POOL.pop(sig, None) LAST_USED.pop(sig, None) USAGE_COUNT.pop(sig, None) @@ -191,19 +225,21 @@ async def janitor(): try: from monitor import get_monitor await get_monitor().track_janitor_event("close_cold", sig, {"idle_seconds": int(idle_time), "ttl": cold_ttl}) - except: + except Exception: # bare except would swallow CancelledError and outlive shutdown pass # Clean hot pool (more conservative) for sig in list(HOT_POOL.keys()): if now - LAST_USED.get(sig, now) > hot_ttl: crawler = HOT_POOL[sig] - if getattr(crawler, 'active_requests', 0) > 0: - continue # still serving requests, skip idle_time = now - LAST_USED[sig] - logger.info(f"🧹 Closing hot browser (sig={sig[:8]}, idle={idle_time:.0f}s)") - with suppress(Exception): - await crawler.close() + if getattr(crawler, 'active_requests', 0) > 0: + if idle_time <= STALE_CEILING: + continue # still serving requests, skip + logger.error(f"🚨 Leaked request counter (sig={sig[:8]}, active={crawler.active_requests}, idle={idle_time:.0f}s > {STALE_CEILING}s) - force-closing") + else: + logger.info(f"🧹 Closing hot browser (sig={sig[:8]}, idle={idle_time:.0f}s)") + _close_in_background(crawler) # close() can hang; never await it under LOCK HOT_POOL.pop(sig, None) LAST_USED.pop(sig, None) USAGE_COUNT.pop(sig, None) @@ -212,7 +248,7 @@ async def janitor(): try: from monitor import get_monitor await get_monitor().track_janitor_event("close_hot", sig, {"idle_seconds": int(idle_time), "ttl": hot_ttl}) - except: + except Exception: # bare except would swallow CancelledError and outlive shutdown pass # Log pool stats diff --git a/deploy/docker/utils.py b/deploy/docker/utils.py index 8f8ecb180..b8c5d59de 100644 --- a/deploy/docker/utils.py +++ b/deploy/docker/utils.py @@ -1,4 +1,3 @@ -import dns.resolver import logging import yaml import os @@ -400,6 +399,7 @@ def validate_webhook_url(url: str) -> None: def verify_email_domain(email: str) -> bool: + import dns.resolver # imported here so this module loads without dnspython try: domain = email.split('@')[1] # Try to resolve MX records for the domain. diff --git a/tests/docker/test_pool_release.py b/tests/docker/test_pool_release.py index 6c81b3e52..0ef710849 100644 --- a/tests/docker/test_pool_release.py +++ b/tests/docker/test_pool_release.py @@ -1,155 +1,386 @@ -"""Tests for crawler pool release_crawler() and active_requests tracking. +"""Tests for crawler pool release_crawler() and the janitor's stale-lease backstop. -These tests validate the pool lifecycle without requiring Docker or a running -server. They test the release logic directly using mock crawler objects. +These exercise the real deploy/docker/crawler_pool module (no Docker, no server) +using lightweight stand-in crawler objects. """ import asyncio +import copy +import importlib +import os +import sys + import pytest -from unittest.mock import MagicMock +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "deploy", "docker"))) -# --------------------------------------------------------------------------- -# Standalone release_crawler implementation for testing -# (mirrors the logic that will be added to deploy/docker/crawler_pool.py) -# --------------------------------------------------------------------------- +import crawler_pool # noqa: E402 +import utils # noqa: E402 + + +class FakeCrawler: + """Stand-in for AsyncWebCrawler; the pool only touches .active_requests.""" + + def __init__(self, active_requests=0): + self.active_requests = active_requests + self.closed = False -_TEST_LOCK = asyncio.Lock() + async def close(self): + self.closed = True -async def _release_crawler(crawler, lock=None): - """Standalone release logic matching crawler_pool.release_crawler().""" - lock = lock or _TEST_LOCK - async with lock: - if hasattr(crawler, "active_requests"): - crawler.active_requests = max(0, crawler.active_requests - 1) +def _reset_pool(): + crawler_pool.HOT_POOL.clear() + crawler_pool.COLD_POOL.clear() + crawler_pool.LAST_USED.clear() + crawler_pool.USAGE_COUNT.clear() + crawler_pool.PERMANENT = None + crawler_pool.DEFAULT_CONFIG_SIG = None + + +@pytest.fixture(autouse=True) +def clean_pool(): + """Reset module globals so tests can't leak state into each other.""" + _reset_pool() + yield + _reset_pool() + for t in list(crawler_pool._CLOSE_TASKS): + t.cancel() + + +async def _drain_close_tasks(): + """Wait for the janitor's fire-and-forget close tasks to finish.""" + tasks = list(crawler_pool._CLOSE_TASKS) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) # --------------------------------------------------------------------------- -# Tests +# release_crawler # --------------------------------------------------------------------------- - class TestReleaseCrawler: - """Tests for the release_crawler function.""" @pytest.mark.asyncio - async def test_release_decrements_active_requests(self): - """release_crawler should decrement active_requests by 1.""" - crawler = MagicMock() - crawler.active_requests = 3 + async def test_decrements(self): + c = FakeCrawler(3) + await crawler_pool.release_crawler(c) + assert c.active_requests == 2 - await _release_crawler(crawler) - assert crawler.active_requests == 2 + @pytest.mark.asyncio + async def test_floors_at_zero(self): + c = FakeCrawler(0) + await crawler_pool.release_crawler(c) + assert c.active_requests == 0 @pytest.mark.asyncio - async def test_release_floors_at_zero(self): - """active_requests should never go below 0.""" - crawler = MagicMock() - crawler.active_requests = 0 + async def test_missing_attribute_is_noop(self): + class Bare: + pass - await _release_crawler(crawler) - assert crawler.active_requests == 0 + await crawler_pool.release_crawler(Bare()) # must not raise @pytest.mark.asyncio - async def test_release_from_one_to_zero(self): - """Standard case: single request finishes.""" - crawler = MagicMock() - crawler.active_requests = 1 + async def test_does_not_take_the_pool_lock(self): + """Regression: a release must not block behind a slow janitor/start holding LOCK.""" + c = FakeCrawler(1) + async with crawler_pool.LOCK: + await asyncio.wait_for(crawler_pool.release_crawler(c), timeout=0.5) + assert c.active_requests == 0 - await _release_crawler(crawler) - assert crawler.active_requests == 0 @pytest.mark.asyncio - async def test_release_handles_missing_attribute(self): - """Should not crash if crawler has no active_requests attribute.""" - crawler = MagicMock(spec=[]) # no attributes at all - # Should not raise - await _release_crawler(crawler) + async def test_wall_clock_timeout_still_releases(self): + """AC1's mechanism: asyncio.wait_for expiring must still run the finally that releases.""" + c = FakeCrawler(1) + + async def handler(): + try: + await asyncio.wait_for(asyncio.sleep(60), timeout=0.01) # the hung crawl + finally: + await crawler_pool.release_crawler(c) + + with pytest.raises(asyncio.TimeoutError): + await handler() + assert c.active_requests == 0 + + +# --------------------------------------------------------------------------- +# janitor sweep: idle TTL + stale-lease backstop, driven through the real loop +# --------------------------------------------------------------------------- + +def _one_pass_only(real_sleep): + """Stand-in for asyncio.sleep that lets the janitor run exactly one pass.""" + passes = {"n": 0} + + async def fake(_delay): + passes["n"] += 1 + if passes["n"] > 1: + raise asyncio.CancelledError + await real_sleep(0) + + return fake + + +class TestJanitorSweep: + """Drives the real janitor loop with sleeps made instant.""" + + def _patch(self, monkeypatch): + real_sleep = asyncio.sleep + monkeypatch.setattr(crawler_pool, "get_container_memory_percent", lambda: 10.0) + monkeypatch.setattr(crawler_pool.asyncio, "sleep", _one_pass_only(real_sleep)) @pytest.mark.asyncio - async def test_multiple_releases_decrement_correctly(self): - """Multiple sequential releases should each decrement by 1.""" - crawler = MagicMock() - crawler.active_requests = 5 + async def test_force_closes_a_cold_browser_busy_past_the_ceiling(self, monkeypatch, caplog): + c = FakeCrawler(1) # counter stuck at 1, nobody using it + crawler_pool.COLD_POOL["deadbeefcafe"] = c + crawler_pool.LAST_USED["deadbeefcafe"] = ( + crawler_pool.time.time() - crawler_pool.STALE_CEILING - 1 + ) + self._patch(monkeypatch) + + with caplog.at_level("ERROR", logger="crawler_pool"): + with pytest.raises(asyncio.CancelledError): + await crawler_pool.janitor() + await _drain_close_tasks() + + assert c.closed is True, "janitor never reclaimed the pinned browser" + assert "deadbeefcafe" not in crawler_pool.COLD_POOL + assert "deadbeef" in caplog.text, "force-close must log an ERROR naming the signature" - for expected in [4, 3, 2, 1, 0, 0]: # last one should floor at 0 - await _release_crawler(crawler) - assert crawler.active_requests == expected + @pytest.mark.asyncio + async def test_force_closes_a_hot_browser_busy_past_the_ceiling(self, monkeypatch): + """Both pools must apply the backstop, not just the cold one.""" + c = FakeCrawler(2) + crawler_pool.HOT_POOL["sig-hot"] = c + crawler_pool.LAST_USED["sig-hot"] = ( + crawler_pool.time.time() - crawler_pool.STALE_CEILING - 1 + ) + self._patch(monkeypatch) + + with pytest.raises(asyncio.CancelledError): + await crawler_pool.janitor() + await _drain_close_tasks() + + assert c.closed is True + assert "sig-hot" not in crawler_pool.HOT_POOL @pytest.mark.asyncio - async def test_concurrent_releases_are_safe(self): - """Concurrent releases should not corrupt the counter.""" - crawler = MagicMock() - crawler.active_requests = 100 - lock = asyncio.Lock() + async def test_skips_a_browser_idle_past_ttl_but_still_busy(self, monkeypatch): + """A slow-but-legitimate crawl: past the idle TTL, under the leak ceiling, must not be closed.""" + c = FakeCrawler(1) + crawler_pool.HOT_POOL["sig-slow"] = c + # 700s: over hot_ttl (600) so the TTL sweep looks at it, under STALE_CEILING (>= 21600) + crawler_pool.LAST_USED["sig-slow"] = crawler_pool.time.time() - 700 + self._patch(monkeypatch) + + with pytest.raises(asyncio.CancelledError): + await crawler_pool.janitor() + await _drain_close_tasks() + + assert c.closed is False, "closed a browser that was still serving a request" + assert c.active_requests == 1 + assert crawler_pool.HOT_POOL.get("sig-slow") is c - async def release_n_times(n): - for _ in range(n): - await _release_crawler(crawler, lock=lock) + @pytest.mark.asyncio + async def test_closes_an_idle_clean_browser_past_ttl(self, monkeypatch): + """The ordinary reap path must still work through the background-close helper.""" + c = FakeCrawler(0) + crawler_pool.COLD_POOL["sig-idle"] = c + crawler_pool.LAST_USED["sig-idle"] = crawler_pool.time.time() - 400 # > cold_ttl (300) + self._patch(monkeypatch) - # 10 concurrent tasks each releasing 10 times = 100 total - tasks = [asyncio.create_task(release_n_times(10)) for _ in range(10)] - await asyncio.gather(*tasks) + with pytest.raises(asyncio.CancelledError): + await crawler_pool.janitor() + await _drain_close_tasks() - assert crawler.active_requests == 0 + assert c.closed is True + assert "sig-idle" not in crawler_pool.COLD_POOL + @pytest.mark.asyncio + async def test_a_hung_close_does_not_block_the_sweep(self, monkeypatch): + """Regression: close() on a wedged browser must never freeze the janitor (it used + to be awaited while holding the pool LOCK).""" + class WedgedCrawler(FakeCrawler): + def __init__(self): + super().__init__(0) + self.blocker = asyncio.Event() + + async def close(self): + await self.blocker.wait() + self.closed = True + + c = WedgedCrawler() + crawler_pool.COLD_POOL["sig-wedge"] = c + crawler_pool.LAST_USED["sig-wedge"] = crawler_pool.time.time() - 400 + self._patch(monkeypatch) + + # If close() were awaited inline this would hang forever instead of raising. + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(crawler_pool.janitor(), timeout=5) + + assert "sig-wedge" not in crawler_pool.COLD_POOL, "pool entry must go even if close hangs" + assert not crawler_pool.LOCK.locked(), "LOCK must not be held while a close is pending" + c.blocker.set() + await _drain_close_tasks() + assert c.closed is True + + +class TestStaleCeiling: + + def test_ceiling_derives_from_the_wall_clock(self): + assert crawler_pool.STALE_CEILING == max(2 * crawler_pool._WALL_CLOCK, 21600) + # 6h floor: streaming has no deadline; a lower floor kills legitimate long streams + assert crawler_pool.STALE_CEILING >= 21600 + + @pytest.mark.parametrize("configured,expected", [ + (0, 21600), (None, 21600), (False, 21600), # unset -> auto: max(2 x 1800, 21600) + (True, 21600), # YAML `true` must not become a 1-second ceiling + (-1, 21600), ("30m", 21600), # nonsense -> auto, never break the janitor + (45, 45), (86400, 86400), # a real value is used as written + ]) + def test_stale_lease_override_resolves_through_the_real_module(self, configured, expected): + cfg = copy.deepcopy(utils.load_config()) + cfg["limits"]["wall_clock_s"] = 1800 + cfg["crawler"]["pool"]["stale_lease_s"] = configured + orig = utils.load_config + utils.load_config = lambda: cfg + try: + assert importlib.reload(crawler_pool).STALE_CEILING == expected + finally: + utils.load_config = orig + importlib.reload(crawler_pool) + + def test_bad_wall_clock_value_never_breaks_import(self): + cfg = copy.deepcopy(utils.load_config()) + cfg["limits"]["wall_clock_s"] = "10m" # nonsense -> fall back, never break the janitor + cfg["crawler"]["pool"]["stale_lease_s"] = 0 + orig = utils.load_config + utils.load_config = lambda: cfg + try: + assert importlib.reload(crawler_pool).STALE_CEILING == 21600 + finally: + utils.load_config = orig + importlib.reload(crawler_pool) + + def test_shipped_config_enables_the_crawl_deadline(self): + """Without a deadline a hung crawl never reaches its finally, so nothing releases the browser.""" + assert crawler_pool._WALL_CLOCK > 0 + + +class TestCloseAllDrainsBackgroundCloses: -class TestActiveRequestsTracking: - """Tests for the get/release lifecycle pattern.""" + @pytest.mark.asyncio + async def test_close_all_waits_for_pending_close_tasks(self): + """Shutdown must not destroy live close tasks (no 'Task was destroyed' noise).""" + started = asyncio.Event() + finished = asyncio.Event() + + class SlowClose(FakeCrawler): + async def close(self): + started.set() + await asyncio.sleep(0.05) + self.closed = True + finished.set() + + c = SlowClose() + crawler_pool._close_in_background(c) + await started.wait() + + await crawler_pool.close_all() + + assert finished.is_set(), "close_all returned while a background close was still running" + assert c.closed is True + assert not crawler_pool._CLOSE_TASKS @pytest.mark.asyncio - async def test_get_sets_active_requests(self): - """Simulated get_crawler should set active_requests to 1 for new crawlers.""" - crawler = MagicMock() - # Simulate what get_crawler does for a new browser - crawler.active_requests = 1 + async def test_close_all_never_holds_the_lock_while_a_close_hangs(self): + """Regression: a wedged browser in the pool at shutdown must not freeze close_all + while it holds LOCK (closes are routed through the background helper).""" + blocker = asyncio.Event() + + class Wedged(FakeCrawler): + async def close(self): + await blocker.wait() + self.closed = True - assert crawler.active_requests == 1 + c = Wedged() + crawler_pool.HOT_POOL["sig-wedge"] = c + + shutdown = asyncio.create_task(crawler_pool.close_all()) + await asyncio.sleep(0.05) # close_all is now waiting on the drain + assert not shutdown.done() + assert not crawler_pool.LOCK.locked(), "LOCK held while waiting on a wedged close" + assert not crawler_pool.HOT_POOL, "pool must be cleared even while the close hangs" + + blocker.set() + await asyncio.wait_for(shutdown, timeout=5) + assert c.closed is True + + +# --------------------------------------------------------------------------- +# get_crawler: the acquire side of the counter +# --------------------------------------------------------------------------- + +class TestActiveRequestsTracking: + """Exercises the real get_crawler on pool hits, which never start a browser.""" + + @staticmethod + def _cfg(): + from crawl4ai import BrowserConfig + return BrowserConfig(headless=True) @pytest.mark.asyncio - async def test_get_increments_existing(self): - """Simulated get_crawler should increment for existing pooled crawlers.""" - crawler = MagicMock() - crawler.active_requests = 2 + async def test_hot_pool_hit_increments(self): + cfg = self._cfg() + c = FakeCrawler(0) + crawler_pool.HOT_POOL[crawler_pool._sig(cfg)] = c - # Simulate another get_crawler call returning same browser - crawler.active_requests += 1 - assert crawler.active_requests == 3 + assert await crawler_pool.get_crawler(cfg) is c + assert c.active_requests == 1 @pytest.mark.asyncio - async def test_full_get_release_lifecycle(self): - """Full lifecycle: get -> use -> release -> get -> release.""" - crawler = MagicMock() + async def test_cold_pool_hit_increments(self): + cfg = self._cfg() + c = FakeCrawler(0) + crawler_pool.COLD_POOL[crawler_pool._sig(cfg)] = c - # First request gets the crawler - crawler.active_requests = 1 + assert await crawler_pool.get_crawler(cfg) is c + assert c.active_requests == 1 - # Second concurrent request gets same crawler - crawler.active_requests += 1 - assert crawler.active_requests == 2 + @pytest.mark.asyncio + async def test_permanent_hit_increments(self): + cfg = self._cfg() + c = FakeCrawler(0) + crawler_pool.PERMANENT = c + crawler_pool.DEFAULT_CONFIG_SIG = crawler_pool._sig(cfg) - # First request finishes - await _release_crawler(crawler) - assert crawler.active_requests == 1 + assert await crawler_pool.get_crawler(cfg) is c + assert c.active_requests == 1 - # Second request finishes - await _release_crawler(crawler) - assert crawler.active_requests == 0 @pytest.mark.asyncio - async def test_janitor_safety_check(self): - """Janitor should only close browsers with active_requests == 0.""" - crawler = MagicMock() - crawler.active_requests = 1 + async def test_third_use_promotes_cold_to_hot_and_keeps_counter(self): + cfg = self._cfg() + sig = crawler_pool._sig(cfg) + c = FakeCrawler(0) + crawler_pool.COLD_POOL[sig] = c + crawler_pool.USAGE_COUNT[sig] = 2 # this call is the third - # Janitor check: should NOT close - should_close = getattr(crawler, "active_requests", 0) == 0 - assert should_close is False + assert await crawler_pool.get_crawler(cfg) is c + assert crawler_pool.HOT_POOL.get(sig) is c + assert sig not in crawler_pool.COLD_POOL + assert c.active_requests == 1 - # Request finishes - await _release_crawler(crawler) - # Janitor check: now safe to close - should_close = getattr(crawler, "active_requests", 0) == 0 - assert should_close is True + @pytest.mark.asyncio + async def test_acquire_refreshes_last_used_so_a_busy_browser_is_not_reaped(self): + """The backstop keys off LAST_USED, so every acquire must bump it.""" + cfg = self._cfg() + sig = crawler_pool._sig(cfg) + c = FakeCrawler(0) + crawler_pool.HOT_POOL[sig] = c + crawler_pool.LAST_USED[sig] = 0.0 # ancient + + await crawler_pool.get_crawler(cfg) + assert crawler_pool.time.time() - crawler_pool.LAST_USED[sig] < 5