From e993925279b9e4cc00d52e7331cc5fa3cd4ae59f Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 25 Aug 2016 17:35:37 +0100 Subject: Add store-and-forward direct-to-device messaging --- synapse/storage/deviceinbox.py | 136 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 synapse/storage/deviceinbox.py (limited to 'synapse/storage/deviceinbox.py') diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py new file mode 100644 index 0000000000..0b71a5cd90 --- /dev/null +++ b/synapse/storage/deviceinbox.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +# Copyright 2016 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import ujson + +from twisted.internet import defer + +from ._base import SQLBaseStore + + +logger = logging.getLogger(__name__) + + +class DeviceInboxStore(SQLBaseStore): + + @defer.inlineCallbacks + def add_messages_to_device_inbox(self, messages_by_user_then_device): + """ + Args: + messages_by_user_and_device(dict): + Dictionary of user_id to device_id to message. + Returns: + A deferred that resolves when the messages have been inserted. + """ + + def select_devices_txn(txn, user_id, devices): + if not devices: + return [] + sql = ( + "SELECT user_id, device_id FROM devices" + " WHERE user_id = ? AND device_id IN (" + + ",".join("?" * len(devices)) + + ")" + ) + # TODO: Maybe this needs to be done in batches if there are + # too many local devices for a given user. + args = [user_id] + devices + txn.execute(sql, args) + return [tuple(row) for row in txn.fetchall()] + + def add_messages_to_device_inbox_txn(txn, stream_id): + local_users_and_devices = set() + for user_id, messages_by_device in messages_by_user_then_device.items(): + local_users_and_devices.update( + select_devices_txn(txn, user_id, messages_by_device.keys()) + ) + + sql = ( + "INSERT INTO device_inbox" + " (user_id, device_id, stream_id, message_json)" + " VALUES (?,?,?,?)" + ) + rows = [] + for user_id, messages_by_device in messages_by_user_then_device.items(): + for device_id, message in messages_by_device.items(): + message_json = ujson.dumps(message) + # Only insert into the local inbox if the device exists on + # this server + if (user_id, device_id) in local_users_and_devices: + rows.append((user_id, device_id, stream_id, message_json)) + + txn.executemany(sql, rows) + + with self._device_inbox_id_gen.get_next() as stream_id: + yield self.runInteraction( + "add_messages_to_device_inbox", + add_messages_to_device_inbox_txn, + stream_id + ) + + def get_new_messages_for_device( + self, user_id, device_id, current_stream_id, limit=100 + ): + """ + Args: + user_id(str): The recipient user_id. + device_id(str): The recipient device_id. + current_stream_id(int): The current position of the to device + message stream. + Returns: + Deferred ([dict], int): List of messages for the device and where + in the stream the messages got to. + """ + def get_new_messages_for_device_txn(txn): + sql = ( + "SELECT stream_id, message_json FROM device_inbox" + " WHERE user_id = ? AND device_id = ?" + " AND stream_id <= ?" + " ORDER BY stream_id ASC" + " LIMIT ?" + ) + txn.execute(sql, (user_id, device_id, current_stream_id, limit)) + messages = [] + for row in txn.fetchall(): + stream_pos = row[0] + messages.append(ujson.loads(row[1])) + if len(messages) < limit: + stream_pos = current_stream_id + return (messages, stream_pos) + + return self.runInteraction( + "get_new_messages_for_device", get_new_messages_for_device_txn, + ) + + def delete_messages_for_device(self, user_id, device_id, up_to_stream_id): + """ + Args: + user_id(str): The recipient user_id. + device_id(str): The recipient device_id. + up_to_stream_id(int): Where to delete messages up to. + Returns: + A deferred that resolves when the messages have been deleted. + """ + def delete_messages_for_device_txn(txn): + sql = ( + "DELETE FROM device_inbox" + " WHERE user_id = ? AND device_id = ?" + " AND stream_id <= ?" + ) + txn.execute(sql, (user_id, device_id, up_to_stream_id)) + + def get_to_device_stream_token(self): + return self._device_inbox_id_gen.get_current_token() -- cgit 1.5.1 From 641efb6a39f0d913895cec73b1bda8dbb548c2a3 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 25 Aug 2016 18:14:02 +0100 Subject: Fix the deduplication of incoming direct-to-device messages --- synapse/rest/client/v2_alpha/sendtodevice.py | 13 ++++++++++++- synapse/storage/deviceinbox.py | 4 ++++ 2 files changed, 16 insertions(+), 1 deletion(-) (limited to 'synapse/storage/deviceinbox.py') diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 6180266499..42754e8774 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -19,6 +19,7 @@ from twisted.internet import defer from synapse.http.servlet import parse_json_object_from_request from synapse.http import servlet +from synapse.rest.client.v1.transactions import HttpTransactionStore from ._base import client_v2_patterns logger = logging.getLogger(__name__) @@ -40,9 +41,17 @@ class SendToDeviceRestServlet(servlet.RestServlet): self.auth = hs.get_auth() self.store = hs.get_datastore() self.is_mine_id = hs.is_mine_id + self.txns = HttpTransactionStore() @defer.inlineCallbacks def on_PUT(self, request, message_type, txn_id): + try: + defer.returnValue( + self.txns.get_client_transaction(request, txn_id) + ) + except KeyError: + pass + requester = yield self.auth.get_user_by_req(request) content = parse_json_object_from_request(request) @@ -62,7 +71,9 @@ class SendToDeviceRestServlet(servlet.RestServlet): yield self.store.add_messages_to_device_inbox(local_messages) - defer.returnValue((200, {})) + response = (200, {}) + self.txns.store_client_transaction(request, txn_id, response) + defer.returnValue(response) def register_servlets(hs, http_server): diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index 0b71a5cd90..e682fe1bd1 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -132,5 +132,9 @@ class DeviceInboxStore(SQLBaseStore): ) txn.execute(sql, (user_id, device_id, up_to_stream_id)) + return self.runInteraction( + "delete_messages_for_device", delete_messages_for_device_txn + ) + def get_to_device_stream_token(self): return self._device_inbox_id_gen.get_current_token() -- cgit 1.5.1 From 1aa3e1d2874370fbb6fa9f4fd2b8a110d81981fc Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 31 Aug 2016 10:38:58 +0100 Subject: Add a replication stream for direct to device messages --- synapse/replication/resource.py | 19 +++++++++++- synapse/replication/slave/storage/deviceinbox.py | 12 ++++++++ synapse/rest/client/v2_alpha/sendtodevice.py | 10 ++++-- synapse/storage/deviceinbox.py | 39 ++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 3 deletions(-) (limited to 'synapse/storage/deviceinbox.py') diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py index 84993b33b3..533676814a 100644 --- a/synapse/replication/resource.py +++ b/synapse/replication/resource.py @@ -42,6 +42,7 @@ STREAM_NAMES = ( ("pushers",), ("state",), ("caches",), + ("to_device",), ) @@ -144,6 +145,7 @@ class ReplicationResource(Resource): pushers_token, state_token, caches_token, + int(stream_token.to_device_key), )) @request_handler() @@ -193,6 +195,7 @@ class ReplicationResource(Resource): yield self.pushers(writer, current_token, limit, request_streams) yield self.state(writer, current_token, limit, request_streams) yield self.caches(writer, current_token, limit, request_streams) + yield self.to_device(writer, current_token, limit, request_streams) self.streams(writer, current_token, request_streams) logger.info("Replicated %d rows", writer.total) @@ -398,6 +401,20 @@ class ReplicationResource(Resource): "position", "cache_func", "keys", "invalidation_ts" )) + @defer.inlineCallbacks + def to_device(self, writer, current_token, limit, request_streams): + current_position = current_token.to_device + + to_device = request_streams.get("to_device") + + if to_device is not None: + to_device_rows = yield self.store.get_all_new_device_messages( + to_device, current_position, limit + ) + writer.write_header_and_rows("to_device", to_device_rows, ( + "position", "user_id", "device_id", "message_json" + )) + class _Writer(object): """Writes the streams as a JSON object as the response to the request""" @@ -426,7 +443,7 @@ class _Writer(object): class _ReplicationToken(collections.namedtuple("_ReplicationToken", ( "events", "presence", "typing", "receipts", "account_data", "backfill", - "push_rules", "pushers", "state", "caches", + "push_rules", "pushers", "state", "caches", "to_device", ))): __slots__ = [] diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py index 7583d23708..64d8eb2af1 100644 --- a/synapse/replication/slave/storage/deviceinbox.py +++ b/synapse/replication/slave/storage/deviceinbox.py @@ -28,3 +28,15 @@ class SlavedDeviceInboxStore(BaseSlavedStore): get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__ get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__ delete_messages_for_device = DataStore.delete_messages_for_device.__func__ + + def stream_positions(self): + result = super(SlavedDeviceInboxStore, self).stream_positions() + result["to_device"] = self._device_inbox_id_gen.get_current_token() + return result + + def process_replication(self, result): + stream = result.get("to_device") + if stream: + self._device_inbox_id_gen.advance(int(stream["position"])) + + return super(SlavedDeviceInboxStore, self).process_replication(result) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 00533741af..7c0991ca55 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -40,6 +40,7 @@ class SendToDeviceRestServlet(servlet.RestServlet): self.hs = hs self.auth = hs.get_auth() self.store = hs.get_datastore() + self.notifier = hs.get_notifier() self.is_mine_id = hs.is_mine_id self.txns = HttpTransactionStore() @@ -71,9 +72,14 @@ class SendToDeviceRestServlet(servlet.RestServlet): } for device_id, message_content in by_device.items() } - local_messages[user_id] = messages_by_device + if messages_by_device: + local_messages[user_id] = messages_by_device - yield self.store.add_messages_to_device_inbox(local_messages) + stream_id = yield self.store.add_messages_to_device_inbox(local_messages) + + self.notifier.on_new_event( + "to_device", stream_id, users=local_messages.keys() + ) response = (200, {}) self.txns.store_client_transaction(request, txn_id, response) diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index e682fe1bd1..27ed1004da 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -136,5 +136,44 @@ class DeviceInboxStore(SQLBaseStore): "delete_messages_for_device", delete_messages_for_device_txn ) + def get_all_new_device_messages(self, last_pos, current_pos, limit): + """ + Args: + last_pos(int): + current_pos(int): + limit(int): + Returns: + A deferred list of rows from the device inbox + """ + if last_pos == current_pos: + return defer.succeed([]) + + def get_all_new_device_messages_txn(txn): + sql = ( + "SELECT stream_id FROM device_inbox" + " WHERE ? < stream_id AND stream_id <= ?" + " GROUP BY stream_id" + " ORDER BY stream_id ASC" + " LIMIT ?" + ) + txn.execute(sql, (last_pos, current_pos, limit)) + stream_ids = txn.fetchall() + if not stream_ids: + return [] + max_stream_id_in_limit = stream_ids[-1] + + sql = ( + "SELECT stream_id, user_id, device_id, message_json" + " FROM device_inbox" + " WHERE ? < stream_id AND stream_id <= ?" + " ORDER BY stream_id ASC" + ) + txn.execute(sql, (last_pos, max_stream_id_in_limit)) + return txn.fetchall() + + return self.runInteraction( + "get_all_new_device_messages", get_all_new_device_messages_txn + ) + def get_to_device_stream_token(self): return self._device_inbox_id_gen.get_current_token() -- cgit 1.5.1 From a66225275896c00287e4f627ebbacd6f7b34e6ba Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 31 Aug 2016 10:42:52 +0100 Subject: Return the current stream position from add_messages_to_device_inbox --- synapse/storage/deviceinbox.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'synapse/storage/deviceinbox.py') diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index 27ed1004da..2fa0a218b9 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -33,7 +33,8 @@ class DeviceInboxStore(SQLBaseStore): messages_by_user_and_device(dict): Dictionary of user_id to device_id to message. Returns: - A deferred that resolves when the messages have been inserted. + A deferred stream_id that resolves when the messages have been + inserted. """ def select_devices_txn(txn, user_id, devices): @@ -81,6 +82,8 @@ class DeviceInboxStore(SQLBaseStore): stream_id ) + defer.returnValue(self._device_inbox_id_gen.get_current_token()) + def get_new_messages_for_device( self, user_id, device_id, current_stream_id, limit=100 ): -- cgit 1.5.1 From 65fa37ac5e8b3e661b853b71cfe6a264d5369bdc Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 2 Sep 2016 15:50:37 +0100 Subject: Only return new device messages in /sync --- synapse/handlers/sync.py | 2 +- synapse/storage/deviceinbox.py | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) (limited to 'synapse/storage/deviceinbox.py') diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 14f2032afa..b5962f4f5a 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -576,7 +576,7 @@ class SyncHandler(object): logger.debug("Getting messages up to %d", now_token.to_device_key) messages, stream_id = yield self.store.get_new_messages_for_device( - user_id, device_id, now_token.to_device_key + user_id, device_id, since_stream_id, now_token.to_device_key ) logger.debug("Got messages up to %d: %r", stream_id, messages) sync_result_builder.now_token = now_token.copy_and_replace( diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index 2fa0a218b9..68116b0394 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -85,7 +85,7 @@ class DeviceInboxStore(SQLBaseStore): defer.returnValue(self._device_inbox_id_gen.get_current_token()) def get_new_messages_for_device( - self, user_id, device_id, current_stream_id, limit=100 + self, user_id, device_id, last_stream_id, current_stream_id, limit=100 ): """ Args: @@ -101,11 +101,13 @@ class DeviceInboxStore(SQLBaseStore): sql = ( "SELECT stream_id, message_json FROM device_inbox" " WHERE user_id = ? AND device_id = ?" - " AND stream_id <= ?" + " AND ? < stream_id AND stream_id <= ?" " ORDER BY stream_id ASC" " LIMIT ?" ) - txn.execute(sql, (user_id, device_id, current_stream_id, limit)) + txn.execute(sql, ( + user_id, device_id, last_stream_id, current_stream_id, limit + )) messages = [] for row in txn.fetchall(): stream_pos = row[0] -- cgit 1.5.1