mirror of
https://github.com/element-hq/synapse
synced 2024-08-19 04:30:53 +00:00
Fix up sending of m.device_list_update edus
This commit is contained in:
parent
2367c5568c
commit
51e9fe36e4
3 changed files with 83 additions and 81 deletions
|
@ -306,62 +306,74 @@ class TransactionQueue(object):
|
||||||
yield run_on_reactor()
|
yield run_on_reactor()
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
pending_pdus = self.pending_pdus_by_dest.pop(destination, [])
|
pending_pdus = self.pending_pdus_by_dest.pop(destination, [])
|
||||||
pending_edus = self.pending_edus_by_dest.pop(destination, [])
|
pending_edus = self.pending_edus_by_dest.pop(destination, [])
|
||||||
pending_presence = self.pending_presence_by_dest.pop(destination, {})
|
pending_presence = self.pending_presence_by_dest.pop(destination, {})
|
||||||
pending_failures = self.pending_failures_by_dest.pop(destination, [])
|
pending_failures = self.pending_failures_by_dest.pop(destination, [])
|
||||||
|
|
||||||
pending_edus.extend(
|
pending_edus.extend(
|
||||||
self.pending_edus_keyed_by_dest.pop(destination, {}).values()
|
self.pending_edus_keyed_by_dest.pop(destination, {}).values()
|
||||||
|
)
|
||||||
|
|
||||||
|
limiter = yield get_retry_limiter(
|
||||||
|
destination,
|
||||||
|
self.clock,
|
||||||
|
self.store,
|
||||||
|
)
|
||||||
|
|
||||||
|
device_message_edus, device_stream_id, dev_list_id = (
|
||||||
|
yield self._get_new_device_messages(destination)
|
||||||
|
)
|
||||||
|
|
||||||
|
pending_edus.extend(device_message_edus)
|
||||||
|
if pending_presence:
|
||||||
|
pending_edus.append(
|
||||||
|
Edu(
|
||||||
|
origin=self.server_name,
|
||||||
|
destination=destination,
|
||||||
|
edu_type="m.presence",
|
||||||
|
content={
|
||||||
|
"push": [
|
||||||
|
format_user_presence_state(
|
||||||
|
presence, self.clock.time_msec()
|
||||||
|
)
|
||||||
|
for presence in pending_presence.values()
|
||||||
|
]
|
||||||
|
},
|
||||||
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
limiter = yield get_retry_limiter(
|
if pending_pdus:
|
||||||
destination,
|
logger.debug("TX [%s] len(pending_pdus_by_dest[dest]) = %d",
|
||||||
self.clock,
|
destination, len(pending_pdus))
|
||||||
self.store,
|
|
||||||
)
|
|
||||||
|
|
||||||
device_message_edus, device_stream_id = (
|
if not pending_pdus and not pending_edus and not pending_failures:
|
||||||
yield self._get_new_device_messages(destination)
|
logger.debug("TX [%s] Nothing to send", destination)
|
||||||
|
self.last_device_stream_id_by_dest[destination] = (
|
||||||
|
device_stream_id
|
||||||
)
|
)
|
||||||
|
return
|
||||||
|
|
||||||
pending_edus.extend(device_message_edus)
|
success = yield self._send_new_transaction(
|
||||||
if pending_presence:
|
destination, pending_pdus, pending_edus, pending_failures,
|
||||||
pending_edus.append(
|
limiter=limiter,
|
||||||
Edu(
|
)
|
||||||
origin=self.server_name,
|
if success:
|
||||||
destination=destination,
|
# Remove the acknowledged device messages from the database
|
||||||
edu_type="m.presence",
|
# Only bother if we actually sent some device messages
|
||||||
content={
|
if device_message_edus:
|
||||||
"push": [
|
yield self.store.delete_device_msgs_for_remote(
|
||||||
format_user_presence_state(
|
destination, device_stream_id
|
||||||
presence, self.clock.time_msec()
|
)
|
||||||
)
|
logger.info("Marking as sent %r %r", destination, dev_list_id)
|
||||||
for presence in pending_presence.values()
|
yield self.store.mark_as_sent_devices_by_remote(
|
||||||
]
|
destination, dev_list_id
|
||||||
},
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
if pending_pdus:
|
self.last_device_stream_id_by_dest[destination] = device_stream_id
|
||||||
logger.debug("TX [%s] len(pending_pdus_by_dest[dest]) = %d",
|
self.last_device_list_stream_id_by_dest[destination] = dev_list_id
|
||||||
destination, len(pending_pdus))
|
else:
|
||||||
|
break
|
||||||
if not pending_pdus and not pending_edus and not pending_failures:
|
|
||||||
logger.debug("TX [%s] Nothing to send", destination)
|
|
||||||
self.last_device_stream_id_by_dest[destination] = (
|
|
||||||
device_stream_id
|
|
||||||
)
|
|
||||||
return
|
|
||||||
|
|
||||||
success = yield self._send_new_transaction(
|
|
||||||
destination, pending_pdus, pending_edus, pending_failures,
|
|
||||||
device_stream_id,
|
|
||||||
includes_device_messages=bool(device_message_edus),
|
|
||||||
limiter=limiter,
|
|
||||||
)
|
|
||||||
if not success:
|
|
||||||
break
|
|
||||||
except NotRetryingDestination:
|
except NotRetryingDestination:
|
||||||
logger.debug(
|
logger.debug(
|
||||||
"TX [%s] not ready for retry yet - "
|
"TX [%s] not ready for retry yet - "
|
||||||
|
@ -374,8 +386,6 @@ class TransactionQueue(object):
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _get_new_device_messages(self, destination):
|
def _get_new_device_messages(self, destination):
|
||||||
# TODO: Send appropriate device list messages
|
|
||||||
|
|
||||||
last_device_stream_id = self.last_device_stream_id_by_dest.get(destination, 0)
|
last_device_stream_id = self.last_device_stream_id_by_dest.get(destination, 0)
|
||||||
to_device_stream_id = self.store.get_to_device_stream_token()
|
to_device_stream_id = self.store.get_to_device_stream_token()
|
||||||
contents, stream_id = yield self.store.get_new_device_msgs_for_remote(
|
contents, stream_id = yield self.store.get_new_device_msgs_for_remote(
|
||||||
|
@ -404,13 +414,12 @@ class TransactionQueue(object):
|
||||||
)
|
)
|
||||||
for content in results
|
for content in results
|
||||||
)
|
)
|
||||||
defer.returnValue((edus, stream_id))
|
defer.returnValue((edus, stream_id, now_stream_id))
|
||||||
|
|
||||||
@measure_func("_send_new_transaction")
|
@measure_func("_send_new_transaction")
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def _send_new_transaction(self, destination, pending_pdus, pending_edus,
|
def _send_new_transaction(self, destination, pending_pdus, pending_edus,
|
||||||
pending_failures, device_stream_id,
|
pending_failures, limiter):
|
||||||
includes_device_messages, limiter):
|
|
||||||
|
|
||||||
# Sort based on the order field
|
# Sort based on the order field
|
||||||
pending_pdus.sort(key=lambda t: t[1])
|
pending_pdus.sort(key=lambda t: t[1])
|
||||||
|
@ -521,14 +530,6 @@ class TransactionQueue(object):
|
||||||
"Failed to send event %s to %s", p.event_id, destination
|
"Failed to send event %s to %s", p.event_id, destination
|
||||||
)
|
)
|
||||||
success = False
|
success = False
|
||||||
else:
|
|
||||||
# Remove the acknowledged device messages from the database
|
|
||||||
# Only bother if we actually sent some device messages
|
|
||||||
if includes_device_messages:
|
|
||||||
yield self.store.delete_device_msgs_for_remote(
|
|
||||||
destination, device_stream_id
|
|
||||||
)
|
|
||||||
self.last_device_stream_id_by_dest[destination] = device_stream_id
|
|
||||||
except RuntimeError as e:
|
except RuntimeError as e:
|
||||||
# We capture this here as there as nothing actually listens
|
# We capture this here as there as nothing actually listens
|
||||||
# for this finishing functions deferred.
|
# for this finishing functions deferred.
|
||||||
|
|
|
@ -29,6 +29,7 @@ class DeviceHandler(BaseHandler):
|
||||||
super(DeviceHandler, self).__init__(hs)
|
super(DeviceHandler, self).__init__(hs)
|
||||||
|
|
||||||
self.state = hs.get_state_handler()
|
self.state = hs.get_state_handler()
|
||||||
|
self.federation = hs.get_federation_sender()
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def check_device_registered(self, user_id, device_id,
|
def check_device_registered(self, user_id, device_id,
|
||||||
|
|
|
@ -141,11 +141,11 @@ class DeviceStore(SQLBaseStore):
|
||||||
def get_devices_by_remote(self, destination, from_stream_id):
|
def get_devices_by_remote(self, destination, from_stream_id):
|
||||||
now_stream_id = self._device_list_id_gen.get_current_token()
|
now_stream_id = self._device_list_id_gen.get_current_token()
|
||||||
|
|
||||||
has_changed = self._device_list_stream_cache.has_entity_changed(
|
has_changed = self._device_list_federation_stream_cache.has_entity_changed(
|
||||||
destination, int(from_stream_id)
|
destination, int(from_stream_id)
|
||||||
)
|
)
|
||||||
if not has_changed:
|
if not has_changed:
|
||||||
defer.returnValue((now_stream_id, []))
|
return (now_stream_id, [])
|
||||||
|
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
"get_devices_by_remote", self._get_devices_by_remote_txn,
|
"get_devices_by_remote", self._get_devices_by_remote_txn,
|
||||||
|
@ -165,7 +165,7 @@ class DeviceStore(SQLBaseStore):
|
||||||
rows = txn.fetchall()
|
rows = txn.fetchall()
|
||||||
|
|
||||||
if not rows:
|
if not rows:
|
||||||
return now_stream_id, []
|
return (now_stream_id, [])
|
||||||
|
|
||||||
# maps (user_id, device_id) -> stream_id
|
# maps (user_id, device_id) -> stream_id
|
||||||
query_map = {(r[0], r[1]): r[2] for r in rows}
|
query_map = {(r[0], r[1]): r[2] for r in rows}
|
||||||
|
@ -189,7 +189,7 @@ class DeviceStore(SQLBaseStore):
|
||||||
result = {
|
result = {
|
||||||
"user_id": user_id,
|
"user_id": user_id,
|
||||||
"device_id": device_id,
|
"device_id": device_id,
|
||||||
"prev_id": prev_id,
|
"prev_id": [prev_id] if prev_id else [],
|
||||||
"stream_id": stream_id,
|
"stream_id": stream_id,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -202,9 +202,9 @@ class DeviceStore(SQLBaseStore):
|
||||||
if device_display_name:
|
if device_display_name:
|
||||||
result["device_display_name"] = device_display_name
|
result["device_display_name"] = device_display_name
|
||||||
|
|
||||||
results.setdefault(user_id, {})[device_id] = result
|
results.append(result)
|
||||||
|
|
||||||
return now_stream_id, results
|
return (now_stream_id, results)
|
||||||
|
|
||||||
def mark_as_sent_devices_by_remote(self, destination, stream_id):
|
def mark_as_sent_devices_by_remote(self, destination, stream_id):
|
||||||
return self.runInteraction(
|
return self.runInteraction(
|
||||||
|
@ -212,19 +212,6 @@ class DeviceStore(SQLBaseStore):
|
||||||
destination, stream_id,
|
destination, stream_id,
|
||||||
)
|
)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
|
||||||
def get_user_whose_devices_changed(self, from_key):
|
|
||||||
from_key = int(from_key)
|
|
||||||
changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
|
|
||||||
if changed is not None:
|
|
||||||
defer.returnValue(set(changed))
|
|
||||||
|
|
||||||
sql = """
|
|
||||||
SELECT user_id FROM device_lists_stream WHERE stream_id > ?
|
|
||||||
"""
|
|
||||||
rows = yield self._execute("get_user_whose_devices_changed", None, sql, from_key)
|
|
||||||
defer.returnValue(set(row["user_id"] for row in rows))
|
|
||||||
|
|
||||||
def _mark_as_sent_devices_by_remote_txn(self, txn, destination, stream_id):
|
def _mark_as_sent_devices_by_remote_txn(self, txn, destination, stream_id):
|
||||||
sql = """
|
sql = """
|
||||||
DELETE FROM device_lists_outbound_pokes
|
DELETE FROM device_lists_outbound_pokes
|
||||||
|
@ -239,7 +226,20 @@ class DeviceStore(SQLBaseStore):
|
||||||
UPDATE device_lists_outbound_pokes SET sent = ?
|
UPDATE device_lists_outbound_pokes SET sent = ?
|
||||||
WHERE destination = ? AND stream_id <= ?
|
WHERE destination = ? AND stream_id <= ?
|
||||||
"""
|
"""
|
||||||
txn.execute(sql, (destination, True,))
|
txn.execute(sql, (True, destination, stream_id,))
|
||||||
|
|
||||||
|
@defer.inlineCallbacks
|
||||||
|
def get_user_whose_devices_changed(self, from_key):
|
||||||
|
from_key = int(from_key)
|
||||||
|
changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
|
||||||
|
if changed is not None:
|
||||||
|
defer.returnValue(set(changed))
|
||||||
|
|
||||||
|
sql = """
|
||||||
|
SELECT user_id FROM device_lists_stream WHERE stream_id > ?
|
||||||
|
"""
|
||||||
|
rows = yield self._execute("get_user_whose_devices_changed", None, sql, from_key)
|
||||||
|
defer.returnValue(set(row["user_id"] for row in rows))
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def add_device_change_to_streams(self, user_id, device_id, hosts):
|
def add_device_change_to_streams(self, user_id, device_id, hosts):
|
||||||
|
|
Loading…
Reference in a new issue