diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py
index d097e65ea7..b3f65e8237 100644
--- a/synapse/federation/sender/per_destination_queue.py
+++ b/synapse/federation/sender/per_destination_queue.py
@@ -156,7 +156,6 @@ class PerDestinationQueue:
# Each receipt can only have a single receipt per
# (room ID, receipt type, user ID, thread ID) tuple.
self._pending_receipt_edus: List[Dict[str, Dict[str, Dict[str, dict]]]] = []
- self._rrs_pending_flush = False
# stream_id of last successfully sent to-device message.
# NB: may be a long or an int.
@@ -258,15 +257,7 @@ class PerDestinationQueue:
}
)
- def flush_read_receipts_for_room(self, room_id: str) -> None:
- # If there are any pending receipts for this room then force-flush them
- # in a new transaction.
- for edu in self._pending_receipt_edus:
- if room_id in edu:
- self._rrs_pending_flush = True
- self.attempt_new_transaction()
- # No use in checking remaining EDUs if the room was found.
- break
+ self.mark_new_data()
def send_keyed_edu(self, edu: Edu, key: Hashable) -> None:
self._pending_edus_keyed[(edu.edu_type, key)] = edu
@@ -603,12 +594,9 @@ class PerDestinationQueue:
self._destination, last_successful_stream_ordering
)
- def _get_receipt_edus(self, force_flush: bool, limit: int) -> Iterable[Edu]:
+ def _get_receipt_edus(self, limit: int) -> Iterable[Edu]:
if not self._pending_receipt_edus:
return
- if not force_flush and not self._rrs_pending_flush:
- # not yet time for this lot
- return
# Send at most limit EDUs for receipts.
for content in self._pending_receipt_edus[:limit]:
@@ -747,7 +735,7 @@ class _TransactionQueueManager:
)
# Add read receipt EDUs.
- pending_edus.extend(self.queue._get_receipt_edus(force_flush=False, limit=5))
+ pending_edus.extend(self.queue._get_receipt_edus(limit=5))
edu_limit = MAX_EDUS_PER_TRANSACTION - len(pending_edus)
# Next, prioritize to-device messages so that existing encryption channels
@@ -795,13 +783,6 @@ class _TransactionQueueManager:
if not self._pdus and not pending_edus:
return [], []
- # if we've decided to send a transaction anyway, and we have room, we
- # may as well send any pending RRs
- if edu_limit:
- pending_edus.extend(
- self.queue._get_receipt_edus(force_flush=True, limit=edu_limit)
- )
-
if self._pdus:
self._last_stream_ordering = self._pdus[
-1
|
