sytest: Make 'Inbound federation can backfill events' pass (#1051)

* sytest: Make 'Inbound federation can backfill events' pass

This breaks 'Outbound federation can backfill events' because now
we are returning the right number of events, which the previous
test was relying on.

Previously, /messages was backfilling the membership event, causing
the test to pass. Now we are no longer backfilling the membership
event due to the change in this commit, causing the test to fail.

The test should instead be returning the membership event locally
from synacpis database, but it doesn't do it fast enough, resulting
in a no-op /sync response with a next_batch=s0_0 which will never
pick up the local membership event when it rolls in. The test
does attempt to retry, but doesn't take the new next_batch=s1_0
resulting in it missing from the /messages response.

* Linting

roomserver/api/query.go

@@ -21,6 +21,7 @@ import (
 
 	commonHTTP "github.com/matrix-org/dendrite/common/http"
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
 	opentracing "github.com/opentracing/opentracing-go"
 )
 
@@ -228,14 +229,24 @@ type QueryStateAndAuthChainResponse struct {
 type QueryBackfillRequest struct {
 	// The room to backfill
 	RoomID string `json:"room_id"`
-	// Events to start paginating from.
-	EarliestEventsIDs []string `json:"earliest_event_ids"`
+	// A map of backwards extremity event ID to a list of its prev_event IDs.
+	BackwardsExtremities map[string][]string `json:"backwards_extremities"`
 	// The maximum number of events to retrieve.
 	Limit int `json:"limit"`
 	// The server interested in the events.
 	ServerName gomatrixserverlib.ServerName `json:"server_name"`
 }
 
+// PrevEventIDs returns the prev_event IDs of all backwards extremities, de-duplicated in a lexicographically sorted order.
+func (r *QueryBackfillRequest) PrevEventIDs() []string {
+	var prevEventIDs []string
+	for _, pes := range r.BackwardsExtremities {
+		prevEventIDs = append(prevEventIDs, pes...)
+	}
+	prevEventIDs = util.UniqueStrings(prevEventIDs)
+	return prevEventIDs
+}
+
 // QueryBackfillResponse is a response to QueryBackfill.
 type QueryBackfillResponse struct {
 	// Missing events, arbritrary order.

roomserver/internal/query.go

@@ -495,14 +495,8 @@ func (r *RoomserverInternalAPI) QueryBackfill(
 	// defines the highest number of elements in the map below.
 	visited := make(map[string]bool, request.Limit)
 
-	// The provided event IDs have already been seen by the request's emitter,
-	// and will be retrieved anyway, so there's no need to care about them if
-	// they appear in our exploration of the event tree.
-	for _, id := range request.EarliestEventsIDs {
-		visited[id] = true
-	}
-
-	front = request.EarliestEventsIDs
+	// this will include these events which is what we want
+	front = request.PrevEventIDs()
 
 	// Scan the event tree for events to send back.
 	resultNIDs, err := r.scanEventTree(ctx, front, visited, request.Limit, request.ServerName)
@@ -534,7 +528,7 @@ func (r *RoomserverInternalAPI) backfillViaFederation(ctx context.Context, req *
 	if err != nil {
 		return fmt.Errorf("backfillViaFederation: unknown room version for room %s : %w", req.RoomID, err)
 	}
-	requester := newBackfillRequester(r.DB, r.FedClient, r.ServerName)
+	requester := newBackfillRequester(r.DB, r.FedClient, r.ServerName, req.BackwardsExtremities)
 	// Request 100 items regardless of what the query asks for.
 	// We don't want to go much higher than this.
 	// We can't honour exactly the limit as some sytests rely on requesting more for tests to pass
@@ -542,7 +536,7 @@ func (r *RoomserverInternalAPI) backfillViaFederation(ctx context.Context, req *
 	// Specifically the test "Outbound federation can backfill events"
 	events, err := gomatrixserverlib.RequestBackfill(
 		ctx, requester,
-		r.KeyRing, req.RoomID, roomVer, req.EarliestEventsIDs, 100)
+		r.KeyRing, req.RoomID, roomVer, req.PrevEventIDs(), 100)
 	if err != nil {
 		return err
 	}

roomserver/internal/query_backfill.go

@@ -16,6 +16,7 @@ type backfillRequester struct {
 	db         storage.Database
 	fedClient  *gomatrixserverlib.FederationClient
 	thisServer gomatrixserverlib.ServerName
+	bwExtrems  map[string][]string
 
 	// per-request state
 	servers                 []gomatrixserverlib.ServerName
@@ -23,13 +24,14 @@ type backfillRequester struct {
 	eventIDMap              map[string]gomatrixserverlib.Event
 }
 
-func newBackfillRequester(db storage.Database, fedClient *gomatrixserverlib.FederationClient, thisServer gomatrixserverlib.ServerName) *backfillRequester {
+func newBackfillRequester(db storage.Database, fedClient *gomatrixserverlib.FederationClient, thisServer gomatrixserverlib.ServerName, bwExtrems map[string][]string) *backfillRequester {
 	return &backfillRequester{
 		db:                      db,
 		fedClient:               fedClient,
 		thisServer:              thisServer,
 		eventIDToBeforeStateIDs: make(map[string][]string),
 		eventIDMap:              make(map[string]gomatrixserverlib.Event),
+		bwExtrems:               bwExtrems,
 	}
 }
 
@@ -160,26 +162,44 @@ func (b *backfillRequester) StateBeforeEvent(ctx context.Context, roomVer gomatr
 // It returns a list of servers which can be queried for backfill requests. These servers
 // will be servers that are in the room already. The entries at the beginning are preferred servers
 // and will be tried first. An empty list will fail the request.
-func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID string) (servers []gomatrixserverlib.ServerName) {
+func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID string) []gomatrixserverlib.ServerName {
+	// eventID will be a prev_event ID of a backwards extremity, meaning we will not have a database entry for it. Instead, use
+	// its successor, so look it up.
+	successor := ""
+FindSuccessor:
+	for sucID, prevEventIDs := range b.bwExtrems {
+		for _, pe := range prevEventIDs {
+			if pe == eventID {
+				successor = sucID
+				break FindSuccessor
+			}
+		}
+	}
+	if successor == "" {
+		logrus.WithField("event_id", eventID).Error("ServersAtEvent: failed to find successor of this event to determine room state")
+		return nil
+	}
+	eventID = successor
+
 	// getMembershipsBeforeEventNID requires a NID, so retrieving the NID for
 	// the event is necessary.
 	NIDs, err := b.db.EventNIDs(ctx, []string{eventID})
 	if err != nil {
 		logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get event NID for event")
-		return
+		return nil
 	}
 
 	stateEntries, err := stateBeforeEvent(ctx, b.db, NIDs[eventID])
 	if err != nil {
 		logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to load state before event")
-		return
+		return nil
 	}
 
 	// possibly return all joined servers depending on history visiblity
 	memberEventsFromVis, err := joinEventsFromHistoryVisibility(ctx, b.db, roomID, stateEntries)
 	if err != nil {
 		logrus.WithError(err).Error("ServersAtEvent: failed calculate servers from history visibility rules")
-		return
+		return nil
 	}
 	logrus.Infof("ServersAtEvent including %d current events from history visibility", len(memberEventsFromVis))
 
@@ -189,7 +209,7 @@ func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID
 	memberEvents, err := getMembershipsAtState(ctx, b.db, stateEntries, true)
 	if err != nil {
 		logrus.WithField("event_id", eventID).WithError(err).Error("ServersAtEvent: failed to get memberships before event")
-		return
+		return nil
 	}
 	memberEvents = append(memberEvents, memberEventsFromVis...)
 
@@ -198,6 +218,7 @@ func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID
 	for _, event := range memberEvents {
 		serverSet[event.Origin()] = true
 	}
+	var servers []gomatrixserverlib.ServerName
 	for server := range serverSet {
 		if server == b.thisServer {
 			continue
@@ -205,7 +226,7 @@ func (b *backfillRequester) ServersAtEvent(ctx context.Context, roomID, eventID
 		servers = append(servers, server)
 	}
 	b.servers = servers
-	return
+	return servers
 }
 
 // Backfill performs a backfill request to the given server.