Ensure (room_id, next_batch_id) is unique to avoid cross-talk/conflic…

…ts between chunks

Part of [MSC2716](matrix-org/matrix-spec-proposals#2716)

Part of #10737

synapse/storage/databases/main/events.py

@@ -1795,6 +1795,25 @@ def _handle_insertion_event(self, txn: LoggingTransaction, event: EventBase):
             # Invalid insertion event without next batch ID
             return
 
+        conflicting_insertion_event_id = self.db_pool.simple_select_one_onecol_txn(
+            txn,
+            table="insertion_events",
+            keyvalues={
+                "room_id": event.room_id,
+                "next_batch_id": next_batch_id,
+            },
+            retcol="event_id",
+            desc="get_device_list_last_stream_id_for_remote",
+            allow_none=True,
+        )
+        if conflicting_insertion_event_id is not None:
+            # The new  insertion event is invalid because there already exists
+            # and insertion event in the room with the same next_batch_id. We
+            # can't allow multiple because the batch pointing will get weird,
+            # e.g. we can't determine which insertion event the batch event is
+            # pointing to.
+            return
+
         logger.debug(
             "_handle_insertion_event (next_batch_id=%s) %s", next_batch_id, event
         )