diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 497744b9af..426a93b8dd 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -202,19 +202,64 @@ class SyncHandler(BaseHandler):
return self.incremental_sync_with_gap(sync_config, batch_token)
@defer.inlineCallbacks
- def _get_room_timestamps_at_token(self, room_ids, token):
- room_to_last_ts = {}
+ def _get_room_timestamps_at_token(self, room_ids, token, sync_config,
+ limit):
+ room_to_entries = {}
@defer.inlineCallbacks
def _get_last_ts(room_id):
- ts = yield self.store.get_last_ts_for_room(
+ entry = yield self.store.get_last_ts_for_room(
room_id, token.room_key
)
- room_to_last_ts[room_id] = ts if ts else 0
- logger.info("room_to_last_ts: %r", room_to_last_ts)
+ # TODO: Is this ever possible?
+ room_to_entries[room_id] = entry if entry else {
+ "origin_server_ts": 0,
+ }
+
yield concurrently_execute(_get_last_ts, room_ids, 10)
- defer.returnValue(room_to_last_ts)
+
+ if len(room_to_entries) <= limit:
+ defer.returnValue({
+ room_id: entry["origin_server_ts"]
+ for room_id, entry in room_to_entries.items()
+ })
+
+ queued_events = sorted(
+ room_to_entries.items(),
+ key=lambda e: e[1]["origin_server_ts"]
+ )
+
+ to_return = {}
+
+ while len(to_return) < limit and len(queued_events) > 0:
+ to_fetch = queued_events[:limit - len(to_return)]
+ event_to_q = {
+ e["event_id"]: (room_id, e) for room_id, e in to_fetch.items()
+ if "event_id" in e
+ }
+
+ # Now we fetch each event to check if its been filtered out
+ event_map = yield self.store.get_events(event_to_q.keys())
+
+ recents = sync_config.filter_collection.filter_room_timeline(
+ event_map.values()
+ )
+ recents = yield filter_events_for_client(
+ self.store,
+ sync_config.user.to_string(),
+ recents,
+ )
+
+ to_return.update({r.room_id: r.origin_server_ts for r in recents})
+
+ for ev_id in set(event_map.keys()) - set(r.event_id for r in recents):
+ queued_events.append(event_to_q[ev_id])
+
+ # FIXME: Need to refetch TS
+ queued_events.sort(key=lambda e: e[1]["origin_server_ts"])
+
+ defer.returnValue(to_return)
@defer.inlineCallbacks
def full_state_sync(self, sync_config, batch_token):
@@ -292,6 +337,8 @@ class SyncHandler(BaseHandler):
e.room_id for e in room_list if e.membership == Membership.JOIN
],
token=now_token,
+ sync_config=sync_config,
+ limit=pagination_limit,
)
if room_to_last_ts:
@@ -512,8 +559,13 @@ class SyncHandler(BaseHandler):
)
@defer.inlineCallbacks
- def _get_rooms_that_need_full_state(self, room_ids, since_token, pa_ts):
- start_ts = yield self._get_room_timestamps_at_token(room_ids, since_token)
+ def _get_rooms_that_need_full_state(self, room_ids, since_token, pa_ts,
+ sync_config, pagination_limit):
+ start_ts = yield self._get_room_timestamps_at_token(
+ room_ids, since_token,
+ sync_config=sync_config,
+ limit=pagination_limit,
+ )
missing_list = frozenset(
room_id for room_id, ts in
@@ -675,10 +727,13 @@ class SyncHandler(BaseHandler):
p_room_token = room_pagination_config.get("t", None)
if p_room_token:
+ pa_limit = room_pagination_config["l"]
needing_full_state = yield self._get_rooms_that_need_full_state(
[room_id],
since_token,
room_pagination_config.get("ts", 0),
+ sync_config=sync_config,
+ pagination_limit=pa_limit,
)
need_full_state = room_id in needing_full_state
else:
@@ -716,10 +771,13 @@ class SyncHandler(BaseHandler):
if not room_sync.timeline:
p_room_token = room_pagination_config.get("t", None)
if p_room_token:
+ pa_limit = room_pagination_config["l"]
needing_full_state = yield self._get_rooms_that_need_full_state(
[room_id],
since_token,
room_pagination_config.get("ts", 0),
+ sync_config=sync_config,
+ pagination_limit=pa_limit,
)
if room_id in needing_full_state:
continue
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 60a8384fae..51f8b9e017 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -533,7 +533,7 @@ class StreamStore(SQLBaseStore):
stream_ordering = RoomStreamToken.parse_stream_token(token).stream
sql = (
- "SELECT origin_server_ts FROM events"
+ "SELECT event_id, origin_server_ts FROM events"
" WHERE room_id = ? AND stream_ordering <= ?"
" ORDER BY topological_ordering DESC, stream_ordering DESC"
" LIMIT 1"
@@ -541,9 +541,9 @@ class StreamStore(SQLBaseStore):
def f(txn):
txn.execute(sql, (room_id, stream_ordering))
- rows = txn.fetchall()
+ rows = self.cursor_to_dict(txn)
if rows:
- return rows[0][0]
+ return rows[0]
else:
return None
|
