diff --git a/synapse/replication/slave/storage/client_ips.py b/synapse/replication/slave/storage/client_ips.py
index 60641f1a49..5b8521c770 100644
--- a/synapse/replication/slave/storage/client_ips.py
+++ b/synapse/replication/slave/storage/client_ips.py
@@ -43,6 +43,8 @@ class SlavedClientIpStore(BaseSlavedStore):
if last_seen is not None and (now - last_seen) < LAST_SEEN_GRANULARITY:
return
+ self.client_ip_last_seen.prefill(key, now)
+
self.hs.get_tcp_replication().send_user_ip(
user_id, access_token, ip, user_agent, device_id, now
)
diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py
index 0b3fe6cbf5..75c9e8355f 100644
--- a/synapse/replication/tcp/protocol.py
+++ b/synapse/replication/tcp/protocol.py
@@ -52,6 +52,7 @@ indicate which side is sending, these are *not* included on the wire::
import fcntl
import logging
import struct
+import traceback
from collections import defaultdict
from six import iteritems, iterkeys
@@ -241,6 +242,7 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
def send_error(self, error_string, *args):
"""Send an error to remote and close the connection.
"""
+ logger.error("[%s] Sending error: %s", self.id(), error_string % args)
self.send_command(ErrorCommand(error_string % args))
self.close()
@@ -323,6 +325,8 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
we or the remote has closed the connection)
"""
logger.info("[%s] Stop producing", self.id())
+ # debug for #4733
+ logger.info("Traceback: %s", "".join(traceback.format_stack()))
self.on_connection_closed()
def connectionLost(self, reason):
diff --git a/synapse/replication/tcp/streams.py b/synapse/replication/tcp/streams.py
index c1e626be3f..d49973634e 100644
--- a/synapse/replication/tcp/streams.py
+++ b/synapse/replication/tcp/streams.py
@@ -32,7 +32,7 @@ from twisted.internet import defer
logger = logging.getLogger(__name__)
-MAX_EVENTS_BEHIND = 10000
+MAX_EVENTS_BEHIND = 500000
EventStreamRow = namedtuple("EventStreamRow", (
@@ -265,8 +265,8 @@ class PresenceStream(Stream):
store = hs.get_datastore()
presence_handler = hs.get_presence_handler()
- self.current_token = store.get_current_presence_token
- self.update_function = presence_handler.get_all_presence_updates
+ self.current_token = lambda: 0
+ self.update_function = lambda _a, _b: []
super(PresenceStream, self).__init__(hs)
|
