fix(lark): retry the Windows install lock instead of failing after ~10s (#6881)

* fix(lark): retry the Windows install lock instead of failing after ~10s

msvcrt.LK_LOCK gives up after roughly ten seconds and surfaces contention
as Permission denied, while a real install (download + setup of the
lark/sandbox CLI) can hold the lock far longer. Retry the non-blocking
acquisition on the Windows fallback branch so concurrent Gateway workers
serialize instead of failing, mirroring the extensions manager's lock
introduced in #6020. The POSIX flock branch is unchanged.

Closes #6879

* test(lark): inject msvcrt stub with raising=False for Linux runners

The regression test patches the module-level ``msvcrt`` attribute, which only
exists on Windows (the POSIX branch imports ``fcntl`` instead), so
``monkeypatch.setattr`` with the default ``raising=True`` raised AttributeError
on Linux CI (backend-unit-tests shard 4).

Pass ``raising=False`` so the stub is injected on POSIX runners: ``fcntl`` is
patched to ``None`` first, and ``_exclusive_install_lock`` then takes the
Windows fallback branch against the fake. Windows behavior is unchanged.

Scope: backend/tests/test_lark_cli_install_lock.py, 1 line.
Verification: pytest tests/test_lark_cli_install_lock.py on Windows
(Python 3.12, pytest 9.0.3) -- 3 passed incl. the real-contention case;
Linux CI rerun is the authoritative check.

* fix(lark): harden and share advisory lock retries

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
sjr666666 2026-10-15 08:03:10 +08:00 committed by GitHub
parent a63856f690
commit 2fe325f8ae
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 424 additions and 43 deletions

View File

@ -1459,6 +1459,8 @@ OAuth data remain isolated under
`{DEER_FLOW_HOME}/users/{user_id}/integrations/lark-cli/{config,data}`. These
secret directories are restricted to `0700`, regular credential files to
`0600`, and symlinks are rejected.
On Windows, waiting for another process's integration file lock retries for up
to 15 minutes; other file-lock errors are reported immediately.
After installation, users can click **Connect Lark** to open a browser
authorization link; no terminal authorization is required. The same UI can

View File

@ -13,10 +13,9 @@ import stat
import subprocess
import sys
import tempfile
import time
import tomllib
import urllib.parse
from collections.abc import Callable, Iterator
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
@ -25,10 +24,11 @@ from typing import Any
import yaml
from packaging.requirements import InvalidRequirement, Requirement
from deerflow.utils.file_lock import retry_until_locked as _retry_until_locked
logger = logging.getLogger(__name__)
_ENTRY_POINT_GROUP = "deerflow.extensions"
_LOCK_RETRY_INTERVAL_SECONDS = 0.2
_SNAPSHOT_IGNORES = (".git", ".venv", "venv", "__pycache__", "*.pyc")
_DISTRIBUTION_NAME = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._-]*[A-Za-z0-9])?$")
_SENSITIVE_FILENAMES = {".npmrc", ".pypirc", "credentials.json"}
@ -531,22 +531,6 @@ def _normalize_distribution(name: str) -> str:
return re.sub(r"[-_.]+", "-", name).lower()
def _retry_until_locked(acquire: Callable[[], None], *, sleep: Callable[[float], None] = time.sleep) -> None:
"""Retry a non-blocking lock acquisition until the region is free.
Windows' blocking mode (``msvcrt.LK_LOCK``) gives up after roughly ten
seconds. A real install holds this lock across ``uv add`` plus a full
``uv sync``, so blocking mode reports contention as ``Permission denied``
instead of serializing the two operations.
"""
while True:
try:
acquire()
return
except OSError:
sleep(_LOCK_RETRY_INTERVAL_SECONDS)
@contextmanager
def _manager_lock(project_root: Path) -> Iterator[None]:
"""Serialize extension mutations across processes for one checkout."""

View File

