diff --git a/README.md b/README.md index ac61ac1fe2..d14f48396c 100644 --- a/README.md +++ b/README.md @@ -1101,6 +1101,10 @@ Invalid UTF-8 in `channels/store.json` follows the same recovery policy as malfo SQL migration warns and skips the file, leaving it unchanged so channels can start; the memory backend's JSON store warns and starts empty, replacing the file on its next write. +On Windows, JSON runtime channel-config saves wait for a competing file-lock holder +without the operating system's roughly ten-second cutoff. A wait reaching 50 retries +(about ten seconds) logs the lock-file path once; permanent lock errors fail immediately. + Conversation-binding, runtime channel-credential and WeChat auth-state saves preserve the original persistence failure when staged-file cleanup also fails. Remaining cleanup still runs, and warnings identify copies that could not be removed. Buzz keeps failed snapshots available for retry, even if diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index bf8205c9c4..8b95a6dc37 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -76,6 +76,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - Frontend APIs: `GET /api/channels/providers`, `GET /api/channels/connections`, `POST /api/channels/{provider}/connect`, and `DELETE /api/channels/connections/{connection_id}`. - Browser APIs remain protected by normal Gateway auth/CSRF. Provider messages arrive through the already-configured channel workers. - Runtime credentials (`runtime_config_store.py`): encrypted `channel_runtime_configs` rows (JSON only for `memory`), re-read per request, never cached on `app.state`; a save is one cancellation-owned commit. +- The JSON runtime-config sidecar uses `utils.file_lock.retry_until_locked` on Windows: retry only EACCES/EAGAIN/EDEADLK every 0.2s without a deadline; propagate other errors before entering the mutation or unlocking. Warn once per acquisition after 50 contention retries with the sidecar path; logging failures must not interrupt the wait. POSIX keeps blocking `flock`. Coverage: `tests/test_channel_runtime_config_store_lock.py`. - Provider-level `connection_status` reflects the user's newest connection row. With no binding it is `not_connected`, except in auth-disabled local mode where a configured running channel reports `connected` because all channel messages already route to the default user. - Slack replies use the configured operator bot token from `channels.slack` unless per-connection credentials are present; unreadable or corrupt stored credentials are treated as unavailable. - Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom workers resolve incoming platform identities to connection records before reaching `ChannelManager`. @@ -87,7 +88,16 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - Custom agents declare a `github:` block in their `config.yaml` to bind to repos and event triggers; the webhook route is fail-closed by default (mounted only when `GITHUB_WEBHOOK_SECRET` is set) and exempt from auth/CSRF because authenticity is enforced by HMAC. - Registry caching is keyed by the configured agent store's opaque signature: file storage uses agent config mtimes, while database storage hashes the ordered owner/name/config/soul contents so same-timestamp writes still invalidate webhook routing. - Outbound is **log-only** by design: each agent posts its own reply mid-run via the `gh` CLI from its sandbox, so the manager uses `fire_and_forget=True` and `runs.create()` returns once pending. -- **Follow-up buffering while busy** (issue #4121): outbound is log-only, so the `THREAD_BUSY_MESSAGE` reply on a `ConflictError` was invisible to the commenter — a comment posted during an active run looked silently ignored. When `ChannelRunPolicy.buffer_followups_on_busy=True` (GitHub's default), a `ConflictError` on `runs.create()` also appends the triggering message to a per-thread in-memory buffer (`ChannelManager._followup_buffers`) — deduped by GitHub delivery id, capped at `FOLLOWUP_BUFFER_MAX_PER_THREAD` (20, oldest dropped with a WARNING). The first successful `runs.create()` on a thread captures its `run_id` and spawns a watcher on that run's `StreamBridge` stream; on `END_SENTINEL` it drains up to `FOLLOWUP_DRAIN_BATCH_SIZE` (10) entries into one ``-wrapped input and fires a follow-up `runs.create()` — itself watched, so a deeper backlog chains into further drain cycles instead of one unbounded input. If that follow-up hits `ConflictError` (a concurrent turn raced onto the thread), the batch is requeued and retried when this manager next creates and watches a run on that thread. Reactions on buffered comments are out of scope; comments coalesce silently. **Plumbing**: the watcher reaches the Gateway's `StreamBridge` through a zero-arg closure threaded from the lifespan via `start_channel_service(get_stream_bridge=...)` → `ChannelService` → `ChannelManager` (like `ScheduledTaskService.launch_run`); without it (tests) buffering works without auto-drain. **Scope limitation**: buffer and watcher are per-process and webhooks reach any instance (the leader lease does not arbitrate GitHub), so a follow-up routed to another worker misses the buffer until a shared buffer store exists (#4120 shape). +- **Busy follow-ups** (#4121): `ChannelRunPolicy.buffer_followups_on_busy=True` + (GitHub default) buffers `runs.create()` conflicts per thread, deduplicated by + delivery ID. Cap at 20 entries, dropping the oldest with WARNING. Watch each + successful run's `StreamBridge`; `END_SENTINEL` drains up to 10 entries into + `` and watches the successor. Requeue drain conflicts + until this manager next creates and watches a run. Buffers/watchers are + process-local; GitHub has no leader-lease arbitration. Without the zero-arg + lifespan → service → manager bridge closure, buffering has no auto-drain. + Coalesce silently without reactions. Before changing buffering or watcher + plumbing, read [the full contract](../../docs/GITHUB_AGENTS.md#follow-up-buffering-while-busy-4121). - See [backend/docs/GITHUB_AGENTS.md](../../docs/GITHUB_AGENTS.md) for the architecture diagrams: webhook → fan-out → `InboundMessage` dispatch, `preferred_thread_id = UUID5(repo, number, agent_name)` thread determinism, mention-handle precedence chain, GH token lifecycle via `GH_TOKEN`/`GITHUB_TOKEN` per-call `extra_env`, and the narrow `ConflictError` (HTTP 409) thread-create race recovery. diff --git a/backend/app/channels/runtime_config_store.py b/backend/app/channels/runtime_config_store.py index a5c38089fc..0d58521485 100644 --- a/backend/app/channels/runtime_config_store.py +++ b/backend/app/channels/runtime_config_store.py @@ -36,15 +36,25 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Protocol from app.channels.credential_staging import cleanup_stale_credential_staging, credential_staging_prefix +from deerflow.utils.file_lock import retry_until_locked if TYPE_CHECKING: from deerflow.config.credentials_key import CredentialsCipher from deerflow.persistence.channel_runtime_configs import SqlChannelRuntimeConfigRepository, StoredChannelRuntimeConfig +try: + import fcntl +except ImportError: # pragma: no cover - Windows fallback + fcntl = None # type: ignore[assignment] + import msvcrt + logger = logging.getLogger(__name__) RUNTIME_CHANNEL_DISABLED_FLAG = "_runtime_disabled" +# About ten seconds at the shared helper's 0.2-second retry cadence. +_LOCK_WAIT_WARNING_ATTEMPTS = 50 + class ChannelRuntimeConfigEncryptionUnavailable(RuntimeError): """The database store has no credentials key, so it refuses to persist credentials.""" @@ -83,28 +93,40 @@ def _runtime_config_file_lock(path: Path) -> Iterator[None]: target_path = path.resolve(strict=False) if path.is_symlink() else path.absolute() lock_path = target_path.parent / f".{target_path.name}.lock" with open(lock_path, "a+b") as lock_file: - if os.name == "nt": - import msvcrt - + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + else: # pragma: no cover - Windows fallback lock_file.seek(0, os.SEEK_END) if lock_file.tell() == 0: lock_file.write(b"\0") lock_file.flush() + # LK_LOCK gives up after roughly ten seconds. Share the + # contention-only retry policy, retaining an indefinite wait. lock_file.seek(0) - msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) - else: - import fcntl + retries = 0 - fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + def wait_for_lock(delay: float) -> None: + nonlocal retries + retries += 1 + if retries == _LOCK_WAIT_WARNING_ATTEMPTS: + # Diagnostics must not abort acquisition or flood the log. + with suppress(Exception): + logger.warning("Still waiting for channel runtime config lock %s after %d contention retries", lock_path, retries) + time.sleep(delay) + + retry_until_locked( + lambda: msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1), + sleep=wait_for_lock, + ) try: yield finally: - if os.name == "nt": + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + else: # pragma: no cover - Windows fallback lock_file.seek(0) msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) - else: - fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) class ChannelRuntimeConfigStore: diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 3d06189f1f..1adf565ab5 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -173,12 +173,16 @@ 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. +Windows Lark, extension-manager and channel runtime-config 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. +extension management and channel runtime config wait 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. +The channel caller wraps its injected sleep to warn once after 50 contention +retries; the helper remains responsible for error classification and cadence. `parse_retry_after_date_ms(raw, now=...)` owns HTTP-date parsing for Claude, Codex and LLM error middleware. Interpret timezone-less asctime as UTC and apply the diff --git a/backend/tests/test_channel_runtime_config_store_lock.py b/backend/tests/test_channel_runtime_config_store_lock.py new file mode 100644 index 0000000000..6b57ff24e1 --- /dev/null +++ b/backend/tests/test_channel_runtime_config_store_lock.py @@ -0,0 +1,249 @@ +"""Behavior tests for the cross-process file lock in the runtime config store. + +Windows' blocking lock mode (``msvcrt.LK_LOCK``) gives up after roughly ten +seconds and surfaces contention as ``Permission denied``, while the whole +point of the sidecar lock is to serialize concurrent Gateway workers' +read-modify-write cycles of one store file. The Windows fallback must retry +the non-blocking acquisition until the region frees, mirroring the extensions +manager's lock. +""" + +from __future__ import annotations + +import errno +import logging +import os +import sys +import threading +import time +from types import SimpleNamespace + +import pytest + +from app.channels import runtime_config_store +from app.channels.runtime_config_store import ChannelRuntimeConfigStore, _runtime_config_file_lock + +TELEGRAM = {"enabled": True, "bot_token": "tg-token"} + + +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 contended") + return None + + +class _BlockingModeFakeMsvcrt: + """Fake ``msvcrt`` that fails only the blocking ``LK_LOCK`` mode. + + Models a region contended past the blocking mode's patience: the + unpatched module calls ``LK_LOCK`` once and fails, while a retry loop + succeeds on its first non-blocking attempt. + """ + + LK_UNLCK = 0 + LK_LOCK = 1 + LK_NBLCK = 2 + + def __init__(self) -> None: + 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 + if mode == self.LK_LOCK: + raise OSError(errno.EDEADLK, "Resource deadlock avoided") + return None + + +@pytest.mark.parametrize("error_number", [errno.EACCES, errno.EAGAIN, errno.EDEADLK]) +def test_file_lock_retries_until_contention_clears(tmp_path, monkeypatch, caplog, error_number) -> None: + """Contention is retried on the Windows fallback instead of failing.""" + fake = _ContendedFakeMsvcrt(fail_times=3, error_number=error_number) + monkeypatch.setattr(runtime_config_store, "fcntl", None, raising=False) + monkeypatch.setattr(runtime_config_store, "msvcrt", fake, raising=False) + sleeps: list[float] = [] + monkeypatch.setattr(runtime_config_store, "time", SimpleNamespace(sleep=sleeps.append)) + + entered = False + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + entered = True + + assert entered + assert fake.attempts == [fake.LK_NBLCK] * 4 + [fake.LK_UNLCK] + assert fake.unlocked == 1 + assert sleeps == [0.2] * 3 + assert not caplog.records + + +@pytest.mark.parametrize("error_number", [errno.EBADF, errno.EINVAL, errno.EIO, errno.EPERM, errno.ENOSYS, None]) +def test_file_lock_propagates_permanent_errors_without_waiting_or_unlocking(tmp_path, monkeypatch, error_number) -> None: + primary = OSError(error_number, "Lock unavailable") + calls = [] + sleeps = [] + + def locking(fd, mode, size): + calls.append(mode) + if len(calls) > 1: + pytest.fail("A failed acquisition must not be retried or unlocked") + raise primary + + monkeypatch.setattr(runtime_config_store, "fcntl", None) + monkeypatch.setattr(runtime_config_store, "msvcrt", SimpleNamespace(LK_NBLCK=2, LK_UNLCK=0, locking=locking), raising=False) + monkeypatch.setattr(runtime_config_store, "time", SimpleNamespace(sleep=sleeps.append)) + + with pytest.raises(OSError) as caught: + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + pytest.fail("A failed acquisition must not enter the protected section") + + assert caught.value is primary + assert calls == [2] + assert sleeps == [] + + +@pytest.mark.parametrize("fail_times", [49, 50, 100]) +def test_file_lock_warns_once_for_prolonged_contention(tmp_path, monkeypatch, caplog, fail_times) -> None: + fake = _ContendedFakeMsvcrt(fail_times=fail_times) + sleeps = [] + monkeypatch.setattr(runtime_config_store, "fcntl", None) + monkeypatch.setattr(runtime_config_store, "msvcrt", fake, raising=False) + monkeypatch.setattr(runtime_config_store, "time", SimpleNamespace(sleep=sleeps.append)) + + with caplog.at_level(logging.WARNING, logger=runtime_config_store.__name__): + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + pass + + warnings = [record for record in caplog.records if record.name == runtime_config_store.__name__] + assert len(warnings) == (1 if fail_times >= 50 else 0) + if warnings: + assert str(tmp_path / ".runtime-config.json.lock") in warnings[0].getMessage() + assert "50" in warnings[0].getMessage() + assert sleeps == [0.2] * fail_times + assert fake.unlocked == 1 + + +def test_file_lock_diagnostic_failure_does_not_abort_the_wait(tmp_path, monkeypatch) -> None: + fake = _ContendedFakeMsvcrt(fail_times=51) + warnings = [] + sleeps = [] + + def failing_warning(*args, **kwargs): + warnings.append(args) + raise RuntimeError("Log handler failed") + + monkeypatch.setattr(runtime_config_store, "fcntl", None) + monkeypatch.setattr(runtime_config_store, "msvcrt", fake, raising=False) + monkeypatch.setattr(runtime_config_store, "time", SimpleNamespace(sleep=sleeps.append)) + monkeypatch.setattr(runtime_config_store.logger, "warning", failing_warning) + + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + pass + + assert len(warnings) == 1 + assert sleeps == [0.2] * 51 + assert fake.unlocked == 1 + + +def test_file_lock_creates_the_sidecar_byte_on_the_windows_fallback(tmp_path, monkeypatch) -> None: + """The empty sidecar file still gets its lockable byte before locking.""" + fake = _ContendedFakeMsvcrt(fail_times=0) + monkeypatch.setattr(runtime_config_store, "fcntl", None, raising=False) + monkeypatch.setattr(runtime_config_store, "msvcrt", fake, raising=False) + + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + pass + + assert (tmp_path / ".runtime-config.json.lock").read_bytes() == b"\0" + + +def test_file_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[tuple[int, int]] = [] + + def flock(self, fd: int, operation: int) -> None: + self.ops.append((fd, operation)) + + fake_fcntl = _FakeFcntl() + monkeypatch.setattr(runtime_config_store, "fcntl", fake_fcntl, raising=False) + + entered = False + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + entered = True + + assert entered + assert [operation for _, operation in fake_fcntl.ops] == [_FakeFcntl.LOCK_EX, _FakeFcntl.LOCK_UN] + + +@pytest.mark.skipif(os.name != "nt", reason="exercises the Windows import path directly") +def test_windows_fallback_survives_contention_the_blocking_mode_cannot(tmp_path, monkeypatch) -> None: + """Direct regression: the fallback retries contention instead of failing. + + ``sys.modules`` catches the function-local ``import msvcrt`` the unpatched + module performs; the module attribute covers the fixed module's global + binding, so this test demonstrates the base failure mode (one ``LK_LOCK`` + call raising ``OSError``) and the fixed retry behavior on either structure. + """ + fake = _BlockingModeFakeMsvcrt() + monkeypatch.setitem(sys.modules, "msvcrt", fake) + monkeypatch.setattr(runtime_config_store, "msvcrt", fake, raising=False) + + with _runtime_config_file_lock(tmp_path / "runtime-config.json"): + pass + + assert fake.attempts == [fake.LK_NBLCK, fake.LK_UNLCK] + assert fake.unlocked == 1 + + +def test_save_serializes_with_a_peer_holding_the_real_lock(tmp_path) -> None: + """A real peer holding the sidecar lock delays the save instead of failing it. + + The advisory file lock is per open file, so two lock contexts on one path + exclude each other inside this process, as the peer tests for #6844 note. + """ + store_path = tmp_path / "channels" / "runtime-config.json" + store_path.parent.mkdir(parents=True, exist_ok=True) + holder_ready = threading.Event() + + def hold(): + with _runtime_config_file_lock(store_path): + holder_ready.set() + time.sleep(0.6) + + holder = threading.Thread(target=hold) + holder.start() + assert holder_ready.wait(timeout=5) + + ChannelRuntimeConfigStore(store_path).set_provider_config("telegram", TELEGRAM) + + holder.join(timeout=5) + assert ChannelRuntimeConfigStore(store_path).get_provider_config("telegram") == TELEGRAM