diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index c350c93c7e..ca9c48cabc 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -211,29 +211,17 @@ class MessageHandler(BaseHandler):
)
if source_config.direction == 'b':
- # if we're going backwards, we might need to backfill. This
- # requires that we have a topo token.
- if room_token.topological:
- max_topo = room_token.topological
- else:
- max_topo = yield self.store.get_max_topological_token(
- room_id, room_token.stream
- )
-
if membership == Membership.LEAVE:
# If they have left the room then clamp the token to be before
# they left the room, to save the effort of loading from the
# database.
+
leave_token = yield self.store.get_topological_token_for_event(
- member_event_id
+ member_event_id,
+ )
+ source_config.from_key = yield self.store.clamp_token_before(
+ room_id, source_config.from_key, leave_token,
)
- leave_token = RoomStreamToken.parse(leave_token)
- if leave_token.topological < max_topo:
- source_config.from_key = str(leave_token)
-
- yield self.hs.get_handlers().federation_handler.maybe_backfill(
- room_id, max_topo
- )
events, next_key, extremities = yield self.store.paginate_room_events(
room_id=room_id,
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 0d32a3a498..00da329924 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -915,6 +915,82 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
defer.returnValue((events, token, extremities))
+ def clamp_token_before(self, room_id, token_str, clamp_to_token_str):
+ """For a given room returns the given token if its before
+ clamp_to, otherwise returns clamp_to.
+
+ Args:
+ room_id (str)
+ token_str (str)
+ clamp_to_token_str(str): Must be topological token
+
+ Returns:
+ Deferred[str]
+ """
+
+ token = RoomStreamToken.parse(token_str)
+ clamp_to_token = RoomStreamToken.parse(clamp_to_token_str)
+
+ def clamp_token_before_txn(txn, token):
+ # If we're given a stream ordering, convert to topological token
+ if not token.chunk:
+ row = self._simple_select_one_txn(
+ txn,
+ table="events",
+ keyvalues={
+ "stream_ordering": token.stream,
+ },
+ retcols=("chunk_id", "topological_ordering", "stream_ordering",),
+ )
+ token = RoomStreamToken(*row)
+
+ # If both tokens have chunk_ids, we can use that.
+ if token.chunk and clamp_to_token.chunk:
+ if token.chunk == clamp_to_token.chunk:
+ if token.topological < clamp_to_token.topological:
+ return token_str
+ else:
+ return clamp_to_token_str
+
+ table = ChunkDBOrderedListStore(
+ txn, room_id, self.clock,
+ )
+
+ if table.is_before(token.chunk, clamp_to_token.chunk):
+ return token_str
+ else:
+ return clamp_to_token_str
+
+ # Ok, so we're dealing with events that haven't been chunked yet,
+ # lets just cheat and fallback to depth.
+
+ token_depth = self._simple_select_one_onecol_txn(
+ txn,
+ table="events",
+ keyvalues={
+ "stream_ordering": token.stream,
+ },
+ retcol="depth",
+ )
+
+ clamp_depth = self._simple_select_one_onecol_txn(
+ txn,
+ table="events",
+ keyvalues={
+ "stream_ordering": clamp_to_token.stream,
+ },
+ retcol="depth",
+ )
+
+ if token_depth < clamp_depth:
+ return token_str
+ else:
+ return clamp_to_token_str
+
+ return self.runInteraction(
+ "clamp_token_before", clamp_token_before_txn, token
+ )
+
class StreamStore(StreamWorkerStore):
def get_room_max_stream_ordering(self):
|
