Merge pull request #6212 from MervinPraison/claude/issue-6208-20261103

fix: gateway single-instance guard is state-scoped, not only port-scoped

src/praisonai-bot/praisonai_bot/gateway/port_utils.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import os
 import socket
 import time
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional, Tuple
 
@@ -86,19 +87,21 @@ def _process_create_time(pid: int) -> Optional[float]:
 class GatewayPIDLock:
     """Manages PID lock file for single-instance gateway enforcement."""
     
-    def __init__(self, lock_dir: Optional[Path] = None, host: str = "127.0.0.1", port: int = 8765):
+    def __init__(self, lock_dir: Optional[Path] = None, host: str = "127.0.0.1", port: int = 8765,
+                 lock_name: Optional[str] = None):
         """Initialize PID lock manager.
         
         Args:
             lock_dir: Directory for lock file (default: ~/.praisonai/)
             host: Gateway host
             port: Gateway port
+            lock_name: Override the lock file name (used by the state lock)
         """
         self.lock_dir = lock_dir or Path.home() / ".praisonai"
         # Make lock file specific to host and port to allow multiple instances
         safe_host = host.replace(":", "_").replace(".", "_")
-        self.lock_file = self.lock_dir / f"gateway-{safe_host}-{port}.pid"
-        self.lock_dir.mkdir(exist_ok=True)
+        self.lock_file = self.lock_dir / (lock_name or f"gateway-{safe_host}-{port}.pid")
+        self.lock_dir.mkdir(parents=True, exist_ok=True)
     
     def acquire_lock(self, host: str, port: int) -> bool:
         """Acquire the PID lock for the gateway.
@@ -110,6 +113,27 @@ class GatewayPIDLock:
         Returns:
             True if lock acquired successfully, False if conflict
         """
+        # Serialise check-then-claim across processes so two simultaneous
+        # starts can never both observe "no live holder" and both win.
+        with self._claim_guard():
+            return self._acquire_locked(host, port)
+
+    @contextmanager
+    def _claim_guard(self):
+        """Hold an OS advisory lock on a sidecar file for the claim (POSIX)."""
+        try:
+            import fcntl
+        except ImportError:  # Windows: exclusive link() below still applies
+            yield
+            return
+        guard = open(self.lock_file.with_name(self.lock_file.name + ".guard"), "a")
+        try:
+            fcntl.flock(guard, fcntl.LOCK_EX)
+            yield
+        finally:
+            guard.close()  # closing drops the flock
+
+    def _acquire_locked(self, host: str, port: int) -> bool:
         current_pid = os.getpid()
         
         # Check if lock file exists
@@ -120,6 +144,9 @@ class GatewayPIDLock:
                     lines = lock_content.split('\n')
                     if lines:
                         existing_pid = int(lines[0])
+                        if existing_pid == current_pid:
+                            # Re-acquire by the same process (in-process restart).
+                            return True
                         existing_create_time = None
                         if len(lines) >= 5 and lines[4]:
                             try:
@@ -145,14 +172,29 @@ class GatewayPIDLock:
         create_time = _process_create_time(current_pid)
         create_time_line = "" if create_time is None else repr(create_time)
         lock_content = f"{current_pid}\n{host}\n{port}\n{int(time.time())}\n{create_time_line}\n"
+        # Exclusive claim: write a private temp file, then hard-link it into
+        # place. ``link`` fails if the lock already exists, so a racing claimant
+        # can never overwrite a live holder (unlike rename/replace).
+        temp_lock_file = self.lock_file.with_name(f"{self.lock_file.name}.{current_pid}.tmp")
         try:
-            # Use a temporary file and atomic rename to prevent race conditions
-            temp_lock_file = self.lock_file.with_suffix(".tmp")
             temp_lock_file.write_text(lock_content)
