Bound ephemeral events by key

[Fizzadar]

Spend a couple of hours working through this with @bradtgmurray:

When the appservice-pusher handles a new read receipt, it grabs up to 100 events newer than the current position and then advances to the new current position. However, it doesn't advance past the 100 events it just grabbed. This means that if there are newer events to handle in the database but we're still in the process handling older events, we'll grab the same events over and over again, pushing them to the appservice scheduler queue.

This can happen if the appservice is slow to respond (but does not error), and thus the in memory queue of events-to-send grows and grows full of duplicated events.

Signed off by Nick @ Beeper

Pull Request Checklist

• Pull request is based on the develop branch
• Pull request includes a changelog file. The entry should:
  • Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from EventStore to EventWorkerStore.".
  • Use markdown where necessary, mostly for code blocks.
  • End with either a period (.) or an exclamation mark (!).
  • Start with a capital letter.
  • Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry.
• Pull request includes a sign off
• Code style is correct
  (run the linters)

[bradtgmurray]

The unbounded memory growth came from this array here growing faster than it could be drained:

synapse/synapse/appservice/scheduler.py

Line 145 in c7d5eda

self.queuer.queued_ephemeral.setdefault(appservice.id, []).extend(ephemeral)

We can send about 700~ EDUs to our app service per second (7 transactions of 100 EDUs each) and we receive about 25~ read receipts per second, and each one of these read receipts caused the 100 most recent read receipts to be added to the queue. This means it was growing at a rate of 2500 per second, which outstrips our ability to drain it at 700 per second, resulted in this list growing infinitely.

[bradtgmurray]

Also note we have not tested this change, so consider this a theoretical fix. Would love some feedback if this fix is in the right direction.

[anoadragon453]

Related: #10836

Thanks both for tracking this down. This particular code path is completely untested, so this error condition is not too surprising... I'm in the middle of writing a unit test now to verify the issue and the proposed solution.

This solution looks like it will work, and we should land this sooner rather than later. I do think that this will eliminate the duplication when sending queued up receipts to AS's. It will result in many small transactions though, whereas it would be ideal to instead send batches of 100 as much as possible.

That would simply mean switching the call:

synapse/synapse/handlers/appservice.py

Lines 340 to 343 in 800ba87

	# Persist the latest handled stream token for this appservice
	await self.store.set_appservice_stream_type_pos(
	service, "read_receipt", new_token
	)

from storing new_token to storing the maximum stream_token of the events pulled from the database. Then any further calls with a token less than that will be ignored.
This should result in less transactions overall.

But as mentioned, both solutions should prevent the catastrophic case.

[bradtgmurray]

Awesome, thanks for taking a look over this Andrew.

from storing new_token to storing the maximum stream_token of the events pulled from the database

The downside here is that we'd drop read receipts in the case where were more than 100 events behind, as #8744 flipped this to taking the most recent 100 events as opposed to the next 100 events.

It's awkward, because either we just take up to the current new_token and end up with many small transactions, or we take the max stream_token from the events and drop in cases where there's more than 100 to send.

However, maybe this isn't a problem. Events are pulled in small batches and added to the self.queuer.queued_ephemeral list, and then that list is pulled in batches as we get ready to send the next transaction. I think storing new_token will still result in maximally large transactions if the pusher can't keep up in that case.

[anoadragon453]

Hrm, re-reading #8744 again, I'm not sure if the problem is exactly as described. We pull read read receipts from the receipts_linearized table, yet that table only ever contains the latest read receipt for user in a room. So hooking up a new AS will indeed send a lot of read receipts, but only num. user x num. room receipts in total - not every read receipt ever sent.

It probably doesn't make sense to send all of the latest read receipts to the AS when first hooking it up (in fact, should we send any receipts at all from before the AS was connected?) but sending all of the latest read receipts since the last time the AS was online may make sense to avoid edge cases.

However, maybe this isn't a problem. Events are pulled in small batches and added to the self.queuer.queued_ephemeral list, and then that list is pulled in batches as we get ready to send the next transaction. I think storing new_token will still result in maximally large transactions if the pusher can't keep up in that case.

Good point, there is a buffer, so transaction sizes will already go out as fast as possible. If there's a lot of traffic, transactions will be >1 in size.

[bradtgmurray]

Hrm, re-reading #8744 again, I'm not sure if the problem is exactly as described. We pull read read receipts from the receipts_linearized table, yet that table only ever contains the latest read receipt for user in a room. So hooking up a new AS will indeed send a lot of read receipts, but only num. user x num. room receipts in total - not every read receipt ever sent.

Interesting. So our worker was probably much further behind.

When we were having this issue, we were running the following query:

SELECT COUNT(*)
FROM receipts_linearized rl
         JOIN application_services_state ass ON 1 = 1
WHERE rl.stream_id > ass.read_receipt_stream_id

This was returning a count of 3000 or so. Since we were only getting the latest 100, and this was much less than 3000, we were getting the amplifcation sadness.

[bradtgmurray]

Oh, you were discussing the fix in #8744, in which I agree that doing something around "since the AS was connected", but that doesn't handle long-lived AS connections that recently have emphemeral read receipts enabled. You kind of need MAX(since AS connected, since emphemeral read receipts enabled) which is funky to implement, hence the hack.

Another aside, SELECT COUNT(*) FROM receipts_linearized gives 1885415 on beeper.com

[anoadragon453]

Right, that's quite a lot of read receipts. Out of interest I checked the count on matrix.org and it's over 9 times that! Perhaps we'll leave the hack in place for now.

[anoadragon453]

I've added a commit to your branch with a unit test which verify that the proposed fix works 🎉 In the test, we have 300 pending read receipts, and with the fix 300 are eventually sent out the other side to the fake AS.

For reference, on the develop branch we try to send 25050 read receipts out(!).

[bradtgmurray]

Thank you all!