Limit per-room mutexes to Postgres

roomserver/internal/input.go

@@ -73,7 +73,11 @@ func (r *RoomserverInternalAPI) InputRoomEvents(
 	response *api.InputRoomEventsResponse,
 ) (err error) {
 	for i, e := range request.InputRoomEvents {
-		mutex, _ := r.mutexes.LoadOrStore(e.Event.RoomID(), &sync.Mutex{})
+		roomID := e.Event.RoomID()
+		if !r.DB.SupportsConcurrentRoomInputs() {
+			roomID = "global"
+		}
+		mutex, _ := r.mutexes.LoadOrStore(roomID, &sync.Mutex{})
 		mutex.(*sync.Mutex).Lock()
 		if response.EventID, err = r.processRoomEvent(ctx, request.InputRoomEvents[i]); err != nil {
 			mutex.(*sync.Mutex).Unlock()

roomserver/storage/interface.go

@@ -24,6 +24,8 @@ import (
 )
 
 type Database interface {
+	// Do we support processing input events for more than one room at a time?
+	SupportsConcurrentRoomInputs() bool
 	// Store the room state at an event in the database
 	AddState(
 		ctx context.Context,

roomserver/storage/shared/storage.go

@@ -44,6 +44,10 @@ type Database struct {
 	RedactionsTable     tables.Redactions
 }
 
+func (d *Database) SupportsConcurrentRoomInputs() bool {
+	return true
+}
+
 func (d *Database) EventTypeNIDs(
 	ctx context.Context, eventTypes []string,
 ) (map[string]types.EventTypeNID, error) {

roomserver/storage/sqlite3/storage.go

@@ -139,6 +139,14 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	return &d, nil
 }
 
+func (d *Database) SupportsConcurrentRoomInputs() bool {
+	// This isn't supported in SQLite mode yet because of issues with
+	// database locks.
+	// TODO: Look at this again - the problem is probably to do with
+	// the membership updaters and latest events updaters.
+	return false
+}
+
 func (d *Database) GetLatestEventsForUpdate(
 	ctx context.Context, roomNID types.RoomNID,
 ) (*shared.LatestEventsUpdater, error) {