diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/events.go b/src/github.com/matrix-org/dendrite/roomserver/input/events.go index 91de64353..02cd45066 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/events.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/events.go @@ -69,6 +69,11 @@ type OutputRoomEventWriter interface { WriteOutputEvents(roomID string, updates []api.OutputEvent) error } +// processRoomEvent can only be called once at a time +// +// TODO(#375): This should be rewritten to allow concurrent calls. The +// difficulty is in ensuring that we correctly annotate events with the correct +// state deltas when sending to kafka streams func processRoomEvent( ctx context.Context, db RoomEventDatabase, diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/input.go b/src/github.com/matrix-org/dendrite/roomserver/input/input.go index 253ef9ff3..4d5ba5cb1 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/input.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/input.go @@ -19,6 +19,7 @@ import ( "context" "encoding/json" "net/http" + "sync" "github.com/matrix-org/dendrite/common" "github.com/matrix-org/dendrite/roomserver/api" @@ -33,6 +34,8 @@ type RoomserverInputAPI struct { // The kafkaesque topic to output new room events to. // This is the name used in kafka to identify the stream to write events to. OutputRoomEventTopic string + // Protects calls to processRoomEvent + mutex sync.Mutex } // WriteOutputEvents implements OutputRoomEventWriter @@ -59,6 +62,10 @@ func (r *RoomserverInputAPI) InputRoomEvents( response *api.InputRoomEventsResponse, ) error { for i := range request.InputRoomEvents { + // We lock as processRoomEvent can ony be called once at a time + r.mutex.Lock() + defer r.mutex.Unlock() + if err := processRoomEvent(ctx, r.DB, r, request.InputRoomEvents[i]); err != nil { return err } diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go b/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go index 2b82bcba2..c2f06393f 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go @@ -42,6 +42,7 @@ import ( // | // 7 <----- latest // +// Can only be called once at a time func updateLatestEvents( ctx context.Context, db RoomEventDatabase,