Fix recorder stopping after unserializable state (#36937)

* Ensure unserializable states do not collapse recording

* augment test coverage

* fix wal mode being set every time

homeassistant/components/recorder/__init__.py

@@ -146,13 +146,13 @@ def run_information_with_session(session, point_in_time: Optional[datetime] = No
     """Return information about current run from the database."""
     recorder_runs = RecorderRuns
 
-    res = (
-        session.query(recorder_runs)
-        .filter(
+    query = session.query(recorder_runs)
+    if point_in_time:
+        query = query.filter(
             (recorder_runs.start < point_in_time) & (recorder_runs.end > point_in_time)
         )
-        .first()
-    )
+
+    res = query.first()
     if res:
         session.expunge(res)
     return res
@@ -245,6 +245,7 @@ class Recorder(threading.Thread):
         self._old_state_ids = {}
         self.event_session = None
         self.get_session = None
+        self._completed_database_setup = False
 
     @callback
     def async_initialize(self):
@@ -401,6 +402,10 @@ class Recorder(threading.Thread):
                     dbstate.event_id = dbevent.event_id
                     self.event_session.add(dbstate)
                     self.event_session.flush()
+                    if "new_state" in event.data:
+                        self._old_state_ids[dbstate.entity_id] = dbstate.state_id
+                    elif dbstate.entity_id in self._old_state_ids:
+                        del self._old_state_ids[dbstate.entity_id]
                 except (TypeError, ValueError):
                     _LOGGER.warning(
                         "State is not JSON serializable: %s",
@@ -410,11 +415,6 @@ class Recorder(threading.Thread):
                     # Must catch the exception to prevent the loop from collapsing
                     _LOGGER.exception("Error adding state change: %s", err)
 
-                if "new_state" in event.data:
-                    self._old_state_ids[dbstate.entity_id] = dbstate.state_id
-                elif dbstate.entity_id in self._old_state_ids:
-                    del self._old_state_ids[dbstate.entity_id]
-
             # If they do not have a commit interval
             # than we commit right away
             if not self.commit_interval:
@@ -514,6 +514,9 @@ class Recorder(threading.Thread):
         def setup_recorder_connection(dbapi_connection, connection_record):
             """Dbapi specific connection settings."""
 
+            if self._completed_database_setup:
+                return
+
             # We do not import sqlite3 here so mysql/other
             # users do not have to pay for it to be loaded in
             # memory
@@ -524,6 +527,10 @@ class Recorder(threading.Thread):
                 cursor.execute("PRAGMA journal_mode=WAL")
                 cursor.close()
                 dbapi_connection.isolation_level = old_isolation
+                # WAL mode only needs to be setup once
+                # instead of every time we open the sqlite connection
+                # as its persistent and isn't free to call every time.
+                self._completed_database_setup = True
             elif self.db_url.startswith("mysql"):
                 cursor = dbapi_connection.cursor()
                 cursor.execute("SET session wait_timeout=28800")

tests/components/recorder/test_init.py

@@ -5,9 +5,14 @@ import unittest
 
 import pytest
 
-from homeassistant.components.recorder import Recorder
+from homeassistant.components.recorder import (
+    Recorder,
+    run_information,
+    run_information_from_instance,
+    run_information_with_session,
+)
 from homeassistant.components.recorder.const import DATA_INSTANCE
-from homeassistant.components.recorder.models import Events, States
+from homeassistant.components.recorder.models import Events, RecorderRuns, States
 from homeassistant.components.recorder.util import session_scope
 from homeassistant.const import MATCH_ALL
 from homeassistant.core import ATTR_NOW, EVENT_TIME_CHANGED, callback
@@ -284,3 +289,61 @@ def test_saving_sets_old_state(hass_recorder):
         assert states[1].old_state_id is None
         assert states[2].old_state_id == states[0].state_id
         assert states[3].old_state_id == states[1].state_id
+
+
+def test_saving_state_with_serializable_data(hass_recorder, caplog):
+    """Test saving data that cannot be serialized does not crash."""
+    hass = hass_recorder()
+
+    hass.states.set("test.one", "on", {"fail": CannotSerializeMe()})
+    wait_recording_done(hass)
+    hass.states.set("test.two", "on", {})
+    wait_recording_done(hass)
+    hass.states.set("test.two", "off", {})
+    wait_recording_done(hass)
+
+    with session_scope(hass=hass) as session:
+        states = list(session.query(States))
+        assert len(states) == 2
+
+        assert states[0].entity_id == "test.two"
+        assert states[1].entity_id == "test.two"
+        assert states[0].old_state_id is None
+        assert states[1].old_state_id == states[0].state_id
+
+    assert "State is not JSON serializable" in caplog.text
+
+
+def test_run_information(hass_recorder):
+    """Ensure run_information returns expected data."""
+    before_start_recording = dt_util.utcnow()
+    hass = hass_recorder()
+    run_info = run_information_from_instance(hass)
+    assert isinstance(run_info, RecorderRuns)
+    assert run_info.closed_incorrect is False
+
+    with session_scope(hass=hass) as session:
+        run_info = run_information_with_session(session)
+        assert isinstance(run_info, RecorderRuns)
+        assert run_info.closed_incorrect is False
+
+    run_info = run_information(hass)
+    assert isinstance(run_info, RecorderRuns)
+    assert run_info.closed_incorrect is False
+
+    hass.states.set("test.two", "on", {})
+    wait_recording_done(hass)
+    run_info = run_information(hass)
+    assert isinstance(run_info, RecorderRuns)
+    assert run_info.closed_incorrect is False
+
+    run_info = run_information(hass, before_start_recording)
+    assert run_info is None
+
+    run_info = run_information(hass, dt_util.utcnow())
+    assert isinstance(run_info, RecorderRuns)
+    assert run_info.closed_incorrect is False
+
+
+class CannotSerializeMe:
+    """A class that the JSONEncoder cannot serialize."""