Merge branch 'master' into matthew/peeking-over-fed

federationapi/routing/send.go

@@ -523,10 +523,99 @@ func (t *txnReq) processEventWithMissingState(ctx context.Context, e gomatrixser
 		*gomatrixserverlib.RespState
 	}
 
-	// pass the event along with the state to the roomserver using a background context so we don't
-	// needlessly expire
-	headeredEvent := e.Headered(roomVersion)
-	return api.SendEventWithState(context.Background(), t.rsAPI, resolvedState, headeredEvent, t.haveEventIDs())
+	// at this point we know we're going to have a gap: we need to work out the room state at the new backwards extremity.
+	// Therefore, we cannot just query /state_ids with this event to get the state before. Instead, we need to query
+	// the state AFTER all the prev_events for this event, then apply state resolution to that to get the state before the event.
+	var states []*respState
+	for _, prevEventID := range backwardsExtremity.PrevEventIDs() {
+		// Look up what the state is after the backward extremity. This will either
+		// come from the roomserver, if we know all the required events, or it will
+		// come from a remote server via /state_ids if not.
+		prevState, trustworthy, lerr := t.lookupStateAfterEvent(gmectx, roomVersion, backwardsExtremity.RoomID(), prevEventID)
+		if lerr != nil {
+			util.GetLogger(ctx).WithError(lerr).Errorf("Failed to lookup state after prev_event: %s", prevEventID)
+			return lerr
+		}
+		// Append the state onto the collected state. We'll run this through the
+		// state resolution next.
+		states = append(states, &respState{trustworthy, prevState})
+	}
+
+	// Now that we have collected all of the state from the prev_events, we'll
+	// run the state through the appropriate state resolution algorithm for the
+	// room if needed. This does a couple of things:
+	// 1. Ensures that the state is deduplicated fully for each state-key tuple
+	// 2. Ensures that we pick the latest events from both sets, in the case that
+	//    one of the prev_events is quite a bit older than the others
+	resolvedState := &gomatrixserverlib.RespState{}
+	switch len(states) {
+	case 0:
+		extremityIsCreate := backwardsExtremity.Type() == gomatrixserverlib.MRoomCreate && backwardsExtremity.StateKeyEquals("")
+		if !extremityIsCreate {
+			// There are no previous states and this isn't the beginning of the
+			// room - this is an error condition!
+			util.GetLogger(ctx).Errorf("Failed to lookup any state after prev_events")
+			return fmt.Errorf("expected %d states but got %d", len(backwardsExtremity.PrevEventIDs()), len(states))
+		}
+	case 1:
+		// There's only one previous state - if it's trustworthy (came from a
+		// local state snapshot which will already have been through state res),
+		// use it as-is. There's no point in resolving it again.
+		if states[0].trustworthy {
+			resolvedState = states[0].RespState
+			break
+		}
+		// Otherwise, if it isn't trustworthy (came from federation), run it through
+		// state resolution anyway for safety, in case there are duplicates.
+		fallthrough
+	default:
+		respStates := make([]*gomatrixserverlib.RespState, len(states))
+		for i := range states {
+			respStates[i] = states[i].RespState
+		}
+		// There's more than one previous state - run them all through state res
+		resolvedState, err = t.resolveStatesAndCheck(gmectx, roomVersion, respStates, backwardsExtremity)
+		if err != nil {
+			util.GetLogger(ctx).WithError(err).Errorf("Failed to resolve state conflicts for event %s", backwardsExtremity.EventID())
+			return err
+		}
+	}
+
+	// First of all, send the backward extremity into the roomserver with the
+	// newly resolved state. This marks the "oldest" point in the backfill and
+	// sets the baseline state for any new events after this.
+	err = api.SendEventWithState(
+		context.Background(),
+		t.rsAPI,
+		api.KindOld,
+		resolvedState,
+		backwardsExtremity.Headered(roomVersion),
+		t.haveEventIDs(),
+	)
+	if err != nil {
+		return fmt.Errorf("api.SendEventWithState: %w", err)
+	}
+
+	// Then send all of the newer backfilled events, of which will all be newer
+	// than the backward extremity, into the roomserver without state. This way
+	// they will automatically fast-forward based on the room state at the
+	// extremity in the last step.
+	headeredNewEvents := make([]gomatrixserverlib.HeaderedEvent, len(newEvents))
+	for i, newEvent := range newEvents {
+		headeredNewEvents[i] = newEvent.Headered(roomVersion)
+	}
+	if err = api.SendEvents(
+		context.Background(),
+		t.rsAPI,
+		api.KindOld,
+		append(headeredNewEvents, e.Headered(roomVersion)),
+		api.DoNotSendToOtherServers,
+		nil,
+	); err != nil {
+		return fmt.Errorf("api.SendEvents: %w", err)
+	}
+
+	return nil
 }
 
 // lookupStateAfterEvent returns the room state after `eventID`, which is the state before eventID with the state of `eventID` (if it's a state event)

