mirror of
https://mau.dev/maunium/synapse.git
synced 2024-11-12 04:52:26 +01:00
Store if we fail to fetch an event from a destination
This commit is contained in:
parent
d45489474d
commit
f91df1f761
1 changed files with 36 additions and 1 deletions
|
@ -51,10 +51,34 @@ sent_edus_counter = metrics.register_counter("sent_edus")
|
||||||
sent_queries_counter = metrics.register_counter("sent_queries", labels=["type"])
|
sent_queries_counter = metrics.register_counter("sent_queries", labels=["type"])
|
||||||
|
|
||||||
|
|
||||||
|
PDU_RETRY_TIME_MS = 1 * 60 * 1000
|
||||||
|
|
||||||
|
|
||||||
class FederationClient(FederationBase):
|
class FederationClient(FederationBase):
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
super(FederationClient, self).__init__(hs)
|
super(FederationClient, self).__init__(hs)
|
||||||
|
|
||||||
|
self.pdu_destination_tried = {}
|
||||||
|
self._clock.looping_call(
|
||||||
|
self._clear_tried_cache, 60 * 1000,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _clear_tried_cache(self):
|
||||||
|
"""Clear pdu_destination_tried cache"""
|
||||||
|
now = self._clock.time_msec()
|
||||||
|
|
||||||
|
old_dict = self.pdu_destination_tried
|
||||||
|
self.pdu_destination_tried = {}
|
||||||
|
|
||||||
|
for event_id, destination_dict in old_dict.items():
|
||||||
|
destination_dict = {
|
||||||
|
dest: time
|
||||||
|
for dest, time in destination_dict.items()
|
||||||
|
if time + PDU_RETRY_TIME_MS > now
|
||||||
|
}
|
||||||
|
if destination_dict:
|
||||||
|
self.pdu_destination_tried[event_id] = destination_dict
|
||||||
|
|
||||||
def start_get_pdu_cache(self):
|
def start_get_pdu_cache(self):
|
||||||
self._get_pdu_cache = ExpiringCache(
|
self._get_pdu_cache = ExpiringCache(
|
||||||
cache_name="get_pdu_cache",
|
cache_name="get_pdu_cache",
|
||||||
|
@ -240,8 +264,15 @@ class FederationClient(FederationBase):
|
||||||
if ev:
|
if ev:
|
||||||
defer.returnValue(ev)
|
defer.returnValue(ev)
|
||||||
|
|
||||||
|
pdu_attempts = self.pdu_destination_tried.setdefault(event_id, {})
|
||||||
|
|
||||||
pdu = None
|
pdu = None
|
||||||
for destination in destinations:
|
for destination in destinations:
|
||||||
|
now = self._clock.time_msec()
|
||||||
|
last_attempt = pdu_attempts.get(destination, 0)
|
||||||
|
if last_attempt + PDU_RETRY_TIME_MS > now:
|
||||||
|
continue
|
||||||
|
|
||||||
try:
|
try:
|
||||||
limiter = yield get_retry_limiter(
|
limiter = yield get_retry_limiter(
|
||||||
destination,
|
destination,
|
||||||
|
@ -276,9 +307,11 @@ class FederationClient(FederationBase):
|
||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
except CodeMessageException as e:
|
except CodeMessageException as e:
|
||||||
if 400 <= e.code < 500:
|
if 400 <= e.code < 500 and e.code != 404:
|
||||||
raise
|
raise
|
||||||
|
|
||||||
|
pdu_attempts[destination] = now
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
"Failed to get PDU %s from %s because %s",
|
"Failed to get PDU %s from %s because %s",
|
||||||
event_id, destination, e,
|
event_id, destination, e,
|
||||||
|
@ -288,6 +321,8 @@ class FederationClient(FederationBase):
|
||||||
logger.info(e.message)
|
logger.info(e.message)
|
||||||
continue
|
continue
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
pdu_attempts[destination] = now
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
"Failed to get PDU %s from %s because %s",
|
"Failed to get PDU %s from %s because %s",
|
||||||
event_id, destination, e,
|
event_id, destination, e,
|
||||||
|
|
Loading…
Reference in a new issue