Reuse the existing lazyload cache on /context and /messages (#2367)

syncapi/routing/context.go

@@ -22,6 +22,7 @@ import (
 	"strconv"
 
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/internal/caching"
 	roomserver "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
@@ -44,6 +45,7 @@ func Context(
 	rsAPI roomserver.RoomserverInternalAPI,
 	syncDB storage.Database,
 	roomID, eventID string,
+	lazyLoadCache *caching.LazyLoadCache,
 ) util.JSONResponse {
 	filter, err := parseRoomEventFilter(req)
 	if err != nil {
@@ -129,7 +131,7 @@ func Context(
 
 	eventsBeforeClient := gomatrixserverlib.HeaderedToClientEvents(eventsBefore, gomatrixserverlib.FormatAll)
 	eventsAfterClient := gomatrixserverlib.HeaderedToClientEvents(eventsAfter, gomatrixserverlib.FormatAll)
-	newState := applyLazyLoadMembers(filter, eventsAfterClient, eventsBeforeClient, state)
+	newState := applyLazyLoadMembers(device, filter, eventsAfterClient, eventsBeforeClient, state, lazyLoadCache)
 
 	response := ContextRespsonse{
 		Event:        gomatrixserverlib.HeaderedToClientEvent(&requestedEvent, gomatrixserverlib.FormatAll),
@@ -148,15 +150,25 @@ func Context(
 	}
 }
 
-func applyLazyLoadMembers(filter *gomatrixserverlib.RoomEventFilter, eventsAfter, eventsBefore []gomatrixserverlib.ClientEvent, state []*gomatrixserverlib.HeaderedEvent) []*gomatrixserverlib.HeaderedEvent {
+func applyLazyLoadMembers(
+	device *userapi.Device,
+	filter *gomatrixserverlib.RoomEventFilter,
+	eventsAfter, eventsBefore []gomatrixserverlib.ClientEvent,
+	state []*gomatrixserverlib.HeaderedEvent,
+	lazyLoadCache *caching.LazyLoadCache,
+) []*gomatrixserverlib.HeaderedEvent {
 	if filter == nil || !filter.LazyLoadMembers {
 		return state
 	}
 	allEvents := append(eventsBefore, eventsAfter...)
-	x := make(map[string]bool)
+	x := make(map[string]struct{})
 	// get members who actually send an event
 	for _, e := range allEvents {
-		x[e.Sender] = true
+		// Don't add membership events the client should already know about
+		if _, cached := lazyLoadCache.IsLazyLoadedUserCached(device, e.RoomID, e.Sender); cached {
+			continue
+		}
+		x[e.Sender] = struct{}{}
 	}
 
 	newState := []*gomatrixserverlib.HeaderedEvent{}
@@ -166,8 +178,9 @@ func applyLazyLoadMembers(filter *gomatrixserverlib.RoomEventFilter, eventsAfter
 			newState = append(newState, event)
 		} else {
 			// did the user send an event?
-			if x[event.Sender()] {
+			if _, ok := x[event.Sender()]; ok {
 				membershipEvents = append(membershipEvents, event)
+				lazyLoadCache.StoreLazyLoadedUser(device, event.RoomID(), event.Sender(), event.EventID())
 			}
 		}
 	}

syncapi/routing/messages.go

@@ -21,6 +21,7 @@ import (
 	"sort"
 
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/syncapi/storage"
@@ -64,6 +65,7 @@ func OnIncomingMessagesRequest(
 	rsAPI api.RoomserverInternalAPI,
 	cfg *config.SyncAPI,
 	srp *sync.RequestPool,
+	lazyLoadCache *caching.LazyLoadCache,
 ) util.JSONResponse {
 	var err error
 
@@ -200,6 +202,10 @@ func OnIncomingMessagesRequest(
 	if filter.LazyLoadMembers {
 		membershipToUser := make(map[string]*gomatrixserverlib.HeaderedEvent)
 		for _, evt := range clientEvents {
+			// Don't add membership events the client should already know about
+			if _, cached := lazyLoadCache.IsLazyLoadedUserCached(device, roomID, evt.Sender); cached {
+				continue
+			}
 			membership, err := db.GetStateEvent(req.Context(), roomID, gomatrixserverlib.MRoomMember, evt.Sender)
 			if err != nil {
 				util.GetLogger(req.Context()).WithError(err).Error("failed to get membership event for user")
@@ -207,10 +213,11 @@ func OnIncomingMessagesRequest(
 			}
 			if membership != nil {
 				membershipToUser[evt.Sender] = membership
+				lazyLoadCache.StoreLazyLoadedUser(device, roomID, evt.Sender, membership.EventID())
 			}
 		}
 		for _, evt := range membershipToUser {
-			state = append(state, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatAll))
+			state = append(state, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatSync))
 		}
 	}
 

syncapi/routing/routing.go

@@ -18,6 +18,7 @@ import (
 	"net/http"
 
 	"github.com/gorilla/mux"
+	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/httputil"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
@@ -38,6 +39,7 @@ func Setup(
 	userAPI userapi.UserInternalAPI, federation *gomatrixserverlib.FederationClient,
 	rsAPI api.RoomserverInternalAPI,
 	cfg *config.SyncAPI,
+	lazyLoadCache *caching.LazyLoadCache,
 ) {
 	v3mux := csMux.PathPrefix("/{apiversion:(?:r0|v3)}/").Subrouter()
 
@@ -51,7 +53,7 @@ func Setup(
 		if err != nil {
 			return util.ErrorResponse(err)
 		}
-		return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], device, federation, rsAPI, cfg, srp)
+		return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], device, federation, rsAPI, cfg, srp, lazyLoadCache)
 	})).Methods(http.MethodGet, http.MethodOptions)
 
 	v3mux.Handle("/user/{userId}/filter",
@@ -89,6 +91,7 @@ func Setup(
 				req, device,
 				rsAPI, syncDB,
 				vars["roomId"], vars["eventId"],
+				lazyLoadCache,
 			)
 		}),
 	).Methods(http.MethodGet, http.MethodOptions)

syncapi/syncapi.go

@@ -148,5 +148,5 @@ func AddPublicRoutes(
 		logrus.WithError(err).Panicf("failed to start presence consumer")
 	}
 
-	routing.Setup(router, requestPool, syncDB, userAPI, federation, rsAPI, cfg)
+	routing.Setup(router, requestPool, syncDB, userAPI, federation, rsAPI, cfg, lazyLoadCache)
 }