Per-room input mutex

roomserver/internal/api.go

@@ -21,6 +21,6 @@ type RoomserverInternalAPI struct {
 	KeyRing              gomatrixserverlib.JSONVerifier
 	FedClient            *gomatrixserverlib.FederationClient
 	OutputRoomEventTopic string   // Kafka topic for new output room events
-	mutex                sync.Mutex // Protects calls to processRoomEvent
+	mutexes              sync.Map // room ID -> *sync.Mutex, protects calls to processRoomEvent
 	fsAPI                fsAPI.FederationSenderInternalAPI
 }

roomserver/internal/input.go

@@ -18,6 +18,7 @@ package internal
 import (
 	"context"
 	"encoding/json"
+	"sync"
 
 	"github.com/Shopify/sarama"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -71,13 +72,14 @@ func (r *RoomserverInternalAPI) InputRoomEvents(
 	request *api.InputRoomEventsRequest,
 	response *api.InputRoomEventsResponse,
 ) (err error) {
-	// We lock as processRoomEvent can only be called once at a time
+	for i, e := range request.InputRoomEvents {
-	r.mutex.Lock()
+		mutex, _ := r.mutexes.LoadOrStore(e.Event.RoomID(), &sync.Mutex{})
-	defer r.mutex.Unlock()
+		mutex.(*sync.Mutex).Lock()
-	for i := range request.InputRoomEvents {
 		if response.EventID, err = r.processRoomEvent(ctx, request.InputRoomEvents[i]); err != nil {
+			mutex.(*sync.Mutex).Unlock()
 			return err
 		}
+		mutex.(*sync.Mutex).Unlock()
 	}
 	return nil
 }