memory (disco): track pending queries to re-use them:

When many queries are run in parallel (e.g. during initial sync on startup), the same
entity may be disco queries several times in parallel. To avoid network request
duplication, we now track pending queries, and re-use them in this case.
This commit is contained in:
Goffi 2026-05-03 10:51:41 +02:00
commit cc4d9b9bd8

View file

@ -113,6 +113,9 @@ class Discovery:
# - (disco.DiscoInfo, float): successful query, expires at timestamp
# - (None, float): failed query, retry after DISCO_RETRY_DELAY
self._jid_disco_cache: dict[jid.JID, tuple[disco.DiscoInfo | None, float]] = {}
# Track in-flight disco queries to avoid duplicate requests
# Maps (jid, node) tuple to query's Deferred
self._pending_queries: dict[tuple[jid.JID, str], defer.Deferred] = {}
def load(self):
"""Load persistent hashes"""
@ -261,6 +264,17 @@ class Discovery:
# No cache is used, we will do a query to return disco infos.
# The query result will be cached in JID cache.
query_key = (jid_, node)
if query_key in self._pending_queries:
log.debug(f"Reusing pending disco query for {jid_.full()}")
return self._pending_queries[query_key]
def cleanup_query(result):
"""Remove the pending query on success or failure."""
self._pending_queries.pop(query_key, None)
return result
def infos_cb(disco_infos):
for ext_form in disco_infos.extensions.values():
# tx_xmpp doesn't call typeCheck on reception, so we do it here
@ -288,7 +302,9 @@ class Discovery:
d = client.disco.requestInfo(jid_, nodeIdentifier=node)
d.addCallback(infos_cb)
d.addBoth(cleanup_query)
d.addErrback(infos_eb)
self._pending_queries[query_key] = d
return d
else:
disco_infos = self.hashes[cap_hash]