BUGFIX:

- Improved error handling on pool timeouts, retries added.

Author: karel26

core/pubsub.py

@@ -118,8 +118,9 @@ async def do_read():
                 query = sql.SQL("DELETE FROM {table} WHERE id = ANY(%s::int[])").format(table=sql.Identifier(self.name))
                 await conn.execute(query, (ids_to_delete,))
 
-        await asyncio.sleep(1)  # Ensure the rest of __init__ has finished
         while not self._stop_event.is_set():
+            # wait one second to clean up any issue
+            await asyncio.sleep(1)
             with suppress(OperationalError):
                 async with await AsyncConnection.connect(self.url, autocommit=True) as conn:
                     while not self._stop_event.is_set():

plugins/cloud/logger.py

@@ -28,6 +28,7 @@ def __init__(self, node: Node, url: str):
         self.node = node
         self.url = url
         self.cwd = os.getcwd()
+        self.pending_futures = set()
 
     def format_traceback(self, trace: traceback) -> tuple[str, int, list[str]]:
         ret = []
@@ -80,4 +81,15 @@ async def send_post(self, record: logging.LogRecord):
     def emit(self, record: logging.LogRecord):
         if record.levelno in [logging.ERROR, logging.CRITICAL] and record.exc_info is not None:
             with suppress(Exception):
-                asyncio.create_task(self.send_post(record))
+                loop = asyncio.get_event_loop()
+                future = asyncio.run_coroutine_threadsafe(self.send_post(record), loop)
+                self.pending_futures.add(future)
+                future.add_done_callback(lambda f: self.pending_futures.discard(f))
+
+    def close(self):
+        for future in list(self.pending_futures):
+            try:
+                future.result(timeout=1.0)
+            except Exception:
+                pass
+        super().close()

plugins/mission/commands.py

@@ -135,10 +135,13 @@ def __init__(self, bot: DCSServerBot, listener: Type[MissionEventListener] = Non
         self.update_channel_name.add_exception_type(AttributeError)
         self.update_channel_name.start()
         self.afk_check.start()
+        self.check_for_unban.add_exception_type(psycopg.DatabaseError)
         self.check_for_unban.start()
         self.expire_token.add_exception_type(psycopg.DatabaseError)
         self.expire_token.start()
         if self.bot.locals.get('autorole', {}):
+            self.check_roles.add_exception_type(psycopg.DatabaseError)
+            self.check_roles.add_exception_type(discord.errors.DiscordException)
             self.check_roles.start()
 
     async def cog_unload(self):
@@ -1872,25 +1875,22 @@ async def expire_token(self):
 
     @tasks.loop(minutes=1.0)
     async def check_for_unban(self):
-        try:
-            async with self.apool.connection() as conn:
-                async with conn.transaction():
-                    cursor = await conn.execute("""
-                        SELECT ucid FROM bans WHERE banned_until < (NOW() AT TIME ZONE 'utc')
-                    """)
-                    rows = await cursor.fetchall()
-                    for row in rows:
-                        for server in self.bot.servers.values():
-                            if server.status not in [Status.PAUSED, Status.RUNNING, Status.STOPPED]:
-                                continue
-                            await server.send_to_dcs({
-                                "command": "unban",
-                                "ucid": row[0]
-                            })
-                        # delete unbanned accounts from the database
-                        await conn.execute("DELETE FROM bans WHERE ucid = %s", (row[0], ))
-        except Exception as ex:
-            self.log.exception(ex)
+        async with self.apool.connection() as conn:
+            async with conn.transaction():
+                cursor = await conn.execute("""
+                    SELECT ucid FROM bans WHERE banned_until < (NOW() AT TIME ZONE 'utc')
+                """)
+                rows = await cursor.fetchall()
+                for row in rows:
+                    for server in self.bot.servers.values():
+                        if server.status not in [Status.PAUSED, Status.RUNNING, Status.STOPPED]:
+                            continue
+                        await server.send_to_dcs({
+                            "command": "unban",
+                            "ucid": row[0]
+                        })
+                    # delete unbanned accounts from the database
+                    await conn.execute("DELETE FROM bans WHERE ucid = %s", (row[0], ))
 
     @check_for_unban.before_loop
     async def before_check_unban(self):

run.py

@@ -15,6 +15,7 @@
 
 from datetime import datetime
 from psycopg import OperationalError
+from psycopg_pool import PoolTimeout
 
 # DCSServerBot imports
 try:
@@ -265,6 +266,9 @@ async def run_node(name, config_dir=None, no_autoupdate=False) -> int:
         # do not restart again
         exit(-2)
     except psycopg.OperationalError as ex:
+        if isinstance(ex, PoolTimeout):
+            # try again on pool timeouts
+            exit(-1)
         log.error(f"Database Error: {ex}", exc_info=True)
         input("Press any key to continue ...")
         # do not restart again

services/monitoring/service.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import psutil
+import psycopg
 import shutil
 import sys
 
@@ -51,6 +52,7 @@ async def start(self):
             self.space_warning_sent['C:'] = False
             self.space_alert_sent['C:'] = False
         self.check_autoexec()
+        self.monitoring.add_exception_type(psycopg.DatabaseError)
         self.monitoring.start()
         if self.get_config().get('time_sync', False):
             time_server = self.get_config().get('time_server', None)