-            temp_lock_file.replace(self.lock_file)
+            try:
+                os.link(temp_lock_file, self.lock_file)
+            except FileExistsError:
+                return False
+            except OSError:
+                # Filesystem without hard links: fall back to O_EXCL create.
+                fd = os.open(self.lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+                with os.fdopen(fd, "w") as f:
+                    f.write(lock_content)
             return True
         except OSError:
             return False
+        finally:
+            try:
+                temp_lock_file.unlink()
+            except OSError:
+                pass
     
     def release_lock(self) -> None:
         """Release the PID lock."""
@@ -253,6 +295,43 @@ class GatewayPIDLock:
             pass
 
 
+def resolve_state_dir() -> Path:
+    """Return the gateway's shared mutable state root (``~/.praisonai``).
+
+    The durable outbox, DLQ, pairing and hook stores are hard-wired under
+    ``~/.praisonai`` (they do not honour ``PRAISONAI_HOME``), so the home root -
+    not the TCP port or ``PRAISONAI_HOME`` - is what identifies a gateway.
+    """
+    return Path.home() / ".praisonai"
+
+
+def state_lock(state_dir: Optional[Path] = None, host: str = "127.0.0.1",
+               port: int = 8765) -> GatewayPIDLock:
+    """Single-instance lock keyed on the shared state directory, regardless of port.
+
+    Two gateways sharing one state root would poll the same bot tokens and write
+    the same SQLite stores concurrently (split-brain), so only one may own it.
+    """
+    return GatewayPIDLock(lock_dir=state_dir or resolve_state_dir(), host=host,
+                          port=port, lock_name="gateway.state.pid")
+
+
+def format_state_collision_error(state_dir: Path, lock_info: Optional[dict] = None) -> str:
+    """Error shown when another live gateway already owns the shared state."""
+    holder = ""
+    if lock_info:
+        holder = f" (PID {lock_info['pid']} on {lock_info['host']}:{lock_info['port']})"
+    return "\n".join([
+        f"Error: another gateway already owns the state at {state_dir}{holder}.",
+        "",
+        "  Two gateways sharing state poll the same bot tokens and write the same",
+        "  queues, causing 409 Conflict errors and duplicate delivery.",
+        "  Stop it:  praisonai gateway stop",
+        "  Or run an isolated gateway with its own state and bot tokens:",
+        "    HOME=/srv/gateway-b praisonai gateway start",
+    ])
+
+
 def format_collision_error(host: str, port: int, lock_info: Optional[dict] = None) -> str:
     """Format a user-friendly error message for port collisions.
     
@@ -275,10 +354,11 @@ def format_collision_error(host: str, port: int, lock_info: Optional[dict] = Non
         pid = lock_info['pid']
         lines.extend([
             f"  Another gateway may be running (PID {pid}).",
-            f"  Stop it:  praisonai gateway stop",
-            f"  Or use a different port:  GATEWAY_PORT={port + 1} praisonai gateway start",
+            "  Stop it:  praisonai gateway stop",
             "",
-            "  Only ONE gateway process should poll each Telegram bot token.",
+            "  Only ONE gateway process may poll each bot token. To run a second,",
+            "  isolated gateway give it its own state, bot tokens and port:",
+            f"    HOME=/srv/gateway-b GATEWAY_PORT={port + 1} praisonai gateway start",
         ])
     else:
         lines.extend([

src/praisonai-bot/praisonai_bot/gateway/server.py

@@ -1412,6 +1412,7 @@ class WebSocketGateway:
         
         # PID lock for single-instance enforcement
         self._pid_lock: Optional[Any] = None
+        self._state_lock: Optional[Any] = None
         
         # Channel supervisor for resilient bot management. Share the gateway's
         # degraded-capability registry so the fleet crash-loop breaker (Issue
@@ -1544,9 +1545,46 @@ class WebSocketGateway:
             logger.error("Gateway startup failed - another instance is running")
             raise RuntimeError(error_msg)
         
-        # Store PID lock for cleanup
         self._pid_lock = pid_lock
-        
+        # When ``start_with_config`` already claimed the state root, its channel
+        # bots are still polling if we fail here; only it may release that lock
+        # (after channels stop), or a second gateway could claim the state.
+        owns_state_lock = self._state_lock is None
+        try:
+            self._acquire_state_lock()
+            await self._start_locked()
+        except BaseException:
+            # Any failure after claiming (bad config, bind error, cancel) must
+            # not leave a live process holding the locks and blocking restarts.
+            self._release_locks(state=owns_state_lock)
+            raise
+
+    def _acquire_state_lock(self) -> None:
+        """Claim the shared state root (idempotent).
+
+        The port lock alone admits a second gateway on another port sharing the
+        same state + bot tokens (split-brain), so the state root is locked too.
+        """
+        if self._state_lock is not None:
+            return
+        from .port_utils import state_lock, resolve_state_dir, format_state_collision_error
+        state_dir = resolve_state_dir()
+        st_lock = state_lock(state_dir, host=self._host, port=self._port)
+        if not st_lock.acquire_lock(self._host, self._port):
+            logger.error("Gateway startup failed - another instance owns the shared state")
+            raise RuntimeError(format_state_collision_error(state_dir, st_lock.get_lock_info()))
+        self._state_lock = st_lock
+
+    def _release_locks(self, state: bool = True) -> None:
+        """Release the port lock and (if ``state``) the state lock (idempotent)."""
+        for attr in ("_pid_lock", "_state_lock") if state else ("_pid_lock",):
+            lock = getattr(self, attr, None)
+            if lock:
+                lock.release_lock()
+                setattr(self, attr, None)
+
+    async def _start_locked(self) -> None:
+        """Body of :meth:`start` once the single-instance locks are held."""
         # Validate bind-aware auth configuration before starting
         from .auth import assert_external_bind_safe
         self.config.host = self._host
@@ -2768,14 +2806,8 @@ class WebSocketGateway:
                 self._ledger_heartbeat_loop(), name="gateway-lifecycle-ledger"
             )
         try:
+            # Lock cleanup on failure is handled by ``start()``.
             await self._server.serve()
-        except Exception as e:
-            # Clean up PID lock on any startup failure
-            if hasattr(self, '_pid_lock') and self._pid_lock:
-                self._pid_lock.release_lock()
-                self._pid_lock = None
-            # Re-raise the original exception
-            raise
         finally:
             self._disarm_watchdog()
             await self._cancel_liveness_task()
@@ -3993,9 +4025,7 @@ class WebSocketGateway:
                 logger.debug("Failed to shut down pooled MCP client")
         self._mcp_pool.clear()
 
-        # Release PID lock
-        if hasattr(self, '_pid_lock') and self._pid_lock:
-            self._pid_lock.release_lock()
+        self._release_locks()
         
         logger.info("Gateway stopped")
     
@@ -12436,6 +12466,9 @@ class WebSocketGateway:
             except (NotImplementedError, OSError, ValueError):
                 pass
 
+        # Claim the state root BEFORE channels start: channel setup replays the
+        # inbound journal, which a rejected second gateway must never touch.
+        self._acquire_state_lock()
         try:
             await _run_all()
         finally:
@@ -12486,3 +12519,4 @@ class WebSocketGateway:
                             logger.warning("Error draining websocket sessions: %s", e)
             finally:
                 self._cancel_shutdown_deadline()
+                self._release_locks()

src/praisonai-bot/tests/unit/gateway/test_state_scoped_lock.py

@@ -0,0 +1,125 @@
+"""A gateway's single-instance identity is its shared state, not its port (#6208)."""
+
+import asyncio
+import multiprocessing
+import os
+from pathlib import Path
+
+import pytest
+
+from praisonai_bot.gateway.port_utils import (
+    format_collision_error,
+    format_state_collision_error,
+    resolve_state_dir,
+    state_lock,
+)
+
+
+def _claim(d, start, done, results):
+    start.wait()  # release every worker at once so the claims genuinely race
+    results.put(state_lock(Path(d)).acquire_lock("127.0.0.1", 8765))
+    done.wait()  # stay alive until all claimed, so no lock is seen as stale
+
+
+def _seed_foreign_holder(lock, pid):
+    lock.lock_file.write_text(f"{pid}\n127.0.0.1\n8765\n0\n\n")
+
+
+def test_second_gateway_on_another_port_is_refused(tmp_path, monkeypatch):
+    first = state_lock(tmp_path, port=8765)
+    _seed_foreign_holder(first, os.getppid())  # a live process other than us
+    second = state_lock(tmp_path, port=8766)
+    assert second.lock_file == first.lock_file
+    assert second.acquire_lock("127.0.0.1", 8766) is False
+    msg = format_state_collision_error(tmp_path, second.get_lock_info())
+    assert str(tmp_path) in msg and "HOME=" in msg and "PRAISONAI_HOME" not in msg
+
+
+def test_distinct_state_dirs_are_independent(tmp_path):
+    a = state_lock(tmp_path / "a")
+    _seed_foreign_holder(a, os.getppid())
+    assert state_lock(tmp_path / "b").acquire_lock("127.0.0.1", 8766) is True
+
+
+def test_stale_state_lock_is_reclaimed(tmp_path, monkeypatch):
+    lock = state_lock(tmp_path)
+    _seed_foreign_holder(lock, 999999)
+    monkeypatch.setattr(lock, "_is_process_running", lambda pid: False)
+    assert lock.acquire_lock("127.0.0.1", 8765) is True
+
+
+def test_state_dir_ignores_praisonai_home(tmp_path, monkeypatch):
+    # The stores live under ~/.praisonai regardless of PRAISONAI_HOME, so a
+    # different PRAISONAI_HOME must not escape the lock.
+    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
+    assert resolve_state_dir() == Path.home() / ".praisonai"
+
+
+def test_concurrent_claims_have_exactly_one_winner(tmp_path):
+    ctx = multiprocessing.get_context("spawn")
+    n = 8
+    start, done, results = ctx.Barrier(n), ctx.Barrier(n), ctx.Queue()
+    procs = [ctx.Process(target=_claim, args=(str(tmp_path), start, done, results)) for _ in range(n)]
+    for p in procs:
+        p.start()
+    outcomes = [results.get(timeout=60) for _ in range(n)]
+    for p in procs:
+        p.join(timeout=60)
+    assert sum(outcomes) == 1
+
+
+def test_claim_never_overwrites_existing_lock(tmp_path):
+    lock = state_lock(tmp_path)
+    _seed_foreign_holder(lock, os.getppid())
+    assert lock.acquire_lock("127.0.0.1", 8766) is False
+    assert lock.lock_file.read_text().startswith(str(os.getppid()))
+
+
+def test_failed_startup_releases_both_locks(tmp_path, monkeypatch):
+    from praisonai_bot.gateway import server as srv
+
+    monkeypatch.setenv("HOME", str(tmp_path))  # isolate both lock files
+    gw = srv.WebSocketGateway.__new__(srv.WebSocketGateway)
+    gw._is_running, gw._host, gw._port = False, "127.0.0.1", 0
+    gw._pid_lock = gw._state_lock = None
+
+    async def boom():
+        raise RuntimeError("bad config")
+
+    gw._start_locked = boom
+    monkeypatch.setattr("praisonai_bot.gateway.port_utils.check_port_available", lambda h, p: (True, None))
+    with pytest.raises(RuntimeError, match="bad config"):
+        asyncio.run(gw.start())
+    assert gw._pid_lock is None and gw._state_lock is None
+    assert not list((tmp_path / ".praisonai").glob("*.pid"))
+
+
+def test_start_failure_keeps_state_lock_owned_by_start_with_config(tmp_path, monkeypatch):
+    # start_with_config claims the state lock before channels start; if start()
+    # then fails, channels are still polling, so start() must not drop it.
+    from praisonai_bot.gateway import server as srv
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    gw = srv.WebSocketGateway.__new__(srv.WebSocketGateway)
+    gw._is_running, gw._host, gw._port = False, "127.0.0.1", 0
+    gw._pid_lock = gw._state_lock = None
+    gw._acquire_state_lock()
+    held = gw._state_lock
+
+    async def boom():
+        raise RuntimeError("bind error")
+
+    gw._start_locked = boom
+    monkeypatch.setattr("praisonai_bot.gateway.port_utils.check_port_available", lambda h, p: (True, None))
+    with pytest.raises(RuntimeError, match="bind error"):
+        asyncio.run(gw.start())
+    assert gw._pid_lock is None
+    assert gw._state_lock is held and held.lock_file.exists()
+    gw._release_locks()  # start_with_config's finally, after channels stop
+    assert gw._state_lock is None and not held.lock_file.exists()
+
+
+def test_collision_error_no_longer_suggests_bare_port_change():
+    msg = format_collision_error("127.0.0.1", 8765, {"pid": 1, "is_running": True})
+    assert "Or use a different port" not in msg
+    assert "HOME=" in msg and "PRAISONAI_HOME" not in msg