From dded554332b87d843a29d028703e7c1613c58a2f Mon Sep 17 00:00:00 2001 From: Allard Date: Thu, 16 Jul 2026 05:11:29 +0800 Subject: [PATCH] Fix #65176: Auto-terminate live gateway on platform lock conflict When a new gateway detects a platform lock conflict (e.g. Telegram bot token already in use) and the holder is a live gateway process, it now terminates the holder and retries the lock acquisition instead of retrying indefinitely with a fatal error. This mirrors the --replace behavior but applies it at platform-connect time when the scoped token lock detects a conflict, resolving the issue where the gateway would wait for manual SIGKILL of the conflicting process. Changes: - gateway/platforms/base.py: Modified _acquire_platform_lock() to detect live gateway holders via _looks_like_gateway_process(), write takeover marker, force-terminate with terminate_pid(), wait 0.5s, and retry lock acquisition. Falls back to fatal error only if retry still fails. - tests/gateway/test_platform_lock_auto_terminate.py: New test file with three test cases covering termination of live gateway holders, non-termination of non-gateway holders, and fallback to fatal error when termination fails. --- gateway/platforms/base.py | 41 ++++- .../test_platform_lock_auto_terminate.py | 149 ++++++++++++++++++ 2 files changed, 189 insertions(+), 1 deletion(-) create mode 100644 tests/gateway/test_platform_lock_auto_terminate.py diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index d595d7033b7b..b248505d9823 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2759,7 +2759,13 @@ async def _notify_fatal_error(self) -> None: def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) -> bool: """Acquire a scoped lock for this adapter. Returns True on success.""" - from gateway.status import acquire_scoped_lock + from gateway.status import ( + acquire_scoped_lock, + terminate_pid, + _looks_like_gateway_process, + write_takeover_marker, + _pid_exists, + ) self._platform_lock_scope = scope self._platform_lock_identity = identity acquired, existing = acquire_scoped_lock( @@ -2768,6 +2774,39 @@ def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) if acquired: return True owner_pid = existing.get('pid') if isinstance(existing, dict) else None + if owner_pid and _looks_like_gateway_process(owner_pid): + # The conflict holder is a live gateway process. Terminate it + # and retry the lock acquisition, mirroring the --replace behavior. + logger.info( + '[%s] %s already in use (PID %d). Terminating holder and retrying.', + self.name, resource_desc, owner_pid + ) + try: + write_takeover_marker(owner_pid) + except Exception as e: + logger.debug('[%s] Could not write takeover marker: %s', self.name, e) + try: + terminate_pid(owner_pid, force=True) + except ProcessLookupError: + logger.debug('[%s] Holder PID %d already exited', self.name, owner_pid) + except Exception as e: + logger.warning('[%s] Failed to terminate holder PID %d: %s', self.name, owner_pid, e) + # Wait for the holder to exit before retrying. Poll with a short timeout + # to avoid blocking indefinitely if the holder is wedged. + import time + max_wait = 5.0 # Maximum time to wait for holder to exit + poll_interval = 0.1 + waited = 0.0 + while waited < max_wait: + if not _pid_exists(owner_pid): + break + time.sleep(poll_interval) + waited += poll_interval + acquired, existing = acquire_scoped_lock( + scope, identity, metadata={'platform': self.platform.value} + ) + if acquired: + return True message = ( f'{resource_desc} already in use' + (f' (PID {owner_pid})' if owner_pid else '') diff --git a/tests/gateway/test_platform_lock_auto_terminate.py b/tests/gateway/test_platform_lock_auto_terminate.py new file mode 100644 index 000000000000..f70bc6317396 --- /dev/null +++ b/tests/gateway/test_platform_lock_auto_terminate.py @@ -0,0 +1,149 @@ +"""Test for #65176 — platform lock conflict should auto-terminate live gateway holder. + +When a new gateway detects a platform lock conflict (e.g. bot token already in use) +and the holder is a live gateway process, it should terminate the holder and retry the lock +acquisition instead of retrying indefinitely with a fatal error. + +This fix applies to all platform adapters (Telegram, Discord, WhatsApp, Weixin, Signal, etc.) +since it's implemented in BasePlatformAdapter._acquire_platform_lock(). +""" + +from typing import Any, Dict +from unittest.mock import MagicMock, patch, AsyncMock + +import pytest + +from gateway.platforms.base import BasePlatformAdapter + + +class _StubAdapter(BasePlatformAdapter): + """Minimal concrete subclass for testing _acquire_platform_lock.""" + + platform = MagicMock(value="test-platform") + + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + pass + + async def send(self, *args: Any, **kwargs: Any) -> None: + pass + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + return {} + + +@pytest.fixture() +def adapter(): + """Create a stub adapter with __init__ bypassed.""" + obj = _StubAdapter.__new__(_StubAdapter) + obj._running = True + obj._fatal_error_code = None + obj._fatal_error_message = None + obj._fatal_error_retryable = True + obj._fatal_error_handler = None + obj._platform_lock_scope = None + obj._platform_lock_identity = None + obj._status_write_logged = None + return obj + + +def test_platform_lock_conflict_with_live_gateway_terminates_holder(adapter): + """When lock conflict holder is a live gateway, terminate it and retry (#65176).""" + call_count = {"n": 0} + holder_pid = 12345 + test_scope = "test-platform-token" + test_identity = "test-identity" + test_desc = "Test platform token" + + def mock_acquire_lock(scope, identity, metadata=None): + call_count["n"] += 1 + if call_count["n"] == 1: + # First call: conflict with live gateway + return (False, {"pid": holder_pid, "start_time": "2026-01-01T00:00:00Z"}) + else: + # Second call (after termination): success + return (True, None) + + with patch( + "gateway.status.acquire_scoped_lock", + side_effect=mock_acquire_lock, + ), patch( + "gateway.status._looks_like_gateway_process", + return_value=True, # Holder is a live gateway + ), patch( + "gateway.status.terminate_pid", + ) as mock_terminate, patch( + "gateway.status.write_takeover_marker", + ), patch( + "gateway.status._pid_exists", + return_value=False, # Process already exited + ), patch.object(adapter, "_write_runtime_status_safe"): + result = adapter._acquire_platform_lock( + test_scope, test_identity, test_desc + ) + + # Should have terminated the holder + mock_terminate.assert_called_once_with(holder_pid, force=True) + # Should have succeeded after retry + assert result is True + assert adapter._fatal_error_code is None, "Should not set fatal error on successful retry" + assert call_count["n"] == 2, "Should have called acquire_scoped_lock twice (initial + retry)" + + +def test_platform_lock_conflict_with_non_gateway_does_not_terminate(adapter): + """When lock conflict holder is NOT a gateway, do NOT terminate it.""" + holder_pid = 12345 + test_scope = "test-platform-token" + test_identity = "test-identity" + test_desc = "Test platform token" + + with patch( + "gateway.status.acquire_scoped_lock", + return_value=(False, {"pid": holder_pid, "start_time": "2026-01-01T00:00:00Z"}), + ), patch( + "gateway.status._looks_like_gateway_process", + return_value=False, # Holder is NOT a gateway + ), patch( + "gateway.status.terminate_pid", + ) as mock_terminate, patch.object(adapter, "_write_runtime_status_safe"): + result = adapter._acquire_platform_lock( + test_scope, test_identity, test_desc + ) + + # Should NOT have terminated the holder + mock_terminate.assert_not_called() + # Should fail with fatal error + assert result is False + assert adapter._fatal_error_code == f"{test_scope}_lock" + assert adapter._fatal_error_retryable is True + + +def test_platform_lock_conflict_terminate_fails_still_sets_fatal_error(adapter): + """If termination fails, still set fatal error as fallback.""" + holder_pid = 12345 + test_scope = "test-platform-token" + test_identity = "test-identity" + test_desc = "Test platform token" + + with patch( + "gateway.status.acquire_scoped_lock", + return_value=(False, {"pid": holder_pid, "start_time": "2026-01-01T00:00:00Z"}), + ), patch( + "gateway.status._looks_like_gateway_process", + return_value=True, + ), patch( + "gateway.status.terminate_pid", + side_effect=Exception("Termination failed"), + ), patch( + "gateway.status.write_takeover_marker", + ), patch.object(adapter, "_write_runtime_status_safe"): + result = adapter._acquire_platform_lock( + test_scope, test_identity, test_desc + ) + + # Should fail with fatal error when termination fails + assert result is False + assert adapter._fatal_error_code == f"{test_scope}_lock" + assert adapter._fatal_error_retryable is True