@ -74,6 +74,8 @@ from typing import Any
from uuid import uuid4
from weakref import WeakValueDictionary
from deerflow.utils.file_lock import retry_until_locked
try:
import fcntl
except ImportError: # pragma: no cover - Windows fallback
@ -182,6 +184,8 @@ LARK_SKILL_NAMES: tuple[str, ...] = (
"lark-workflow-standup-report",
)
LARK_SKILL_NAME_SET = frozenset(LARK_SKILL_NAMES)
# Allow long installs to finish without waiting indefinitely on a stuck holder.
_INSTALL_LOCK_TIMEOUT_SECONDS = 15 * 60
_LARK_INSTALL_THREAD_LOCK = threading.Lock()
_LARK_RUNTIME_INSTALL_THREAD_LOCK = threading.Lock()
_LARK_CREDENTIAL_LOCKS_GUARD = threading.Lock()
@ -1335,7 +1339,18 @@ def _exclusive_install_lock(lock_path: Path, thread_lock):
if fcntl is not None:
fcntl.flock(lock_file, fcntl.LOCK_EX)
else: # pragma: no cover - Windows fallback
msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
def acquire_region() -> None:
lock_file.seek(0)
msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
retry_until_locked(
acquire_region,
deadline=time.monotonic() + _INSTALL_LOCK_TIMEOUT_SECONDS,
sleep=time.sleep,
monotonic=time.monotonic,
timeout_message="Lark install lock timed out",
)
try:
yield
finally:

View File

@ -18,6 +18,7 @@ from typing import TYPE_CHECKING
from deerflow.skills.parser import parse_skill_file
from deerflow.skills.types import SKILL_MD_FILE, Skill, SkillCategory
from deerflow.utils.file_lock import retry_until_locked
if TYPE_CHECKING:
from deerflow.skills.storage.skill_storage import SkillStorage
@ -836,22 +837,24 @@ def skill_projection_read_lock(storage: SkillStorage, *, timeout: float = 5.0, c
with lock_path.open("a", encoding="utf-8") as lock_file:
locked = False
try:
while not locked:
if check:
check()
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)
locked = True
except OSError as error:
if error.errno not in (errno.EACCES, errno.EAGAIN, errno.EDEADLK):
raise
if time.monotonic() >= deadline:
raise TimeoutError("Skill projection lock timeout") from None
time.sleep(0.02)
def acquire_region() -> None:
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)
retry_until_locked(
acquire_region,
interval=0.02,
deadline=deadline,
sleep=time.sleep,
monotonic=time.monotonic,
check=check,
timeout_message="Skill projection lock timeout",
)
locked = True
yield
finally:
if locked:

View File

@ -133,3 +133,13 @@ non-ASCII whitespace remain unchanged.
Do not normalize arbitrary Unicode digits into device numbers: names such as
`COM⁴.txt` and `COM¹notes.txt` are ordinary portable names. Read/removal callers
retain their existing portability exemptions.
### Advisory File Lock Retries
`file_lock.retry_until_locked` owns contention-only acquisition retries for the
Windows Lark and extension-manager locks and bounded skill-projection reads.
Retry only EACCES/EAGAIN/EDEADLK; other errors propagate without sleeping.
Keep per-caller policies: Lark has a 15-minute Windows file-lock deadline,
extension management waits indefinitely, and skill export retains its existing
deadline, 0.02-second cadence and cancellation check. Clocks and sleeps are
injectable; tests must not patch the shared stdlib time module.

View File

@ -0,0 +1,44 @@
"""Contention-only retries for non-blocking advisory file locks."""
from __future__ import annotations
import errno
import time
from collections.abc import Callable
LOCK_RETRY_INTERVAL_SECONDS = 0.2
def retry_until_locked(
acquire: Callable[[], None],
*,
interval: float = LOCK_RETRY_INTERVAL_SECONDS,
deadline: float | None = None,
sleep: Callable[[float], None] = time.sleep,
monotonic: Callable[[], float] = time.monotonic,
check: Callable[[], None] | None = None,
timeout_message: str = "File lock acquisition timed out",
) -> None:
"""Retry only contention, optionally bounded by an absolute monotonic deadline.
Windows' blocking ``msvcrt.LK_LOCK`` gives up after about ten seconds.
Callers instead acquire with ``LK_NBLCK`` and choose their own deadline;
omitting it preserves an indefinite wait. Sleep/clock injection keeps
tests local rather than replacing functions on the shared time module.
"""
while True:
if check is not None:
check()
try:
acquire()
return
except OSError as error:
if error.errno not in (errno.EACCES, errno.EAGAIN, errno.EDEADLK):
raise
delay = interval
if deadline is not None:
remaining = deadline - monotonic()
if remaining <= 0:
raise TimeoutError(timeout_message) from None
delay = min(delay, remaining)
sleep(delay)

View File

