Roomserver/federation input refactor (#2104)

* Put federation client functions into their own file

* Look for missing auth events in RS input

* Remove retrieveMissingAuthEvents from federation API

* Logging

* Sorta transplanted the code over

* Use event origin failing all else

* Don't get stuck on mutexes:

* Add verifier

* Don't mark state events with zero snapshot NID as not existing

* Check missing state if not an outlier before storing the event

* Reject instead of soft-fail, don't copy roominfo so much

* Use synchronous contexts, limit time to fetch missing events

* Clean up some commented out bits

* Simplify `/send` endpoint significantly

* Submit async

* Report errors on sending to RS input

* Set max payload in NATS to 16MB

* Tweak metrics

* Add `workerForRoom` for tidiness

* Try skipping unmarshalling errors for RespMissingEvents

* Track missing prev events separately to avoid calculating state when not possible

* Tweak logic around checking missing state

* Care about state when checking missing prev events

* Don't check missing state for create events

* Try that again

* Handle create events better

* Send create room events as new

* Use given event kind when sending auth/state events

* Revert "Use given event kind when sending auth/state events"

This reverts commit 089d64d271b5fca8c104e1554711187420dbebca.

* Only search for missing prev events or state for new events

* Tweaks

* We only have missing prev if we don't supply state

* Room version tweaks

* Allow async inputs again

* Apply backpressure to consumers/synchronous requests to hopefully stop things being overwhelmed

* Set timeouts on roomserver input tasks (need to decide what timeout makes sense)

* Use work queue policy, deliver all on restart

* Reduce chance of duplicates being sent by NATS

* Limit the number of servers we attempt to reduce backpressure

* Some review comment fixes

* Tidy up a couple things

* Don't limit servers, randomise order using map

* Some context refactoring

* Update gmsl

* Don't resend create events

* Set stateIDs length correctly or else the roomserver thinks there are missing events when there aren't

* Exclude our own servername

* Try backing off servers

* Make excluding self behaviour optional

* Exclude self from g_m_e

* Update sytest-whitelist

* Update consumers for the roomserver output stream

* Remember to send outliers for state returned from /gme

* Make full HTTP tests less upsetti

* Remove 'If a device list update goes missing, the server resyncs on the next one' from the sytest blacklist

* Remove debugging test

* Fix blacklist again, remove unnecessary duplicate context

* Clearer contexts, don't use background in case there's something happening there

* Don't queue up events more than once in memory

* Correctly identify create events when checking for state

* Fill in gaps again in /gme code

* Remove `AuthEventIDs` from `InputRoomEvent`

* Remove stray field

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

6 files changed, 19 insertions, 19 deletions

diff --git a/clientapi/routing/createroom.go b/clientapi/routing/createroom.go
index 85331192..e89d8ff2 100644
--- a/clientapi/routing/createroom.go
+++ b/clientapi/routing/createroom.go
@@ -451,23 +451,20 @@ func createRoom(
 			util.GetLogger(req.Context()).WithError(err).Error("authEvents.AddEvent failed")
 			return jsonerror.InternalServerError()
 		}
+	}
 
-		accumulated := gomatrixserverlib.UnwrapEventHeaders(builtEvents)
-		if err = roomserverAPI.SendEventWithState(
-			req.Context(),
-			rsAPI,
-			roomserverAPI.KindNew,
-			&gomatrixserverlib.RespState{
-				StateEvents: accumulated,
-				AuthEvents:  accumulated,
-			},
-			ev.Headered(roomVersion),
-			nil,
-			false,
-		); err != nil {
-			util.GetLogger(req.Context()).WithError(err).Error("SendEventWithState failed")
-			return jsonerror.InternalServerError()
-		}
+	inputs := make([]roomserverAPI.InputRoomEvent, 0, len(builtEvents))
+	for _, event := range builtEvents {
+		inputs = append(inputs, roomserverAPI.InputRoomEvent{
+			Kind:         roomserverAPI.KindNew,
+			Event:        event,
+			Origin:       cfg.Matrix.ServerName,
+			SendAsServer: roomserverAPI.DoNotSendToOtherServers,
+		})
+	}
+	if err = roomserverAPI.SendInputRoomEvents(req.Context(), rsAPI, inputs, false); err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("roomserverAPI.SendInputRoomEvents failed")
+		return jsonerror.InternalServerError()
 	}
 
 	// TODO(#269): Reserve room alias while we create the room. This stops us
