Ensure recorder event loop recovers if the database server dis… (home-assistant#33253)

If the database server disconnects there were exceptions
that were not trapped which would cause the recorder event
loop to collapse.  As we never want the loop to end
we trap exceptions broadly.

Fix a bug in the new commit interval setting which caused
it to always commit after 1s

Author: bdraco

homeassistant/components/recorder/__init__.py

@@ -342,7 +342,6 @@ def async_purge(now):
         # has changed.  This reduces the disk io.
         while True:
             event = self.queue.get()
-
             if event is None:
                 self._close_run()
                 self._close_connection()
@@ -356,7 +355,7 @@ def async_purge(now):
                 self.queue.task_done()
                 if self.commit_interval:
                     self._timechanges_seen += 1
-                    if self.commit_interval >= self._timechanges_seen:
+                    if self._timechanges_seen >= self.commit_interval:
                         self._timechanges_seen = 0
                         self._commit_event_session_or_retry()
                 continue
@@ -376,6 +375,9 @@ def async_purge(now):
                 self.event_session.flush()
             except (TypeError, ValueError):
                 _LOGGER.warning("Event is not JSON serializable: %s", event)
+            except Exception as err:  # pylint: disable=broad-except
+                # Must catch the exception to prevent the loop from collapsing
+                _LOGGER.exception("Error adding event: %s", err)
 
             if dbevent and event.event_type == EVENT_STATE_CHANGED:
                 try:
@@ -387,6 +389,9 @@ def async_purge(now):
                         "State is not JSON serializable: %s",
                         event.data.get("new_state"),
                     )
+                except Exception as err:  # pylint: disable=broad-except
+                    # Must catch the exception to prevent the loop from collapsing
+                    _LOGGER.exception("Error adding state change: %s", err)
 
             # If they do not have a commit interval
             # than we commit right away
@@ -404,17 +409,26 @@ def _commit_event_session_or_retry(self):
             try:
                 self._commit_event_session()
                 return
-
-            except exc.OperationalError as err:
-                _LOGGER.error(
-                    "Error in database connectivity: %s. " "(retrying in %s seconds)",
-                    err,
-                    self.db_retry_wait,
-                )
+            except (exc.InternalError, exc.OperationalError) as err:
+                if err.connection_invalidated:
+                    _LOGGER.error(
+                        "Database connection invalidated: %s. "
+                        "(retrying in %s seconds)",
+                        err,
+                        self.db_retry_wait,
+                    )
+                else:
+                    _LOGGER.error(
+                        "Error in database connectivity: %s. "
+                        "(retrying in %s seconds)",
+                        err,
+                        self.db_retry_wait,
+                    )
                 tries += 1
 
-            except exc.SQLAlchemyError:
-                _LOGGER.exception("Error saving events")
+            except Exception as err:  # pylint: disable=broad-except
+                # Must catch the exception to prevent the loop from collapsing
+                _LOGGER.exception("Error saving events: %s", err)
                 return
 
         _LOGGER.error(
@@ -423,10 +437,15 @@ def _commit_event_session_or_retry(self):
         )
         try:
             self.event_session.close()
-        except exc.SQLAlchemyError:
-            _LOGGER.exception("Failed to close event session.")
+        except Exception as err:  # pylint: disable=broad-except
+            # Must catch the exception to prevent the loop from collapsing
+            _LOGGER.exception("Error while closing event session: %s", err)
 
-        self.event_session = self.get_session()
+        try:
+            self.event_session = self.get_session()
+        except Exception as err:  # pylint: disable=broad-except
+            # Must catch the exception to prevent the loop from collapsing
+            _LOGGER.exception("Error while creating new event session: %s", err)
 
     def _commit_event_session(self):
         try: