Merge branch 'development'

Author: karel26

core/data/impl/nodeimpl.py

@@ -38,7 +38,7 @@
 from packaging.version import parse
 from pathlib import Path
 from psycopg import sql
-from psycopg.errors import InFailedSqlTransaction, ConnectionTimeout, UniqueViolation, UndefinedTable, UndefinedColumn
+from psycopg.errors import ConnectionTimeout, UniqueViolation, UndefinedTable, UndefinedColumn
 from psycopg.types.json import Json
 from psycopg_pool import ConnectionPool, AsyncConnectionPool
 from typing import Awaitable, Callable, Any
@@ -651,6 +651,10 @@ async def upgrade(self):
         async with self.cpool.connection() as conn:
             await self._upgrade(self.master, conn)
 
+    @override
+    async def is_alive(self, timeout: int = 30) -> bool:
+        return True
+
     @override
     async def get_dcs_branch_and_version(self) -> tuple[str, str]:
         if not self.dcs_branch or not self.dcs_version:
@@ -1097,8 +1101,8 @@ async def get_master() -> tuple[str | None, str | None, str, bool]:
             except psycopg.errors.UndefinedColumn:
                 # add the column if missing
                 await conn.rollback()
-                await conn.execute("ALTER TABLE cluster ADD COLUMN IF NOT EXISTS takeover_requested_by TEXT NULL")
-                await conn.commit()
+                async with conn.transaction():
+                    await conn.execute("ALTER TABLE cluster ADD COLUMN IF NOT EXISTS takeover_requested_by TEXT NULL")
                 return await get_master()
 
         async def is_node_alive(node: str, timeout: int) -> bool:
