Fix historical messages backfilling in random order on remote homeservers (MSC2716)

[MadLittleMods]

Started as an investigation of #11091 but then realized the backfilled messages are really weird on remote homeservers.

---

Fix historical messages backfilling in random order on remote homeservers (MSC2716).

Fix #11091
Fix #10764

1. Made the /backfill response return messages in (depth, stream_ordering) order (previously only sorted by depth)
   • Technically, it shouldn't really matter how /backfill returns things but I'm just trying to make the stream_ordering a little more consistent from the origin to the remote homeservers in order to get the order of messages from /messages consistent (sorted by (topological_ordering, stream_ordering)).
   • Even now that we return backfilled messages in order, it still doesn't guarantee the same stream_ordering (and more importantly the /messages order) on the other server. For example, if a room has a bunch of history imported and someone visits a permalink to a historical message back in time, their homeserver will skip over the historical messages in between and insert the permalink as the next message in the stream_order and totally throw off the sort.
     • This will be even more the case when we add the MSC3030 jump to date API endpoint so the static archives can navigate and jump to a certain date.
     • We're solving this in the future by switching to online topological ordering and chunking which by its nature will apply retroactively to fix any inconsistencies introduced by people permalinking
2. As we're navigating prev_events to return in /backfill, we order by depth first (newest -> oldest) and now also tie-break based on the stream_ordering (newest -> oldest). This is technically important because MSC2716 inserts a bunch of historical messages at the same depth so it's best to be prescriptive about which ones we should process first. In reality, I think the code already looped over the historical messages as expected because the database is already in order.
3. Process the events returned by /backfill from oldest -> newest (reverse-chronological) which causes a lot less missing prev_event fetching. Doing all of the prev_event fetching is not only inefficient because that "missing" event is already available later in same the backfill chunk but also causes bad stream_ordering values because fetched missing prev_events aren't considered backfilled and will increment the stream_ordering. We expect all of the MSC2716 historical events to be backfilled and decrement stream_ordering to appear in the correct order.
   • Tracking this separately in MSC2716 leftover todo tasks #10737
4. Making the historical state chain and historical event chain float on their own by having no prev_events instead of a fake prev_event which caused backfill to get clogged with an unresolvable event. Fixes Backfill gets clogged up with fake prev_events when using the MSC2716 /batch_send endpoint #11091 and Backfill should gracefully handle the case where no one has the complete state for a given event #10764
5. We no longer find connected insertion events by finding a potential prev_event connection to the current event we're iterating over. We now solely rely on marker events which when processed, add the insertion event as an extremity and the federating homeserver can ask about it when time calls.
   • Related discussion, Fix historical messages backfilling in random order on remote homeservers (MSC2716) #11114 (comment)

Before	After
Mermaid live editor playground link	Mermaid live editor playground link

The real solution around event ordering is related to these issues where we don't rely on depth or stream_ordering and it's just up to the shape of the DAG itself (chunking and online topological ordering):

• https://github.com/matrix-org/matrix-doc/issues/3263
• https://github.com/matrix-org/matrix-doc/issues/1229
• Project: Chunking #3785

Why aren't we sorting topologically when receiving backfill events?

See #11114 (comment)

Dev notes

Tested with matrix-org/complement#214. I am able to import 11 batches of 100 historical messages and fetch them all on the other server via /messages?dir=b&limit=100 pagination 🎉

COMPLEMENT_ALWAYS_PRINT_SERVER_LOGS=1 COMPLEMENT_DIR=../complement ./scripts-dev/complement.sh TestImportHistoricalMessages/parallel/Backfill_still_works_after_many_batches_are_imported

---

• depth is consistent across homeservers. But stream_ordering will be different between homeservers depending on when the event is processed by them.

---

To get streaming log output in Complement, use go test -v -tags synapse_blacklist,msc2403,msc2716 -count=1 $EXTRA_COMPLEMENT_ARGS ./tests/... (so it only looks at one package, avoid the csapi tests), see matrix-org/complement#215

