From c276bd996916adce899410b9c4c891892f51b992 Mon Sep 17 00:00:00 2001 From: Will Hunt Date: Thu, 15 Oct 2020 17:33:28 +0100 Subject: Send some ephemeral events to appservices (#8437) Optionally sends typing, presence, and read receipt information to appservices. --- synapse/handlers/receipts.py | 35 ++++++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) (limited to 'synapse/handlers/receipts.py') diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py index 7225923757..c242c409cf 100644 --- a/synapse/handlers/receipts.py +++ b/synapse/handlers/receipts.py @@ -13,9 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging +from typing import List, Tuple +from synapse.appservice import ApplicationService from synapse.handlers._base import BaseHandler -from synapse.types import ReadReceipt, get_domain_from_id +from synapse.types import JsonDict, ReadReceipt, get_domain_from_id from synapse.util.async_helpers import maybe_awaitable logger = logging.getLogger(__name__) @@ -140,5 +142,36 @@ class ReceiptEventSource: return (events, to_key) + async def get_new_events_as( + self, from_key: int, service: ApplicationService + ) -> Tuple[List[JsonDict], int]: + """Returns a set of new receipt events that an appservice + may be interested in. + + Args: + from_key: the stream position at which events should be fetched from + service: The appservice which may be interested + """ + from_key = int(from_key) + to_key = self.get_current_key() + + if from_key == to_key: + return [], to_key + + # We first need to fetch all new receipts + rooms_to_events = await self.store.get_linearized_receipts_for_all_rooms( + from_key=from_key, to_key=to_key + ) + + # Then filter down to rooms that the AS can read + events = [] + for room_id, event in rooms_to_events.items(): + if not await service.matches_user_in_member_list(room_id, self.store): + continue + + events.append(event) + + return (events, to_key) + def get_current_key(self, direction="f"): return self.store.get_max_receipt_stream_id() -- cgit 1.5.1 From 51338491c9bedcdfb5f9babad2a34cbfea6b57d2 Mon Sep 17 00:00:00 2001 From: Will Hunt Date: Wed, 18 Nov 2020 18:54:09 +0000 Subject: Improve appservice handler to send only the most recent read receipts when no stream_id is stored. (#8744) * Make this line debug (it's noisy) * Don't include from_key for presence if we are at 0 * Limit read receipts for all rooms to 100 * changelog.d/8744.bugfix * Allow from_key to be None * Update 8744.bugfix * The from_key is superflous * Update comment --- changelog.d/8744.bugfix | 1 + synapse/handlers/appservice.py | 2 +- synapse/handlers/receipts.py | 3 ++- synapse/storage/databases/main/receipts.py | 7 ++++++- 4 files changed, 10 insertions(+), 3 deletions(-) create mode 100644 changelog.d/8744.bugfix (limited to 'synapse/handlers/receipts.py') diff --git a/changelog.d/8744.bugfix b/changelog.d/8744.bugfix new file mode 100644 index 0000000000..f8f9630bd6 --- /dev/null +++ b/changelog.d/8744.bugfix @@ -0,0 +1 @@ +Fix a bug where appservices may be sent an excessive amount of read receipts and presence. Broke in v1.22.0. diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index 9fc8444228..5c6458eb52 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -226,7 +226,7 @@ class ApplicationServicesHandler: new_token: Optional[int], users: Collection[Union[str, UserID]], ): - logger.info("Checking interested services for %s" % (stream_key)) + logger.debug("Checking interested services for %s" % (stream_key)) with Measure(self.clock, "notify_interested_services_ephemeral"): for service in services: # Only handle typing if we have the latest token diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py index c242c409cf..153cbae7b9 100644 --- a/synapse/handlers/receipts.py +++ b/synapse/handlers/receipts.py @@ -158,7 +158,8 @@ class ReceiptEventSource: if from_key == to_key: return [], to_key - # We first need to fetch all new receipts + # Fetch all read receipts for all rooms, up to a limit of 100. This is ordered + # by most recent. rooms_to_events = await self.store.get_linearized_receipts_for_all_rooms( from_key=from_key, to_key=to_key ) diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index ca7917c989..1e7949a323 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -278,7 +278,8 @@ class ReceiptsWorkerStore(SQLBaseStore, metaclass=abc.ABCMeta): async def get_linearized_receipts_for_all_rooms( self, to_key: int, from_key: Optional[int] = None ) -> Dict[str, JsonDict]: - """Get receipts for all rooms between two stream_ids. + """Get receipts for all rooms between two stream_ids, up + to a limit of the latest 100 read receipts. Args: to_key: Max stream id to fetch receipts upto. @@ -294,12 +295,16 @@ class ReceiptsWorkerStore(SQLBaseStore, metaclass=abc.ABCMeta): sql = """ SELECT * FROM receipts_linearized WHERE stream_id > ? AND stream_id <= ? + ORDER BY stream_id DESC + LIMIT 100 """ txn.execute(sql, [from_key, to_key]) else: sql = """ SELECT * FROM receipts_linearized WHERE stream_id <= ? + ORDER BY stream_id DESC + LIMIT 100 """ txn.execute(sql, [to_key]) -- cgit 1.5.1