diff --git a/crawl4ai/browser_manager.py b/crawl4ai/browser_manager.py index e0b0cd059..a710c244c 100644 --- a/crawl4ai/browser_manager.py +++ b/crawl4ai/browser_manager.py @@ -1598,6 +1598,79 @@ async def _get_page_by_target_id(self, context: BrowserContext, target_id: str): ) return None + + async def _new_page_cancel_safe(self, context): + """ + Create a page, shielding the Playwright call so a CancelledError + cannot orphan a page that finishes after the caller is cancelled. + """ + task = asyncio.create_task(context.new_page()) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + # Caller was cancelled. If Playwright still finishes the page, + # close it so it is not orphaned; if it is still running, cancel it. + if task.done(): + if task.cancelled(): + raise + exc = task.exception() + if exc is not None: + raise + page = task.result() + try: + await page.close() + except Exception: + pass + else: + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + raise + except Exception: + if not task.done(): + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + elif not task.cancelled() and task.exception() is None: + try: + await task.result().close() + except Exception: + pass + raise + + async def _release_unreturned_context_acquire( + self, config_signature: str, page=None + ): + """ + Undo a get_page() acquisition that never returned to the caller. + + Decrements the context refcount and closes any page we created. + Must not raise — used from BaseException cleanup paths. + """ + if page is not None: + try: + self._page_to_sig.pop(page, None) + except Exception: + pass + try: + await page.close() + except Exception: + pass + if config_signature is None: + return + try: + async with self._contexts_lock: + if config_signature in self._context_refcounts: + self._context_refcounts[config_signature] = max( + 0, self._context_refcounts[config_signature] - 1 + ) + except Exception: + pass + async def get_page(self, crawlerRunConfig: CrawlerRunConfig): """ Get a page for the given session ID, creating a new one if needed. @@ -1609,6 +1682,7 @@ async def get_page(self, crawlerRunConfig: CrawlerRunConfig): (page, context): The Page and its BrowserContext """ self._cleanup_expired_sessions() + pending_acquire = None # (config_signature, page) while acquire is unreturned # If a session_id is provided and we already have it, reuse that page + context if crawlerRunConfig.session_id and crawlerRunConfig.session_id in self.sessions: @@ -1643,25 +1717,28 @@ async def get_page(self, crawlerRunConfig: CrawlerRunConfig): ) self._context_last_used[config_signature] = time.monotonic() - # Close evicted context OUTSIDE lock - if to_close is not None: - try: - await to_close.close() - except Exception: - pass - - # Always create a new page for each crawl (isolation for navigation) + # Always create a new page for each crawl (isolation for navigation). + # pending_acquire keeps the refcount until the common return path + # successfully hands the page to the caller (covers cancel during bump). + page = None + pending_acquire = (config_signature, None) try: - page = await context.new_page() - except Exception: - async with self._contexts_lock: - if config_signature in self._context_refcounts: - self._context_refcounts[config_signature] = max( - 0, self._context_refcounts[config_signature] - 1 - ) + # Close evicted context OUTSIDE lock + if to_close is not None: + try: + await to_close.close() + except Exception: + pass + to_close = None + page = await self._new_page_cancel_safe(context) + pending_acquire = (config_signature, page) + await self._apply_stealth_to_page(page) + self._page_to_sig[page] = config_signature + except BaseException: + await self._release_unreturned_context_acquire( + pending_acquire[0], pending_acquire[1] + ) raise - await self._apply_stealth_to_page(page) - self._page_to_sig[page] = config_signature elif self.config.storage_state: tmp_context = await self.create_browser_context(crawlerRunConfig) ctx = self.default_context # default context, one window only @@ -1743,37 +1820,74 @@ async def get_page(self, crawlerRunConfig: CrawlerRunConfig): ) self._context_last_used[config_signature] = time.monotonic() - # Close evicted context OUTSIDE lock - if to_close is not None: - try: - await to_close.close() - except Exception: - pass - - # Create a new page from the chosen context + # Create a new page from the chosen context. + # Hold the acquire until we successfully return so cancellation + # during new_page / stealth / recycle cannot leak the refcount. + page = None + acquired = True try: - page = await context.new_page() - except Exception: - async with self._contexts_lock: - if config_signature in self._context_refcounts: - self._context_refcounts[config_signature] = max( - 0, self._context_refcounts[config_signature] - 1 - ) - raise - await self._apply_stealth_to_page(page) - self._page_to_sig[page] = config_signature + if to_close is not None: + try: + await to_close.close() + except Exception: + pass + to_close = None + page = await self._new_page_cancel_safe(context) + await self._apply_stealth_to_page(page) + self._page_to_sig[page] = config_signature - # If a session_id is specified, store this session so we can reuse later - if crawlerRunConfig.session_id: - self.sessions[crawlerRunConfig.session_id] = (context, page, time.time()) + # If a session_id is specified, store this session so we can reuse later + if crawlerRunConfig.session_id: + self.sessions[crawlerRunConfig.session_id] = (context, page, time.time()) - self._pages_served += 1 + self._pages_served += 1 - # Check if browser recycle threshold is hit — bump version for next requests - # This happens AFTER incrementing counter so concurrent requests see correct count - await self._maybe_bump_browser_version() + # Check if browser recycle threshold is hit — bump version for next requests + # Non-blocking: if too many old versions are draining, defer the bump. + await self._maybe_bump_browser_version() + acquired = False # caller owns release via release_page_async + return page, context + finally: + if acquired: + # Never returned — drop session bookkeeping if we set it. + if ( + crawlerRunConfig.session_id + and crawlerRunConfig.session_id in self.sessions + ): + sess = self.sessions.get(crawlerRunConfig.session_id) + if sess is not None and sess[1] is page: + self.sessions.pop(crawlerRunConfig.session_id, None) + await self._release_unreturned_context_acquire( + config_signature, page + ) + + # Managed-browser paths above that did not early-return still land here. + # pending_acquire is set only for the isolated-context path (refcount tracking). + try: + # If a session_id is specified, store this session so we can reuse later + if crawlerRunConfig.session_id: + self.sessions[crawlerRunConfig.session_id] = (context, page, time.time()) - return page, context + self._pages_served += 1 + + # Check if browser recycle threshold is hit — bump version for next requests + # Non-blocking: if too many old versions are draining, defer the bump. + await self._maybe_bump_browser_version() + + pending_acquire = None # handed off to caller + return page, context + finally: + if pending_acquire is not None: + if ( + crawlerRunConfig.session_id + and crawlerRunConfig.session_id in self.sessions + ): + sess = self.sessions.get(crawlerRunConfig.session_id) + if sess is not None and sess[1] is page: + self.sessions.pop(crawlerRunConfig.session_id, None) + await self._release_unreturned_context_acquire( + pending_acquire[0], pending_acquire[1] + ) async def kill_session(self, session_id: str): """ @@ -1847,108 +1961,96 @@ async def _maybe_bump_browser_version(self): New requests automatically get a new browser (via new signature). Old browser drains naturally and gets cleaned up when refcount hits 0. + + Never blocks inside get_page(): if too many old versions are still + draining, opportunistically drop entries whose refcount is already 0 + and otherwise defer the bump to a later request. Blocking here was the + second half of the Docker recycle deadlock in #2419 — cancelled waits + leaked context refcounts and permanently filled the pending drain. """ if not self._should_recycle(): return - # Safety cap: wait if too many old browsers are draining - while True: - async with self._pending_cleanup_lock: - # Re-check threshold under lock (another request may have bumped already) - if not self._should_recycle(): - return + idle_sigs = [] + old_version = None - # Check safety cap - if len(self._pending_cleanup) >= self._max_pending_browsers: + async with self._pending_cleanup_lock: + # Re-check threshold under lock (another request may have bumped already) + if not self._should_recycle(): + return + + # Opportunistically clear drained (refcount 0) pending entries so a + # full drain queue does not permanently block recycling. + if len(self._pending_cleanup) >= self._max_pending_browsers: + drained = [ + s + for s in list(self._pending_cleanup.keys()) + if self._context_refcounts.get(s, 0) == 0 + ] + for sig in drained: + self._pending_cleanup.pop(sig, None) + if drained: + # Close drained contexts outside this lock below. + idle_sigs.extend(drained) if self.logger: self.logger.debug( - message="Waiting for old browser to drain (pending: {count})", + message="Cleared {count} drained pending entries before recycle", tag="BROWSER", - params={"count": len(self._pending_cleanup)}, + params={"count": len(drained)}, ) - self._cleanup_slot_available.clear() - # Release lock and wait - else: - # We have a slot — do the bump inside this lock hold - old_version = self._browser_version - active_sigs = [] - idle_sigs = [] - async with self._contexts_lock: - for sig in list(self._context_refcounts.keys()): - if self._context_refcounts.get(sig, 0) > 0: - active_sigs.append(sig) - else: - idle_sigs.append(sig) + if len(self._pending_cleanup) < self._max_pending_browsers: + self._cleanup_slot_available.set() - if self.logger: - self.logger.info( - message="Bumping browser version {old} -> {new} after {count} pages ({active} active, {idle} idle sigs)", - tag="BROWSER", - params={ - "old": old_version, - "new": old_version + 1, - "count": self._pages_served, - "active": len(active_sigs), - "idle": len(idle_sigs), - }, - ) + if len(self._pending_cleanup) >= self._max_pending_browsers: + if self.logger: + self.logger.debug( + message="Deferring browser recycle; pending drain full ({count})", + tag="BROWSER", + params={"count": len(self._pending_cleanup)}, + ) + # Fall through to close any drained contexts we collected, then return. + else: + # We have a slot — do the bump inside this lock hold + old_version = self._browser_version + active_sigs = [] + bump_idle = [] + async with self._contexts_lock: + for sig in list(self._context_refcounts.keys()): + if self._context_refcounts.get(sig, 0) > 0: + active_sigs.append(sig) + else: + bump_idle.append(sig) - # Only add sigs with active crawls to pending cleanup. - # Sigs with refcount 0 are cleaned up immediately below - # to avoid them being stuck in _pending_cleanup forever - # (no future release would trigger their cleanup). - done_event = asyncio.Event() - for sig in active_sigs: - self._pending_cleanup[sig] = { - "version": old_version, - "done": done_event, - } - - # Bump version — new get_page() calls will create new contexts - self._browser_version += 1 - self._pages_served = 0 - - # Clean up idle sigs immediately (outside pending_cleanup_lock below) - break # exit while loop to do cleanup outside locks - - # Safety cap path: wait for a cleanup slot, then retry. - # Timeout prevents permanent deadlock if stuck entries never drain. - try: - await asyncio.wait_for( - self._cleanup_slot_available.wait(), timeout=30.0 - ) - except asyncio.TimeoutError: - # Force-clean any pending entries that have refcount 0 - # (they're stuck and will never drain naturally) - async with self._pending_cleanup_lock: - stuck_sigs = [ - s for s in list(self._pending_cleanup.keys()) - if self._context_refcounts.get(s, 0) == 0 - ] - for sig in stuck_sigs: - self._pending_cleanup.pop(sig, None) - if stuck_sigs: - if self.logger: - self.logger.warning( - message="Force-cleaned {count} stuck pending entries after timeout", - tag="BROWSER", - params={"count": len(stuck_sigs)}, - ) - # Clean up the stuck contexts - for sig in stuck_sigs: - async with self._contexts_lock: - context = self.contexts_by_config.pop(sig, None) - self._context_refcounts.pop(sig, None) - self._context_last_used.pop(sig, None) - if context is not None: - try: - await context.close() - except Exception: - pass - if len(self._pending_cleanup) < self._max_pending_browsers: - self._cleanup_slot_available.set() - - # Reached via break — clean up idle sigs immediately (outside locks) + if self.logger: + self.logger.info( + message="Bumping browser version {old} -> {new} after {count} pages ({active} active, {idle} idle sigs)", + tag="BROWSER", + params={ + "old": old_version, + "new": old_version + 1, + "count": self._pages_served, + "active": len(active_sigs), + "idle": len(bump_idle), + }, + ) + + # Only add sigs with active crawls to pending cleanup. + # Sigs with refcount 0 are cleaned up immediately below + # to avoid them being stuck in _pending_cleanup forever + # (no future release would trigger their cleanup). + done_event = asyncio.Event() + for sig in active_sigs: + self._pending_cleanup[sig] = { + "version": old_version, + "done": done_event, + } + + # Bump version — new get_page() calls will create new contexts + self._browser_version += 1 + self._pages_served = 0 + idle_sigs.extend(bump_idle) + + # Close idle / drained contexts outside locks for sig in idle_sigs: async with self._contexts_lock: context = self.contexts_by_config.pop(sig, None) @@ -1959,13 +2061,14 @@ async def _maybe_bump_browser_version(self): await context.close() except Exception: pass - if idle_sigs and self.logger: + if idle_sigs and self.logger and old_version is not None: self.logger.debug( message="Immediately cleaned up {count} idle contexts from version {version}", tag="BROWSER", params={"count": len(idle_sigs), "version": old_version}, ) + async def _maybe_cleanup_old_browser(self, sig: str): """Clean up an old browser's context if its refcount hit 0 and it's pending cleanup.""" async with self._pending_cleanup_lock: diff --git a/tests/browser/test_get_page_refcount_cancel_2419.py b/tests/browser/test_get_page_refcount_cancel_2419.py new file mode 100644 index 000000000..309242e21 --- /dev/null +++ b/tests/browser/test_get_page_refcount_cancel_2419.py @@ -0,0 +1,195 @@ +""" +Regression tests for #2419: get_page() must not leak context refcounts on +cancellation / BaseException before the page is returned, and browser recycle +must not block forever inside get_page when the pending drain is full. +""" +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from crawl4ai.async_configs import BrowserConfig, CrawlerRunConfig +from crawl4ai.browser_manager import BrowserManager + + +def _bm(**kwargs): + cfg = BrowserConfig(**kwargs) + bm = BrowserManager(cfg, logger=None) + return bm + + +@pytest.mark.asyncio +async def test_cancel_during_new_page_releases_refcount(): + """CancelledError after the refcount bump must return the refcount to 0.""" + bm = _bm() + sig = "sig_cancel_new_page" + context = MagicMock() + started = asyncio.Event() + + async def hang_new_page(): + started.set() + await asyncio.sleep(3600) + return MagicMock() + + context.new_page = hang_new_page + bm.contexts_by_config[sig] = context + bm._context_refcounts[sig] = 0 + bm._context_last_used[sig] = 0.0 + bm._make_config_signature = MagicMock(return_value=sig) + + task = asyncio.create_task(bm.get_page(CrawlerRunConfig())) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert bm._context_refcounts.get(sig, 0) == 0 + + +@pytest.mark.asyncio +async def test_exception_after_new_page_releases_refcount_and_closes_page(): + """Exception from stealth after new_page must drop refcount and close the page.""" + bm = _bm() + sig = "sig_stealth_boom" + context = MagicMock() + page = MagicMock() + page.close = AsyncMock() + + async def ok_new_page(): + return page + + context.new_page = ok_new_page + bm.contexts_by_config[sig] = context + bm._context_refcounts[sig] = 0 + bm._context_last_used[sig] = 0.0 + bm._make_config_signature = MagicMock(return_value=sig) + bm._apply_stealth_to_page = AsyncMock(side_effect=RuntimeError("stealth failed")) + + with pytest.raises(RuntimeError, match="stealth failed"): + await bm.get_page(CrawlerRunConfig()) + + assert bm._context_refcounts.get(sig, 0) == 0 + page.close.assert_awaited() + assert page not in bm._page_to_sig + + +@pytest.mark.asyncio +async def test_cancel_during_bump_releases_refcount(): + """Cancel while recycle bump runs after the page was created must not leak.""" + bm = _bm() + bm.config.max_pages_before_recycle = 1 + sig = "sig_bump_cancel" + context = MagicMock() + page = MagicMock() + page.close = AsyncMock() + + async def ok_new_page(): + return page + + context.new_page = ok_new_page + bm.contexts_by_config[sig] = context + bm._context_refcounts[sig] = 0 + bm._context_last_used[sig] = 0.0 + bm._pages_served = 0 + bm._make_config_signature = MagicMock(return_value=sig) + bm._apply_stealth_to_page = AsyncMock() + + entered = asyncio.Event() + + async def blocking_bump(): + entered.set() + await asyncio.sleep(3600) + + bm._maybe_bump_browser_version = blocking_bump + + task = asyncio.create_task(bm.get_page(CrawlerRunConfig())) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert bm._context_refcounts.get(sig, 0) == 0 + page.close.assert_awaited() + + +@pytest.mark.asyncio +async def test_recycle_defers_when_pending_drain_full(): + """Full pending drain must defer the bump instead of waiting inside get_page.""" + bm = _bm() + bm.config.max_pages_before_recycle = 1 + bm._pages_served = 1 + bm._max_pending_browsers = 3 + # Three active (refcount > 0) pending entries — cannot opportunistically clear. + for i in range(3): + sig = f"old_{i}" + bm._pending_cleanup[sig] = {"version": 1, "done": asyncio.Event()} + bm._context_refcounts[sig] = 1 + bm.contexts_by_config[sig] = MagicMock() + + await asyncio.wait_for(bm._maybe_bump_browser_version(), timeout=1.0) + + assert bm._browser_version == 1 # deferred, not bumped + assert bm._pages_served == 1 + assert len(bm._pending_cleanup) == 3 + + +@pytest.mark.asyncio +async def test_recycle_clears_drained_pending_then_bumps(): + """Refcount-0 pending entries are cleared so recycle can proceed.""" + bm = _bm() + bm.config.max_pages_before_recycle = 1 + bm._pages_served = 1 + bm._max_pending_browsers = 3 + bm._browser_version = 5 + + for i in range(3): + sig = f"drained_{i}" + ctx = MagicMock() + ctx.close = AsyncMock() + bm._pending_cleanup[sig] = {"version": 4, "done": asyncio.Event()} + bm._context_refcounts[sig] = 0 + bm.contexts_by_config[sig] = ctx + bm._context_last_used[sig] = 0.0 + + # Also seed a live context that will be active during the bump. + live_sig = "live" + live_ctx = MagicMock() + live_ctx.close = AsyncMock() + bm.contexts_by_config[live_sig] = live_ctx + bm._context_refcounts[live_sig] = 1 + bm._context_last_used[live_sig] = 1.0 + + await asyncio.wait_for(bm._maybe_bump_browser_version(), timeout=1.0) + + assert bm._browser_version == 6 + assert bm._pages_served == 0 + # drained_* removed; live moved into pending cleanup for the old version + assert "live" in bm._pending_cleanup + assert all(s not in bm._pending_cleanup for s in ("drained_0", "drained_1", "drained_2")) + + +@pytest.mark.asyncio +async def test_successful_get_page_keeps_refcount_for_caller(): + """A successful get_page must leave refcount=1 for release_page_async.""" + bm = _bm() + sig = "sig_ok" + context = MagicMock() + page = MagicMock() + page.close = AsyncMock() + + async def ok_new_page(): + return page + + context.new_page = ok_new_page + bm.contexts_by_config[sig] = context + bm._context_refcounts[sig] = 0 + bm._context_last_used[sig] = 0.0 + bm._make_config_signature = MagicMock(return_value=sig) + bm._apply_stealth_to_page = AsyncMock() + + got_page, got_ctx = await bm.get_page(CrawlerRunConfig()) + assert got_page is page + assert got_ctx is context + assert bm._context_refcounts[sig] == 1 + assert bm._page_to_sig[page] == sig + page.close.assert_not_awaited()