Mariadb fix of dead server issue due to unrecycled sessions (#204)

* improve database session recycling

* version increased

* MariaDB unrecycled sessions fix to webserver database

* Removed mariadbconnector

Author: stefanodellosa-personal

wadas/_version.py

@@ -17,5 +17,5 @@
 # Date: 2024-08-14
 # Description: module to keep track of WADAS version
 
-__version__ = "v1.0.0.a6"
+__version__ = "v1.0.0.a7"
 __dbversion__ = __version__

wadas/domain/database.py

@@ -193,8 +193,8 @@ def get_engine(cls):
                     if (DataBase.wadas_db.type == DataBase.DBTypes.SQLITE)
                     else create_engine(
                         DataBase.wadas_db.get_connection_string(),
-                        pool_recycle=1800,  # Recycle connection every 30 minutes
-                        pool_pre_ping=True,  # Check if connection is still valid
+                        pool_recycle=300,  # Recycle connection every 5 minutes (under MariaDB wait_timeout)
+                        pool_pre_ping=True,  # Check if connection is still valid before use
                     )
                 )
         return DataBase.wadas_db_engine
@@ -237,12 +237,18 @@ def create_session(cls, retry_count=0):
             if engine := cls.get_engine():
                 # If db is not SQLite, check engine status (SQLite has no pre-existing sessions)
                 if DataBase.wadas_db.type != DataBase.DBTypes.SQLITE:
-                    with engine.connect() as connection:
-                        if connection.invalidated:
-                            logger.warning("Connection invalidated, disposing engine...")
-                            engine.dispose()
-                            DataBase.wadas_db_engine = None  # Force new engine
-                            engine = cls.get_engine()
+                    try:
+                        with engine.connect() as connection:
+                            if connection.invalidated:
+                                logger.warning("Connection invalidated, disposing engine...")
+                                engine.dispose()
+                                DataBase.wadas_db_engine = None  # Force new engine
+                                engine = cls.get_engine()
+                    except (InterfaceError, SQLAlchemyOperationalError, mariadbOperationalerror):
+                        logger.warning("Pre-ping connection check failed, disposing engine...")
+                        engine.dispose()
+                        DataBase.wadas_db_engine = None
+                        engine = cls.get_engine()
 
                 Session = sessionmaker(bind=engine)
                 return Session()
@@ -277,6 +283,8 @@ def get_db_uuid(cls):
                 return result[0] if result else None
             except Exception:
                 return None
+            finally:
+                session.close()
         else:
             return None
 
@@ -290,6 +298,8 @@ def get_db_version(cls):
                 return result[0] if result else None
             except Exception:
                 return None
+            finally:
+                session.close()
         else:
             return None
 
@@ -767,36 +777,40 @@ def remove_actuator_from_camera(cls, camera, actuator):
 
         if cls.get_instance():
             logger.debug("Removing actuator %s from camera %s in db.", actuator.id, camera.id)
