Merge branch 'master' into neilalexander/federationinput

roomserver/api/query.go

@@ -226,6 +226,10 @@ type QueryStateAndAuthChainRequest struct {
 	PrevEventIDs []string `json:"prev_event_ids"`
 	// The list of auth events for the event. Used to calculate the auth chain
 	AuthEventIDs []string `json:"auth_event_ids"`
+	// If true, the auth chain events for the auth event IDs given will be fetched only. Prev event IDs are ignored.
+	// If false, state and auth chain events for the prev event IDs and entire current state will be included.
+	// TODO: not a great API shape. It serves 2 main uses: false=>response for send_join, true=>response for /event_auth
+	OnlyFetchAuthChain bool `json:"only_fetch_auth_chain"`
 	// Should state resolution be ran on the result events?
 	// TODO: check call sites and remove if we always want to do state res
 	ResolveState bool `json:"resolve_state"`

roomserver/internal/api.go

@@ -78,9 +78,6 @@ func NewRoomserverAPI(
 		},
 		// perform-er structs get initialised when we have a federation sender to use
 	}
-	if err := a.Inputer.Start(); err != nil {
-		logrus.WithError(err).Panic("failed to start roomserver input API")
-	}
 	return a
 }
 
@@ -157,6 +154,10 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.FederationInternalA
 	r.Forgetter = &perform.Forgetter{
 		DB: r.DB,
 	}
+
+	if err := r.Inputer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start roomserver input API")
+	}
 }
 
 func (r *RoomserverInternalAPI) SetAppserviceAPI(asAPI asAPI.AppServiceQueryAPI) {

roomserver/internal/input/input.go

@@ -65,14 +65,15 @@ func (r *Inputer) Start() error {
 		// later, possibly with an error response to the inputter if synchronous.
 		func(msg *nats.Msg) {
 			roomID := msg.Header.Get("room_id")
-			defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
 			var inputRoomEvent api.InputRoomEvent
 			if err := json.Unmarshal(msg.Data, &inputRoomEvent); err != nil {
 				_ = msg.Term()
 				return
 			}
 			inbox, _ := r.workers.LoadOrStore(roomID, &phony.Inbox{})
+			roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
 			inbox.(*phony.Inbox).Act(nil, func() {
+				defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
 				if err := r.processRoomEvent(context.TODO(), &inputRoomEvent); err != nil {
 					sentry.CaptureException(err)
 				} else {
@@ -117,15 +118,17 @@ func (r *Inputer) InputRoomEvents(
 			if _, err = r.JetStream.PublishMsg(msg); err != nil {
 				return
 			}
-			roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
 		}
 	} else {
 		responses := make(chan error, len(request.InputRoomEvents))
 		defer close(responses)
 		for _, e := range request.InputRoomEvents {
 			inputRoomEvent := e
-			inbox, _ := r.workers.LoadOrStore(inputRoomEvent.Event.RoomID(), &phony.Inbox{})
+			roomID := inputRoomEvent.Event.RoomID()
+			inbox, _ := r.workers.LoadOrStore(roomID, &phony.Inbox{})
+			roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
 			inbox.(*phony.Inbox).Act(nil, func() {
+				defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
 				err := r.processRoomEvent(ctx, &inputRoomEvent)
 				if err != nil {
 					sentry.CaptureException(err)

roomserver/internal/query/query.go

@@ -457,6 +457,21 @@ func (r *Queryer) QueryStateAndAuthChain(
 	response.RoomExists = true
 	response.RoomVersion = info.RoomVersion
 
+	// handle this entirely separately to the other case so we don't have to pull out
+	// the entire current state of the room
+	// TODO: this probably means it should be a different query operation...
+	if request.OnlyFetchAuthChain {
+		var authEvents []*gomatrixserverlib.Event
+		authEvents, err = GetAuthChain(ctx, r.DB.EventsFromIDs, request.AuthEventIDs)
+		if err != nil {
+			return err
+		}
+		for _, event := range authEvents {
+			response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
+		}
+		return nil
+	}
+
 	var stateEvents []*gomatrixserverlib.Event
 	stateEvents, err = r.loadStateAtEventIDs(ctx, info, request.PrevEventIDs)
 	if err != nil {