Merge pull request #1140 from matrix-org/erikj/typing_fed_timeout

Time out typing over federation
This commit is contained in:
Erik Johnston 2016-09-26 11:24:14 +01:00 committed by GitHub
commit a38d46249e
6 changed files with 121 additions and 82 deletions

View file

@ -136,9 +136,7 @@ class FederationClient(FederationBase):
sent_edus_counter.inc() sent_edus_counter.inc()
# TODO, add errback, etc.
self._transaction_queue.enqueue_edu(edu, key=key) self._transaction_queue.enqueue_edu(edu, key=key)
return defer.succeed(None)
@log_function @log_function
def send_device_messages(self, destination): def send_device_messages(self, destination):

View file

@ -16,10 +16,9 @@
from twisted.internet import defer from twisted.internet import defer
from synapse.api.errors import SynapseError, AuthError from synapse.api.errors import SynapseError, AuthError
from synapse.util.logcontext import ( from synapse.util.logcontext import preserve_fn
PreserveLoggingContext, preserve_fn, preserve_context_over_deferred,
)
from synapse.util.metrics import Measure from synapse.util.metrics import Measure
from synapse.util.wheel_timer import WheelTimer
from synapse.types import UserID, get_domain_from_id from synapse.types import UserID, get_domain_from_id
import logging import logging
@ -35,6 +34,13 @@ logger = logging.getLogger(__name__)
RoomMember = namedtuple("RoomMember", ("room_id", "user_id")) RoomMember = namedtuple("RoomMember", ("room_id", "user_id"))
# How often we expect remote servers to resend us presence.
FEDERATION_TIMEOUT = 60 * 1000
# How often to resend typing across federation.
FEDERATION_PING_INTERVAL = 40 * 1000
class TypingHandler(object): class TypingHandler(object):
def __init__(self, hs): def __init__(self, hs):
self.store = hs.get_datastore() self.store = hs.get_datastore()
@ -44,7 +50,10 @@ class TypingHandler(object):
self.notifier = hs.get_notifier() self.notifier = hs.get_notifier()
self.state = hs.get_state_handler() self.state = hs.get_state_handler()
self.hs = hs
self.clock = hs.get_clock() self.clock = hs.get_clock()
self.wheel_timer = WheelTimer(bucket_size=5000)
self.federation = hs.get_replication_layer() self.federation = hs.get_replication_layer()
@ -53,7 +62,7 @@ class TypingHandler(object):
hs.get_distributor().observe("user_left_room", self.user_left_room) hs.get_distributor().observe("user_left_room", self.user_left_room)
self._member_typing_until = {} # clock time we expect to stop self._member_typing_until = {} # clock time we expect to stop
self._member_typing_timer = {} # deferreds to manage theabove self._member_last_federation_poke = {}
# map room IDs to serial numbers # map room IDs to serial numbers
self._room_serials = {} self._room_serials = {}
@ -61,12 +70,41 @@ class TypingHandler(object):
# map room IDs to sets of users currently typing # map room IDs to sets of users currently typing
self._room_typing = {} self._room_typing = {}
def tearDown(self): self.clock.looping_call(
"""Cancels all the pending timers. self._handle_timeouts,
Normally this shouldn't be needed, but it's required from unit tests 5000,
to avoid a "Reactor was unclean" warning.""" )
for t in self._member_typing_timer.values():
self.clock.cancel_call_later(t) def _handle_timeouts(self):
logger.info("Checking for typing timeouts")
now = self.clock.time_msec()
members = set(self.wheel_timer.fetch(now))
for member in members:
if not self.is_typing(member):
# Nothing to do if they're no longer typing
continue
until = self._member_typing_until.get(member, None)
if not until or until < now:
logger.info("Timing out typing for: %s", member.user_id)
preserve_fn(self._stopped_typing)(member)
continue
# Check if we need to resend a keep alive over federation for this
# user.
if self.hs.is_mine_id(member.user_id):
last_fed_poke = self._member_last_federation_poke.get(member, None)
if not last_fed_poke or last_fed_poke + FEDERATION_PING_INTERVAL < now:
preserve_fn(self._push_remote)(
member=member,
typing=True
)
def is_typing(self, member):
return member.user_id in self._room_typing.get(member.room_id, [])
@defer.inlineCallbacks @defer.inlineCallbacks
def started_typing(self, target_user, auth_user, room_id, timeout): def started_typing(self, target_user, auth_user, room_id, timeout):
@ -85,23 +123,17 @@ class TypingHandler(object):
"%s has started typing in %s", target_user_id, room_id "%s has started typing in %s", target_user_id, room_id
) )
until = self.clock.time_msec() + timeout
member = RoomMember(room_id=room_id, user_id=target_user_id) member = RoomMember(room_id=room_id, user_id=target_user_id)
was_present = member in self._member_typing_until was_present = member.user_id in self._room_typing.get(room_id, set())
if member in self._member_typing_timer: now = self.clock.time_msec()
self.clock.cancel_call_later(self._member_typing_timer[member]) self._member_typing_until[member] = now + timeout
def _cb(): self.wheel_timer.insert(
logger.debug( now=now,
"%s has timed out in %s", target_user.to_string(), room_id obj=member,
) then=now + timeout,
self._stopped_typing(member)
self._member_typing_until[member] = until
self._member_typing_timer[member] = self.clock.call_later(
timeout / 1000.0, _cb
) )
if was_present: if was_present:
@ -109,8 +141,7 @@ class TypingHandler(object):
defer.returnValue(None) defer.returnValue(None)
yield self._push_update( yield self._push_update(
room_id=room_id, member=member,
user_id=target_user_id,
typing=True, typing=True,
) )
@ -133,10 +164,6 @@ class TypingHandler(object):
member = RoomMember(room_id=room_id, user_id=target_user_id) member = RoomMember(room_id=room_id, user_id=target_user_id)
if member in self._member_typing_timer:
self.clock.cancel_call_later(self._member_typing_timer[member])
del self._member_typing_timer[member]
yield self._stopped_typing(member) yield self._stopped_typing(member)
@defer.inlineCallbacks @defer.inlineCallbacks
@ -148,50 +175,52 @@ class TypingHandler(object):
@defer.inlineCallbacks @defer.inlineCallbacks
def _stopped_typing(self, member): def _stopped_typing(self, member):
if member not in self._member_typing_until: if member.user_id not in self._room_typing.get(member.room_id, set()):
# No point # No point
defer.returnValue(None) defer.returnValue(None)
self._member_typing_until.pop(member, None)
self._member_last_federation_poke.pop(member, None)
yield self._push_update( yield self._push_update(
room_id=member.room_id, member=member,
user_id=member.user_id,
typing=False, typing=False,
) )
del self._member_typing_until[member]
if member in self._member_typing_timer:
# Don't cancel it - either it already expired, or the real
# stopped_typing() will cancel it
del self._member_typing_timer[member]
@defer.inlineCallbacks @defer.inlineCallbacks
def _push_update(self, room_id, user_id, typing): def _push_update(self, member, typing):
users = yield self.state.get_current_user_in_room(room_id) if self.hs.is_mine_id(member.user_id):
domains = set(get_domain_from_id(u) for u in users) # Only send updates for changes to our own users.
yield self._push_remote(member, typing)
deferreds = [] self._push_update_local(
for domain in domains: member=member,
if domain == self.server_name:
preserve_fn(self._push_update_local)(
room_id=room_id,
user_id=user_id,
typing=typing typing=typing
) )
else:
deferreds.append(preserve_fn(self.federation.send_edu)( @defer.inlineCallbacks
def _push_remote(self, member, typing):
users = yield self.state.get_current_user_in_room(member.room_id)
self._member_last_federation_poke[member] = self.clock.time_msec()
now = self.clock.time_msec()
self.wheel_timer.insert(
now=now,
obj=member,
then=now + FEDERATION_PING_INTERVAL,
)
for domain in set(get_domain_from_id(u) for u in users):
if domain != self.server_name:
self.federation.send_edu(
destination=domain, destination=domain,
edu_type="m.typing", edu_type="m.typing",
content={ content={
"room_id": room_id, "room_id": member.room_id,
"user_id": user_id, "user_id": member.user_id,
"typing": typing, "typing": typing,
}, },
key=(room_id, user_id), key=member,
))
yield preserve_context_over_deferred(
defer.DeferredList(deferreds, consumeErrors=True)
) )
@defer.inlineCallbacks @defer.inlineCallbacks
@ -199,6 +228,8 @@ class TypingHandler(object):
room_id = content["room_id"] room_id = content["room_id"]
user_id = content["user_id"] user_id = content["user_id"]
member = RoomMember(user_id=user_id, room_id=room_id)
# Check that the string is a valid user id # Check that the string is a valid user id
user = UserID.from_string(user_id) user = UserID.from_string(user_id)
@ -213,25 +244,31 @@ class TypingHandler(object):
domains = set(get_domain_from_id(u) for u in users) domains = set(get_domain_from_id(u) for u in users)
if self.server_name in domains: if self.server_name in domains:
logger.info("Got typing update from %s: %r", user_id, content)
now = self.clock.time_msec()
self._member_typing_until[member] = now + FEDERATION_TIMEOUT
self.wheel_timer.insert(
now=now,
obj=member,
then=now + FEDERATION_TIMEOUT,
)
self._push_update_local( self._push_update_local(
room_id=room_id, member=member,
user_id=user_id,
typing=content["typing"] typing=content["typing"]
) )
def _push_update_local(self, room_id, user_id, typing): def _push_update_local(self, member, typing):
room_set = self._room_typing.setdefault(room_id, set()) room_set = self._room_typing.setdefault(member.room_id, set())
if typing: if typing:
room_set.add(user_id) room_set.add(member.user_id)
else: else:
room_set.discard(user_id) room_set.discard(member.user_id)
self._latest_room_serial += 1 self._latest_room_serial += 1
self._room_serials[room_id] = self._latest_room_serial self._room_serials[member.room_id] = self._latest_room_serial
with PreserveLoggingContext():
self.notifier.on_new_event( self.notifier.on_new_event(
"typing_key", self._latest_room_serial, rooms=[room_id] "typing_key", self._latest_room_serial, rooms=[member.room_id]
) )
def get_all_typing_updates(self, last_id, current_id): def get_all_typing_updates(self, last_id, current_id):