---

Call stacks for backfilling

From the remote server

• RoomMessageListRestServlet.on_GET: /rooms/(?P<room_id>[^/]*)/messages$ (hs2 asks for more messages)
• get_messages
• maybe_backfill
• _maybe_backfill_inner
  • get_oldest_events_with_depth_in_room
  • try_backfill
• backfill
• federation_client.backfill
• transport_layer.backfill -> calls /backfill

From the origin server

• FederationBackfillServlet.on_GET: /backfill/(?P<room_id>[^/]*)/?
• on_backfill_request
• get_backfill_events
• _get_backfill_events

-- #10498

---

backfill
_process_pulled_events
_process_pulled_event
_resolve_state_at_missing_prevs
_get_state_after_missing_prev_event


_get_events_and_persist
_auth_and_persist_fetched_events
_auth_and_persist_fetched_events_inner
persist_events_and_notify

---

FederationEventServlet.on_GET (`/_matrix/federation/v1/event/{eventId}`)
on_pdu_request
get_persisted_pdu
get_event
get_events_as_list

---

insert_insertion_extremity

insertion_event_extremities



on_receive_pdu (`Process a PDU received via a federation /send/ transaction`)
_process_received_pdu
_handle_marker_event



backfill
_process_pulled_events
_process_pulled_event
_process_received_pdu



persist_events_and_notify
persist_events
_persist_event_batch
_persist_events_and_state_updates
_persist_events_txn
_update_metadata_tables_txn
_handle_insertion_event and _handle_batch_event

---

update_membership
update_membership_locked
_local_membership_update
create_event
create_new_client_event

create_and_send_nonmember_event
create_event
create_new_client_event

---

for ev in events:
            event_after_persisted = await self._store.get_event(
                ev.event_id, allow_none=True
            )

            if event_after_persisted:
                logger.info(
                    "from remote server: processed backfilled event_id=%s type=%s depth=%s stream_ordering=%s content=%s",
                    ev.event_id,
                    event_after_persisted["type"],
                    event_after_persisted["depth"],
                    event_after_persisted.internal_metadata.stream_ordering,
                    event_after_persisted["content"].get("body", None),
                )
            else:
                logger.info(
                    "from remote server: processed backfilled event_id=%s failed to lookup",
                    ev.event_id,
                )

event_lookup_result = self.db_pool.simple_select_one_txn(
                txn,
                table="events",
                keyvalues={"event_id": event_id},
                retcols=["type", "depth", "stream_ordering", "content"],
                allow_none=True,
            )

            event_json_lookup_result = self.db_pool.simple_select_one_onecol_txn(
                txn,
                table="event_json",
                keyvalues={"event_id": event_id},
                retcol="json",
                allow_none=True,
            )

            ev = db_to_json(event_json_lookup_result)

            if event_lookup_result:
                logger.info(
                    "_get_backfill_events: event_results add event_id=%s type=%s depth=%s stream_ordering=%s content=%s",
                    event_id,
                    ev["type"],
                    ev["depth"],
                    event_lookup_result["stream_ordering"],
                    ev["content"].get("body", None),
                )
            else:
                logger.info(
                    "_get_backfill_events: event_results event_id=%s failed to lookup",
                    event_id,
                )

---

logger.info(
            "backfill events=%s",
            [
                "event_id=%s,depth=%d,body=%s"
                % (event.event_id, event.depth, event.content.get("body", event.type))
                for event in events
            ],
        )

Todo

