diff --git a/changelog.d/16397.bugfix b/changelog.d/16397.bugfix
new file mode 100644
index 000000000000..e4b27221a56e
--- /dev/null
+++ b/changelog.d/16397.bugfix
@@ -0,0 +1 @@
+Enforce that old background updates have run when starting Synapse.
diff --git a/docs/development/database_schema.md b/docs/development/database_schema.md
index 37a06acc1282..9fedebda6163 100644
--- a/docs/development/database_schema.md
+++ b/docs/development/database_schema.md
@@ -25,20 +25,37 @@ updated. They work as follows:
 
  * The Synapse codebase defines a constant `synapse.storage.schema.SCHEMA_VERSION`
    which represents the expectations made about the database by that version. For
-   example, as of Synapse v1.36, this is `59`.
+   example, as of Synapse v1.36, this is `59`. This version should be incremented
+   whenever a backwards-incompatible change is made to the database format (normally
+   via a `delta` file)
+
+ * The Synapse codebase defines a constant `synapse.storage.schema.SCHEMA_COMPAT_VERSION`
+   which represents the minimum database versions the current code supports.
+   Whenever the Synapse code is updated to assume backwards-incompatible changes
+   made to the database format, `synapse.storage.schema.SCHEMA_COMPAT_VERSION` is also updated
+   so that administrators can not accidentally roll back to a too-old version of Synapse.
 
- * The database stores a "compatibility version" in
+   The database stores a "compatibility version" in
    `schema_compat_version.compat_version` which defines the `SCHEMA_VERSION` of the
    oldest version of Synapse which will work with the database. On startup, if
    `compat_version` is found to be newer than `SCHEMA_VERSION`, Synapse will refuse to
    start.
 
-   Synapse automatically updates this field from
-   `synapse.storage.schema.SCHEMA_COMPAT_VERSION`.
+   Synapse automatically updates `schema_compat_version.compat_version` from
+   `synapse.storage.schema.SCHEMA_COMPAT_VERSION` during start-up.
 
- * Whenever a backwards-incompatible change is made to the database format (normally
-   via a `delta` file), `synapse.storage.schema.SCHEMA_COMPAT_VERSION` is also updated
-   so that administrators can not accidentally roll back to a too-old version of Synapse.
+ * The Synapse codebase defines a constant `synapse.storage.schema.BACKGROUND_UPDATES_COMPAT_VERSION`
+   which represents the earliest supported background updates.
+
+   On startup, if there exists any background update (via the
+   `background_updates.ordering` column) older than `BACKGROUND_UPDATES_COMPAT_VERSION`,
+   Synpase will refuse to start.
+
+   This is useful for adding delta files which assume background updates have
+   finished; overall maintenance of Synapse (by allowing for removal of code
+   supporting old background updates); among other things.
+
+   `BACKGROUND_UPDATES_COMPAT_VERSION` must be < the latest [full schema dump](#full-schema-dumps).
 
 Generally, the goal is to maintain compatibility with at least one or two previous
 releases of Synapse, so any substantial change tends to require multiple releases and a
diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py
index 31501fd57331..598ded33b5e1 100644
--- a/synapse/storage/prepare_database.py
+++ b/synapse/storage/prepare_database.py
@@ -25,6 +25,7 @@
     Optional,
     TextIO,
     Tuple,
+    cast,
 )
 
 import attr
@@ -32,7 +33,11 @@
 from synapse.config.homeserver import HomeServerConfig
 from synapse.storage.database import LoggingDatabaseConnection, LoggingTransaction
 from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine
-from synapse.storage.schema import SCHEMA_COMPAT_VERSION, SCHEMA_VERSION
+from synapse.storage.schema import (
+    BACKGROUND_UPDATES_COMPAT_VERSION,
+    SCHEMA_COMPAT_VERSION,
+    SCHEMA_VERSION,
+)
 from synapse.storage.types import Cursor
 
 logger = logging.getLogger(__name__)
@@ -80,6 +85,9 @@ class _SchemaState:
     applied_deltas: Collection[str] = attr.ib(factory=tuple)
     """Any delta files for `current_version` which have already been applied"""
 
+    background_updates: Collection[Tuple[str, int]] = attr.ib(factory=tuple)
+    """Any (pending) updates in the `background_updates` table."""
+
     upgraded: bool = attr.ib(default=False)
     """Whether the current state was reached by applying deltas.
 
@@ -359,6 +367,7 @@ def _upgrade_existing_database(
     """
     if is_empty:
         assert not current_schema_state.applied_deltas
+        assert not current_schema_state.background_updates
     else:
         assert config
 
@@ -413,6 +422,24 @@ def _upgrade_existing_database(
         start_ver += 1
 
     logger.debug("applied_delta_files: %s", current_schema_state.applied_deltas)
+    logger.debug(
+        "pending background_updates: %s",
+        (name for name, ordering in current_schema_state.background_updates),
+    )
+
+    # Bail if there are any pending background updates from before the background schema compat version.
+    for update_name, ordering in sorted(
+        current_schema_state.background_updates, key=lambda b: b[1]
+    ):
+        # ordering is an int based on when the background update was added:
+        #
+        #   (schema version when added * 100) + (schema delta when added).
+        update_schema_version = ordering // 100
+        if update_schema_version < BACKGROUND_UPDATES_COMPAT_VERSION:
+            raise UpgradeDatabaseException(
+                "Database has old pending background updates for version %d: %s"
+                % (update_schema_version, update_name)
+            )
 
     if isinstance(database_engine, PostgresEngine):
         specific_engine_extension = ".postgres"
@@ -705,10 +732,14 @@ def _get_or_create_schema_state(
     )
     applied_deltas = tuple(d for d, in txn)
 
+    txn.execute("SELECT update_name, ordering FROM background_updates")
+    background_Updates = cast(Tuple[Tuple[str, int], ...], tuple(txn))
+
     return _SchemaState(
         current_version=current_version,
         compat_version=compat_version,
         applied_deltas=applied_deltas,
+        background_updates=background_Updates,
         upgraded=upgraded,
     )
 
diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py
index 158b528dce18..7438337f457f 100644
--- a/synapse/storage/schema/__init__.py
+++ b/synapse/storage/schema/__init__.py
@@ -136,3 +136,17 @@
 This value is stored in the database, and checked on startup. If the value in the
 database is greater than SCHEMA_VERSION, then Synapse will refuse to start.
 """
+
+BACKGROUND_UPDATES_COMPAT_VERSION = (
+    # The replace_stream_ordering_column from 6001 must have run.
+    61
+)
+"""Limit on how far the syanpse can be rolled forward without breaking db compat
+
+This value is checked on startup against any pending background updates. If there
+are any pending background updates less than BACKGROUND_UPDATES_COMPAT_VERSION, then
+Synapse will refuse to start.
+
+In order to work with *new* databases this *must* be smaller than the latest full
+dump of the database.
+"""