@ -0,0 +1,97 @@
"""Portable contention/error/deadline semantics for advisory lock retries."""
import asyncio
import errno
import pytest
from deerflow.utils.file_lock import retry_until_locked
@pytest.mark.parametrize("error_number", [errno.EACCES, errno.EAGAIN, errno.EDEADLK])
def test_contention_retries_until_acquisition_succeeds(error_number):
attempts = 0
sleeps = []
def acquire():
nonlocal attempts
attempts += 1
if attempts < 3:
raise OSError(error_number, "Contended")
retry_until_locked(acquire, sleep=sleeps.append)
assert attempts == 3
assert sleeps == [0.2, 0.2]
@pytest.mark.parametrize("error_number", [errno.EBADF, errno.EINVAL, errno.EIO, errno.EPERM, None])
def test_permanent_error_is_propagated_without_retry(error_number):
error = OSError(error_number, "Permanent error")
sleeps = []
def acquire():
raise error
with pytest.raises(OSError) as caught:
retry_until_locked(acquire, sleep=sleeps.append)
assert caught.value is error
assert sleeps == []
def test_deadline_caps_the_final_sleep_and_reports_timeout():
now = 10.0
sleeps = []
def acquire():
raise OSError(errno.EAGAIN, "Contended")
def sleep(delay):
nonlocal now
sleeps.append(delay)
now += delay
with pytest.raises(TimeoutError, match="test deadline"):
retry_until_locked(acquire, deadline=10.45, sleep=sleep, monotonic=lambda: now, timeout_message="test deadline")
assert sleeps == pytest.approx([0.2, 0.2, 0.05])
assert now == pytest.approx(10.45)
def test_check_can_cancel_between_contended_attempts():
attempts = 0
checks = 0
sleeps = []
def check():
nonlocal checks
checks += 1
if checks == 2:
raise asyncio.CancelledError
def acquire():
nonlocal attempts
attempts += 1
raise OSError(errno.EAGAIN, "Contended")
with pytest.raises(asyncio.CancelledError):
retry_until_locked(acquire, check=check, sleep=sleeps.append)
assert attempts == 1
assert sleeps == [0.2]
def test_check_errors_are_not_treated_as_lock_contention():
error = OSError(errno.EACCES, "Denied by check")
attempts = []
sleeps = []
def check():
raise error
with pytest.raises(OSError) as caught:
retry_until_locked(lambda: attempts.append(True), check=check, sleep=sleeps.append)
assert caught.value is error
assert attempts == sleeps == []

View File