@@ -1283,8 +1287,9 @@ async def check_nodes():
                                 ON CONFLICT (guild_id, node) DO UPDATE 
                                 SET last_seen = (NOW() AT TIME ZONE 'UTC')
                             """, (self.guild_id, self.name))
-        except InFailedSqlTransaction:
-            # we should only be here when the CLUSTER table does not exist yet
+        except UndefinedTable:
+            # we should only be here when the CLUSTER table does not exist
+            # it will be created directly afterward
             return True
 
     @tasks.loop(seconds=5.0)

core/data/node.py

@@ -280,3 +280,7 @@ async def info(self) -> dict:
     @abstractmethod
     async def get_config(self) -> dict:
         raise NotImplementedError()
+
+    @abstractmethod
+    async def is_alive(self, timeout: int = 30) -> bool:
+        raise NotImplementedError()

core/data/proxy/nodeproxy.py

@@ -7,6 +7,7 @@
 from core.services.registry import ServiceRegistry
 from core.utils import async_cache, cache_with_expiration
 from pathlib import Path
+from psycopg import sql
 from typing import TYPE_CHECKING
 from typing_extensions import override
 
@@ -29,6 +30,7 @@ def __init__(self, local_node: NodeImpl, name: str, public_ip: str, dcs_version:
         self.local_node = local_node
         self.pool = self.local_node.pool
         self.apool = self.local_node.apool
+        self.cpool = self.local_node.cpool
         self.log = self.local_node.log
         self._public_ip = public_ip
         self.locals = self.read_locals()
@@ -467,3 +469,14 @@ async def get_config(self) -> dict:
             "object": "Node",
             "method": "get_config"
         }, timeout=timeout, node=self.name)
+
+    @override
+    async def is_alive(self, timeout: int = 30) -> bool:
+        async with self.cpool.connection() as conn:
+            query = sql.SQL("""
+                SELECT COUNT(*) FROM nodes 
+                WHERE guild_id = %s AND node = %s 
+                AND last_seen > (NOW() AT TIME ZONE 'UTC' - interval {interval})
+            """).format(interval=sql.Literal(f"{timeout} seconds"))
+            cursor = await conn.execute(query, (self.guild_id, self.name))
+            return (await cursor.fetchone())[0] == 1

requirements.in

@@ -27,7 +27,7 @@ discord.py==2.7.1
 eyeD3==0.9.9
 
 # FastAPI: A modern, fast (high-performance), web framework for building APIs with Python 3.6+ based on standard Python type hints
-fastapi==0.135.3
+fastapi==0.136.0
 
 # FuzzyWuzzy: Python library that uses Levenshtein Distance to calculate the differences between strings
 fuzzywuzzy==0.18.0
@@ -45,7 +45,7 @@ jinja2==3.1.6
 jsonschema==4.26.0
 
 # Fast LUA parser
-lupa==2.7
+lupa==2.8
 
 # Matplotlib: Python 2D plotting library
 matplotlib==3.10.8

requirements.txt

@@ -31,7 +31,7 @@ discord-py==2.7.1
 docopt==0.6.2
 et-xmlfile==2.0.0
 eyed3==0.9.9
-fastapi==0.135.3
+fastapi==0.136.0
 filetype==1.2.0
 fonttools==4.62.1
 frozenlist==1.8.0
@@ -47,7 +47,7 @@ jsonschema==4.26.0
 jsonschema-specifications==2025.9.1
 kiwisolver==1.5.0
 levenshtein==0.27.3
-lupa==2.7
+lupa==2.8
 markdown-it-py==4.0.0
 markupsafe==3.0.3
 matplotlib==3.10.8

services/bot/service.py

@@ -190,25 +190,28 @@ async def stop(self):
         await super().stop()
 
     async def alert(self, title: str, message: str, server: Server | None = None) -> None:
-        # if we have dedicated managers of a server, send the alerts to them
-        if server and server.locals.get('managed_by'):
-            alert_roles = server.locals['managed_by']
-        # use the default Alert role otherwise
-        else:
-            alert_roles = self.bot.roles['Alert']
         try:
-            mentions = ''.join([self.bot.get_role(role).mention for role in alert_roles if role is not None])
-        except AttributeError:
-            self.log.error(f"Alert-Role {alert_roles} not found.")
-            mentions = ""
-        embed = utils.create_warning_embed(title=title, text=utils.escape_string(message))
-        admin_channel = self.bot.get_admin_channel(server)
-        audit_channel = self.bot.get_channel(self.bot.locals.get('channels', {}).get('audit', -1))
-        channel = admin_channel or audit_channel
-        if channel:
-            await channel.send(content=mentions, embed=embed)
-        else:
-            self.log.critical(f"{title}: {message}")
+            # if we have dedicated managers of a server, send the alerts to them
+            if server and server.locals.get('managed_by'):
+                alert_roles = server.locals['managed_by']
+            # use the default Alert role otherwise
+            else:
+                alert_roles = self.bot.roles['Alert']
+            try:
+                mentions = ''.join([self.bot.get_role(role).mention for role in alert_roles if role is not None])
+            except AttributeError:
+                self.log.error(f"Alert-Role {alert_roles} not found.")
+                mentions = ""
+            embed = utils.create_warning_embed(title=title, text=utils.escape_string(message))
+            admin_channel = self.bot.get_admin_channel(server)
+            audit_channel = self.bot.get_channel(self.bot.locals.get('channels', {}).get('audit', -1))
+            channel = admin_channel or audit_channel
+            if channel:
+                await channel.send(content=mentions, embed=embed)
+            else:
+                self.log.critical(f"{title}: {message}")
+        except Exception:
+            self.log.warning("Audit message discarded due to master takeover: " + message)
 
     async def install_fonts(self):
         font_dir = Path('fonts')

services/servicebus/service.py

@@ -117,6 +117,19 @@ async def switch(self, master: bool):
                     }
                 }, node=node)
         else:
+            # figure the master node
+            async with self.node.cpool.connection() as conn:
+                cursor = await conn.execute("SELECT master FROM cluster WHERE guild_id = %s", (self.node.guild_id, ))
+                row = await cursor.fetchone()
+                if not row:
+                    self.log.warning("No master available, can't register.")
+                    return
+                master = row[0]
+
+            if master not in await self.node.get_active_nodes():
+                self.log.debug(f"Master node {master} is not active (yet), waiting ...")
+                return
+
             await self.send_to_node({
                 "command": "rpc",
                 "service": self.__class__.__name__,
@@ -274,8 +287,16 @@ async def register_remote_node(self, name: str, public_ip: str, dcs_version: str
         if name == self.node.name:
             return
 
-        self.log.info(f"- Registering remote node {name} ...")
+        if self.node.all_nodes.get(name):
+            self.log.debug(f"Node {name} already registered, skipping registration request.")
+            return
+
         node = NodeProxy(self.node, name, public_ip, dcs_version)
+        if not await node.is_alive(self.node.config.get('cluster', {}).get('heartbeat', 30)):
+            self.log.warning(f"Node {name} is not alive (anymore). Skipping registration request.")
+            return
+
+        self.log.info(f"- Registering remote node {name} ...")
         # we did not find a configuration for this node, load it from remote
         if not node.locals:
             node.locals = await node.get_config()
@@ -466,12 +487,12 @@ async def is_banned(self, ucid: str) -> dict | None:
                 return await cursor.fetchone()
 
     async def init_remote_server(self, server_name: str, status: str, instance: str, home: str,
-                                 settings: dict, options: dict, node: Node, channels: dict, dcs_port: int,
+                                 settings: dict, options: dict, node: Node | str, channels: dict, dcs_port: int,
                                  webgui_port: int, maintenance: bool) -> None:
         from core import InstanceProxy
 
         # init event for an unregistered remote node received or a race condition due to master switches, ignoring
-        if not node or node == self.node:
+        if not node or node == self.node or isinstance(node, str):
             return
         try:
             server: ServerProxy = cast(ServerProxy, self.servers.get(server_name))