roomserver/api/wrapper.go

@@ -82,96 +82,6 @@ func SendEventWithState(
 	return SendInputRoomEvents(ctx, rsAPI, ires)
 }
 
-// SendEventWithRewrite writes an event with KindNew to the roomserver along
-// with a number of rewrite and outlier events for state and auth events
-// respectively.
-func SendEventWithRewrite(
-	ctx context.Context, rsAPI RoomserverInternalAPI, state *gomatrixserverlib.RespState,
-	event gomatrixserverlib.HeaderedEvent, haveEventIDs map[string]bool,
-) error {
-	isCurrentState := map[string]struct{}{}
-	for _, se := range state.StateEvents {
-		isCurrentState[se.EventID()] = struct{}{}
-	}
-
-	authAndStateEvents, err := state.Events()
-	if err != nil {
-		return err
-	}
-
-	var ires []InputRoomEvent
-	var stateIDs []string
-
-	// This function generates three things:
-	// A - A set of "rewrite" events, which will form the newly rewritten
-	//     state before the event, which includes every rewrite event that
-	//     came before it in its state
-	// B - A set of "outlier" events, which are auth events but not part
-	//     of the rewritten state
-	// C - A "new" event, which include all of the rewrite events in its
-	//     state
-	for _, authOrStateEvent := range authAndStateEvents {
-		if authOrStateEvent.StateKey() == nil {
-			continue
-		}
-		if haveEventIDs[authOrStateEvent.EventID()] {
-			continue
-		}
-
-		// We will handle an event as if it's an outlier if one of the
-		// following conditions is true:
-		storeAsOutlier := false
-		if _, ok := isCurrentState[authOrStateEvent.EventID()]; !ok {
-			// The event is an auth event and isn't a part of the state set.
-			// We'll send it as an outlier because we need it to be stored
-			// in case something is referring to it as an auth event.
-			storeAsOutlier = true
-		}
-
-		if storeAsOutlier {
-			ires = append(ires, InputRoomEvent{
-				Kind:         KindOutlier,
-				Event:        authOrStateEvent.Headered(event.RoomVersion),
-				AuthEventIDs: authOrStateEvent.AuthEventIDs(),
-			})
-			continue
-		}
-
-		// If the event isn't an outlier then we'll instead send it as a
-		// rewrite event, so that it'll form part of the rewritten state.
-		// These events will go through the membership and latest event
-		// updaters and we will generate output events, but they will be
-		// flagged as non-current (i.e. didn't just happen) events.
-		// Each of these rewrite events includes all of the rewrite events
-		// that came before in their StateEventIDs.
-		ires = append(ires, InputRoomEvent{
-			Kind:          KindRewrite,
-			Event:         authOrStateEvent.Headered(event.RoomVersion),
-			AuthEventIDs:  authOrStateEvent.AuthEventIDs(),
-			HasState:      true,
-			StateEventIDs: stateIDs,
-		})
-
-		// Add the event ID into the StateEventIDs of all subsequent
-		// rewrite events, and the new event.
-		stateIDs = append(stateIDs, authOrStateEvent.EventID())
-	}
-
-	// Send the final event as a new event, which will generate
-	// a timeline output event for it. All of the rewrite events
-	// that came before will be sent as StateEventIDs, forming a
-	// new clean state before the event.
-	ires = append(ires, InputRoomEvent{
-		Kind:          KindNew,
-		Event:         event,
-		AuthEventIDs:  event.AuthEventIDs(),
-		HasState:      true,
-		StateEventIDs: stateIDs,
-	})
-
-	return SendInputRoomEvents(ctx, rsAPI, ires)
-}
-
 // SendInputRoomEvents to the roomserver.
 func SendInputRoomEvents(
 	ctx context.Context, rsAPI RoomserverInternalAPI, ires []InputRoomEvent,