diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py
index df99c53041..29d341835a 100644
--- a/synapse/http/matrixfederationclient.py
+++ b/synapse/http/matrixfederationclient.py
@@ -358,6 +358,9 @@ class MatrixFederationHttpClient:
self._sleeper = AwakenableSleeper(self.reactor)
+ self._notifier = hs.get_notifier()
+ self._replication_client = hs.get_replication_command_handler()
+
def wake_destination(self, destination: str) -> None:
"""Called when the remote server may have come back online."""
@@ -484,6 +487,8 @@ class MatrixFederationHttpClient:
self._store,
backoff_on_404=backoff_on_404,
ignore_backoff=ignore_backoff,
+ notifier=self._notifier,
+ replication_client=self._replication_client,
)
method_bytes = request.method.encode("ascii")
diff --git a/synapse/util/retryutils.py b/synapse/util/retryutils.py
index 352e110859..a93d7936b6 100644
--- a/synapse/util/retryutils.py
+++ b/synapse/util/retryutils.py
@@ -14,13 +14,17 @@
import logging
import random
from types import TracebackType
-from typing import Any, Optional, Type
+from typing import TYPE_CHECKING, Any, Optional, Type
import synapse.logging.context
from synapse.api.errors import CodeMessageException
from synapse.storage import DataStore
from synapse.util import Clock
+if TYPE_CHECKING:
+ from synapse.notifier import Notifier
+ from synapse.replication.tcp.handler import ReplicationCommandHandler
+
logger = logging.getLogger(__name__)
# the initial backoff, after the first transaction fails
@@ -131,6 +135,8 @@ class RetryDestinationLimiter:
retry_interval: int,
backoff_on_404: bool = False,
backoff_on_failure: bool = True,
+ notifier: Optional["Notifier"] = None,
+ replication_client: Optional["ReplicationCommandHandler"] = None,
):
"""Marks the destination as "down" if an exception is thrown in the
context, except for CodeMessageException with code < 500.
@@ -160,6 +166,9 @@ class RetryDestinationLimiter:
self.backoff_on_404 = backoff_on_404
self.backoff_on_failure = backoff_on_failure
+ self.notifier = notifier
+ self.replication_client = replication_client
+
def __enter__(self) -> None:
pass
@@ -239,6 +248,19 @@ class RetryDestinationLimiter:
retry_last_ts,
self.retry_interval,
)
+
+ if self.notifier:
+ # Inform the relevant places that the remote server is back up.
+ self.notifier.notify_remote_server_up(self.destination)
+
+ if self.replication_client:
+ # If we're on a worker we try and inform master about this. The
+ # replication client doesn't hook into the notifier to avoid
+ # infinite loops where we send a `REMOTE_SERVER_UP` command to
+ # master, which then echoes it back to us which in turn pokes
+ # the notifier.
+ self.replication_client.send_remote_server_up(self.destination)
+
except Exception:
logger.exception("Failed to store destination_retry_timings")
|
