diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 14f2032afa8d..b5962f4f5a19 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -576,7 +576,7 @@ def _generate_sync_entry_for_to_device(self, sync_result_builder): 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 2fa0a218b95e..68116b0394a3 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -85,7 +85,7 @@ def add_messages_to_device_inbox_txn(txn, 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 + self, user_id, device_id, last_stream_id, current_stream_id, limit=100 ): """ Args: @@ -101,11 +101,13 @@ 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 <= ?" + " 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]