@ -0,0 +1,185 @@
"""Behavior tests for the cross-process install lock in ``lark_cli``.
Windows' blocking lock mode (``msvcrt.LK_LOCK``) gives up after roughly ten
seconds and surfaces contention as ``Permission denied``. A real install
(download + setup of the lark/sandbox CLI) can hold the lock far longer, so
the fallback branch must retry the non-blocking acquisition until the region
frees, mirroring the extensions manager's lock.
"""
from __future__ import annotations
import errno
import threading
import time
from types import SimpleNamespace
import pytest
from deerflow.integrations import lark_cli
class _ContendedFakeMsvcrt:
"""Fake ``msvcrt`` whose lock region stays busy for the first N attempts.
Records every ``locking`` call so tests can assert the retry cadence and
the final unlock, independently of the real Windows semantics.
"""
LK_UNLCK = 0
LK_LOCK = 1
LK_NBLCK = 2
def __init__(self, fail_times: int, *, error_number: int = errno.EDEADLK) -> None:
self.fail_times = fail_times
self.error_number = error_number
self.attempts: list[int] = []
self.unlocked = 0
def locking(self, fd: int, mode: int, nbytes: int) -> None:
self.attempts.append(mode)
if mode == self.LK_UNLCK:
self.unlocked += 1
return None
blocking_attempts = len(self.attempts) - self.unlocked
if blocking_attempts <= self.fail_times:
raise OSError(self.error_number, "Lock unavailable")
return None
def test_install_lock_retries_until_contention_clears(tmp_path, monkeypatch) -> None:
"""Contention is retried on the Windows fallback instead of failing fast."""
fake = _ContendedFakeMsvcrt(fail_times=3)
monkeypatch.setattr(lark_cli, "fcntl", None)
monkeypatch.setattr(lark_cli, "msvcrt", fake, raising=False)
sleeps: list[float] = []
monkeypatch.setattr(lark_cli, "time", SimpleNamespace(sleep=sleeps.append, monotonic=time.monotonic))
lock_path = tmp_path / ".lark-cli.install.lock"
entered = False
with lark_cli._exclusive_install_lock(lock_path, threading.Lock()):
entered = True
assert entered
assert fake.attempts == [fake.LK_NBLCK] * 4 + [fake.LK_UNLCK]
assert fake.unlocked == 1
assert sleeps == [0.2] * 3
def test_install_lock_posix_branch_still_flocks(tmp_path, monkeypatch) -> None:
"""The POSIX branch keeps using ``fcntl.flock`` unchanged."""
class _FakeFcntl:
LOCK_EX = 2
LOCK_UN = 8
def __init__(self) -> None:
self.ops: list[int] = []
def flock(self, fd: int, operation: int) -> None:
self.ops.append(operation)
fake_fcntl = _FakeFcntl()
monkeypatch.setattr(lark_cli, "fcntl", fake_fcntl)
lock_path = tmp_path / ".sandbox-cli.install.lock"
entered = False
with lark_cli._exclusive_install_lock(lock_path, threading.Lock()):
entered = True
assert entered
assert fake_fcntl.ops == [_FakeFcntl.LOCK_EX, _FakeFcntl.LOCK_UN]
@pytest.mark.skipif(lark_cli.fcntl is not None, reason="exercises the real msvcrt fallback branch")
def test_install_lock_waits_for_real_windows_contention(tmp_path) -> None:
"""A real second contender acquires the lock once the holder releases.
The holder keeps the region past ``LK_LOCK``'s ~10s give-up window: a
blocking-mode contender dies with ``Permission denied`` there, while a
retrying contender acquires as soon as the holder lets go.
"""
import msvcrt
lock_path = tmp_path / ".lark-cli.install.lock"
lock_path.write_bytes(b"\0")
holder = lock_path.open("a+b")
holder.seek(0)
msvcrt.locking(holder.fileno(), msvcrt.LK_NBLCK, 1)
acquired = threading.Event()
release = threading.Event()
errors: list[BaseException] = []
def contender() -> None:
try:
with lark_cli._exclusive_install_lock(lock_path, threading.Lock()):
acquired.set()
release.wait(timeout=30)
except BaseException as exc: # surfaced via the assertion below
errors.append(exc)
thread = threading.Thread(target=contender, daemon=True)
thread.start()
try:
# Keep the region locked past LK_LOCK's ~10s give-up window before
# releasing it: a blocking-mode contender dies there, a retrying one
# acquires as soon as the holder lets go.
time.sleep(12)
holder.seek(0)
msvcrt.locking(holder.fileno(), msvcrt.LK_UNLCK, 1)
holder.close()
assert acquired.wait(timeout=5), f"contender failed to acquire the contended lock; errors={errors!r}"
finally:
release.set()
thread.join(timeout=10)
assert not errors
@pytest.mark.parametrize("error_number", [errno.EBADF, errno.EINVAL, errno.EIO, errno.EPERM])
def test_install_lock_propagates_non_contention_errors(tmp_path, monkeypatch, error_number):
fake = _ContendedFakeMsvcrt(fail_times=1, error_number=error_number)
monkeypatch.setattr(lark_cli, "fcntl", None)
monkeypatch.setattr(lark_cli, "msvcrt", fake, raising=False)
def unexpected_sleep(_delay):
raise AssertionError("Permanent lock errors must not be retried")
monkeypatch.setattr(lark_cli, "time", SimpleNamespace(sleep=unexpected_sleep, monotonic=lambda: 0.0))
thread_lock = threading.Lock()
with pytest.raises(OSError) as caught:
with lark_cli._exclusive_install_lock(tmp_path / ".lock", thread_lock):
pytest.fail("Must not enter after a failed acquisition")
assert caught.value.errno == error_number
assert fake.attempts == [fake.LK_NBLCK]
assert fake.unlocked == 0
assert not thread_lock.locked()
def test_install_lock_contention_has_a_deadline(tmp_path, monkeypatch):
fake = _ContendedFakeMsvcrt(fail_times=100)
monkeypatch.setattr(lark_cli, "fcntl", None)
monkeypatch.setattr(lark_cli, "msvcrt", fake, raising=False)
monkeypatch.setattr(lark_cli, "_INSTALL_LOCK_TIMEOUT_SECONDS", 0.5, raising=False)
now = 0.0
sleeps = []
def sleep(delay):
nonlocal now
sleeps.append(delay)
now += delay
assert now <= 1, "Lock acquisition must not retry forever"
monkeypatch.setattr(lark_cli, "time", SimpleNamespace(sleep=sleep, monotonic=lambda: now))
thread_lock = threading.Lock()
with pytest.raises(TimeoutError, match="Lark install lock timed out"):
with lark_cli._exclusive_install_lock(tmp_path / ".lock", thread_lock):
pytest.fail("Must not enter while the lock is held")
assert sleeps == pytest.approx([0.2, 0.2, 0.1])
assert fake.attempts == [fake.LK_NBLCK] * 4
assert fake.unlocked == 0
assert not thread_lock.locked()