• Verify still works with avatar/displayname (at least when scrolled up into the historical messages, other scenarios tracked by MSC2716 leftover todo tasks #10737)
• Sort on the backfilling receiving side before processing, Fix historical messages backfilling in random order on remote homeservers (MSC2716) #11114 (comment)
  • Not doing this anymore.
• Put events with no prev_events but some auth_events behind the new room version
• Comment on MSC2716
  • Document change to allow events with no prev_events but with some auth_events
  • Update MSC about not being able to support existing room versions
• To fix up the Complement tests where they are missing the historical messages in the /messages chunk because it requires the homeserver to ask about the insertion event extremity next time; when we're assembling events to send over and see a marker event, also add the insertion event as an extremity

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.
• Pull request includes a sign off
• Code style is correct (run the linters)

[MadLittleMods]

@erikjohnston After experimenting with bigger datasets using the Gitter import script and writing bigger Complement tests, I'm seeing problems with our whole stream_ordering approach to sorting historical messages between depth. depth is the same for all homeservers but stream_ordering is different across homeservers and in Synapse land, just means whenever the event was processed.

With all of the massaging I'm trying to do in this PR to make it work for the basic scrollback case, it won't solve all of the problems. Technically, it shouldn't really matter how /backfill returns things but I'm just trying to make the stream_ordering a little more consistent from the origin to the remote homeservers in order to get the order of messages from /messages consistent (sorted by (topological_ordering, stream_ordering)).

Even if we can backfill messages in order, it still doesn't guarantee the same stream_ordering (and more importantly the /messages order) on the other server. For example, if a room has a bunch of history imported and someone visits a permalink to a historical message back in time, their homeserver will skip over the historical messages in between and insert the permalink as the next message in the stream_order and totally throw off the sort.

This will be even more of an exact use case when we add the MSC3030 jump to date API endpoint so the static archives can navigate and jump to a certain date.

The real solution around event ordering looks related to these issues where we don't rely on depth or stream_ordering and it's just up to the shape of the DAG itself. Any thoughts for making this work?

• https://github.com/matrix-org/matrix-doc/issues/3263
• https://github.com/matrix-org/matrix-doc/issues/1229

[erikjohnston]

The real solution around event ordering looks related to these issues where we don't rely on depth or stream_ordering and it's just up to the shape of the DAG itself. Any thoughts for making this work?

Can we sort the events returned by /backfill before persisting them based on the DAG? We have a sorted_topologically function that can be used to sort a set of events somewhere. Or is that not enough?

[MadLittleMods]

Can we sort the events returned by /backfill before persisting them based on the DAG? We have a sorted_topologically function that can be used to sort a set of events somewhere. Or is that not enough?

That may help for the specific backfill case I'm sorta optimizing here in this PR where the user is scrolling back in time sequentially.

But it doesn't help the case where you jump back in time via permalink or jump to date endpoint and start the stream_ordering from the middle the historical messages. It's currently important that stream_ordering starts from our newest historical messages at -1 and decrements to our oldest historical messages -939394, etc. If we have the middle historical messages at -1, then the sorting is all off.

That's the most obvious case of it going wrong, but could also go wrong in /backfill if some other homeserver backfills in their own not Synapse way where we try to provide it in perfect order (like an older Synapse or other homeserver implementation).

[erikjohnston]

I was thinking of doing the sorted_topologically on the receiving side? That way it doesn't matter what order the other server returns them in.

Ideally ideally, we wouldn't use depths and instead use an "online topological sorting" algorithm (e.g. Katriel–Bodlaender algorithm) to ensure that we correctly handle disjoint chunks of the graph becoming connected. But that is fiddly to do and I'm not quite sure how things like pagination tokens would work if the ordering could change.

[MadLittleMods]

I was thinking of doing the sorted_topologically on the receiving side? That way it doesn't matter what order the other server returns them in.

I think sorting on both the sending and receiving sides would be good to try to ensure best chance of it working (just for server differences between sending and receiving). But it's still not bullet-proof in the permalink case where stream_ordering becomes out of order.

---

Ideally ideally, we wouldn't use depths and instead use an "online topological sorting" algorithm (e.g. Katriel–Bodlaender algorithm) to ensure that we correctly handle disjoint chunks of the graph becoming connected. But that is fiddly to do and I'm not quite sure how things like pagination tokens would work if the ordering could change.

Does the chunking you worked on achieve this? #3785 Worth reviving?

Any good references for the Katriel–Bodlaender algorithm? Something more grokable over text PDF 😵 -- maybe how that algorithm works doesn't matter and the aspect I'm curious about is how we store this topological order and how it's updated if we insert an event in the middle. I assume it's more than just depth: 9 and we update every single row before or after when we insert something in the middle?

I see matrix-org/gomatrixserverlib#187 which discusses the "online topological sorting" and additionally the chunking stuff.

I'm not quite sure how things like pagination tokens would work if the ordering could change.

For reference: current pagination tokens, https://github.com/matrix-org/matrix-doc/issues/574

t8-8_0_0_0_0_0_0_0_0 seems to represent a single event already, i.e. the combo of depth and stream_ordering only point to a single event.

Could pagination tokens just become the event ID itself and then you just use the direction to paginate from there?

[erikjohnston]

Any good references for the Katriel–Bodlaender algorithm? Something more grokable over text PDF dizzy_face -- maybe how that algorithm works doesn't matter and the aspect I'm curious about is how we store this topological order and how it's updated if we insert an event in the middle. I assume it's more than just depth: 9 and we update every single row before or after when we insert something in the middle?

I have not found any helpful references to that algorithm beyond compsci papers. The actual algorithm is relatively fine, the major problem is picking a data structure for the labels that satisfy the necessary properties and is efficient to store in the DB:

The ord labels of the nodes are maintained by an Ordered List data structure ORD,
which is a data structure that allows to maintain a total order over a list of items and to perform
the following operations in constant amortized time [2, 4]: InsertAfter (x, y) (InsertBefore(x, y))
inserts the item x immediately after (before) the item y in the total order, Delete(x) removes the
item x, the query Order (x, y) determines whether x precedes y or y precedes x in the total order
and Next(x) (Prev (x)) returns the item that appears immediately after (before) x in the total
order.

While they suggest just using integers, that is a bit problematic for us as that would require potentially rewriting a lot of rows in the DB. There's no reason that they need to be integers, the previous work used fractions to allow inserting nodes in between existing nodes without renumbering everything (and also operated on chunks rather than individual events, iirc).

Could pagination tokens just become the event ID itself and then you just use the direction to paginate from there?

It would need to be a set of events to handle forks in the graph, annoyingly, as you basically need to serialise the state of a breadth-first search algorithm, iyswim. I.e. if you have a DAG A -> B -> D, A -> C -> D then if you have a token at B you don't know if you then need to still send down C or not.

[erikjohnston]

I think it would be valuable to resurrect the chunking project fwiw, its just a faff without concrete benefits to end users so its just never been prioritised.

[MadLittleMods]

The conversation for topological sorting when we receive backfill events continues in #11114 (comment)

---

For chunking and online topological sorting, we discussed this in an out of band call:

Plan forward:

1. Go forward with [this PR] so it fixes the scrollback case in federation
2. For Gitter and the static archives, we can have a script manually scrollback across all of the rooms in the archive server before anyone else or Google spider crawls in some weird way. This way it will lock the sort in place for all of the historical messages
3. Online topological ordering can happen in the future and by its nature will apply retroactively to fix any inconsistencies introduced by people permalinking

-- meeting summary notes 2021-10-28

[MadLittleMods]

Instead of using fake prev_events to make the historical events float, I've moved to just making an event with an empty array of prev_events. Which also stops the backfill mechanism from getting clogged with unresolvable backward extremities, see #11091

This is to allow prev_event_ids = [] if it has auth_event_ids present

-- #11114 (comment)

[MadLittleMods]

We no longer connect the historical state chain to the historical event chain so we don't need to specify any initial prev_events.

[erikjohnston]

I think this is making sense now!

[erikjohnston]

Why did we move this up?

[MadLittleMods]

To group it next to prev_event_ids for easier association.

[erikjohnston]

Woop!

[MadLittleMods]

Woop! Woop!

Thanks for the review and sticking it out to understand the details @erikjohnston 🦩