Roomserver per-room input lock

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,18 @@ 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()
+		// Take the lock for a given room ID. The roomserver should only
-	defer r.mutex.Unlock()
+		// process one event for a given room at a time to ensure that
-	for i := range request.InputRoomEvents {
+		// forward extremities, previous events etc are stored properly
+		// and are not corrupted by race conditions.
+		mutex, _ := r.mutexes.LoadOrStore(e.Event.RoomID(), &sync.Mutex{})
+		mutex.(*sync.Mutex).Lock()
 		if response.EventID, err = r.processRoomEvent(ctx, request.InputRoomEvents[i]); err != nil {
+			mutex.(*sync.Mutex).Unlock()
 			return err
 		}
+		mutex.(*sync.Mutex).Unlock()
 	}
 	return nil
 }