Raise on database error in recorder.migration function (#123644)

* Raise on database error in recorder.migration._update_states_table_with_foreign_key_options

* Improve test coverage

* Fix test

* Fix test

homeassistant/components/recorder/migration.py

@@ -639,6 +639,7 @@ def _update_states_table_with_foreign_key_options(
                 _LOGGER.exception(
                     "Could not update foreign options in %s table", TABLE_STATES
                 )
+                raise
 
 
 def _drop_foreign_key_constraints(

tests/components/recorder/test_migrate.py

@@ -199,6 +199,64 @@ async def test_database_migration_failed(
     assert len(mock_dismiss.mock_calls) == expected_pn_dismiss
 
 
+@pytest.mark.parametrize(
+    (
+        "func_to_patch",
+        "expected_setup_result",
+        "expected_pn_create",
+        "expected_pn_dismiss",
+    ),
+    [
+        ("DropConstraint", False, 1, 0),  # This makes migration to step 11 fail
+    ],
+)
+@pytest.mark.skip_on_db_engine(["sqlite"])
+@pytest.mark.usefixtures("skip_by_db_engine")
+async def test_database_migration_failed_step_11(
+    hass: HomeAssistant,
+    async_setup_recorder_instance: RecorderInstanceGenerator,
+    func_to_patch: str,
+    expected_setup_result: bool,
+    expected_pn_create: int,
+    expected_pn_dismiss: int,
+) -> None:
+    """Test we notify if the migration fails."""
+    assert recorder.util.async_migration_in_progress(hass) is False
+
+    with (
+        patch(
+            "homeassistant.components.recorder.core.create_engine",
+            new=create_engine_test,
+        ),
+        patch(
+            f"homeassistant.components.recorder.migration.{func_to_patch}",
+            side_effect=OperationalError(
+                None, None, OSError("No space left on device")
+            ),
+        ),
+        patch(
+            "homeassistant.components.persistent_notification.create",
+            side_effect=pn.create,
+        ) as mock_create,
+        patch(
+            "homeassistant.components.persistent_notification.dismiss",
+            side_effect=pn.dismiss,
+        ) as mock_dismiss,
+    ):
+        await async_setup_recorder_instance(
+            hass, wait_recorder=False, expected_setup_result=expected_setup_result
+        )
+        hass.states.async_set("my.entity", "on", {})
+        hass.states.async_set("my.entity", "off", {})
+        await hass.async_block_till_done()
+        await hass.async_add_executor_job(recorder.get_instance(hass).join)
+        await hass.async_block_till_done()
+
+    assert recorder.util.async_migration_in_progress(hass) is False
+    assert len(mock_create.mock_calls) == expected_pn_create
+    assert len(mock_dismiss.mock_calls) == expected_pn_dismiss
+
+
 @pytest.mark.skip_on_db_engine(["mysql", "postgresql"])
 @pytest.mark.usefixtures("skip_by_db_engine")
 async def test_live_database_migration_encounters_corruption(