diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index ebd3e98105..bfe2583002 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -16,7 +16,7 @@
import itertools
import logging
-from typing import Any, Dict, FrozenSet, List, Optional, Set, Tuple
+from typing import TYPE_CHECKING, Any, Dict, FrozenSet, List, Optional, Set, Tuple
import attr
from prometheus_client import Counter
@@ -31,6 +31,7 @@ from synapse.storage.state import StateFilter
from synapse.types import (
Collection,
JsonDict,
+ MutableStateMap,
RoomStreamToken,
StateMap,
StreamToken,
@@ -43,6 +44,9 @@ from synapse.util.caches.response_cache import ResponseCache
from synapse.util.metrics import Measure, measure_func
from synapse.visibility import filter_events_for_client
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
logger = logging.getLogger(__name__)
# Debug logger for https://github.com/matrix-org/synapse/issues/4422
@@ -85,16 +89,19 @@ class TimelineBatch:
events = attr.ib(type=List[EventBase])
limited = attr.ib(bool)
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
"""Make the result appear empty if there are no updates. This is used
to tell if room needs to be part of the sync result.
"""
return bool(self.events)
- __bool__ = __nonzero__ # python3
-
-@attr.s(slots=True, frozen=True)
+# We can't freeze this class, because we need to update it after it's instantiated to
+# update its unread count. This is because we calculate the unread count for a room only
+# if there are updates for it, which we check after the instance has been created.
+# This should not be a big deal because we update the notification counts afterwards as
+# well anyway.
+@attr.s(slots=True)
class JoinedSyncResult:
room_id = attr.ib(type=str)
timeline = attr.ib(type=TimelineBatch)
@@ -103,8 +110,9 @@ class JoinedSyncResult:
account_data = attr.ib(type=List[JsonDict])
unread_notifications = attr.ib(type=JsonDict)
summary = attr.ib(type=Optional[JsonDict])
+ unread_count = attr.ib(type=int)
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
"""Make the result appear empty if there are no updates. This is used
to tell if room needs to be part of the sync result.
"""
@@ -117,8 +125,6 @@ class JoinedSyncResult:
# else in the result, we don't need to send it.
)
- __bool__ = __nonzero__ # python3
-
@attr.s(slots=True, frozen=True)
class ArchivedSyncResult:
@@ -127,26 +133,22 @@ class ArchivedSyncResult:
state = attr.ib(type=StateMap[EventBase])
account_data = attr.ib(type=List[JsonDict])
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
"""Make the result appear empty if there are no updates. This is used
to tell if room needs to be part of the sync result.
"""
return bool(self.timeline or self.state or self.account_data)
- __bool__ = __nonzero__ # python3
-
@attr.s(slots=True, frozen=True)
class InvitedSyncResult:
room_id = attr.ib(type=str)
invite = attr.ib(type=EventBase)
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
"""Invited rooms should always be reported to the client"""
return True
- __bool__ = __nonzero__ # python3
-
@attr.s(slots=True, frozen=True)
class GroupsSyncResult:
@@ -154,11 +156,9 @@ class GroupsSyncResult:
invite = attr.ib(type=JsonDict)
leave = attr.ib(type=JsonDict)
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
return bool(self.join or self.invite or self.leave)
- __bool__ = __nonzero__ # python3
-
@attr.s(slots=True, frozen=True)
class DeviceLists:
@@ -171,13 +171,11 @@ class DeviceLists:
changed = attr.ib(type=Collection[str])
left = attr.ib(type=Collection[str])
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
return bool(self.changed or self.left)
- __bool__ = __nonzero__ # python3
-
-@attr.s
+@attr.s(slots=True)
class _RoomChanges:
"""The set of room entries to include in the sync, plus the set of joined
and left room IDs since last sync.
@@ -217,7 +215,7 @@ class SyncResult:
device_one_time_keys_count = attr.ib(type=JsonDict)
groups = attr.ib(type=Optional[GroupsSyncResult])
- def __nonzero__(self) -> bool:
+ def __bool__(self) -> bool:
"""Make the result appear empty if there are no updates. This is used
to tell if the notifier needs to wait for more events when polling for
events.
@@ -233,11 +231,9 @@ class SyncResult:
or self.groups
)
- __bool__ = __nonzero__ # python3
-
-class SyncHandler(object):
- def __init__(self, hs):
+class SyncHandler:
+ def __init__(self, hs: "HomeServer"):
self.hs_config = hs.config
self.store = hs.get_datastore()
self.notifier = hs.get_notifier()
@@ -368,7 +364,7 @@ class SyncHandler(object):
sync_config = sync_result_builder.sync_config
with Measure(self.clock, "ephemeral_by_room"):
- typing_key = since_token.typing_key if since_token else "0"
+ typing_key = since_token.typing_key if since_token else 0
room_ids = sync_result_builder.joined_room_ids
@@ -392,7 +388,7 @@ class SyncHandler(object):
event_copy = {k: v for (k, v) in event.items() if k != "room_id"}
ephemeral_by_room.setdefault(room_id, []).append(event_copy)
- receipt_key = since_token.receipt_key if since_token else "0"
+ receipt_key = since_token.receipt_key if since_token else 0
receipt_source = self.event_sources.sources["receipt"]
receipts, receipt_key = await receipt_source.get_new_events(
@@ -588,7 +584,7 @@ class SyncHandler(object):
room_id: str,
sync_config: SyncConfig,
batch: TimelineBatch,
- state: StateMap[EventBase],
+ state: MutableStateMap[EventBase],
now_token: StreamToken,
) -> Optional[JsonDict]:
""" Works out a room summary block for this room, summarising the number
@@ -710,9 +706,8 @@ class SyncHandler(object):
]
missing_hero_state = await self.store.get_events(missing_hero_event_ids)
- missing_hero_state = missing_hero_state.values()
- for s in missing_hero_state:
+ for s in missing_hero_state.values():
cache.set(s.state_key, s.event_id)
state[(EventTypes.Member, s.state_key)] = s
@@ -736,7 +731,7 @@ class SyncHandler(object):
since_token: Optional[StreamToken],
now_token: StreamToken,
full_state: bool,
- ) -> StateMap[EventBase]:
+ ) -> MutableStateMap[EventBase]:
""" Works out the difference in state between the start of the timeline
and the previous sync.
@@ -930,7 +925,7 @@ class SyncHandler(object):
async def unread_notifs_for_room_id(
self, room_id: str, sync_config: SyncConfig
- ) -> Optional[Dict[str, str]]:
+ ) -> Dict[str, int]:
with Measure(self.clock, "unread_notifs_for_room_id"):
last_unread_event_id = await self.store.get_last_receipt_event_id_for_user(
user_id=sync_config.user.to_string(),
@@ -938,15 +933,10 @@ class SyncHandler(object):
receipt_type="m.read",
)
- if last_unread_event_id:
- notifs = await self.store.get_unread_event_push_actions_by_room_for_user(
- room_id, sync_config.user.to_string(), last_unread_event_id
- )
- return notifs
-
- # There is no new information in this period, so your notification
- # count is whatever it was last time.
- return None
+ notifs = await self.store.get_unread_event_push_actions_by_room_for_user(
+ room_id, sync_config.user.to_string(), last_unread_event_id
+ )
+ return notifs
async def generate_sync_result(
self,
@@ -960,7 +950,7 @@ class SyncHandler(object):
# this is due to some of the underlying streams not supporting the ability
# to query up to a given point.
# Always use the `now_token` in `SyncResultBuilder`
- now_token = await self.event_sources.get_current_token()
+ now_token = self.event_sources.get_current_token()
logger.debug(
"Calculating sync response for %r between %s and %s",
@@ -977,7 +967,7 @@ class SyncHandler(object):
raise NotImplementedError()
else:
joined_room_ids = await self.get_rooms_for_user_at(
- user_id, now_token.room_stream_id
+ user_id, now_token.room_key
)
sync_result_builder = SyncResultBuilder(
sync_config,
@@ -1306,12 +1296,11 @@ class SyncHandler(object):
presence_source = self.event_sources.sources["presence"]
since_token = sync_result_builder.since_token
+ presence_key = None
+ include_offline = False
if since_token and not sync_result_builder.full_state:
presence_key = since_token.presence_key
include_offline = True
- else:
- presence_key = None
- include_offline = False
presence, presence_key = await presence_source.get_new_events(
user=user,
@@ -1319,6 +1308,7 @@ class SyncHandler(object):
is_guest=sync_config.is_guest,
include_offline=include_offline,
)
+ assert presence_key
sync_result_builder.now_token = now_token.copy_and_replace(
"presence_key", presence_key
)
@@ -1481,7 +1471,7 @@ class SyncHandler(object):
if rooms_changed:
return True
- stream_id = RoomStreamToken.parse_stream_token(since_token.room_key).stream
+ stream_id = since_token.room_key.stream
for room_id in sync_result_builder.joined_room_ids:
if self.store.has_room_changed_since(room_id, stream_id):
return True
@@ -1605,16 +1595,24 @@ class SyncHandler(object):
if leave_events:
leave_event = leave_events[-1]
- leave_stream_token = await self.store.get_stream_token_for_event(
+ leave_position = await self.store.get_position_for_event(
leave_event.event_id
)
- leave_token = since_token.copy_and_replace(
- "room_key", leave_stream_token
- )
- if since_token and since_token.is_after(leave_token):
+ # If the leave event happened before the since token then we
+ # bail.
+ if since_token and not leave_position.persisted_after(
+ since_token.room_key
+ ):
continue
+ # We can safely convert the position of the leave event into a
+ # stream token as it'll only be used in the context of this
+ # room. (c.f. the docstring of `to_room_stream_token`).
+ leave_token = since_token.copy_and_replace(
+ "room_key", leave_position.to_room_stream_token()
+ )
+
# If this is an out of band message, like a remote invite
# rejection, we include it in the recents batch. Otherwise, we
# let _load_filtered_recents handle fetching the correct
@@ -1747,7 +1745,7 @@ class SyncHandler(object):
continue
leave_token = now_token.copy_and_replace(
- "room_key", "s%d" % (event.stream_ordering,)
+ "room_key", RoomStreamToken(None, event.stream_ordering)
)
room_entries.append(
RoomSyncResultBuilder(
@@ -1769,7 +1767,7 @@ class SyncHandler(object):
ignored_users: Set[str],
room_builder: "RoomSyncResultBuilder",
ephemeral: List[JsonDict],
- tags: Optional[List[JsonDict]],
+ tags: Optional[Dict[str, Dict[str, Any]]],
account_data: Dict[str, JsonDict],
always_include: bool = False,
):
@@ -1885,7 +1883,7 @@ class SyncHandler(object):
)
if room_builder.rtype == "joined":
- unread_notifications = {} # type: Dict[str, str]
+ unread_notifications = {} # type: Dict[str, int]
room_sync = JoinedSyncResult(
room_id=room_id,
timeline=batch,
@@ -1894,14 +1892,16 @@ class SyncHandler(object):
account_data=account_data_events,
unread_notifications=unread_notifications,
summary=summary,
+ unread_count=0,
)
if room_sync or always_include:
notifs = await self.unread_notifs_for_room_id(room_id, sync_config)
- if notifs is not None:
- unread_notifications["notification_count"] = notifs["notify_count"]
- unread_notifications["highlight_count"] = notifs["highlight_count"]
+ unread_notifications["notification_count"] = notifs["notify_count"]
+ unread_notifications["highlight_count"] = notifs["highlight_count"]
+
+ room_sync.unread_count = notifs["unread_count"]
sync_result_builder.joined.append(room_sync)
@@ -1924,7 +1924,7 @@ class SyncHandler(object):
raise Exception("Unrecognized rtype: %r", room_builder.rtype)
async def get_rooms_for_user_at(
- self, user_id: str, stream_ordering: int
+ self, user_id: str, room_key: RoomStreamToken
) -> FrozenSet[str]:
"""Get set of joined rooms for a user at the given stream ordering.
@@ -1950,15 +1950,15 @@ class SyncHandler(object):
# If the membership's stream ordering is after the given stream
# ordering, we need to go and work out if the user was in the room
# before.
- for room_id, membership_stream_ordering in joined_rooms:
- if membership_stream_ordering <= stream_ordering:
+ for room_id, event_pos in joined_rooms:
+ if not event_pos.persisted_after(room_key):
joined_room_ids.add(room_id)
continue
logger.info("User joined room after current token: %s", room_id)
extrems = await self.store.get_forward_extremeties_for_room(
- room_id, stream_ordering
+ room_id, event_pos.stream
)
users_in_room = await self.state.get_current_users_in_room(room_id, extrems)
if user_id in users_in_room:
@@ -2032,7 +2032,7 @@ def _calculate_state(
return {event_id_to_key[e]: e for e in state_ids}
-@attr.s
+@attr.s(slots=True)
class SyncResultBuilder:
"""Used to help build up a new SyncResult for a user
@@ -2068,8 +2068,8 @@ class SyncResultBuilder:
to_device = attr.ib(type=List[JsonDict], default=attr.Factory(list))
-@attr.s
-class RoomSyncResultBuilder(object):
+@attr.s(slots=True)
+class RoomSyncResultBuilder:
"""Stores information needed to create either a `JoinedSyncResult` or
`ArchivedSyncResult`.
|
