diff --git a/python/rustwright/_async_generated.py b/python/rustwright/_async_generated.py
index 35b3133..6e2670f 100644
--- a/python/rustwright/_async_generated.py
+++ b/python/rustwright/_async_generated.py
@@ -1,5 +1,5 @@
# This file is generated by tools/generate_async_api.py. Do not edit.
-# sync_api.py sha256: 86dbd1ae4130f44b5d7fdcc0756a6b0c36f5aa57b75e7e4249f64142e29a49c3
+# sync_api.py sha256: 49b3f768b7d4abd70196e6caf57ef90a50b1d2ad57e0a826b0029a707e70dc90
from __future__ import annotations
from pathlib import Path
diff --git a/python/rustwright/async_api.py b/python/rustwright/async_api.py
index b95c158..355b280 100644
--- a/python/rustwright/async_api.py
+++ b/python/rustwright/async_api.py
@@ -75,11 +75,16 @@
_default_timeout_for_method,
_emit_event,
_event_handler_positional_args,
+ _EVENT_DIALOG_CLAIM,
+ _SYNC_CLOSE_CLOSED,
+ _SYNC_CLOSE_CLOSING,
+ _raise_close_error,
_EVENT_DISPATCH_OWNER,
_EVENT_DISPATCH_REGISTRATION,
_EVENT_DISPATCH_SEQUENCE,
_json,
_is_ignorable_close_error,
+ _validate_timeout_value,
_sync_close_wait_deadline,
_raise_close_wait_timeout,
_update_sync_cleanup_complete,
@@ -137,6 +142,9 @@
_DEFAULT_ASYNCIO_TO_THREAD = asyncio.to_thread
+_ASYNC_CLOSE_OWNER: contextvars.ContextVar[Any] = contextvars.ContextVar(
+ "_rustwright_async_close_owner", default=None
+)
_ASYNC_EXECUTOR_LOCK = threading.Lock()
_ASYNC_EXECUTOR: Optional[concurrent.futures.Executor] = None
@@ -191,18 +199,36 @@ def async_executor_info() -> dict[str, Any]:
"owned": owns_executor,
"max_workers": getattr(executor, "_max_workers", None),
}
-
-
async def _run_sync_call(func: Callable[..., Any], /, *args: Any, **kwargs: Any) -> Any:
with _ASYNC_EXECUTOR_LOCK:
executor = _ASYNC_EXECUTOR
+ owner = _ASYNC_CLOSE_OWNER.get()
+
+ def call() -> Any:
+ if owner is None:
+ return func(*args, **kwargs)
+ condition = getattr(owner, "_rustwright_sync_close_condition", None)
+ if condition is None or not hasattr(condition, "__enter__"):
+ return func(*args, **kwargs)
+ current_thread = threading.get_ident()
+ with condition:
+ owner._rustwright_async_close_sync_owner = current_thread
+ condition.notify_all()
+ try:
+ return func(*args, **kwargs)
+ finally:
+ with condition:
+ if owner._rustwright_async_close_sync_owner == current_thread:
+ owner._rustwright_async_close_sync_owner = None
+ condition.notify_all()
+
if executor is None:
- return await _DEFAULT_ASYNCIO_TO_THREAD(func, *args, **kwargs)
+ return await _DEFAULT_ASYNCIO_TO_THREAD(call)
loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
- call = functools.partial(ctx.run, func, *args, **kwargs)
- return await loop.run_in_executor(executor, call)
+ wrapped_call = functools.partial(ctx.run, call)
+ return await loop.run_in_executor(executor, wrapped_call)
async def _await_native(awaitable: Any) -> Any:
@@ -246,7 +272,6 @@ async def _await_native_action(method: str, awaitable: Any) -> Any:
error = error_type(f"{method}: {message}")
raise _copy_wire_error_metadata(exc, error) from None
-
async def _await_cleanup_completion(awaitable: Any) -> Any:
"""Finish lifecycle cleanup before propagating cancellation to the caller."""
cleanup_task = asyncio.ensure_future(awaitable)
@@ -266,6 +291,16 @@ async def _await_cleanup_completion(awaitable: Any) -> Any:
_CLOSE_CLOSING = "closing"
_CLOSE_CLOSED = "closed"
+def _async_close_is_terminal(sync_obj: Any) -> bool:
+ condition = getattr(sync_obj, "_rustwright_sync_close_condition", None)
+ if condition is None or not hasattr(condition, "__enter__"):
+ return getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN) == _CLOSE_CLOSED
+ with condition:
+ return (
+ getattr(sync_obj, "_rustwright_sync_close_state", _CLOSE_OPEN) == _SYNC_CLOSE_CLOSED
+ or getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN) == _CLOSE_CLOSED
+ )
+
async def _single_flight_close(sync_obj: Any, cleanup: Callable[[], Any]) -> None:
"""Run one cancellation-safe close task and share its result with every caller."""
@@ -273,48 +308,115 @@ async def _single_flight_close(sync_obj: Any, cleanup: Callable[[], Any]) -> Non
if condition is not None and not hasattr(condition, "__enter__"):
condition = None
- def read_state() -> tuple[str, Any]:
- if condition is None:
- return (
- getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN),
- getattr(sync_obj, "_rustwright_async_close_task", None),
- )
- with condition:
- return (
- getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN),
- getattr(sync_obj, "_rustwright_async_close_task", None),
- )
-
- def write_state(state: str, task: Any) -> None:
- if condition is None:
- sync_obj._rustwright_async_close_state = state
- sync_obj._rustwright_async_close_task = task
+ async def owned_cleanup() -> Any:
+ token = _ASYNC_CLOSE_OWNER.set(sync_obj)
+ try:
+ return await cleanup()
+ finally:
+ _ASYNC_CLOSE_OWNER.reset(token)
+
+ candidate = asyncio.create_task(owned_cleanup())
+ current_thread = threading.get_ident()
+ if condition is None:
+ state = getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN)
+ task = getattr(sync_obj, "_rustwright_async_close_task", None)
+ if state == _CLOSE_CLOSED:
+ candidate.cancel()
+ return
+ if state == _CLOSE_CLOSING and task is not None:
+ candidate.cancel()
+ await _await_cleanup_completion(task)
return
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSING
+ sync_obj._rustwright_async_close_sync_owner = current_thread
+ sync_obj._rustwright_async_close_task = candidate
+ owner = True
+ else:
+ role: str
+ task: Any = None
+ generation: Optional[int] = None
+ closed_error: Any = None
with condition:
- sync_obj._rustwright_async_close_state = state
- sync_obj._rustwright_async_close_task = task
- condition.notify_all()
-
- state, task = read_state()
- if state == _CLOSE_CLOSED:
- return
- if state == _CLOSE_CLOSING and task is not None:
- if task is asyncio.current_task():
+ sync_state = getattr(sync_obj, "_rustwright_sync_close_state", _CLOSE_OPEN)
+ if sync_state == _SYNC_CLOSE_CLOSED:
+ role = "closed"
+ outcomes = getattr(sync_obj, "_rustwright_sync_close_outcomes", {})
+ outcome = outcomes.get(getattr(sync_obj, "_rustwright_sync_close_generation", 0))
+ closed_error = getattr(outcome, "error", None)
+ elif sync_state == _SYNC_CLOSE_CLOSING:
+ owner = getattr(sync_obj, "_rustwright_sync_close_owner", None)
+ if owner == current_thread:
+ role = "reentrant"
+ else:
+ role = "sync"
+ generation = getattr(sync_obj, "_rustwright_sync_close_generation", 0)
+ waiters = getattr(sync_obj, "_rustwright_sync_close_waiters", None)
+ if isinstance(waiters, dict):
+ waiters[generation] = waiters.get(generation, 0) + 1
+ else:
+ async_state = getattr(sync_obj, "_rustwright_async_close_state", _CLOSE_OPEN)
+ if async_state == _CLOSE_CLOSED:
+ role = "closed"
+ elif async_state == _CLOSE_CLOSING:
+ role = "async"
+ task = getattr(sync_obj, "_rustwright_async_close_task", None)
+ else:
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSING
+ sync_obj._rustwright_async_close_sync_owner = current_thread
+ sync_obj._rustwright_async_close_task = candidate
+ role = "owner"
+ owner = True
+ if role != "owner":
+ candidate.cancel()
+ if role == "closed" and closed_error is not None:
+ _raise_close_error(closed_error)
+ if role == "sync" and generation is not None:
+ wait_sync_generation = getattr(sync_obj, "_wait_for_sync_close_generation", None)
+ if callable(wait_sync_generation):
+ await _run_sync_call(wait_sync_generation, generation)
+ return
+ if role == "async" and task is not None:
+ if task is not asyncio.current_task():
+ await _await_cleanup_completion(task)
+ return
return
- await _await_cleanup_completion(task)
- return
- task = asyncio.create_task(cleanup())
- write_state(_CLOSE_CLOSING, task)
try:
- await _await_cleanup_completion(task)
+ await _await_cleanup_completion(candidate)
except BaseException:
- write_state(_CLOSE_CLOSED if task.done() and not task.cancelled() and task.exception() is None else _CLOSE_OPEN, None)
+ native_disposed = getattr(sync_obj, "_rustwright_sync_close_native_disposed", None)
+ if native_disposed is None:
+ native_disposed = getattr(sync_obj, "_rustwright_async_close_native_disposed", None)
+ if native_disposed is None:
+ native_disposed = True
+ if condition is None:
+ complete = bool(
+ native_disposed
+ and getattr(sync_obj, "_rustwright_sync_close_cleanup_complete", False)
+ )
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSED if complete else _CLOSE_OPEN
+ sync_obj._rustwright_async_close_task = None
+ else:
+ with condition:
+ complete = bool(
+ native_disposed
+ and getattr(sync_obj, "_rustwright_sync_close_cleanup_complete", False)
+ )
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSED if complete else _CLOSE_OPEN
+ sync_obj._rustwright_async_close_task = None
+ sync_obj._rustwright_async_close_sync_owner = None
+ condition.notify_all()
raise
else:
- write_state(_CLOSE_CLOSED, None)
-
-
+ if condition is None:
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSED
+ sync_obj._rustwright_async_close_task = None
+ else:
+ with condition:
+ sync_obj._rustwright_async_close_state = _CLOSE_CLOSED
+ sync_obj._rustwright_async_close_task = None
+ sync_obj._rustwright_async_close_sync_owner = None
+ condition.notify_all()
def _native_normalize_selector(selector: str, *, method: str) -> str:
missing_method = "_click" if method == "Page.click" else method.rsplit(".", 1)[-1]
return _normalize_selector_option(
@@ -1320,6 +1422,17 @@ def _wrap_async_event_handler(owner: Any, event: str, handler: Any) -> Any:
def wrapper(*args: Any) -> None:
sync_owner = getattr(owner, "_sync", None)
+ dialog_claim = _EVENT_DIALOG_CLAIM.get() if event == "dialog" else None
+ claim_released = False
+
+ def release_dialog_claim() -> None:
+ nonlocal claim_released
+ if claim_released or dialog_claim is None or not args:
+ return
+ claim_released = True
+ release = getattr(args[0], "_release_handler_claim", None)
+ if callable(release):
+ release(dialog_claim)
def call_handler() -> Any:
mapped_args = tuple(_wrap_async_event_value(event, arg) for arg in args)
@@ -1333,10 +1446,15 @@ async def await_result() -> Any:
if not callable(activate) or not activate(state, task):
if inspect.iscoroutine(result):
result.close()
+ release_dialog_claim()
return None
owner_token = _EVENT_DISPATCH_OWNER.set(task)
try:
- return await result
+ try:
+ return await result
+ except BaseException:
+ release_dialog_claim()
+ return None
finally:
_EVENT_DISPATCH_OWNER.reset(owner_token)
if state is not None:
@@ -1350,19 +1468,26 @@ def invoke_on_loop(callback_owner: Any, state: Any) -> None:
owner_token = _EVENT_DISPATCH_OWNER.set(callback_owner)
active = False
deferred = False
+ handler_raised = False
try:
if state is not None:
activate = getattr(sync_owner, "_activate_deferred_event_handler", None)
if not callable(activate) or not activate(state, callback_owner):
+ release_dialog_claim()
return
active = True
- result = call_handler()
+ try:
+ result = call_handler()
+ except BaseException:
+ handler_raised = True
+ raise
if _should_await_callback_result(result):
if state is not None:
deferred_state = getattr(sync_owner, "_defer_current_event_handler", lambda: None)()
if deferred_state is None:
if inspect.iscoroutine(result):
result.close()
+ release_dialog_claim()
return
deferred = True
try:
@@ -1371,37 +1496,54 @@ def invoke_on_loop(callback_owner: Any, state: Any) -> None:
release = getattr(sync_owner, "_release_deferred_event_handler", None)
if callable(release):
release(deferred_state)
+ release_dialog_claim()
raise
else:
_run_awaitable_on_loop(loop, result)
finally:
- if active and not deferred:
- finish = getattr(sync_owner, "_finish_event_handler", None)
- if callable(finish):
- finish(state, callback_owner)
+ if not deferred:
+ if handler_raised:
+ release_dialog_claim()
+ if active:
+ finish = getattr(sync_owner, "_finish_event_handler", None)
+ if callable(finish):
+ finish(state, callback_owner)
_EVENT_DISPATCH_OWNER.reset(owner_token)
def invoke_in_current_loop() -> None:
- result = call_handler()
- if not _should_await_callback_result(result):
- return
- state = _EVENT_DISPATCH_REGISTRATION.get()
- defer = getattr(sync_owner, "_defer_current_event_handler", None)
- deferred_state = defer() if callable(defer) else None
- if state is not None and deferred_state is None:
- if inspect.iscoroutine(result):
- result.close()
- return
- if deferred_state is not None:
+ deferred = False
+ handler_raised = False
+ try:
try:
- run_awaitable_with_state(result, deferred_state, loop)
+ result = call_handler()
except BaseException:
- release = getattr(sync_owner, "_release_deferred_event_handler", None)
- if callable(release):
- release(deferred_state)
+ handler_raised = True
raise
- else:
- _run_awaitable_on_loop(loop, result)
+ if not _should_await_callback_result(result):
+ return
+ state = _EVENT_DISPATCH_REGISTRATION.get()
+ defer = getattr(sync_owner, "_defer_current_event_handler", None)
+ deferred_state = defer() if callable(defer) else None
+ if state is not None and deferred_state is None:
+ if inspect.iscoroutine(result):
+ result.close()
+ release_dialog_claim()
+ return
+ if deferred_state is not None:
+ deferred = True
+ try:
+ run_awaitable_with_state(result, deferred_state, loop)
+ except BaseException:
+ release = getattr(sync_owner, "_release_deferred_event_handler", None)
+ if callable(release):
+ release(deferred_state)
+ release_dialog_claim()
+ raise
+ else:
+ _run_awaitable_on_loop(loop, result)
+ finally:
+ if not deferred and handler_raised:
+ release_dialog_claim()
loop = getattr(owner, "_loop", None)
if loop is not None and loop.is_running():
@@ -1427,11 +1569,16 @@ def run_on_loop() -> None:
release = getattr(sync_owner, "_release_deferred_event_handler", None)
if callable(release):
release(state)
+ release_dialog_claim()
raise
return
- result = call_handler()
- _run_awaitable(result)
+ try:
+ result = call_handler()
+ _run_awaitable(result)
+ except BaseException:
+ release_dialog_claim()
+ raise
wrappers.append((event, handler, wrapper))
return wrapper
@@ -2215,6 +2362,7 @@ async def new_page(
else:
context._finish_page_creation()
page._owns_context = True
+ page._rustwright_sync_close_default_context_cleaned = False
return _wrap_async_page(page)
return _wrap_async_page(await _run_sync_call(self._sync.new_page, **options))
@@ -2287,6 +2435,8 @@ async def new_context(
return _wrap_async_browser_context(await _run_sync_call(self._sync.new_context, **options))
async def close(self, *, reason: Optional[str] = None) -> None:
+ if _async_close_is_terminal(self._sync):
+ return
if not isinstance(self._sync, SyncBrowser):
await _single_flight_close(
self._sync,
@@ -2382,6 +2532,8 @@ async def new_page(self) -> "AsyncPage":
return _wrap_async_page(await _run_sync_call(self._sync.new_page))
async def close(self, *, reason: Optional[str] = None) -> None:
+ if _async_close_is_terminal(self._sync):
+ return
browser = getattr(self._sync, "_browser", None)
if getattr(self._sync, "_owns_browser", False) and browser is not None:
normalized_reason = None
@@ -2582,7 +2734,7 @@ async def _event_pump(self) -> None:
await _await_native(self._sync._event_stream.wait_batch_async(500.0, 64))
)
except asyncio.CancelledError:
- self._sync._event_stream.rollback_batch()
+ await _await_cleanup_completion(self._sync._event_stream.cancel_wait_async())
return
except RuntimeError as exc:
if "page event stream is already waiting" not in str(exc):
@@ -2597,7 +2749,9 @@ async def _event_pump(self) -> None:
try:
stream_closed = await _await_cleanup_completion(self._consume_event_batch(batch))
except asyncio.CancelledError:
+ await _await_cleanup_completion(self._sync._event_stream.cancel_wait_async())
return
+ self._sync._note_event_batch_consumed(stream_closed)
if stream_closed:
self._sync._stop_event_pump()
return
@@ -2833,6 +2987,11 @@ async def goto(
)
):
raise Error("Page.goto: Download is starting") from None
+ if self._sync._crashed or (
+ target.lower().startswith("chrome://crash")
+ and message.startswith("Page.goto: net::ERR_ABORTED")
+ ):
+ raise Error("Page crashed") from None
raise
response = (
None
@@ -3332,6 +3491,8 @@ async def remove_locator_handler(self, locator: Any) -> None:
await _run_sync_call(self._sync.remove_locator_handler, sync_locator)
async def close(self, *, run_before_unload: Optional[bool] = None, reason: Optional[str] = None) -> None:
+ if _async_close_is_terminal(self._sync):
+ return
if not isinstance(self._sync, SyncPage):
await _single_flight_close(
self._sync,
@@ -3342,10 +3503,6 @@ async def close(self, *, run_before_unload: Optional[bool] = None, reason: Optio
),
)
return
- if self._sync._closed or getattr(
- self._sync, "_rustwright_async_close_state", _CLOSE_OPEN
- ) == _CLOSE_CLOSED:
- return
if (
self._sync._video is not None
or self._sync._har_recordings
@@ -3373,56 +3530,53 @@ async def close(self, *, run_before_unload: Optional[bool] = None, reason: Optio
async def _close_native(self, normalized_reason: Optional[str], unload: bool) -> None:
self._sync._closed_reason = normalized_reason
- if self._sync._owns_context and self._sync._context is not None:
- await _run_sync_call(self._sync._context._cleanup_default_context_state)
- dialog_dispatch_count = self._sync._dialog_dispatch_count
+ first_error: Optional[BaseException] = None
+ native_attempted = self._sync._rustwright_sync_close_native_disposed
try:
- try:
- await _await_native_method(
- "Page.close",
- self._sync._core.close_async(self._sync._default_timeout, unload)
- )
- except Error as exc:
- if not _is_ignorable_close_error(exc):
- raise
- if unload and self._sync._event_handlers.get("dialog"):
- deadline = time.monotonic() + min(self._sync._default_timeout / 1000, 0.5)
- while (
- self._sync._dialog_dispatch_count == dialog_dispatch_count
- and time.monotonic() < deadline
+ await _run_sync_call(self._sync._purge_owned_default_context)
+ if not self._sync._rustwright_sync_close_native_disposed:
+ with self._sync._event_dispatch_condition:
+ beforeunload_dispatch_count = self._sync._beforeunload_dispatch_count
+ if not unload:
+ self._sync._closing = True
+ native_attempted = True
+ try:
+ await _await_native_method(
+ "Page.close",
+ self._sync._core.close_async(self._sync._default_timeout, unload),
+ )
+ except Error as exc:
+ if not _is_ignorable_close_error(exc):
+ raise
+ self._sync._rustwright_sync_close_native_disposed = True
+ if unload:
+ self._sync._rustwright_sync_close_beforeunload_count = beforeunload_dispatch_count
+ if self._sync._rustwright_sync_close_beforeunload_count is not None:
+ if not await _run_sync_call(
+ self._sync._wait_for_close_event_dispatch_or_fallback,
+ self._sync._rustwright_sync_close_beforeunload_count,
):
- await asyncio.sleep(0.01)
+ raise TimeoutError("Page.close: timed out waiting for beforeunload event dispatch; retry close")
+ self._sync._rustwright_sync_close_beforeunload_count = None
+ except BaseException as exc:
+ first_error = exc
finally:
- self._sync._stop_event_pump()
- pump_task = self._event_pump_task
- if pump_task is not None and pump_task is not asyncio.current_task():
- await asyncio.gather(pump_task, return_exceptions=True)
- self._sync._closed = True
- self._sync._closing = True
- self._sync._mark_owned_cdp_sessions_closed()
- if self._sync._context is not None and self._sync in self._sync._context._pages:
- self._sync._context._pages.remove(self._sync)
- _emit_event(self._sync._event_handlers, "close", self._sync)
- self._sync._release_memory_buffers()
- if (
- self._sync._owns_context
- and self._sync._context is not None
- and getattr(
- self._sync._context,
- "_rustwright_sync_close_state",
- _CLOSE_OPEN,
- )
- == _CLOSE_OPEN
- and getattr(
- self._sync._context,
- "_rustwright_async_close_state",
- _CLOSE_OPEN,
- )
- != _CLOSE_CLOSING
- ):
- await _wrap_async_browser_context(self._sync._context).close()
-
-
+ if native_attempted and self._sync._rustwright_sync_close_beforeunload_count is None:
+ try:
+ self._sync._stop_event_pump()
+ pump_task = self._event_pump_task
+ if pump_task is not None and pump_task is not asyncio.current_task():
+ await asyncio.gather(pump_task, return_exceptions=True)
+ except BaseException as exc:
+ if first_error is None:
+ first_error = exc
+ try:
+ await _run_sync_call(self._sync._run_close_cleanup)
+ except BaseException as exc:
+ if first_error is None:
+ first_error = exc
+ if first_error is not None:
+ raise first_error
class AsyncJSHandle(_AsyncJSHandleGeneratedMixin, _AsyncWrapper):
def __str__(self) -> str:
return str(self._sync)
diff --git a/python/rustwright/sync_api.py b/python/rustwright/sync_api.py
index 79a949e..af4df7a 100644
--- a/python/rustwright/sync_api.py
+++ b/python/rustwright/sync_api.py
@@ -1933,6 +1933,22 @@ def _update_sync_cleanup_complete(context: Any) -> None:
context._rustwright_sync_close_cleanup_complete = _is_context_cleanup_complete(context)
+def _is_page_cleanup_complete(page: Any) -> bool:
+ return all(
+ bool(getattr(page, name, False))
+ for name in (
+ "_rustwright_sync_close_sessions_closed",
+ "_rustwright_sync_close_default_context_cleaned",
+ "_rustwright_sync_close_context_removed",
+ "_rustwright_sync_close_close_event_emitted",
+ "_rustwright_sync_close_memory_released",
+ "_rustwright_sync_close_owned_context_closed",
+ )
+ )
+
+def _update_page_cleanup_complete(page: Any) -> None:
+ page._rustwright_sync_close_cleanup_complete = _is_page_cleanup_complete(page)
+
_CLOSE_WAIT_FALLBACK_MS = 30_000.0
@@ -3945,6 +3961,10 @@ def _emit_local_event(self, event: str, *args: Any) -> None:
"rustwright_event_dialog_dispatch",
default=None,
)
+_EVENT_DIALOG_CLAIM: contextvars.ContextVar[Any] = contextvars.ContextVar(
+ "rustwright_event_dialog_claim",
+ default=None,
+)
class _DialogDispatch:
@@ -3954,6 +3974,8 @@ class _DialogDispatch:
"remaining",
"finished",
"captured",
+ "handler_claims",
+ "next_claim",
"handled",
"fallback_started",
"lock",
@@ -3965,6 +3987,8 @@ def __init__(self, page: Any, dialog: Any, remaining: int) -> None:
self.remaining = remaining
self.finished = False
self.captured = False
+ self.handler_claims: set[int] = set()
+ self.next_claim = 0
self.handled = False
self.fallback_started = False
self.lock = threading.Lock()
@@ -3973,6 +3997,26 @@ def capture(self) -> None:
with self.lock:
self.captured = True
+ def release_capture(self) -> None:
+ with self.lock:
+ self.captured = False
+
+ def claim_handler(self) -> int:
+ with self.lock:
+ self.next_claim += 1
+ claim = self.next_claim
+ self.handler_claims.add(claim)
+ return claim
+
+ def release_handler_claim(self, claim: Optional[int]) -> None:
+ if claim is None:
+ return
+ with self.lock:
+ self.handler_claims.discard(claim)
+ finished = self.finished
+ if finished:
+ self.maybe_fallback()
+
def mark_handled(self) -> None:
with self.lock:
self.handled = True
@@ -4004,6 +4048,7 @@ def maybe_fallback(self) -> None:
if (
not self.finished
or self.captured
+ or self.handler_claims
or self.handled
or self.fallback_started
):
@@ -8988,6 +9033,7 @@ def _wait_for_descriptor_event_impl(
timeout,
method=method or descriptor.timeout_method,
)
+ event_state = state if state is not None else {}
if waiter is None:
waiter = _create_descriptor_waiter(target, descriptor, deadline, timeout_display)
if descriptor.competing_waiter_factory is not None:
@@ -8999,7 +9045,6 @@ def _wait_for_descriptor_event_impl(
and descriptor.competing_waiter_factory is None
else None
)
- event_state = state or {}
native_wait_matching = getattr(waiter, "wait_matching", None)
native_wait = kind in {"console", "pageerror"} and callable(native_wait_matching)
native_matcher = _native_event_matcher(kind, matcher) if native_wait else None
@@ -9309,6 +9354,9 @@ def __init__(
self._deadline: Optional[float] = None
self._timeout_display = ""
self._dialog_owner_release: Optional[Callable[[], None]] = None
+ self._state_lock = threading.Lock()
+ self._error: Optional[BaseException] = None
+ self._deadline_expired = False
def __enter__(self) -> "_ListenerEventContextManager":
descriptor = _event_waiter_descriptor(self._kind)
@@ -9322,16 +9370,43 @@ def __enter__(self) -> "_ListenerEventContextManager":
self._dialog_owner_release = self._page._acquire_dialog_waiter()
def handler(value: Any) -> None:
- if self._value is None and descriptor.matches(self._page, self._matcher, value):
- owner_release = self._dialog_owner_release
- self._dialog_owner_release = None
- if self._kind == "dialog" and isinstance(value, Dialog):
- self._page._capture_dialog(value, owner_release)
- elif owner_release is not None:
- owner_release()
- self._value = value
- self._ready.set()
-
+ try:
+ matched = descriptor.matches(self._page, self._matcher, value)
+ except BaseException as exc:
+ with self._state_lock:
+ if self._error is None and self._value is None and not self._deadline_expired:
+ self._error = exc
+ self._ready.set()
+ release = self._dialog_owner_release
+ self._dialog_owner_release = None
+ else:
+ release = self._dialog_owner_release
+ self._dialog_owner_release = None
+ if release is not None:
+ release()
+ return
+ if not matched:
+ return
+ with self._state_lock:
+ if self._value is not None or self._error is not None or self._deadline_expired:
+ release = self._dialog_owner_release
+ self._dialog_owner_release = None
+ capture = False
+ else:
+ self._value = value
+ release = self._dialog_owner_release
+ self._dialog_owner_release = None
+ capture = True
+ if self._kind == "dialog" and isinstance(value, Dialog):
+ self._page._capture_dialog(value, release)
+ release = None
+ self._ready.set()
+ if not capture and release is not None:
+ release()
+ elif capture and release is not None:
+ release()
+ if self._kind == "dialog":
+ setattr(handler, "_rustwright_dialog_predicate_waiter", True)
self._handler = handler
try:
self._page.on(descriptor.event, handler)
@@ -9356,21 +9431,35 @@ def __exit__(self, exc_type, exc, tb) -> None:
if self._deadline is None
else max((self._deadline - time.monotonic()) * 1000, 0)
)
- _wait_for_ready_event_or_owner_close(
- self._ready,
- self._page,
- descriptor.event,
- timeout_ms,
- self._timeout_display,
- deadline=self._deadline,
- )
+ try:
+ _wait_for_ready_event_or_owner_close(
+ self._ready,
+ self._page,
+ descriptor.event,
+ timeout_ms,
+ self._timeout_display,
+ deadline=self._deadline,
+ )
+ except BaseException:
+ with self._state_lock:
+ won = self._value is not None or self._error is not None
+ if not won:
+ self._deadline_expired = True
+ if not won:
+ raise
+ with self._state_lock:
+ error = self._error
+ if error is not None:
+ raise error
finally:
if self._handler is not None:
self._page.remove_listener(descriptor.event, self._handler)
self._handler = None
- if self._dialog_owner_release is not None:
- self._dialog_owner_release()
+ with self._state_lock:
+ release = self._dialog_owner_release
self._dialog_owner_release = None
+ if release is not None:
+ release()
@property
def value(self) -> Any:
@@ -10274,6 +10363,30 @@ def _capture(self, owner_release: Optional[Callable[[], None]] = None) -> None:
dispatch.capture()
if release_now:
owner_release()
+ def _claim_handler(self) -> Optional[int]:
+ dispatch = self._dispatch
+ return dispatch.claim_handler() if dispatch is not None else None
+
+ def _release_handler_claim(self, claim: Optional[int]) -> None:
+ dispatch = self._dispatch
+ if dispatch is not None:
+ dispatch.release_handler_claim(claim)
+
+ def _release_capture(self) -> None:
+ with self._fallback_lock:
+ self._captured = False
+ dispatch = self._dispatch
+ if dispatch is not None:
+ dispatch.release_capture()
+
+ def _release_capture_if_unowned(self) -> None:
+ with self._fallback_lock:
+ if self._handled or self._owner_releases or not self._captured:
+ return
+ self._captured = False
+ dispatch = self._dispatch
+ if dispatch is not None:
+ dispatch.release_capture()
def _mark_handled(self) -> None:
with self._fallback_lock:
@@ -11894,6 +12007,7 @@ def _new_page_from_options(self, options: Dict[str, Any]) -> "Page":
context.close()
raise
page._owns_context = True
+ page._rustwright_sync_close_default_context_cleaned = False
return page
def new_context(
@@ -16307,7 +16421,7 @@ def __init__(
self._auth_waiter: Any = None
self._proxy_auth_credentials: Optional[tuple[str, str]] = None
self._http_auth_credentials: Optional[tuple[str, str, Optional[str]]] = None
- self._dialog_dispatch_count = 0
+ self._beforeunload_dispatch_count = 0
self._dialog_state_lock = threading.Lock()
self._dialog_waiter_count = 0
self._dialog_operation_count = 0
@@ -16379,7 +16493,22 @@ def __init__(
self._clock_initialized: set[str] = set()
self._closing = False
self._closed = False
+ self._rustwright_sync_close_state = _SYNC_CLOSE_OPEN
+ self._rustwright_sync_close_condition = threading.Condition()
+ self._rustwright_sync_close_owner: Optional[int] = None
+ self._rustwright_sync_close_generation = 0
+ self._rustwright_sync_close_outcomes: dict[int, _CloseAttemptOutcome] = {}
+ self._rustwright_sync_close_waiters: dict[int, int] = {}
+ self._rustwright_sync_close_native_disposed = False
+ self._rustwright_sync_close_beforeunload_count: Optional[int] = None
+ self._rustwright_sync_close_close_event_emitted = False
+ self._rustwright_sync_close_default_context_cleaned = True
+ self._rustwright_sync_close_owned_context_closed = False
+ self._rustwright_sync_close_cleanup_complete = False
+ self._rustwright_sync_close_sessions_closed = False
+ self._rustwright_sync_close_context_removed = False
self._rustwright_async_close_state = "open"
+ self._rustwright_async_close_sync_owner: Optional[int] = None
self._rustwright_async_close_task: Any = None
self._closed_reason: Optional[str] = None
self._event_pump_stop_lock = threading.Lock()
@@ -16388,6 +16517,7 @@ def __init__(
self._event_dispatch_condition = threading.Condition(self._event_dispatch_lock)
self._event_handler_registrations: dict[tuple[str, int], _EventHandlerRegistration] = {}
self._event_stream = self._core.combined_event_stream()
+ self._event_pump_terminal_consumed = False
self._event_pump_thread: Optional[threading.Thread] = None
if _start_event_pump:
self._event_pump_thread = threading.Thread(
@@ -17313,9 +17443,6 @@ def goto(
"referer": normalized_referer,
},
)
- single_process_crash_navigation = (
- self._uses_single_process_fallback() and target_url.lower().startswith("chrome://crash")
- )
try:
try:
target_scheme = url_parse.urlparse(target_url).scheme.lower()
@@ -17357,9 +17484,11 @@ def goto(
and self._download_started_for_url(download_waiter, target_url, timeout=1_000.0)
):
raise Error("Page.goto: Download is starting") from None
- if target_url.lower().startswith("chrome://crash") and message.startswith("Page.goto: net::ERR_ABORTED"):
+ if self._crashed or (
+ target_url.lower().startswith("chrome://crash")
+ and message.startswith("Page.goto: net::ERR_ABORTED")
+ ):
crash_error = Error("Page crashed")
- self._mark_crashed()
raise crash_error from None
if message.startswith("Protocol error (Page.navigate):"):
raise Error(f"Page.goto: {message}") from None
@@ -17407,9 +17536,8 @@ def goto(
self._slow_mo()
except Exception as exc:
self._clear_context_pageload_pending()
- if single_process_crash_navigation:
+ if self._crashed:
crash_error = Error("Page crashed")
- self._mark_crashed()
self._trace_end_action(call_id, error=crash_error)
raise crash_error from exc
self._trace_end_action(call_id, error=exc)
@@ -20847,6 +20975,10 @@ def _mark_owned_cdp_sessions_closed(self) -> None:
for session in sessions:
session._mark_owner_closed()
+ def _remove_page_from_context(self) -> None:
+ if self._context is not None and self in self._context._pages:
+ self._context._pages.remove(self)
+
def bring_to_front(self) -> None:
self.evaluate("() => window.focus()")
@@ -21226,69 +21358,307 @@ def aria_snapshot(
depth=_normalize_aria_snapshot_depth(depth, method="Page.aria_snapshot"),
mode=_normalize_aria_snapshot_mode(mode, method="Page.aria_snapshot"),
)
+ def _acquire_sync_close_follower(self) -> Optional[int]:
+ current_thread = threading.get_ident()
+ with self._rustwright_sync_close_condition:
+ if self._rustwright_sync_close_state != _SYNC_CLOSE_CLOSING:
+ return None
+ if self._rustwright_sync_close_owner == current_thread:
+ return None
+ generation = self._rustwright_sync_close_generation
+ self._rustwright_sync_close_waiters[generation] = (
+ self._rustwright_sync_close_waiters.get(generation, 0) + 1
+ )
+ return generation
+
+ def _wait_for_sync_close_generation(self, generation: int) -> None:
+ deadline = _sync_close_wait_deadline(self)
+ with self._rustwright_sync_close_condition:
+ try:
+ while generation not in self._rustwright_sync_close_outcomes:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ _raise_close_wait_timeout("Page.close")
+ self._rustwright_sync_close_condition.wait(timeout=remaining)
+ outcome = self._rustwright_sync_close_outcomes[generation]
+ finally:
+ waiter_count = self._rustwright_sync_close_waiters.get(generation, 0)
+ if waiter_count <= 1:
+ self._rustwright_sync_close_waiters.pop(generation, None)
+ else:
+ self._rustwright_sync_close_waiters[generation] = waiter_count - 1
+ if outcome.error is not None:
+ _raise_close_error(outcome.error)
+
+ def _fallback_close_dialog(self) -> None:
+ self._maybe_fallback_pending_dialogs()
+
+
+ def _wait_for_sync_close_completion(self) -> None:
+ generation = self._acquire_sync_close_follower()
+ if generation is not None:
+ self._wait_for_sync_close_generation(generation)
+
+ def _begin_sync_close(self) -> bool:
+ current_thread = threading.get_ident()
+ deadline = _sync_close_wait_deadline(self)
+ with self._rustwright_sync_close_condition:
+ while (
+ self._rustwright_sync_close_state == _SYNC_CLOSE_CLOSING
+ or (
+ self._rustwright_async_close_state == "closing"
+ and self._rustwright_async_close_sync_owner != current_thread
+ )
+ ):
+ if (
+ self._rustwright_sync_close_state == _SYNC_CLOSE_CLOSING
+ and self._rustwright_sync_close_owner == current_thread
+ ):
+ return False
+ if self._rustwright_sync_close_state == _SYNC_CLOSE_CLOSING:
+ waited_generation = self._rustwright_sync_close_generation
+ self._rustwright_sync_close_waiters[waited_generation] = (
+ self._rustwright_sync_close_waiters.get(waited_generation, 0) + 1
+ )
+ try:
+ while waited_generation not in self._rustwright_sync_close_outcomes:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ _raise_close_wait_timeout("Page.close")
+ self._rustwright_sync_close_condition.wait(timeout=remaining)
+ outcome = self._rustwright_sync_close_outcomes[waited_generation]
+ finally:
+ waiter_count = self._rustwright_sync_close_waiters.get(waited_generation, 0)
+ if waiter_count <= 1:
+ self._rustwright_sync_close_waiters.pop(waited_generation, None)
+ else:
+ self._rustwright_sync_close_waiters[waited_generation] = waiter_count - 1
+ if outcome.error is not None:
+ _raise_close_error(outcome.error)
+ return False
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ _raise_close_wait_timeout("Page.close")
+ self._rustwright_sync_close_condition.wait(timeout=remaining)
+ if self._rustwright_sync_close_state == _SYNC_CLOSE_CLOSED:
+ outcome = self._rustwright_sync_close_outcomes.get(self._rustwright_sync_close_generation)
+ if outcome is not None and outcome.error is not None:
+ _raise_close_error(outcome.error)
+ return False
+ for generation in tuple(self._rustwright_sync_close_outcomes):
+ if not self._rustwright_sync_close_waiters.get(generation, 0):
+ self._rustwright_sync_close_outcomes.pop(generation, None)
+ self._rustwright_sync_close_state = _SYNC_CLOSE_CLOSING
+ self._rustwright_sync_close_owner = current_thread
+ self._rustwright_sync_close_generation += 1
+ return True
+ def _finish_sync_close_success(self) -> None:
+ with self._rustwright_sync_close_condition:
+ generation = self._rustwright_sync_close_generation
+ self._rustwright_sync_close_outcomes[generation] = _CloseAttemptOutcome(generation)
+ self._rustwright_sync_close_state = _SYNC_CLOSE_CLOSED
+ self._rustwright_sync_close_owner = None
+ self._rustwright_sync_close_condition.notify_all()
+
+ def _finish_sync_close_failure(self, error: BaseException) -> None:
+ with self._rustwright_sync_close_condition:
+ generation = self._rustwright_sync_close_generation
+ snapshot = _CloseErrorSnapshot.from_exception(error)
+ self._rustwright_sync_close_outcomes[generation] = _CloseAttemptOutcome(
+ generation, snapshot
+ )
+ terminal = (
+ getattr(self, "_rustwright_sync_close_native_disposed", False)
+ and getattr(self, "_rustwright_sync_close_cleanup_complete", False)
+ )
+ self._closed = terminal
+ self._closing = terminal
+ self._rustwright_sync_close_state = _SYNC_CLOSE_CLOSED if terminal else _SYNC_CLOSE_OPEN
+ self._rustwright_sync_close_owner = None
+ self._rustwright_sync_close_condition.notify_all()
+ def _note_event_batch_consumed(self, terminal: bool) -> None:
+ with self._event_dispatch_condition:
+ # Native cursors track reads; several drained batches can share one.
+ # Only consumption of the terminal envelope completes this barrier.
+ self._event_pump_terminal_consumed |= terminal
+ self._event_dispatch_condition.notify_all()
+
+ def _wait_for_close_event_dispatch_or_fallback(
+ self,
+ initial_beforeunload_count: Optional[int] = None,
+ ) -> bool:
+ deadline = _sync_close_wait_deadline(self)
+ with self._event_dispatch_condition:
+ while True:
+ if (
+ initial_beforeunload_count is not None
+ and self._beforeunload_dispatch_count != initial_beforeunload_count
+ ):
+ return True
+ if self._event_pump_terminal_consumed:
+ return True
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ self._fallback_close_dialog()
+ return False
+ self._event_dispatch_condition.wait(timeout=remaining)
+
+
def close(self, *, run_before_unload: Optional[bool] = None, reason: Optional[str] = None) -> None:
- if self._closed:
- return
+ with self._rustwright_sync_close_condition:
+ if self._rustwright_sync_close_state == _SYNC_CLOSE_CLOSED:
+ return
normalized_reason = None
if reason is not None:
normalized_reason = _normalize_string_option(reason, method="Page.close", name="reason")
- self._closed_reason = normalized_reason
- run_before_unload = bool(run_before_unload or False)
- if not run_before_unload:
- self._closing = True
- if self._video is not None:
- self._video._finalize()
- self._screencast.stop()
- for har_path, url_filter, content_mode, har_mode in self._har_recordings:
- _write_har(har_path, [self], url_filter, content_mode=content_mode, har_mode=har_mode)
- if self._fetch_enabled:
+ if not self._begin_sync_close():
+ return
+ try:
+ self._close_impl(
+ normalized_reason=normalized_reason,
+ run_before_unload=bool(run_before_unload or False),
+ )
+ except BaseException as exc:
+ self._finish_sync_close_failure(exc)
+ raise
+ else:
+ self._finish_sync_close_success()
+
+ def _purge_owned_default_context(self) -> None:
+ if self._rustwright_sync_close_default_context_cleaned:
+ return
+ if self._owns_context and self._context is not None:
+ self._context._cleanup_default_context_state()
+ self._rustwright_sync_close_default_context_cleaned = True
+
+ def _run_close_cleanup(self) -> None:
+ self._closed = False
+ self._closing = True
+ first_error: Optional[BaseException] = None
+
+ def run_cleanup(stage: Callable[[], None], marker: str) -> None:
+ nonlocal first_error
+ if getattr(self, marker, False):
+ return
try:
- _call(self._core.disable_fetch, self._default_timeout)
- except Error:
- pass
- self._fetch_enabled = False
- if self._binding_server is not None:
- self._binding_server.shutdown()
- self._binding_server.server_close()
- self._binding_server = None
- if self._crash_session is not None:
+ stage()
+ except BaseException as exc:
+ if first_error is None:
+ first_error = exc
+ return
+ setattr(self, marker, True)
+ _update_page_cleanup_complete(self)
+
+ run_cleanup(
+ self._mark_owned_cdp_sessions_closed,
+ "_rustwright_sync_close_sessions_closed",
+ )
+ run_cleanup(
+ self._remove_page_from_context,
+ "_rustwright_sync_close_context_removed",
+ )
+ if not self._rustwright_sync_close_close_event_emitted:
+ self._closed = True
try:
- self._crash_session.detach()
- except Error:
- pass
- self._crash_session = None
- self._crash_waiter = None
- dialog_dispatch_count = self._dialog_dispatch_count
+ _emit_event(self._event_handlers, "close", self)
+ except BaseException as exc:
+ first_error = first_error or exc
+ else:
+ self._rustwright_sync_close_close_event_emitted = True
+ _update_page_cleanup_complete(self)
+ run_cleanup(self._release_memory_buffers, "_rustwright_sync_close_memory_released")
+ if not self._rustwright_sync_close_owned_context_closed:
+ def close_owned_context() -> None:
+ if not self._owns_context:
+ return
+ if (
+ self._context is not None
+ and getattr(self._context, "_rustwright_sync_close_state", _SYNC_CLOSE_OPEN)
+ == _SYNC_CLOSE_OPEN
+ and getattr(self._context, "_rustwright_async_close_state", "open") != "closing"
+ ):
+ self._context.close()
+
+ run_cleanup(close_owned_context, "_rustwright_sync_close_owned_context_closed")
+ _update_page_cleanup_complete(self)
+ native_disposed = getattr(self, "_rustwright_sync_close_native_disposed", True)
+ if self._rustwright_sync_close_cleanup_complete and native_disposed:
+ self._closed = True
+ self._closing = True
+ else:
+ self._closed = False
+ self._closing = False
+ if first_error is not None:
+ raise first_error
+
+ def _close_impl(self, *, normalized_reason: Optional[str], run_before_unload: bool) -> None:
+ self._closed_reason = normalized_reason
+ first_error: Optional[BaseException] = None
+ native_attempted = self._rustwright_sync_close_native_disposed
try:
- try:
- _call(self._core.close, self._default_timeout, run_before_unload)
- except Error as exc:
- if not _is_ignorable_close_error(exc):
- raise
- if run_before_unload and self._event_handlers.get("dialog"):
- deadline = time.monotonic() + min(self._default_timeout / 1000, 0.5)
- while self._dialog_dispatch_count == dialog_dispatch_count and time.monotonic() < deadline:
- time.sleep(0.01)
+ if self._video is not None:
+ self._video._finalize()
+ self._screencast.stop()
+ for har_path, url_filter, content_mode, har_mode in self._har_recordings:
+ _write_har(har_path, [self], url_filter, content_mode=content_mode, har_mode=har_mode)
+ if self._fetch_enabled:
+ try:
+ _call(self._core.disable_fetch, self._default_timeout)
+ except Error:
+ pass
+ self._fetch_enabled = False
+ if self._binding_server is not None:
+ self._binding_server.shutdown()
+ self._binding_server.server_close()
+ self._binding_server = None
+ if self._crash_session is not None:
+ try:
+ self._crash_session.detach()
+ except Error:
+ pass
+ self._crash_session = None
+ self._crash_waiter = None
+ if not self._rustwright_sync_close_native_disposed:
+ self._purge_owned_default_context()
+ with self._event_dispatch_condition:
+ beforeunload_dispatch_count = self._beforeunload_dispatch_count
+ if not run_before_unload:
+ self._closing = True
+ native_attempted = True
+ try:
+ _call(self._core.close, self._default_timeout, run_before_unload)
+ except Error as exc:
+ if not _is_ignorable_close_error(exc):
+ raise
+ self._rustwright_sync_close_native_disposed = True
+ if run_before_unload:
+ self._rustwright_sync_close_beforeunload_count = beforeunload_dispatch_count
+ if self._rustwright_sync_close_beforeunload_count is not None:
+ if not self._wait_for_close_event_dispatch_or_fallback(
+ self._rustwright_sync_close_beforeunload_count
+ ):
+ raise TimeoutError("Page.close: timed out waiting for beforeunload event dispatch; retry close")
+ self._rustwright_sync_close_beforeunload_count = None
+ except BaseException as exc:
+ first_error = exc
finally:
- self._stop_event_pump()
- self._detach_worker_console_propagation(force=True)
- self._stop_worker_thread()
- self._reap_worker_event_threads()
- self._closed = True
- self._closing = True
- self._mark_owned_cdp_sessions_closed()
- if self._context is not None and self in self._context._pages:
- self._context._pages.remove(self)
- _emit_event(self._event_handlers, "close", self)
- self._release_memory_buffers()
- if (
- self._owns_context
- and self._context is not None
- and getattr(self._context, "_rustwright_sync_close_state", _SYNC_CLOSE_OPEN) == _SYNC_CLOSE_OPEN
- and getattr(self._context, "_rustwright_async_close_state", "open") != "closing"
- ):
- self._context.close()
+ if native_attempted and self._rustwright_sync_close_beforeunload_count is None:
+ try:
+ self._stop_event_pump()
+ self._detach_worker_console_propagation(force=True)
+ self._stop_worker_thread()
+ self._reap_worker_event_threads()
+ except BaseException as exc:
+ if first_error is None:
+ first_error = exc
+ try:
+ self._run_close_cleanup()
+ except BaseException as exc:
+ if first_error is None:
+ first_error = exc
+ if first_error is not None:
+ raise first_error
def get_by_text(self, text: str, *, exact: bool = False) -> "Locator":
return Locator(self, {"kind": "text", "text": _text_matcher(text), "exact": exact})
@@ -21435,7 +21805,6 @@ def _finish_event_handler(
dispatch = _EVENT_DIALOG_DISPATCH.get()
if isinstance(dispatch, _DialogDispatch):
dispatch.settle()
-
def _wait_for_event_handler_removal(
self,
event: str,
@@ -21497,10 +21866,7 @@ def _listener_registration_cursor(self, event: str) -> Optional[int]:
return None
def _dialog_fallback_suppressed(self, dialog: Dialog) -> bool:
- if getattr(dialog, "_captured", False):
- return True
- with self._dialog_state_lock:
- return self._dialog_waiter_count > 0 or self._dialog_operation_count > 0
+ return bool(getattr(dialog, "_captured", False))
def _dialog_dispatch_finished(self, dispatch: _DialogDispatch) -> None:
if getattr(dispatch.dialog, "_handled", False):
@@ -21595,20 +21961,31 @@ def _dispatch_event_handlers(
dialog_dispatch.settle()
continue
activated = True
- if event == "dialog" and args and self._dialog_operation_active():
- capture = getattr(args[0], "_capture", None)
- if callable(capture):
- capture()
sequence_token = _EVENT_DISPATCH_SEQUENCE.set((self, event_sequence))
owner_token = _EVENT_DISPATCH_OWNER.set(owner)
registration_token = _EVENT_DISPATCH_REGISTRATION.set(state)
deferred_token = _EVENT_DISPATCH_DEFERRED.set(False)
dialog_token = _EVENT_DIALOG_DISPATCH.set(dialog_dispatch)
+ claim: Optional[int] = None
+ if (
+ event == "dialog"
+ and args
+ and not getattr(state.handler, "_rustwright_dialog_predicate_waiter", False)
+ ):
+ claim_handler = getattr(args[0], "_claim_handler", None)
+ if callable(claim_handler):
+ claim = claim_handler()
+ claim_token = _EVENT_DIALOG_CLAIM.set(claim)
+ handler_raised = False
try:
try:
state.handler(*args)
except Exception:
+ handler_raised = True
continue
+ except BaseException:
+ handler_raised = True
+ raise
finally:
deferred = _EVENT_DISPATCH_DEFERRED.get()
_EVENT_DISPATCH_DEFERRED.reset(deferred_token)
@@ -21616,12 +21993,16 @@ def _dispatch_event_handlers(
_EVENT_DISPATCH_OWNER.reset(owner_token)
_EVENT_DISPATCH_SEQUENCE.reset(sequence_token)
if not deferred:
+ if handler_raised:
+ release_claim = getattr(args[0], "_release_handler_claim", None) if args else None
+ if callable(release_claim):
+ release_claim(claim)
self._finish_event_handler(state, owner)
+ _EVENT_DIALOG_CLAIM.reset(claim_token)
_EVENT_DIALOG_DISPATCH.reset(dialog_token)
if dialog_dispatch is not None:
dialog_dispatch.finish_if_idle()
return activated
-
def _dismiss_dialog_if_unhandled(self, dialog: Dialog) -> None:
with dialog._fallback_lock:
if dialog._handled or dialog._fallback_attempted:
@@ -21639,7 +22020,10 @@ def _dismiss_dialog_if_unhandled(self, dialog: Dialog) -> None:
def _handle_dialog_event(self, payload: dict[str, Any]) -> None:
dialog = Dialog(self, payload)
self._dispatch_event_handlers("dialog", dialog)
- self._dialog_dispatch_count += 1
+ with self._event_dispatch_condition:
+ if dialog.type == "beforeunload":
+ self._beforeunload_dispatch_count += 1
+ self._event_dispatch_condition.notify_all()
def _snapshot_event_handlers(
self,
@@ -21762,6 +22146,16 @@ def remove_listener(self, event: str, f: Callable[..., Any]) -> None:
if detach_worker_forwards:
self._detach_worker_console_propagation()
+ def _ack_event_batch(self) -> bool:
+ ack_batch = getattr(self._event_stream, "ack_batch", None)
+ if not callable(ack_batch):
+ return True
+ try:
+ _call(ack_batch)
+ except Error:
+ return False
+ return True
+
def _event_pump(self) -> None:
while self._event_listeners_active():
try:
@@ -21769,13 +22163,19 @@ def _event_pump(self) -> None:
except Error:
break
if not isinstance(batch, list):
+ if not self._ack_event_batch():
+ break
continue
+ stop_after_ack = False
+ terminal_consumed = False
for envelope in batch:
if not isinstance(envelope, dict):
continue
kind = str(envelope.get("kind") or "")
if kind == "_closed":
- return
+ terminal_consumed = True
+ stop_after_ack = True
+ break
if kind == "_overflow":
self._reconcile_event_stream_overflow(envelope.get("payload"))
continue
@@ -21790,7 +22190,13 @@ def _event_pump(self) -> None:
event_sequence=event_sequence,
)
if not self._event_listeners_active():
- return
+ stop_after_ack = True
+ break
+ if not self._ack_event_batch():
+ break
+ self._note_event_batch_consumed(terminal_consumed)
+ if stop_after_ack:
+ return
def _reconcile_event_stream_overflow(self, payload: Any) -> None:
dropped = payload.get("dropped") if isinstance(payload, dict) else None
@@ -21800,7 +22206,7 @@ def _reconcile_event_stream_overflow(self, payload: Any) -> None:
dropped_count = 0
warnings.warn(
f"rustwright page event stream overflow: dropped {dropped_count} event(s); "
- "network-idle and dialog bookkeeping was resynchronized",
+ "network-idle bookkeeping was resynchronized",
RuntimeWarning,
stacklevel=2,
)
@@ -21808,10 +22214,7 @@ def _reconcile_event_stream_overflow(self, payload: Any) -> None:
self._network_idle_active_requests.clear()
self._network_idle_last_activity = time.monotonic()
self._network_idle_condition.notify_all()
- try:
- _call(self._core.handle_dialog, False, None, min(self._default_timeout, 500.0))
- except Error:
- pass
+ # Retained dialog openings follow overflow and use ownership-aware dispatch.
def _handle_observation_event(
self,
diff --git a/src/lib.rs b/src/lib.rs
index 9eddf5d..45c26b6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -266,16 +266,27 @@ impl CloseOutcome {
enum ClosePhase {
Open,
- Closing(watch::Sender