From 6150de6cb3611ffc61ce10ed6714f65e51e38e78 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 3 Sep 2020 15:22:16 +0100 Subject: FIFO ordering of input events (#1386) * Initial FIFOing of roomserver inputs * Remove EventID response from api.InputRoomEventsResponse * Don't send back event ID unnecessarily * Fix ordering hopefully * Reduce copies, use buffered task channel to reduce contention on other rooms * Fix error handling --- clientapi/routing/createroom.go | 3 +-- clientapi/routing/membership.go | 5 ++--- clientapi/routing/profile.go | 4 ++-- clientapi/routing/redaction.go | 3 +-- clientapi/routing/sendevent.go | 9 ++++----- clientapi/threepid/invites.go | 3 +-- 6 files changed, 11 insertions(+), 16 deletions(-) (limited to 'clientapi') diff --git a/clientapi/routing/createroom.go b/clientapi/routing/createroom.go index 57fc3f33..af43064f 100644 --- a/clientapi/routing/createroom.go +++ b/clientapi/routing/createroom.go @@ -342,8 +342,7 @@ func createRoom( } // send events to the room server - _, err = roomserverAPI.SendEvents(req.Context(), rsAPI, builtEvents, cfg.Matrix.ServerName, nil) - if err != nil { + if err = roomserverAPI.SendEvents(req.Context(), rsAPI, builtEvents, cfg.Matrix.ServerName, nil); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } diff --git a/clientapi/routing/membership.go b/clientapi/routing/membership.go index cba19a24..202662ab 100644 --- a/clientapi/routing/membership.go +++ b/clientapi/routing/membership.go @@ -75,13 +75,12 @@ func sendMembership(ctx context.Context, accountDB accounts.Database, device *us return jsonerror.InternalServerError() } - _, err = roomserverAPI.SendEvents( + if err = roomserverAPI.SendEvents( ctx, rsAPI, []gomatrixserverlib.HeaderedEvent{event.Event.Headered(roomVer)}, cfg.Matrix.ServerName, nil, - ) - if err != nil { + ); err != nil { util.GetLogger(ctx).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } diff --git a/clientapi/routing/profile.go b/clientapi/routing/profile.go index 4c7895bd..bc51b0b5 100644 --- a/clientapi/routing/profile.go +++ b/clientapi/routing/profile.go @@ -171,7 +171,7 @@ func SetAvatarURL( return jsonerror.InternalServerError() } - if _, err := api.SendEvents(req.Context(), rsAPI, events, cfg.Matrix.ServerName, nil); err != nil { + if err := api.SendEvents(req.Context(), rsAPI, events, cfg.Matrix.ServerName, nil); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } @@ -289,7 +289,7 @@ func SetDisplayName( return jsonerror.InternalServerError() } - if _, err := api.SendEvents(req.Context(), rsAPI, events, cfg.Matrix.ServerName, nil); err != nil { + if err := api.SendEvents(req.Context(), rsAPI, events, cfg.Matrix.ServerName, nil); 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 a825da64..178bfafc 100644 --- a/clientapi/routing/redaction.go +++ b/clientapi/routing/redaction.go @@ -122,8 +122,7 @@ func SendRedaction( JSON: jsonerror.NotFound("Room does not exist"), } } - _, err = roomserverAPI.SendEvents(context.Background(), rsAPI, []gomatrixserverlib.HeaderedEvent{*e}, cfg.Matrix.ServerName, nil) - if err != nil { + if err = roomserverAPI.SendEvents(context.Background(), rsAPI, []gomatrixserverlib.HeaderedEvent{*e}, cfg.Matrix.ServerName, nil); 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 a25979ea..9744a564 100644 --- a/clientapi/routing/sendevent.go +++ b/clientapi/routing/sendevent.go @@ -90,27 +90,26 @@ func SendEvent( // pass the new event to the roomserver and receive the correct event ID // event ID in case of duplicate transaction is discarded - eventID, err := api.SendEvents( + if err := api.SendEvents( req.Context(), rsAPI, []gomatrixserverlib.HeaderedEvent{ e.Headered(verRes.RoomVersion), }, cfg.Matrix.ServerName, txnAndSessionID, - ) - if err != nil { + ); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } util.GetLogger(req.Context()).WithFields(logrus.Fields{ - "event_id": eventID, + "event_id": e.EventID(), "room_id": roomID, "room_version": verRes.RoomVersion, }).Info("Sent event to roomserver") res := util.JSONResponse{ Code: http.StatusOK, - JSON: sendEventResponse{eventID}, + JSON: sendEventResponse{e.EventID()}, } // Add response to transactionsCache if txnID != nil { diff --git a/clientapi/threepid/invites.go b/clientapi/threepid/invites.go index 2ffb6bb0..b9575a28 100644 --- a/clientapi/threepid/invites.go +++ b/clientapi/threepid/invites.go @@ -359,7 +359,7 @@ func emit3PIDInviteEvent( return err } - _, err = api.SendEvents( + return api.SendEvents( ctx, rsAPI, []gomatrixserverlib.HeaderedEvent{ (*event).Headered(queryRes.RoomVersion), @@ -367,5 +367,4 @@ func emit3PIDInviteEvent( cfg.Matrix.ServerName, nil, ) - return err } -- cgit v1.2.3