From daa10e3e66dadef3b860c31baaeded1da92430be Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Mon, 4 Mar 2019 18:27:32 +0000
Subject: [PATCH 1/3] Remove unused `wait_for_replication` method

I guess this was used once? It's not now, anyway.
---
 synapse/notifier.py | 50 ---------------------------------------------
 1 file changed, 50 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index de02b1017e..2505202e98 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -178,8 +178,6 @@ class Notifier(object):
             self.remove_expired_streams, self.UNUSED_STREAM_EXPIRY_MS
         )
 
-        self.replication_deferred = ObservableDeferred(defer.Deferred())
-
         # This is not a very cheap test to perform, but it's only executed
         # when rendering the metrics page, which is likely once per minute at
         # most when scraping it.
@@ -518,10 +516,6 @@ class Notifier(object):
     def notify_replication(self):
         """Notify the any replication listeners that there's a new event"""
         with PreserveLoggingContext():
-            deferred = self.replication_deferred
-            self.replication_deferred = ObservableDeferred(defer.Deferred())
-            deferred.callback(None)
-
             # the callbacks may well outlast the current request, so we run
             # them in the sentinel logcontext.
             #
@@ -530,47 +524,3 @@ class Notifier(object):
             # accordingly, but that requires more changes)
             for cb in self.replication_callbacks:
                 cb()
-
-    @defer.inlineCallbacks
-    def wait_for_replication(self, callback, timeout):
-        """Wait for an event to happen.
-
-        Args:
-            callback: Gets called whenever an event happens. If this returns a
-                truthy value then ``wait_for_replication`` returns, otherwise
-                it waits for another event.
-            timeout: How many milliseconds to wait for callback return a truthy
-                value.
-
-        Returns:
-            A deferred that resolves with the value returned by the callback.
-        """
-        listener = _NotificationListener(None)
-
-        end_time = self.clock.time_msec() + timeout
-
-        while True:
-            listener.deferred = self.replication_deferred.observe()
-            result = yield callback()
-            if result:
-                break
-
-            now = self.clock.time_msec()
-            if end_time <= now:
-                break
-
-            listener.deferred = timeout_deferred(
-                listener.deferred,
-                timeout=(end_time - now) / 1000.,
-                reactor=self.hs.get_reactor(),
-            )
-
-            try:
-                with PreserveLoggingContext():
-                    yield listener.deferred
-            except defer.TimeoutError:
-                break
-            except defer.CancelledError:
-                break
-
-        defer.returnValue(result)

From c7325776a7a02354657c519eb5169c1f2f3e0872 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Mon, 4 Mar 2019 18:31:18 +0000
Subject: [PATCH 2/3] Remove redundant PreserveLoggingContext

Both (!) things that register as replication listeners do the right thing wrt
logcontexts, so this is redundant.
---
 synapse/notifier.py | 15 +++++----------
 1 file changed, 5 insertions(+), 10 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 2505202e98..ff589660da 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -203,7 +203,9 @@ class Notifier(object):
 
     def add_replication_callback(self, cb):
         """Add a callback that will be called when some new data is available.
-        Callback is not given any arguments.
+        Callback is not given any arguments. It should *not* return a Deferred - if
+        it needs to do any asynchronous work, a background thread should be started and
+        wrapped with run_as_background_process.
         """
         self.replication_callbacks.append(cb)
 
@@ -515,12 +517,5 @@ class Notifier(object):
 
     def notify_replication(self):
         """Notify the any replication listeners that there's a new event"""
-        with PreserveLoggingContext():
-            # the callbacks may well outlast the current request, so we run
-            # them in the sentinel logcontext.
-            #
-            # (ideally it would be up to the callbacks to know if they were
-            # starting off background processes and drop the logcontext
-            # accordingly, but that requires more changes)
-            for cb in self.replication_callbacks:
-                cb()
+        for cb in self.replication_callbacks:
+            cb()

From eaa9f4360336ea5d62fc200aae14130b5fa55c7b Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Mon, 4 Mar 2019 19:01:19 +0000
Subject: [PATCH 3/3] changelog

---
 changelog.d/4799.misc | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changelog.d/4799.misc

diff --git a/changelog.d/4799.misc b/changelog.d/4799.misc
new file mode 100644
index 0000000000..5ab11a5c0b
--- /dev/null
+++ b/changelog.d/4799.misc
@@ -0,0 +1 @@
+Clean up some replication code.