diff --git a/clientapi/routing/membership.go b/clientapi/routing/membership.go
index 7ddb827e..4ce82079 100644
--- a/clientapi/routing/membership.go
+++ b/clientapi/routing/membership.go
@@ -109,6 +109,7 @@ func sendMembership(ctx context.Context, accountDB accounts.Database, device *us
 		roomserverAPI.KindNew,
 		[]*gomatrixserverlib.HeaderedEvent{event.Event.Headered(roomVer)},
 		cfg.Matrix.ServerName,
+		cfg.Matrix.ServerName,
 		nil,
 		false,
 	); err != nil {
diff --git a/clientapi/routing/profile.go b/clientapi/routing/profile.go
index 9de1869b..017facd2 100644
--- a/clientapi/routing/profile.go
+++ b/clientapi/routing/profile.go
@@ -169,7 +169,7 @@ func SetAvatarURL(
 		return jsonerror.InternalServerError()
 	}
 
-	if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil {
+	if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, cfg.Matrix.ServerName, nil, false); err != nil {
 		util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed")
 		return jsonerror.InternalServerError()
 	}
@@ -286,7 +286,7 @@ func SetDisplayName(
 		return jsonerror.InternalServerError()
 	}
 
-	if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil {
+	if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, cfg.Matrix.ServerName, nil, false); err != nil {
 		util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed")
 		return jsonerror.InternalServerError()
 	}
diff --git a/clientapi/routing/redaction.go b/clientapi/routing/redaction.go
index 8492236b..01ea818a 100644
--- a/clientapi/routing/redaction.go
+++ b/clientapi/routing/redaction.go
@@ -120,7 +120,7 @@ func SendRedaction(
 			JSON: jsonerror.NotFound("Room does not exist"),
 		}
 	}
-	if err = roomserverAPI.SendEvents(context.Background(), rsAPI, roomserverAPI.KindNew, []*gomatrixserverlib.HeaderedEvent{e}, cfg.Matrix.ServerName, nil, false); err != nil {
+	if err = roomserverAPI.SendEvents(context.Background(), rsAPI, roomserverAPI.KindNew, []*gomatrixserverlib.HeaderedEvent{e}, cfg.Matrix.ServerName, cfg.Matrix.ServerName, nil, false); err != nil {
 		util.GetLogger(req.Context()).WithError(err).Errorf("failed to SendEvents")
 		return jsonerror.InternalServerError()
 	}
diff --git a/clientapi/routing/sendevent.go b/clientapi/routing/sendevent.go
index f0498312..606107b9 100644
--- a/clientapi/routing/sendevent.go
+++ b/clientapi/routing/sendevent.go
@@ -121,6 +121,7 @@ func SendEvent(
 			e.Headered(verRes.RoomVersion),
 		},
 		cfg.Matrix.ServerName,
+		cfg.Matrix.ServerName,
 		txnAndSessionID,
 		false,
 	); err != nil {
diff --git a/clientapi/threepid/invites.go b/clientapi/threepid/invites.go
index 985cf00c..db62ce06 100644
--- a/clientapi/threepid/invites.go
+++ b/clientapi/threepid/invites.go
@@ -366,6 +366,7 @@ func emit3PIDInviteEvent(
 			event.Headered(queryRes.RoomVersion),
 		},
 		cfg.Matrix.ServerName,
+		cfg.Matrix.ServerName,
 		nil,
 		false,
 	)