View File

@ -3,6 +3,7 @@
from contextlib import asynccontextmanager
import pytest
import pytest_asyncio
from _agent_e2e_helpers import build_single_tool_call_model
from langchain.agents import create_agent
from langchain_core.messages import ToolMessage
@ -13,9 +14,19 @@ from deerflow.agents.middlewares.artifact_capture_middleware import ArtifactCapt
from deerflow.agents.thread_state import ThreadState
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.config.paths import Paths
from deerflow.mcp.session_pool import MCPSessionPool
from deerflow.mcp.tools import _RawServerDiscovery, _wrap_discovered_server_tools
@pytest_asyncio.fixture
async def session_pool():
pool = MCPSessionPool()
try:
yield pool
finally:
await pool.close_all()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"transport,mime,suffix",
@ -25,7 +36,7 @@ from deerflow.mcp.tools import _RawServerDiscovery, _wrap_discovered_server_tool
("streamable_http", "image/png", "png"),
],
)
async def test_resource_link_captured_by_agent_graph(transport, mime, suffix, monkeypatch, tmp_path):
async def test_resource_link_captured_by_agent_graph(transport, mime, suffix, monkeypatch, tmp_path, session_pool):
url = f"https://example.com/report.{suffix}"
calls = []
@ -61,13 +72,10 @@ async def test_resource_link_captured_by_agent_graph(transport, mime, suffix, mo
async def fake_session(*args, **kwargs):
yield Session()
class Pool:
async def get_session(self, *args, **kwargs):
return Session()
# Replace only the external MCP session and the filesystem base directory.
# Discovery conversion, transport dispatch, tool execution, and capture are real.
# Discovery, pooling/call protection, tool execution, and capture are real.
monkeypatch.setattr("langchain_mcp_adapters.tools.create_session", fake_session)
monkeypatch.setattr("langchain_mcp_adapters.sessions.create_session", fake_session)
monkeypatch.setattr("deerflow.mcp.tools.get_paths", lambda: Paths(tmp_path))
connection = {"transport": "stdio", "command": "offline-mcp-fixture"} if transport == "stdio" else {"transport": "streamable_http", "url": "https://example.com/mcp"}
discovered = await load_mcp_tools(
@ -83,7 +91,7 @@ async def test_resource_link_captured_by_agent_graph(transport, mime, suffix, mo
_RawServerDiscovery(discovered, []),
tool_name_prefix=True,
session_init_timeout=None,
pool=Pool(),
pool=session_pool,
binding=None,
ownership_domain="deployment",
connection_scope="deployment",

View File

@ -919,3 +919,36 @@ def test_concurrent_custom_skill_toggles_do_not_lose_state(projection_env) -> No
"skill-b": {"enabled": False},
}
assert list(env.paths.user_custom_skills_view_dir("alice").iterdir()) == []
@pytest.mark.parametrize("permanent_error", [False, True])
def test_projection_read_windows_lock_retries_only_contention(tmp_path, monkeypatch, permanent_error):
import errno
from deerflow.skills import projection
calls = []
class FakeMsvcrt:
LK_NBLCK = 2
LK_UNLCK = 0
@staticmethod
def locking(fd, operation, length):
calls.append(operation)
if operation == FakeMsvcrt.LK_NBLCK and len(calls) == 1:
raise OSError(errno.EBADF if permanent_error else errno.EACCES, "Lock unavailable")
monkeypatch.setattr(projection, "fcntl", None)
monkeypatch.setattr(projection, "msvcrt", FakeMsvcrt, raising=False)
storage = SimpleNamespace(user_id=None, get_skills_root_path=lambda: tmp_path)
if not permanent_error:
with projection.skill_projection_read_lock(storage, timeout=1):
assert calls == [FakeMsvcrt.LK_NBLCK] * 2
assert calls == [FakeMsvcrt.LK_NBLCK] * 2 + [FakeMsvcrt.LK_UNLCK]
else:
with pytest.raises(OSError) as caught:
with projection.skill_projection_read_lock(storage, timeout=1):
pytest.fail("Must not enter after a permanent lock error")
assert caught.value.errno == errno.EBADF
assert calls == [FakeMsvcrt.LK_NBLCK]