-            if cls.create_session():
-                # Retrieve camera and actuator db instances
-                camera_db_id = cls.get_camera_id(camera.id)
-                actuator_db_id = cls.get_actuator_id(actuator.id)
-
-                stmt = delete(camera_actuator_association).where(
-                    and_(
-                        camera_actuator_association.c.actuator_id == actuator_db_id,
-                        camera_actuator_association.c.camera_id == camera_db_id,
-                    )
+            # Retrieve camera and actuator db instances
+            camera_db_id = cls.get_camera_id(camera.id)
+            actuator_db_id = cls.get_actuator_id(actuator.id)
+
+            stmt = delete(camera_actuator_association).where(
+                and_(
+                    camera_actuator_association.c.actuator_id == actuator_db_id,
+                    camera_actuator_association.c.camera_id == camera_db_id,
                 )
-                cls.run_query(stmt)
-            else:
-                logger.debug("Could not create db session, skipping actuator association insert.")
+            )
+            cls.run_query(stmt)
+        else:
+            logger.debug("Could not create db session, skipping actuator association insert.")
 
     @classmethod
     def get_camera_id(cls, camera_id):
         """Method to return camera database id (primary key)"""
 
         if session := cls.create_session():
-            return (
-                session.query(ORMCamera.db_id)
-                .filter(
-                    and_(
-                        ORMCamera.camera_id == camera_id,
-                        ORMCamera.deletion_date.is_(None),  # Avoid to return id of deleted camera
+            try:
+                return (
+                    session.query(ORMCamera.db_id)
+                    .filter(
+                        and_(
+                            ORMCamera.camera_id == camera_id,
+                            ORMCamera.deletion_date.is_(
+                                None
+                            ),  # Avoid to return id of deleted camera
+                        )
                     )
-                )
-                .scalar()
-            )  # Use scalar() to retrieve the value directly
+                    .scalar()
+                )  # Use scalar() to retrieve the value directly
+            finally:
+                session.close()
         else:
             logger.debug(
                 "Could not get camera id %s since session has not been created.", camera_id
@@ -808,16 +822,21 @@ def get_actuator_id(cls, actuator_id):
         """Method to return actuator database id (primary key)"""
 
         if session := cls.create_session():
-            return (
-                session.query(ORMActuator.db_id)
-                .filter(
-                    and_(
-                        ORMActuator.actuator_id == actuator_id,
-                        ORMActuator.deletion_date.is_(None),  # Avoid to return id of deleted camera
+            try:
+                return (
+                    session.query(ORMActuator.db_id)
+                    .filter(
+                        and_(
+                            ORMActuator.actuator_id == actuator_id,
+                            ORMActuator.deletion_date.is_(
+                                None
+                            ),  # Avoid to return id of deleted camera
+                        )
                     )
+                    .scalar()
                 )
-                .scalar()
-            )
+            finally:
+                session.close()
         else:
             logger.debug(
                 "Could not get actuator id %s since session has not been created.", actuator_id
@@ -836,16 +855,19 @@ def get_detection_event_id(cls, detection_event: DetectionEvent):
             return None
 
         if session := cls.create_session():
-            return (
-                session.query(ORMDetectionEvent.db_id)
-                .filter(
-                    and_(
-                        ORMDetectionEvent.camera_id == camera_db_id,
-                        ORMDetectionEvent.time_stamp == detection_event.time_stamp,
+            try:
+                return (
+                    session.query(ORMDetectionEvent.db_id)
+                    .filter(
+                        and_(
+                            ORMDetectionEvent.camera_id == camera_db_id,
+                            ORMDetectionEvent.time_stamp == detection_event.time_stamp,
+                        )
                     )
-                )
-                .scalar()
-            )  # Use scalar() to retrieve the value directly
+                    .scalar()
+                )  # Use scalar() to retrieve the value directly
+            finally:
+                session.close()
         else:
             logger.error(
                 "Could not retrieve detection event id as connection could not be created."
@@ -1379,7 +1401,7 @@ def create_database(self):
             # Connect to MySQL server to check DB connection, credentials and db existence
             test_engine = create_engine(
                 f"mysql+pymysql://{self.username}:{self.get_password()}@{self.host}:{self.port}",
-                pool_recycle=1800,
+                pool_recycle=300,
                 pool_pre_ping=True,
             )
 
@@ -1439,7 +1461,7 @@ def create_database(self):
                 temp_engine = create_engine(
                     f"mysql+pymysql://{self.username}:{self.get_password()}"
                     f"@{self.host}:{self.port}",
-                    pool_recycle=1800,
+                    pool_recycle=300,
                     pool_pre_ping=True,
                 )
 
@@ -1463,7 +1485,7 @@ def create_database(self):
 
             # Recreate engine with correct database
             DataBase.wadas_db_engine = create_engine(
-                self.get_connection_string(), pool_recycle=1800, pool_pre_ping=True
+                self.get_connection_string(), pool_recycle=300, pool_pre_ping=True
             )
 
             # Create tables
@@ -1564,9 +1586,7 @@ def get_connection_string(self, create=False):
         if not self.database_name and not create:
             raise ValueError("Database name is required for MariaDB.")
 
-        base_string = (
-            f"mariadb+mariadbconnector://{self.username}:{password}@{self.host}:{self.port}"
-        )
+        base_string = f"mariadb+pymysql://{self.username}:{password}@{self.host}:{self.port}"
         return base_string if create else f"{base_string}/{self.database_name}"
 
     def create_database(self):
@@ -1578,7 +1598,7 @@ def create_database(self):
         try:
             # Connect to MariaDB server to check DB connection, credentials and db existence
             test_engine = create_engine(
-                self.get_connection_string(create=True), pool_recycle=1800, pool_pre_ping=True
+                self.get_connection_string(create=True), pool_recycle=300, pool_pre_ping=True
             )
 
             with test_engine.connect() as conn:
@@ -1635,7 +1655,7 @@ def create_database(self):
             try:
                 # Connect without specifying the database
                 temp_engine = create_engine(
-                    self.get_connection_string(create=True), pool_recycle=1800, pool_pre_ping=True
+                    self.get_connection_string(create=True), pool_recycle=300, pool_pre_ping=True
                 )
 
                 with temp_engine.connect() as conn:
@@ -1658,7 +1678,7 @@ def create_database(self):
 
             # Recreate engine with correct database
             DataBase.wadas_db_engine = create_engine(
-                self.get_connection_string(create=False), pool_recycle=1800, pool_pre_ping=True
+                self.get_connection_string(create=False), pool_recycle=300, pool_pre_ping=True
             )
 
             # Create tables

wadas_webserver/database.py

@@ -56,7 +56,19 @@ class Database:
 
     def __init__(self, connection_string):
         self.connection_string = connection_string
-        self.engine = create_engine(self.get_connection_string())
+        is_sqlite = connection_string.startswith("sqlite")
+        self.engine = (
+            create_engine(self.get_connection_string())
+            if is_sqlite
+            else create_engine(
+                self.get_connection_string(),
+                pool_recycle=300,  # Recycle connections every 5 min (below MariaDB/MySQL wait_timeout)
+                pool_pre_ping=True,  # Validate connection before use
+                pool_size=5,  # Persistent connections kept in pool
+                max_overflow=10,  # Extra temporary connections allowed under load
+                pool_timeout=30,  # Max seconds to wait for a connection from the pool
+            )
+        )
 
     @contextmanager
     def get_session(self):
@@ -65,7 +77,9 @@ def get_session(self):
         try:
             yield session
         except Exception:
+            session.rollback()
             logger.exception("An error occurred while creating a session")
+            raise
         finally:
             session.close()