Run _upgrade_existing_database on workers if at current schema_version

Workers need to run this method to figure out if any outstanding delta
files exist - even if we're currently on the latest schema version.

#11255 made the correct fix in that _upgrade_existing_database should
not be run if version_info.schema_version > SCHEMA_VERSION. But we
still need to run this method if version_info.schema_version ==
SCHEMA_VERSION.

synapse/storage/prepare_database.py

@@ -131,17 +131,13 @@ def prepare_database(
                     "config==None in prepare_database, but database is not empty"
                 )
 
-            # if it's a worker app, refuse to upgrade the database, to avoid multiple
-            # workers doing it at once.
-            if config.worker.worker_app is None:
-                _upgrade_existing_database(
-                    cur,
-                    version_info,
-                    database_engine,
-                    config,
-                    databases=databases,
-                )
-            elif version_info.current_version < SCHEMA_VERSION:
+            # If the schema version needs to be updated, and we are on a worker, we immediately
+            # know to bail out as workers cannot update the database schema. Only one worker
+            # must update the database at the time, therefore we delegate this task to the master.
+            if (
+                config.worker.worker_app is not None
+                and version_info.current_version < SCHEMA_VERSION
+            ):
                 # If the DB is on an older version than we expect then we refuse
                 # to start the worker (as the main process needs to run first to
                 # update the schema).
@@ -150,6 +146,24 @@ def prepare_database(
                     % (SCHEMA_VERSION, version_info.current_version)
                 )
 
+            # Otherwise if we are on the current schema version, we need to check if there
+            # are any unapplied delta files to process. This should be run on all processes,
+            # master or worker. The master will apply the deltas, workers will check if any
+            # outstanding deltas exist and bail out if they do.
+            if version_info.current_version == SCHEMA_VERSION:
+                _upgrade_existing_database(
+                    cur,
+                    version_info,
+                    database_engine,
+                    config,
+                    databases=databases,
+                )
+
+            # If SCHEMA_VERSION is greater than version_info.current_version (for instance,
+            # if we're rolling back the database) then don't apply any migrations.
+            # Things should work as long as one hasn't rolled back past the
+            # SCHEMA_COMPAT_VERSION.
+
         else:
             logger.info("%r: Initialising new database", databases)