View file

@ -705,12 +705,15 @@ class RoomTypingRestServlet(ClientV1RestServlet):
yield self.presence_handler.bump_presence_active_time(requester.user) yield self.presence_handler.bump_presence_active_time(requester.user)
# Limit timeout to stop people from setting silly typing timeouts.
timeout = min(content.get("timeout", 30000), 120000)
if content["typing"]: if content["typing"]:
yield self.typing_handler.started_typing( yield self.typing_handler.started_typing(
target_user=target_user, target_user=target_user,
auth_user=requester.user, auth_user=requester.user,
room_id=room_id, room_id=room_id,
timeout=content.get("timeout", 30000), timeout=timeout,
) )
else: else:
yield self.typing_handler.stopped_typing( yield self.typing_handler.stopped_typing(

View file

@ -267,10 +267,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
from synapse.handlers.typing import RoomMember from synapse.handlers.typing import RoomMember
member = RoomMember(self.room_id, self.u_apple.to_string()) member = RoomMember(self.room_id, self.u_apple.to_string())
self.handler._member_typing_until[member] = 1002000 self.handler._member_typing_until[member] = 1002000
self.handler._member_typing_timer[member] = ( self.handler._room_typing[self.room_id] = set([self.u_apple.to_string()])
self.clock.call_later(1002, lambda: 0)
)
self.handler._room_typing[self.room_id] = set((self.u_apple.to_string(),))
self.assertEquals(self.event_source.get_current_key(), 0) self.assertEquals(self.event_source.get_current_key(), 0)
@ -330,7 +327,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
}, },
}]) }])
self.clock.advance_time(11) self.clock.advance_time(16)
self.on_new_event.assert_has_calls([ self.on_new_event.assert_has_calls([
call('typing_key', 2, rooms=[self.room_id]), call('typing_key', 2, rooms=[self.room_id]),

View file

@ -105,9 +105,6 @@ class RoomTypingTestCase(RestTestCase):
# Need another user to make notifications actually work # Need another user to make notifications actually work
yield self.join(self.room_id, user="@jim:red") yield self.join(self.room_id, user="@jim:red")
def tearDown(self):
self.hs.get_typing_handler().tearDown()
@defer.inlineCallbacks @defer.inlineCallbacks
def test_set_typing(self): def test_set_typing(self):
(code, _) = yield self.mock_resource.trigger( (code, _) = yield self.mock_resource.trigger(
@ -147,7 +144,7 @@ class RoomTypingTestCase(RestTestCase):
self.assertEquals(self.event_source.get_current_key(), 1) self.assertEquals(self.event_source.get_current_key(), 1)
self.clock.advance_time(31) self.clock.advance_time(36)
self.assertEquals(self.event_source.get_current_key(), 2) self.assertEquals(self.event_source.get_current_key(), 2)

View file

@ -220,6 +220,7 @@ class MockClock(object):
# list of lists of [absolute_time, callback, expired] in no particular # list of lists of [absolute_time, callback, expired] in no particular
# order # order
self.timers = [] self.timers = []
self.loopers = []
def time(self): def time(self):
return self.now return self.now
@ -240,7 +241,7 @@ class MockClock(object):
return t return t
def looping_call(self, function, interval): def looping_call(self, function, interval):
pass self.loopers.append([function, interval / 1000., self.now])
def cancel_call_later(self, timer, ignore_errs=False): def cancel_call_later(self, timer, ignore_errs=False):
if timer[2]: if timer[2]:
@ -269,6 +270,12 @@ class MockClock(object):
else: else:
self.timers.append(t) self.timers.append(t)
for looped in self.loopers:
func, interval, last = looped
if last + interval < self.now:
func()
looped[2] = self.now
def advance_time_msec(self, ms): def advance_time_msec(self, ms):
self.advance_time(ms / 1000.) self.advance_time(ms / 1000.)