Resync remote device list when detected as stale. (#6786)
parent
c3d4ad8afd
commit
b660327056
|
@ -0,0 +1 @@
|
||||||
|
Attempt to resync remote users' devices when detected as stale.
|
|
@ -21,6 +21,7 @@ from canonicaljson import json
|
||||||
from twisted.internet import defer
|
from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.api.errors import SynapseError
|
from synapse.api.errors import SynapseError
|
||||||
|
from synapse.logging.context import run_in_background
|
||||||
from synapse.logging.opentracing import (
|
from synapse.logging.opentracing import (
|
||||||
get_active_span_text_map,
|
get_active_span_text_map,
|
||||||
log_kv,
|
log_kv,
|
||||||
|
@ -48,6 +49,8 @@ class DeviceMessageHandler(object):
|
||||||
"m.direct_to_device", self.on_direct_to_device_edu
|
"m.direct_to_device", self.on_direct_to_device_edu
|
||||||
)
|
)
|
||||||
|
|
||||||
|
self._device_list_updater = hs.get_device_handler().device_list_updater
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def on_direct_to_device_edu(self, origin, content):
|
def on_direct_to_device_edu(self, origin, content):
|
||||||
local_messages = {}
|
local_messages = {}
|
||||||
|
@ -134,8 +137,11 @@ class DeviceMessageHandler(object):
|
||||||
unknown_devices,
|
unknown_devices,
|
||||||
)
|
)
|
||||||
yield self.store.mark_remote_user_device_cache_as_stale(sender_user_id)
|
yield self.store.mark_remote_user_device_cache_as_stale(sender_user_id)
|
||||||
# TODO: Poke something to start trying to refetch user's
|
|
||||||
# keys.
|
# Immediately attempt a resync in the background
|
||||||
|
run_in_background(
|
||||||
|
self._device_list_updater.user_device_resync, sender_user_id
|
||||||
|
)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def send_device_message(self, sender_user_id, message_type, messages):
|
def send_device_message(self, sender_user_id, message_type, messages):
|
||||||
|
|
|
@ -57,6 +57,7 @@ from synapse.logging.context import (
|
||||||
run_in_background,
|
run_in_background,
|
||||||
)
|
)
|
||||||
from synapse.logging.utils import log_function
|
from synapse.logging.utils import log_function
|
||||||
|
from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
|
||||||
from synapse.replication.http.federation import (
|
from synapse.replication.http.federation import (
|
||||||
ReplicationCleanRoomRestServlet,
|
ReplicationCleanRoomRestServlet,
|
||||||
ReplicationFederationSendEventsRestServlet,
|
ReplicationFederationSendEventsRestServlet,
|
||||||
|
@ -156,6 +157,13 @@ class FederationHandler(BaseHandler):
|
||||||
hs
|
hs
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if hs.config.worker_app:
|
||||||
|
self._user_device_resync = ReplicationUserDevicesResyncRestServlet.make_client(
|
||||||
|
hs
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._device_list_updater = hs.get_device_handler().device_list_updater
|
||||||
|
|
||||||
# When joining a room we need to queue any events for that room up
|
# When joining a room we need to queue any events for that room up
|
||||||
self.room_queues = {}
|
self.room_queues = {}
|
||||||
self._room_pdu_linearizer = Linearizer("fed_room_pdu")
|
self._room_pdu_linearizer = Linearizer("fed_room_pdu")
|
||||||
|
@ -759,8 +767,14 @@ class FederationHandler(BaseHandler):
|
||||||
await self.store.mark_remote_user_device_cache_as_stale(
|
await self.store.mark_remote_user_device_cache_as_stale(
|
||||||
event.sender
|
event.sender
|
||||||
)
|
)
|
||||||
# TODO: Poke something to start trying to refetch user's
|
|
||||||
# keys.
|
# Immediately attempt a resync in the background
|
||||||
|
if self.config.worker_app:
|
||||||
|
return run_in_background(self._user_device_resync, event.sender)
|
||||||
|
else:
|
||||||
|
return run_in_background(
|
||||||
|
self._device_list_updater.user_device_resync, event.sender
|
||||||
|
)
|
||||||
|
|
||||||
@log_function
|
@log_function
|
||||||
async def backfill(self, dest, room_id, limit, extremities):
|
async def backfill(self, dest, room_id, limit, extremities):
|
||||||
|
|
|
@ -81,6 +81,9 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# the tests assume that we are starting at unix time 1000
|
||||||
|
reactor.pump((1000,))
|
||||||
|
|
||||||
hs = self.setup_test_homeserver(
|
hs = self.setup_test_homeserver(
|
||||||
notifier=Mock(), http_client=mock_federation_client, keyring=mock_keyring
|
notifier=Mock(), http_client=mock_federation_client, keyring=mock_keyring
|
||||||
)
|
)
|
||||||
|
@ -90,9 +93,6 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase):
|
||||||
return hs
|
return hs
|
||||||
|
|
||||||
def prepare(self, reactor, clock, hs):
|
def prepare(self, reactor, clock, hs):
|
||||||
# the tests assume that we are starting at unix time 1000
|
|
||||||
reactor.pump((1000,))
|
|
||||||
|
|
||||||
mock_notifier = hs.get_notifier()
|
mock_notifier = hs.get_notifier()
|
||||||
self.on_new_event = mock_notifier.on_new_event
|
self.on_new_event = mock_notifier.on_new_event
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue