Optimize `/sync` and history visibility (#2961)

Should fix the following issues or make a lot less worse when using
Postgres:

The main issue behind #2911: The client gives up after a certain time,
causing a cascade of context errors, because the response couldn't be
built up fast enough. This mostly happens on accounts with many rooms,
due to the inefficient way we're getting recent events and current state

For #2777: The queries for getting the membership events for history
visibility were being executed for each room (I think 185?), resulting
in a whooping 2k queries for membership events. (Getting the
statesnapshot -> block nids -> actual wanted membership event)

Both should now be better by:
- Using a LATERAL join to get all recent events for all joined rooms in
one go (TODO: maybe do the same for room summary and current state etc)
- If we're lazy loading on initial syncs, we're now not getting the
whole current state, just to drop the majority of it because we're lazy
loading members - we add a filter to exclude membership events on the
first call to `CurrentState`.
- Using an optimized query to get the membership events needed to
calculate history visibility

---------

Co-authored-by: kegsay <kegan@matrix.org>

1 files changed, 39 insertions, 25 deletions

diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go
index 44013e37..6af25c02 100644
--- a/syncapi/streams/stream_pdu.go
+++ b/syncapi/streams/stream_pdu.go
@@ -82,19 +82,24 @@ func (p *PDUStreamProvider) CompleteSync(
 		req.Log.WithError(err).Error("unable to update event filter with ignored users")
 	}
 
-	// Invalidate the lazyLoadCache, otherwise we end up with missing displaynames/avatars
-	// TODO: This might be inefficient, when joined to many and/or large rooms.
+	recentEvents, err := snapshot.RecentEvents(ctx, joinedRoomIDs, r, &eventFilter, true, true)
+	if err != nil {
+		return from
+	}
+	// Build up a /sync response. Add joined rooms.
 	for _, roomID := range joinedRoomIDs {
+		events := recentEvents[roomID]
+		// Invalidate the lazyLoadCache, otherwise we end up with missing displaynames/avatars
+		// TODO: This might be inefficient, when joined to many and/or large rooms.
 		joinedUsers := p.notifier.JoinedUsers(roomID)
 		for _, sharedUser := range joinedUsers {
 			p.lazyLoadCache.InvalidateLazyLoadedUser(req.Device, roomID, sharedUser)
 		}
-	}
 
-	// Build up a /sync response. Add joined rooms.
-	for _, roomID := range joinedRoomIDs {
+		// get the join response for each room
 		jr, jerr := p.getJoinResponseForCompleteSync(
-			ctx, snapshot, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device, false,
+			ctx, snapshot, roomID, &stateFilter, req.WantFullState, req.Device, false,
+			events.Events, events.Limited,
 		)
 		if jerr != nil {
 			req.Log.WithError(jerr).Error("p.getJoinResponseForCompleteSync failed")
@@ -113,11 +118,25 @@ func (p *PDUStreamProvider) CompleteSync(
 		req.Log.WithError(err).Error("p.DB.PeeksInRange failed")
 		return from
 	}
-	for _, peek := range peeks {
-		if !peek.Deleted {
+	if len(peeks) > 0 {
+		peekRooms := make([]string, 0, len(peeks))
+		for _, peek := range peeks {
+			if !peek.Deleted {
+				peekRooms = append(peekRooms, peek.RoomID)
+			}
+		}
+
+		recentEvents, err = snapshot.RecentEvents(ctx, peekRooms, r, &eventFilter, true, true)
+		if err != nil {
+			return from
+		}
+
+		for _, roomID := range peekRooms {
 			var jr *types.JoinResponse
+			events := recentEvents[roomID]
 			jr, err = p.getJoinResponseForCompleteSync(
-				ctx, snapshot, peek.RoomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device, true,
+				ctx, snapshot, roomID, &stateFilter, req.WantFullState, req.Device, true,
+				events.Events, events.Limited,
 			)
 			if err != nil {
 				req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
@@ -126,7 +145,7 @@ func (p *PDUStreamProvider) CompleteSync(
 				}
 				continue
 			}
-			req.Response.Rooms.Peek[peek.RoomID] = jr
+			req.Response.Rooms.Peek[roomID] = jr
 		}
 	}
 
@@ -227,7 +246,7 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 	stateFilter *gomatrixserverlib.StateFilter,
 	req *types.SyncRequest,
 ) (types.StreamPosition, error) {
-
+	var err error
 	originalLimit := eventFilter.Limit
 	// If we're going backwards, grep at least X events, this is mostly to satisfy Sytest
 	if r.Backwards && originalLimit < recentEventBackwardsLimit {
@@ -238,8 +257,8 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 		}
 	}
 
-	recentStreamEvents, limited, err := snapshot.RecentEvents(
-		ctx, delta.RoomID, r,
+	dbEvents, err := snapshot.RecentEvents(
+		ctx, []string{delta.RoomID}, r,
 		eventFilter, true, true,
 	)
 	if err != nil {
@@ -248,6 +267,10 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 		}
 		return r.From, fmt.Errorf("p.DB.RecentEvents: %w", err)
 	}
+
+	recentStreamEvents := dbEvents[delta.RoomID].Events
+	limited := dbEvents[delta.RoomID].Limited
+
 	recentEvents := gomatrixserverlib.HeaderedReverseTopologicalOrdering(
 		snapshot.StreamEventsToEvents(device, recentStreamEvents),
 		gomatrixserverlib.TopologicalOrderByPrevEvents,
@@ -420,7 +443,7 @@ func applyHistoryVisibilityFilter(
 		"room_id":  roomID,
 		"before":   len(recentEvents),
 		"after":    len(events),
-	}).Trace("Applied history visibility (sync)")
+	}).Debugf("Applied history visibility (sync)")
 	return events, nil
 }
 
@@ -428,25 +451,16 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	ctx context.Context,
 	snapshot storage.DatabaseTransaction,
 	roomID string,
-	r types.Range,
 	stateFilter *gomatrixserverlib.StateFilter,
-	eventFilter *gomatrixserverlib.RoomEventFilter,
 	wantFullState bool,
 	device *userapi.Device,
 	isPeek bool,
+	recentStreamEvents []types.StreamEvent,
+	limited bool,
 ) (jr *types.JoinResponse, err error) {
 	jr = types.NewJoinResponse()
 	// TODO: When filters are added, we may need to call this multiple times to get enough events.
 	//       See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
-	recentStreamEvents, limited, err := snapshot.RecentEvents(
-		ctx, roomID, r, eventFilter, true, true,
-	)
-	if err != nil {
-		if err == sql.ErrNoRows {
-			return jr, nil
-		}
-		return
-	}
 
 	// Work our way through the timeline events and pick out the event IDs
 	// of any state events that appear in the timeline. We'll specifically