Fix potential state reset when trying to join a room (#3040)

When trying to join a room in short sequence, it is possible that a
state reset occurs. This fixes it by using `singleflight`.

Author: S7evinK

clientapi/routing/routing.go

@@ -20,14 +20,16 @@ import (
 	"strings"
 
 	"github.com/gorilla/mux"
-	"github.com/matrix-org/dendrite/setup/base"
-	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib/fclient"
 	"github.com/matrix-org/gomatrixserverlib/spec"
 	"github.com/matrix-org/util"
 	"github.com/nats-io/nats.go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/singleflight"
+
+	"github.com/matrix-org/dendrite/setup/base"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
 
 	appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
 	"github.com/matrix-org/dendrite/clientapi/api"
@@ -84,6 +86,14 @@ func Setup(
 		unstableFeatures["org.matrix."+msc] = true
 	}
 
+	// singleflight protects /join endpoints from being invoked
+	// multiple times from the same user and room, otherwise
+	// a state reset can occur. This also avoids unneeded
+	// state calculations.
+	// TODO: actually fix this in the roomserver, as there are
+	// 		 possibly other ways that can result in a stat reset.
+	sf := singleflight.Group{}
+
 	if cfg.Matrix.WellKnownClientName != "" {
 		logrus.Infof("Setting m.homeserver base_url as %s at /.well-known/matrix/client", cfg.Matrix.WellKnownClientName)
 		wkMux.Handle("/client", httputil.MakeExternalAPI("wellknown", func(r *http.Request) util.JSONResponse {
@@ -264,9 +274,17 @@ func Setup(
 			if err != nil {
 				return util.ErrorResponse(err)
 			}
-			return JoinRoomByIDOrAlias(
-				req, device, rsAPI, userAPI, vars["roomIDOrAlias"],
-			)
+			// Only execute a join for roomIDOrAlias and UserID once. If there is a join in progress
+			// it waits for it to complete and returns that result for subsequent requests.
+			resp, _, _ := sf.Do(vars["roomIDOrAlias"]+device.UserID, func() (any, error) {
+				return JoinRoomByIDOrAlias(
+					req, device, rsAPI, userAPI, vars["roomIDOrAlias"],
+				), nil
+			})
+			// once all joins are processed, drop them from the cache. Further requests
+			// will be processed as usual.
+			sf.Forget(vars["roomIDOrAlias"] + device.UserID)
+			return resp.(util.JSONResponse)
 		}, httputil.WithAllowGuests()),
 	).Methods(http.MethodPost, http.MethodOptions)
 
@@ -300,9 +318,17 @@ func Setup(
 			if err != nil {
 				return util.ErrorResponse(err)
 			}
-			return JoinRoomByIDOrAlias(
-				req, device, rsAPI, userAPI, vars["roomID"],
-			)
+			// Only execute a join for roomID and UserID once. If there is a join in progress
+			// it waits for it to complete and returns that result for subsequent requests.
+			resp, _, _ := sf.Do(vars["roomID"]+device.UserID, func() (any, error) {
+				return JoinRoomByIDOrAlias(
+					req, device, rsAPI, userAPI, vars["roomID"],
+				), nil
+			})
+			// once all joins are processed, drop them from the cache. Further requests
+			// will be processed as usual.
+			sf.Forget(vars["roomID"] + device.UserID)
+			return resp.(util.JSONResponse)
 		}, httputil.WithAllowGuests()),
 	).Methods(http.MethodPost, http.MethodOptions)
 	v3mux.Handle("/rooms/{roomID}/leave",

go.mod

@@ -45,6 +45,7 @@ require (
 	golang.org/x/crypto v0.9.0
 	golang.org/x/image v0.5.0
 	golang.org/x/mobile v0.0.0-20221020085226-b36e6246172e
+	golang.org/x/sync v0.1.0
 	golang.org/x/term v0.8.0
 	gopkg.in/h2non/bimg.v1 v1.1.9
 	gopkg.in/yaml.v2 v2.4.0

go.sum

@@ -614,6 +614,7 @@ golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=