diff --git a/homeassistant/components/recorder/core.py b/homeassistant/components/recorder/core.py
index 61c64be105c..e027922e8c4 100644
--- a/homeassistant/components/recorder/core.py
+++ b/homeassistant/components/recorder/core.py
@@ -719,6 +719,16 @@ class Recorder(threading.Thread):
         if schema_status is None:
             # Give up if we could not validate the schema
             return
+        if schema_status.current_version > SCHEMA_VERSION:
+            _LOGGER.error(
+                "The database schema version %s is newer than %s which is the maximum "
+                "database schema version supported by the installed version of "
+                "Home Assistant Core, either upgrade Home Assistant Core or restore "
+                "the database from a backup compatible with this version",
+                schema_status.current_version,
+                SCHEMA_VERSION,
+            )
+            return
         self.schema_version = schema_status.current_version
 
         if not schema_status.migration_needed and not schema_status.schema_errors:
diff --git a/tests/components/recorder/test_init.py b/tests/components/recorder/test_init.py
index 7e5abf1b514..2e9e9a7c729 100644
--- a/tests/components/recorder/test_init.py
+++ b/tests/components/recorder/test_init.py
@@ -2615,6 +2615,46 @@ async def test_clean_shutdown_when_schema_migration_fails(
     assert instance.engine is None
 
 
+async def test_setup_fails_after_downgrade(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test we fail to setup after a downgrade.
+
+    Also test we shutdown cleanly.
+    """
+    with (
+        patch.object(
+            migration,
+            "_get_current_schema_version",
+            side_effect=[None, SCHEMA_VERSION + 1],
+        ),
+        patch("homeassistant.components.recorder.ALLOW_IN_MEMORY_DB", True),
+    ):
+        if recorder.DOMAIN not in hass.data:
+            recorder_helper.async_initialize_recorder(hass)
+        assert not await async_setup_component(
+            hass,
+            recorder.DOMAIN,
+            {
+                recorder.DOMAIN: {
+                    CONF_DB_URL: "sqlite://",
+                    CONF_DB_RETRY_WAIT: 0,
+                    CONF_DB_MAX_RETRIES: 1,
+                }
+            },
+        )
+        await hass.async_block_till_done()
+
+    instance = recorder.get_instance(hass)
+    await hass.async_stop()
+    assert instance.engine is None
+    assert (
+        f"The database schema version {SCHEMA_VERSION+1} is newer than {SCHEMA_VERSION}"
+        " which is the maximum database schema version supported by the installed "
+        "version of Home Assistant Core"
+    ) in caplog.text
+
+
 async def test_events_are_recorded_until_final_write(
     hass: HomeAssistant,
     async_setup_recorder_instance: RecorderInstanceGenerator,