Make the browser pool un-pinnable by hung crawls

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

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": "…"}`.

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

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

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.