From f2dd189cf9f26aac03389d028306ef1d95189a48 Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sat, 26 Sep 2026 11:18:58 +0800 Subject: [PATCH] fix(sandbox): retain AIO flock across cancellation (#5820) * fix(sandbox): retain AIO flock across cancellation * fix(sandbox): drain async create before flock release * test(sandbox): cover cancellation during AIO create * docs(sandbox): pin create drain ownership * fix(sandbox): keep flock waits cancellable * test(sandbox): pin AIO flock lifecycle drains * docs(sandbox): document cancellable flock wait * test(sandbox): follow nonblocking flock helper * test(sandbox): update create cancellation lock mock * style(sandbox): sort lifecycle imports * test(sandbox): assert cancelled flock wait closes file * docs(sandbox): trim cancellation guidance below chain limit --------- Co-authored-by: Willem Jiang --- .../aio_sandbox/aio_sandbox_provider.py | 63 +++- .../harness/deerflow/sandbox/AGENTS.md | 2 +- backend/tests/test_aio_sandbox_provider.py | 326 ++++++++++++++++++ 3 files changed, 372 insertions(+), 19 deletions(-) diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index 61d9378c4b..528c5b0839 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -13,6 +13,7 @@ The provider itself handles: import asyncio import atexit import contextlib +import errno import hashlib import logging import math @@ -46,9 +47,11 @@ from deerflow.integrations.lark_cli import LARK_CLI_SANDBOX_CONFIG_DIR, LARK_CLI from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.acquire_serialization import AcquireSerializer from deerflow.sandbox.identity import derive_sandbox_scope_token +from deerflow.sandbox.lease import run_sync_lifecycle_operation from deerflow.sandbox.sandbox import Sandbox from deerflow.sandbox.sandbox_provider import SandboxProvider from deerflow.skills.types import SkillCategory +from deerflow.utils.file_io import await_drained from .aio_sandbox import AioSandbox from .backend import SANDBOX_LOCAL_PROVIDER_READY_TIMEOUT, SandboxBackend, wait_for_sandbox_ready, wait_for_sandbox_ready_async @@ -123,6 +126,21 @@ def _lock_file_exclusive(lock_file) -> None: msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) +def _try_lock_file_exclusive(lock_file) -> bool: + """Attempt the cross-process lock once without blocking the worker thread.""" + try: + if fcntl is not None: + fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + else: # pragma: no cover - Windows + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1) + except OSError as error: + if error.errno in (errno.EACCES, errno.EAGAIN, errno.EDEADLK): + return False + raise + return True + + def _unlock_file(lock_file) -> None: if fcntl is not None: fcntl.flock(lock_file, fcntl.LOCK_UN) @@ -2187,20 +2205,22 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): lock_file = await asyncio.to_thread(_open_lock_file, lock_path) locked = False try: - await asyncio.to_thread(_lock_file_exclusive, lock_file) - locked = True + while not locked: + locked = await run_sync_lifecycle_operation(_try_lock_file_exclusive, lock_file) + if not locked: + await asyncio.sleep(0.02) # Re-check in-process caches under the file lock in case another # thread in this process won the race while we were waiting. - cached_id = await asyncio.to_thread(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id) + cached_id = await run_sync_lifecycle_operation(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id) if cached_id is not None: return cached_id # Backend discovery is sync because local discovery may inspect # Docker and perform a health check; keep it off the event loop. - discovered = await asyncio.to_thread(self._backend.discover, sandbox_id) + discovered = await run_sync_lifecycle_operation(self._backend.discover, sandbox_id) if discovered is not None: if discovered.requires_replacement: - replaced = await asyncio.to_thread( + replaced = await run_sync_lifecycle_operation( self._replace_incompatible_sandbox, discovered, time.time(), @@ -2211,13 +2231,20 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # Registration publishes ownership, which is blocking store # IO (filesystem or network depending on the backend) — same # reason every other step in this coroutine is offloaded. - return await asyncio.to_thread(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id) + return await run_sync_lifecycle_operation(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id) - return await self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id) + # Keep the entire async create lifecycle under the cross-process flock. + # Shielding only individual to_thread workers is insufficient: cancellation + # after backend.create() would otherwise discard SandboxInfo before readiness + # and registration/cleanup run, then finally release the flock over an + # unregistered deterministic container. + return await await_drained(self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id)) finally: - if locked: - await asyncio.to_thread(_unlock_file, lock_file) - await asyncio.to_thread(lock_file.close) + try: + if locked: + await run_sync_lifecycle_operation(_unlock_file, lock_file) + finally: + await run_sync_lifecycle_operation(lock_file.close) def _destroy_unready_sandbox(self, sandbox_id: str, info: SandboxInfo) -> None: """Tear down a freshly-created container whose readiness check failed. @@ -2333,10 +2360,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): async def _create_sandbox_async(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str: """Async counterpart to ``_create_sandbox``.""" effective_user_id = self._effective_acquire_user_id(user_id) - extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id, user_id=effective_user_id) - provision_lark_cli_runtime = await asyncio.to_thread(self._lark_integration_active, effective_user_id) - provision_lark_cli_broker = await asyncio.to_thread(self._lark_broker_active, effective_user_id) - config_mount_exclusion_root = await asyncio.to_thread( + extra_mounts = await run_sync_lifecycle_operation(self._get_extra_mounts, thread_id, user_id=effective_user_id) + provision_lark_cli_runtime = await run_sync_lifecycle_operation(self._lark_integration_active, effective_user_id) + provision_lark_cli_broker = await run_sync_lifecycle_operation(self._lark_broker_active, effective_user_id) + config_mount_exclusion_root = await run_sync_lifecycle_operation( self._local_config_mount_exclusion_root, thread_id, user_id=effective_user_id, @@ -2346,7 +2373,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # Active sandboxes are in use by live threads and must not be forcibly stopped. replicas, total = self._replica_count() if total >= replicas: - evicted = await asyncio.to_thread(self._evict_oldest_warm) + evicted = await run_sync_lifecycle_operation(self._evict_oldest_warm) self._log_replicas_soft_cap(replicas, sandbox_id, evicted) create_kwargs = {} @@ -2354,7 +2381,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): create_kwargs["config_mount_exclusion_root"] = config_mount_exclusion_root if isinstance(self._backend, RemoteSandboxBackend): create_kwargs["skills_container_path"] = self._configured_skills_container_path() - info = await asyncio.to_thread( + info = await run_sync_lifecycle_operation( self._backend.create, thread_id, sandbox_id, @@ -2376,12 +2403,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # ``_register_created_sandbox`` after this gate. Claim the teardown # lease before stopping it so a peer cannot adopt the not-yet-ready # Pod in the meantime (#4248). - await asyncio.to_thread(self._destroy_unready_sandbox, sandbox_id, info) + await run_sync_lifecycle_operation(self._destroy_unready_sandbox, sandbox_id, info) raise RuntimeError(f"Sandbox {sandbox_id} failed to become ready within timeout at {info.sandbox_url}") # Registration publishes ownership (blocking store IO), so it is offloaded # like every other blocking step on this path. - return await asyncio.to_thread(self._register_created_sandbox, thread_id, sandbox_id, info, user_id=effective_user_id) + return await run_sync_lifecycle_operation(self._register_created_sandbox, thread_id, sandbox_id, info, user_id=effective_user_id) def get(self, sandbox_id: str) -> Sandbox | None: """Get a sandbox by ID. Updates last activity timestamp. diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 8fee9d4eee..9646cc3f5b 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -19,7 +19,7 @@ to respond. **Implementations**: - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-user/thread `LocalSandbox` (id `local:{user_id}:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Shared runs use category mappings; a policy-scoped run replaces them with one `/mnt/skills` root mapping to the coherent thread view, so structured file tools resolve through one managed boundary. This is not a host filesystem security boundary: an enabled host `bash` subprocess can use canonical paths without `PathMapping`, so `supports_agent_skill_isolation` is dynamic and explicit Agent policies fail closed while host bash is enabled. Host-to-virtual output masking scans dynamic per-user/per-thread roots directly instead of compiling path-specific regexes, so evicted thread IDs do not remain in Python's global regex caches; a separate 256-entry root cache prevents repeated `realpath()` walks for every glob/grep match while bounding dynamic-path retention, and only the small process-stable skill/integration source set uses a bounded compiled cache. The shared `path_patterns.py` tail stops at `:`, so `$PATH`-style lists mask every entry; a mount symlink resolving outside all mounts keeps its mount path. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes; do not restore a blanket conversion disable, because host-native CLI launchers need normal MSYS path conversion for their own installation paths. PowerShell fallback pairs guarded console UTF-8 setters with explicit UTF-8 pipe decoding; cmd/MSYS retain locale decoding. Encoding regressions include real-pipe probes and Windows-only PowerShell roundtrips with and without a console. -- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker isolation. Reuse enforces minimum persisted shell capacity even without image-env overrides. Upgrades require ownership-fenced teardown; provisioner create returns 409 for undersized Pods without deleting them. Acquire/reuse checks active-cache and warm-pool entries with the backend, dropping definitively dead containers from all in-process maps before rediscovery/creation. Health-check failures mean unknown, not dead; local discovery skips unverifiable containers and tries creation. Acquire/reclaim publishes ownership; `_renew_owned_leases` refreshes it on a dedicated thread. Text appends use native AIO append, never read-modify-write that could overwrite after a failed read. `reset()` closes the acquire serializer and its workers; `shutdown()` performs full remote teardown. `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner=False); `sandbox.thread_data_mounts` overrides this for deployments sharing thread user-data directories. `true` skips upload acquire/sync; a false positive makes uploads unavailable. Explicit Agent policies use four thread projection category mounts and a distinct deterministic identity, excluding older shared-mount containers. The startup snapshot of `skills.container_path` drives mounts, identity, remote requests, and provisioner validation; root changes prevent container/Pod reuse. Gateway and provisioner independently require a canonical absolute root outside reserved platform mounts and derive four category allowlist entries from it. All four category overrides suppress the provisioner's default hostPath/skills-PVC mount; with `USERDATA_PVC_NAME`, categories use subpaths on that shared PVC. Readiness probes and `agent_sandbox` clients set `trust_env=False` for direct control-plane destinations (loopback/private IPs, single-label cluster hosts, Docker/Podman internal hostnames); external FQDNs and public IPs retain environment proxies. +- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker isolation. Reuse enforces minimum persisted shell capacity even without image-env overrides. Upgrades require ownership-fenced teardown; provisioner create returns 409 for undersized Pods without deleting them. Acquire/reuse checks active-cache and warm-pool entries with the backend, dropping definitively dead containers from all in-process maps before rediscovery/creation. Health-check failures mean unknown, not dead; local discovery skips unverifiable containers and tries creation. Acquire/reclaim publishes ownership; `_renew_owned_leases` refreshes it on a dedicated thread. Text appends use native AIO append, never read-modify-write that could overwrite after a failed read. `reset()` closes the acquire serializer and its workers; `shutdown()` performs full remote teardown. Async create polls flock acquisition so cancelled waiters leave promptly. Once locked, cancellation drains admitted workers and create readiness/registration/cleanup before unlock/close. `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner=False); `sandbox.thread_data_mounts` overrides this for deployments sharing thread user-data directories. `true` skips upload acquire/sync; a false positive makes uploads unavailable. Explicit Agent policies use four thread projection category mounts and a distinct deterministic identity, excluding older shared-mount containers. The startup snapshot of `skills.container_path` drives mounts, identity, remote requests, and provisioner validation; root changes prevent container/Pod reuse. Gateway and provisioner independently require a canonical absolute root outside reserved platform mounts and derive four category allowlist entries from it. All four category overrides suppress the provisioner's default hostPath/skills-PVC mount; with `USERDATA_PVC_NAME`, categories use subpaths on that shared PVC. Readiness probes and `agent_sandbox` clients set `trust_env=False` for direct control-plane destinations (loopback/private IPs, single-label cluster hosts, Docker/Podman internal hostnames); external FQDNs and public IPs retain environment proxies. - `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) provides E2B remote isolation. New unrestricted sandboxes receive a one-shot upload from the enabled-only public, custom, legacy, and managed integration projections. For a thread diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index 2e2b33d95e..9c3b23d8bf 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -937,6 +937,332 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl assert any(getattr(func, "__name__", "") == "close" for func in to_thread_calls) +@pytest.mark.anyio +async def test_discover_or_create_with_lock_async_cancellation_aborts_flock_wait(tmp_path, monkeypatch): + """A cancelled waiter must not park until a peer releases the cross-process flock.""" + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + provider = _make_provider(tmp_path) + provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__( + provider, + aio_mod.AioSandboxProvider, + ) + provider._backend = SimpleNamespace(discover=MagicMock(return_value=None)) + + peer_lock = threading.Lock() + peer_lock.acquire() + attempt_seen = threading.Event() + closed = threading.Event() + + class TrackedLockFile: + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, name): + return getattr(self._inner, name) + + def close(self): + self._inner.close() + closed.set() + + def open_lock_file(lock_path): + return TrackedLockFile(open(lock_path, "a", encoding="utf-8")) + + def try_lock(_lock_file) -> bool: + attempt_seen.set() + return peer_lock.acquire(blocking=False) + + monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(aio_mod, "_open_lock_file", open_lock_file) + monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", try_lock) + + owner = asyncio.create_task( + provider._discover_or_create_with_lock_async( + "thread-cancel-flock-wait", + "sandbox-cancel-flock-wait", + user_id="default", + ) + ) + try: + assert await asyncio.to_thread(attempt_seen.wait, 2) + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(owner, timeout=0.5) + assert closed.is_set() + provider._backend.discover.assert_not_called() + finally: + peer_lock.release() + if not owner.done(): + owner.cancel() + await asyncio.gather(owner, return_exceptions=True) + + +@pytest.mark.anyio +async def test_discover_or_create_with_lock_async_cancellation_keeps_file_lock_until_worker_finishes(tmp_path, monkeypatch): + """Caller cancellation must not release the cross-process lock over an admitted worker.""" + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + provider = _make_provider(tmp_path) + provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__( + provider, + aio_mod.AioSandboxProvider, + ) + provider._warm_pool = {} + provider._sandbox_infos = {} + provider._thread_sandboxes = {} + provider._sandboxes = {} + provider._last_activity = {} + provider._lock = aio_mod.threading.Lock() + + discover_started = threading.Event() + allow_discover = threading.Event() + contender_acquired = threading.Event() + release_contender = threading.Event() + cross_process_lock = threading.Lock() + + def blocking_discover(_sandbox_id: str): + discover_started.set() + assert allow_discover.wait(timeout=2) + return None + + create_calls: list[tuple[str | None, str, str | None]] = [] + + async def fake_create(thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str: + create_calls.append((thread_id, sandbox_id, user_id)) + return sandbox_id + + def contend_for_lock() -> None: + cross_process_lock.acquire() + try: + contender_acquired.set() + assert release_contender.wait(timeout=2) + finally: + cross_process_lock.release() + + provider._backend = SimpleNamespace(discover=blocking_discover) + monkeypatch.setattr(provider, "_create_sandbox_async", fake_create) + monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False)) + monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release()) + + owner = asyncio.create_task( + provider._discover_or_create_with_lock_async( + "thread-cancel-flock", + "sandbox-cancel-flock", + user_id="default", + ) + ) + contender = None + try: + assert await asyncio.to_thread(discover_started.wait, 2) + owner.cancel() + contender = asyncio.create_task(asyncio.to_thread(contend_for_lock)) + + await asyncio.sleep(0.05) + assert not contender_acquired.is_set(), "cross-process lock released while discover worker was still running" + + allow_discover.set() + with pytest.raises(asyncio.CancelledError): + await owner + + assert await asyncio.to_thread(contender_acquired.wait, 2) + assert create_calls == [] + finally: + allow_discover.set() + release_contender.set() + if not owner.done(): + owner.cancel() + await asyncio.gather(owner, return_exceptions=True) + if contender is not None: + await contender + + +@pytest.mark.anyio +async def test_discover_or_create_with_lock_async_cancellation_drains_create_before_unlock(tmp_path, monkeypatch): + """Cancellation during create must finish the async lifecycle before releasing the flock.""" + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + provider = _make_provider(tmp_path) + provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__( + provider, + aio_mod.AioSandboxProvider, + ) + provider._warm_pool = {} + provider._sandbox_infos = {} + provider._thread_sandboxes = {} + provider._sandboxes = {} + provider._last_activity = {} + provider._lock = aio_mod.threading.Lock() + + create_started = asyncio.Event() + allow_create = asyncio.Event() + create_finished = asyncio.Event() + contender_acquired = threading.Event() + release_contender = threading.Event() + cross_process_lock = threading.Lock() + + async def fake_create(thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str: + assert thread_id == "thread-cancel-create" + assert sandbox_id == "sandbox-cancel-create" + assert user_id == "default" + create_started.set() + await allow_create.wait() + create_finished.set() + return sandbox_id + + def contend_for_lock() -> None: + cross_process_lock.acquire() + try: + contender_acquired.set() + assert release_contender.wait(timeout=2) + finally: + cross_process_lock.release() + + provider._backend = SimpleNamespace(discover=MagicMock(return_value=None)) + monkeypatch.setattr(provider, "_create_sandbox_async", fake_create) + monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False)) + monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release()) + + owner = asyncio.create_task( + provider._discover_or_create_with_lock_async( + "thread-cancel-create", + "sandbox-cancel-create", + user_id="default", + ) + ) + contender = None + try: + await asyncio.wait_for(create_started.wait(), timeout=2) + owner.cancel() + contender = asyncio.create_task(asyncio.to_thread(contend_for_lock)) + + await asyncio.sleep(0.05) + assert not contender_acquired.is_set(), "flock released before the async create lifecycle finished" + + allow_create.set() + with pytest.raises(asyncio.CancelledError): + await owner + + assert create_finished.is_set() + assert await asyncio.to_thread(contender_acquired.wait, 2) + finally: + allow_create.set() + release_contender.set() + if not owner.done(): + owner.cancel() + await asyncio.gather(owner, return_exceptions=True) + if contender is not None: + await contender + + +@pytest.mark.anyio +@pytest.mark.parametrize("blocked_step", ["create", "register"]) +async def test_discover_or_create_with_lock_async_cancellation_drains_create_lifecycle_steps( + blocked_step, + tmp_path, + monkeypatch, +): + """Create/register workers must settle before cancellation releases the flock.""" + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + provider = _make_provider(tmp_path) + provider._discover_or_create_with_lock_async = aio_mod.AioSandboxProvider._discover_or_create_with_lock_async.__get__( + provider, + aio_mod.AioSandboxProvider, + ) + provider._warm_pool = {} + provider._sandbox_infos = {} + provider._thread_sandboxes = {} + provider._sandboxes = {} + provider._last_activity = {} + provider._lock = aio_mod.threading.Lock() + provider._config = {"command_timeout": 600.0, "idle_timeout": 600, "replicas": 3} + + step_started = threading.Event() + allow_step = threading.Event() + step_finished = threading.Event() + register_called = threading.Event() + contender_acquired = threading.Event() + release_contender = threading.Event() + cross_process_lock = threading.Lock() + + info = aio_mod.SandboxInfo( + sandbox_id="sandbox-cancel-lifecycle", + sandbox_url="http://sandbox", + ) + + def create(*_args, **_kwargs): + if blocked_step == "create": + step_started.set() + assert allow_step.wait(timeout=2) + step_finished.set() + return info + + def register(*_args, **_kwargs): + register_called.set() + if blocked_step == "register": + step_started.set() + assert allow_step.wait(timeout=2) + step_finished.set() + return info.sandbox_id + + async def ready(*_args, **_kwargs): + return True + + def contend_for_lock() -> None: + cross_process_lock.acquire() + try: + contender_acquired.set() + assert release_contender.wait(timeout=2) + finally: + cross_process_lock.release() + + provider._backend = SimpleNamespace( + create=create, + destroy=MagicMock(), + discover=MagicMock(return_value=None), + ) + monkeypatch.setattr(provider, "_get_extra_mounts", lambda *_args, **_kwargs: []) + monkeypatch.setattr(provider, "_lark_integration_active", lambda *_args, **_kwargs: False) + monkeypatch.setattr(provider, "_lark_broker_active", lambda *_args, **_kwargs: False) + monkeypatch.setattr(provider, "_local_config_mount_exclusion_root", lambda *_args, **_kwargs: None) + monkeypatch.setattr(provider, "_replica_count", lambda: (3, 0)) + monkeypatch.setattr(provider, "_register_created_sandbox", register) + monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready_async", ready) + monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(aio_mod, "_try_lock_file_exclusive", lambda _lock_file: cross_process_lock.acquire(blocking=False)) + monkeypatch.setattr(aio_mod, "_unlock_file", lambda _lock_file: cross_process_lock.release()) + + owner = asyncio.create_task( + provider._discover_or_create_with_lock_async( + "thread-cancel-lifecycle", + info.sandbox_id, + user_id="default", + ) + ) + contender = None + try: + assert await asyncio.to_thread(step_started.wait, 2) + owner.cancel() + contender = asyncio.create_task(asyncio.to_thread(contend_for_lock)) + + await asyncio.sleep(0.05) + assert not contender_acquired.is_set() + + allow_step.set() + with pytest.raises(asyncio.CancelledError): + await owner + + assert step_finished.is_set() + assert register_called.is_set() + assert await asyncio.to_thread(contender_acquired.wait, 2) + finally: + allow_step.set() + release_contender.set() + if not owner.done(): + owner.cancel() + await asyncio.gather(owner, return_exceptions=True) + if contender is not None: + await contender + + @pytest.mark.anyio async def test_acquire_async_lock_wait_uses_dedicated_executor(tmp_path, monkeypatch): """Per-thread lock waits should not consume the default asyncio.to_thread pool."""