Stream ordering and out of order insertions.

Handle the fact that events can be persisted out of order, and so to get
the "current max" stream token becomes non trivial - as we need to make
sure that *all* stream tokens less than the current max have also
successfully been persisted.

1 files changed, 7 insertions, 2 deletions

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 3b3416716e..f066484c7e 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -52,7 +52,6 @@ class EventsStore(SQLBaseStore):
                 is_new_state=is_new_state,
                 current_state=current_state,
             )
-            self.get_room_events_max_id.invalidate()
         except _RollbackButIsFineException:
             pass
 
@@ -97,7 +96,13 @@ class EventsStore(SQLBaseStore):
         self._get_event_cache.pop(event.event_id)
 
         if stream_ordering is None:
-            stream_ordering = self._stream_id_gen.get_next_txn(txn)
+            with self._stream_id_gen.get_next_txn(txn) as stream_ordering:
+                return self._persist_event_txn(
+                    txn, event, context, backfilled,
+                    stream_ordering=stream_ordering,
+                    is_new_state=is_new_state,
+                    current_state=current_state,
+                )
 
         # We purposefully do this first since if we include a `current_state`
         # key, we *want* to update the `current_state_events` table