Ensure recorder commit can retry after encountering invalid data (#41426)

homeassistant/components/recorder/__init__.py

@@ -494,6 +494,7 @@ class Recorder(threading.Thread):
             for dbstate in self._pending_expunge:
                 # Expunge the state so its not expired
                 # until we use it later for dbstate.old_state
+                if dbstate in self.event_session:
                     self.event_session.expunge(dbstate)
             self._pending_expunge = []
             self.event_session.commit()

tests/components/recorder/test_init.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta
 import unittest
 
 import pytest
+from sqlalchemy.exc import OperationalError
 
 from homeassistant.components.recorder import (
     CONFIG_SCHEMA,
@@ -452,3 +453,41 @@ def test_run_information(hass_recorder):
 
 class CannotSerializeMe:
     """A class that the JSONEncoder cannot serialize."""
+
+
+def test_saving_state_with_exception(hass, hass_recorder, caplog):
+    """Test saving and restoring a state."""
+    hass = hass_recorder()
+
+    entity_id = "test.recorder"
+    state = "restoring_from_db"
+    attributes = {"test_attr": 5, "test_attr_10": "nice"}
+
+    def _throw_if_state_in_session(*args, **kwargs):
+        for obj in hass.data[DATA_INSTANCE].event_session:
+            if isinstance(obj, States):
+                raise OperationalError(
+                    "insert the state", "fake params", "forced to fail"
+                )
+
+    with patch("time.sleep"), patch.object(
+        hass.data[DATA_INSTANCE].event_session,
+        "flush",
+        side_effect=_throw_if_state_in_session,
+    ):
+        hass.states.set(entity_id, "fail", attributes)
+        wait_recording_done(hass)
+
+    assert "Error executing query" in caplog.text
+    assert "Error saving events" not in caplog.text
+
+    caplog.clear()
+    hass.states.set(entity_id, state, attributes)
+    wait_recording_done(hass)
+
+    with session_scope(hass=hass) as session:
+        db_states = list(session.query(States))
+        assert len(db_states) >= 1
+
+    assert "Error executing query" not in caplog.text
+    assert "Error saving events" not in caplog.text