diff --git a/changelog.d/7290.misc b/changelog.d/7290.misc
new file mode 100644
index 0000000000000000000000000000000000000000..676f285377f5d1e651c6b0985727493cd571fdff
--- /dev/null
+++ b/changelog.d/7290.misc
@@ -0,0 +1 @@
+Move catchup of replication streams logic to worker.
diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py
index f56a0fd4b58eb677cd4ef42e01b68194c92f4c75..0d3f0507767c18b4ae3f95ec286b605a5997a2e4 100644
--- a/synapse/replication/tcp/streams/_base.py
+++ b/synapse/replication/tcp/streams/_base.py
@@ -148,8 +148,9 @@ def db_query_to_update_function(
         updates = [(row[0], row[1:]) for row in rows]
         limited = False
         if len(updates) == limit:
-            upto_token = rows[-1][0]
+            upto_token = updates[-1][0]
             limited = True
+        assert len(updates) <= limit
 
         return updates, upto_token, limited