Fix up logic for delaying sending read receipts over federation.

[erikjohnston]

For context of why we delay read receipts, see matrix-org/synapse#4730.

Element Web often sends read receipts in quick succession, if it reloads the timeline it'll send one for the last message in the old timeline and again for the last message in the new timeline. This caused remote users to see a read receipt for older messages come through quickly, but then the second read receipt taking a while to arrive for the most recent message.

There are two things going on in this PR:

1. There was a mismatch between seconds and milliseconds, and so we ended up delaying for far longer than intended.
2. Changing the logic to reuse the DestinationWakeupQueue (used for presence)

The changes in logic are:

• Treat the first receipt and subsequent receipts in a room in the same way
• Whitelist certain classes of receipts to never delay being sent, i.e. receipts in small rooms, receipts for events that were sent within the last 60s, and sending receipts to the event sender's server.
• The maximum delay a receipt can have before being sent to a server is 30s, and we'll send out receipts to remotes at least at 50Hz (by default)

The upshot is that this should make receipts feel more snappy over federation.

This new logic should send roughly between 10%–20% of transactions immediately on matrix.org.

SQL query and output

Looking at the last 100,000 local receipts, we look at the number of servers we need to send those receipts to and bucket them into different classes. First, if they are in a small room (i.e. less than 5 domains), and then by age of event.

     delay     |  sum   
---------------+--------
 00 small room |   2688
 01 <1s        |   1476
 05 <5s        |   4967
 10 <10s       |   1643
 60 <60s       |   9052
 old           | 129735

SELECT
  delay,
  SUM (to_send) 
FROM
  (
    SELECT
      CASE
        WHEN
          domains - 1 < 5 
        THEN
          '00 small room' 
        WHEN
          read_ts - received_ts < '1s' 
        THEN
          '01 <1s' 
        WHEN
          read_ts - received_ts < '5s' 
        THEN
          '05 <5s' 
        WHEN
          read_ts - received_ts < '10s' 
        THEN
          '10 <10s' 
        WHEN
          read_ts - received_ts < '60s' 
        THEN
          '60 <60s' 
        ELSE
          'old' 
      END
      AS delay, domains - 1 AS to_send 
    FROM
      (
        SELECT
          r.room_id,
          event_id,
          to_timestamp((data::json ->> 'ts')::BIGINT / 1000) AS read_ts,
          to_timestamp(received_ts / 1000) AS received_ts 
        FROM
          receipts_linearized AS r 
          INNER JOIN
            events USING (event_id) 
        WHERE
          r.user_id LIKE '%:matrix.org' 
        ORDER BY
          stream_id DESC LIMIT 10000
      )
      AS r2 
      INNER JOIN
        lateral (
        SELECT
          COUNT(DISTINCT SUBSTRING(state_key FROM '@[^:]*:(.*)$')) AS domains 
        FROM
          current_state_events 
        WHERE
          type = 'm.room.member' 
          AND room_id = r2.room_id 
          AND membership = 'join') AS c 
          ON TRUE 
        WHERE
          domains > 1
  )
  AS a 
GROUP BY
  delay 
ORDER BY
  delay;

[reivilibre]

seems to make sense

[reivilibre]

I know it's a silly case, but maybe

Suggested change

	if self.clock.time_msec() - metadata.received_ts < 60_000:
	if 0 <= self.clock.time_msec() - metadata.received_ts < 60_000:

to guard against the case where someone's clock was wrong (in the future) for a while, or is currently in the past, from causing them to treat all events as recent?

Or perhaps it should have some tolerance for workers being out of sync, so perhaps -60_000 <=???...

[reivilibre]

Suggested change

	transaction = mock_send_transaction.call_args_list[0][0][0]
	transaction = mock_send_transaction.call_args_list[0].args[0]

I think this is correct and clearer? (ditto for other assertions)

[reivilibre]

(also for these

Suggested change

	self._assert_edu_in_call(mock_send_transaction.call_args[0][1])
	self._assert_edu_in_call(mock_send_transaction.call_args.args[1])

)