Merge branch 'main' into feature/adfs-unitest

calculus-ask · web-flow · commit bb873a0cc43c · 2026-04-01T10:56:40.000+05:30
diff --git a/.secrets.baseline b/.secrets.baseline
@@ -1,6 +1,6 @@
 {
   "exclude": {
-    "files": "package-lock.json|Cargo.lock|^.secrets.baseline$|scripts/sign_image.sh|scripts/zap|sonar-project.properties",
+    "files": "package-lock.json|Cargo.lock|^.secrets.baseline$|scripts/sign_image.sh|scripts/zap|sonar-project.properties|^/Users/brian/dev/github.ibm.com/contextforge-org/sps-pipeline-config/.secrets.baseline$|^./.secrets.baseline$",
     "lines": null
   },
   "generated_at": "2026-04-01T00:41:17Z",
@@ -6843,6 +6843,32 @@
         "verified_result": null
       }
     ],
+    "plugins_rust/secrets_detection/benches/secrets_detection.rs": [
+      {
+        "hashed_secret": "86de8c52637ec530fe39b0a8471da9b8764d5242",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 53,
+        "type": "AWS Access Key",
+        "verified_result": null
+      },
+      {
+        "hashed_secret": "5b8c02feb3811310ff452e9a812b9f98873f36bf",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 69,
+        "type": "SoftLayer Credentials",
+        "verified_result": null
+      },
+      {
+        "hashed_secret": "0fa6996ddd42e0f9db3f1c04d06453026d56da0f",
+        "is_secret": false,
+        "is_verified": false,
+        "line_number": 969,
+        "type": "Secret Keyword",
+        "verified_result": null
+      }
+    ],
     "podman-compose-sonarqube.yaml": [
       {
         "hashed_secret": "345e9ea7c857e75dedd9edb24c232e1cab297c19",
diff --git a/mcpgateway/bootstrap_db.py b/mcpgateway/bootstrap_db.py
@@ -36,7 +36,9 @@
 import json
 import os
 from pathlib import Path
+import random
 import tempfile
+import time
 from typing import cast
 
 # Third-Party
@@ -102,7 +104,7 @@ def advisory_lock(conn: Connection):
     Acquire a distributed advisory lock to serialize migrations across multiple instances.
 
     Behavior depends on the database backend:
-    - Postgres: Uses `pg_advisory_lock` (blocking)
+    - Postgres: Uses `pg_try_advisory_lock` (non-blocking)
     - SQLite: Fallback to local `FileLock`
 
     Args:
@@ -119,8 +121,34 @@ def advisory_lock(conn: Connection):
     pg_lock_id = 42424242424242
 
     if dialect == "postgresql":
-        logger.info("Acquiring Postgres advisory lock...")
-        conn.execute(text(f"SELECT pg_advisory_lock({pg_lock_id})"))
+        logger.info("Attempting to acquire Postgres advisory lock...")
+
+        # Retry parameters
+        max_retries = 60  # 60 attempts
+        base_delay = 1.0  # Start with 1 second
+        max_delay = 10.0  # Cap at 10 seconds
+
+        acquired = False
+        for attempt in range(max_retries):
+            # Try non-blocking lock
+            result = conn.execute(text(f"SELECT pg_try_advisory_lock({pg_lock_id})"))
+            acquired = result.scalar()
+
+            if acquired:
+                logger.info(f"Acquired Postgres advisory lock on attempt {attempt + 1}")
+                break
+
+            # Exponential backoff with jitter
+            delay = min(base_delay * (1.5**attempt), max_delay)
+            jitter = delay * random.uniform(-0.1, 0.1)  # nosec B311  # noqa: DUO102
+            sleep_time = delay + jitter
+
+            logger.info(f"Lock held by another instance, retrying in {sleep_time:.1f}s (attempt {attempt + 1}/{max_retries})")
+            time.sleep(sleep_time)
+
+        if not acquired:
+            raise TimeoutError(f"Failed to acquire advisory lock after {max_retries} attempts")
+
         try:
             yield
         finally:
diff --git a/mcpgateway/services/gateway_service.py b/mcpgateway/services/gateway_service.py
@@ -476,6 +476,10 @@ def __init__(self) -> None:
             self._leader_ttl = settings.redis_leader_ttl
             self._leader_heartbeat_interval = settings.redis_leader_heartbeat_interval
             self._leader_heartbeat_task: Optional[asyncio.Task] = None
+            self._follower_election_task: Optional[asyncio.Task] = None
+
+            # Log instance mapping for debugging
+            logger.info(f"Instance started: instance_id={self._instance_id}, port={settings.port}, pid={os.getpid()}")
 
         # Always initialize file lock as fallback (used if Redis connection fails at runtime)
         if settings.cache_type != "none":
@@ -588,8 +592,12 @@ async def initialize(self) -> None:
                 logger.info("Acquired Redis leadership. Starting health check and heartbeat tasks.")
                 self._health_check_task = asyncio.create_task(self._run_health_checks(user_email))
                 self._leader_heartbeat_task = asyncio.create_task(self._run_leader_heartbeat())
+            else:
+                # Did not acquire leadership - start follower election loop
+                logger.info("Did not acquire leadership. Starting follower election loop.")
+                self._follower_election_task = asyncio.create_task(self._run_follower_election(user_email))
         else:
-            # Always create the health check task in filelock mode; leader check is handled inside.
+            # No Redis available - always create the health check task in filelock mode
             self._health_check_task = asyncio.create_task(self._run_health_checks(user_email))
 
     async def shutdown(self) -> None:
@@ -608,14 +616,25 @@ async def shutdown(self) -> None:
             >>> len(service._active_gateways)
             0
         """
+        # Cancel follower election FIRST to prevent it from spawning new
+        # health-check / heartbeat tasks while we are tearing down.
+        if getattr(self, "_follower_election_task", None):
+            self._follower_election_task.cancel()
+            try:
+                await self._follower_election_task
+            except asyncio.CancelledError:
+                pass
+
+        # Now safe to cancel health-check and heartbeat (handles may have been
+        # overwritten by follower election just before cancellation — that is fine,
+        # we always cancel whichever task the attribute currently points to).
         if self._health_check_task:
             self._health_check_task.cancel()
             try:
                 await self._health_check_task
             except asyncio.CancelledError:
                 pass
 
-        # Cancel leader heartbeat task if running
         if getattr(self, "_leader_heartbeat_task", None):
             self._leader_heartbeat_task.cancel()
             try:
@@ -3901,34 +3920,87 @@ def get_first_gateway_by_url(self, db: Session, url: str, team_id: Optional[str]
         return self.convert_gateway_to_read(result)
 
     async def _run_leader_heartbeat(self) -> None:
-        """Run leader heartbeat loop to keep leader key alive.
-
-        This runs independently from health checks to ensure the leader key
-        is refreshed frequently enough (every redis_leader_heartbeat_interval seconds)
-        to prevent expiration during long-running health check operations.
+        """Run leader heartbeat loop with Redis reconnection support.
 
-        The loop exits if this instance loses leadership.
+        Refreshes the leader key TTL every heartbeat interval. Exits and starts
+        follower election if leadership is lost or after consecutive failures.
         """
+        consecutive_failures = 0
+        max_failures = 3
+
         while True:
             try:
                 await asyncio.sleep(self._leader_heartbeat_interval)
 
                 if not self._redis_client:
-                    return
+                    logger.warning("Redis client unavailable in heartbeat")
+                    consecutive_failures += 1
+                    if consecutive_failures >= max_failures:
+                        logger.error("Lost Redis connection, stopping heartbeat")
+                        return
+                    continue
 
                 # Check if we're still the leader
                 current_leader = await self._redis_client.get(self._leader_key)
                 if current_leader != self._instance_id:
                     logger.info("Lost Redis leadership, stopping heartbeat")
+                    self._start_follower_election()
                     return
 
                 # Refresh the leader key TTL
                 await self._redis_client.expire(self._leader_key, self._leader_ttl)
                 logger.debug(f"Leader heartbeat: refreshed TTL to {self._leader_ttl}s")
+                consecutive_failures = 0
+
+            except Exception as e:
+                consecutive_failures += 1
+                logger.warning(f"Leader heartbeat error (failure {consecutive_failures}/{max_failures}): {e}")
+                if consecutive_failures >= max_failures:
+                    logger.error("Too many consecutive heartbeat failures, starting follower election")
+                    self._start_follower_election()
+                    return
+
+    def _start_follower_election(self) -> None:
+        """Start a follower election task if one is not already running."""
+        if self._follower_election_task is None or self._follower_election_task.done():
+            self._follower_election_task = asyncio.create_task(self._run_follower_election(settings.platform_admin_email))
+
+    async def _run_follower_election(self, user_email: str) -> None:
+        """Continuously attempt to acquire leadership when not the leader.
+
+        This runs on follower instances and polls Redis to claim leadership
+        when the current leader key expires or becomes available.
+
+        Args:
+            user_email: Email of the user for OAuth token lookup
+        """
+        retry_interval = max(1, self._leader_ttl // 3)  # Poll at 1/3 of TTL
+
+        while True:
+            try:
+                await asyncio.sleep(retry_interval)
+
+                if not self._redis_client:
+                    logger.warning("Redis client unavailable, cannot attempt election.")
+                    continue
+
+                # Attempt to acquire leadership
+                is_leader = await self._redis_client.set(self._leader_key, self._instance_id, ex=self._leader_ttl, nx=True)
+
+                if is_leader:
+                    logger.info("Acquired Redis leadership via follower election. Starting health check and heartbeat.")
+                    # Cancel stale tasks from a previous leadership period to prevent
+                    # orphaned loops running alongside the new ones.
+                    if self._health_check_task and not self._health_check_task.done():
+                        self._health_check_task.cancel()
+                    if getattr(self, "_leader_heartbeat_task", None) and not self._leader_heartbeat_task.done():
+                        self._leader_heartbeat_task.cancel()
+                    self._health_check_task = asyncio.create_task(self._run_health_checks(user_email))
+                    self._leader_heartbeat_task = asyncio.create_task(self._run_leader_heartbeat())
+                    return  # Exit follower loop, now running as leader
 
             except Exception as e:
-                logger.warning(f"Leader heartbeat error: {e}")
-                # Continue trying - the main health check loop will handle leadership loss
+                logger.warning(f"Follower election error: {e}", exc_info=True)
 
     async def _run_health_checks(self, user_email: str) -> None:
         """Run health checks periodically,
diff --git a/tests/js/admin.test.js b/tests/js/admin.test.js
@@ -10,7 +10,7 @@ let escapeHtml;
 beforeAll(() => {
     const win = loadAdminJs();
     escapeHtml = win.escapeHtml;
-});
+}, 5000);
 
 afterAll(() => {
     cleanupAdminJs();
diff --git a/tests/js/helpers/admin-env.js b/tests/js/helpers/admin-env.js
@@ -37,15 +37,26 @@ let instrumentedCode = null;
 export function loadAdminJs(options = {}) {
     if (!instrumentedCode) {
         const adminJsContent = fs.readFileSync(adminJsPath, "utf8");
-        const instrumenter = createInstrumenter({
-            compact: false,
-            esModules: false,
-            coverageVariable: "__coverage__",
-        });
-        instrumentedCode = instrumenter.instrumentSync(
-            adminJsContent,
-            adminJsPath,
-        );
+        // Skip instrumentation when not collecting coverage — instrumenting a
+        // ~1.3 MB file takes several seconds and is only needed for coverage reports.
+        // process.argv is unreliable in Vitest workers (spawned as separate processes),
+        // so fall back to Vitest's internal worker state.
+        const isCoverageRun =
+            process.argv.includes("--coverage") ||
+            globalThis.__vitest_worker__?.config?.coverage?.enabled === true;
+        if (isCoverageRun) {
+            const instrumenter = createInstrumenter({
+                compact: false,
+                esModules: false,
+                coverageVariable: "__coverage__",
+            });
+            instrumentedCode = instrumenter.instrumentSync(
+                adminJsContent,
+                adminJsPath,
+            );
+        } else {
+            instrumentedCode = adminJsContent;
+        }
     }
 
     dom = new JSDOM("<!DOCTYPE html><html><body></body></html>", {
diff --git a/tests/unit/mcpgateway/services/test_gateway_service.py b/tests/unit/mcpgateway/services/test_gateway_service.py
@@ -7240,6 +7240,7 @@ async def test_heartbeat_no_redis(self, gateway_service):
         gateway_service._leader_ttl = 30
         gateway_service._redis_client = None
         gateway_service._leader_heartbeat_interval = 0
+        gateway_service._follower_election_task = None
         await gateway_service._run_leader_heartbeat()
 
     @pytest.mark.asyncio
@@ -7251,14 +7252,17 @@ async def test_heartbeat_lost_leadership(self, gateway_service):
         gateway_service._redis_client = AsyncMock()
         gateway_service._redis_client.get = AsyncMock(return_value="other-leader")
         gateway_service._leader_heartbeat_interval = 0
-        await gateway_service._run_leader_heartbeat()
+        gateway_service._follower_election_task = None
+        with patch.object(gateway_service, "_start_follower_election"):
+            await gateway_service._run_leader_heartbeat()
 
     @pytest.mark.asyncio
     async def test_heartbeat_refreshes_ttl(self, gateway_service):
         """Heartbeat refreshes TTL then exits when losing leadership."""
         gateway_service._instance_id = "test-id"
         gateway_service._leader_key = "leader:health_check"
         gateway_service._leader_ttl = 30
+        gateway_service._follower_election_task = None
         call_count = 0
 
         async def mock_get(*args):
@@ -7272,7 +7276,8 @@ async def mock_get(*args):
         gateway_service._redis_client.get = mock_get
         gateway_service._redis_client.expire = AsyncMock()
         gateway_service._leader_heartbeat_interval = 0
-        await gateway_service._run_leader_heartbeat()
+        with patch.object(gateway_service, "_start_follower_election"):
+            await gateway_service._run_leader_heartbeat()
         gateway_service._redis_client.expire.assert_awaited_once()
 
 
diff --git a/tests/unit/mcpgateway/services/test_gateway_service_redis_leadership.py b/tests/unit/mcpgateway/services/test_gateway_service_redis_leadership.py
diff --git a/tests/unit/mcpgateway/test_bootstrap_db_advisory_lock.py b/tests/unit/mcpgateway/test_bootstrap_db_advisory_lock.py
