mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-01 03:03:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Don't queue up events more than once in memory

Browse source
This commit is contained in:
Neil Alexander 2022-01-26 17:30:34 +00:00
parent 3150f6e64e
commit ffe93f103f
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 26 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
roomserver/internal/input/input.go
View file

@ -63,6 +63,13 @@ func (r *Inputer) workerForRoom(roomID string) *phony.Inbox {
return inbox.(*phony.Inbox) return inbox.(*phony.Inbox)
} }
// eventsInProgress is an in-memory map to keep a track of which events we have
// queued up for processing. If we get a redelivery from NATS and we still have
// the queued up item then we won't do anything with the redelivered message. If
// we've restarted Dendrite and now this map is empty then it means that we will
// reload pending work from NATS.
var eventsInProgress sync.Map
// onMessage is called when a new event arrives in the roomserver input stream. // onMessage is called when a new event arrives in the roomserver input stream.
func (r *Inputer) Start() error { func (r *Inputer) Start() error {
_, err := r.JetStream.Subscribe( _, err := r.JetStream.Subscribe(
@ -79,9 +86,19 @@ func (r *Inputer) Start() error {
} }
_ = msg.InProgress() _ = msg.InProgress()
index := roomID + "\000" + inputRoomEvent.Event.EventID()
if _, ok := eventsInProgress.LoadOrStore(index, struct{}{}); ok {
// We're already waiting to deal with this event, so there's no
// point in queuing it up again. We've notified NATS that we're
// working on the message still, so that will have deferred the
// redelivery by a bit.
return
}
roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc() roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
r.workerForRoom(roomID).Act(nil, func() { r.workerForRoom(roomID).Act(nil, func() {
_ = msg.InProgress() // resets the acknowledgement wait timer _ = msg.InProgress() // resets the acknowledgement wait timer
defer eventsInProgress.Delete(index)
defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec() defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
if err := r.processRoomEvent(context.Background(), &inputRoomEvent); err != nil { if err := r.processRoomEvent(context.Background(), &inputRoomEvent); err != nil {
sentry.CaptureException(err) sentry.CaptureException(err)
@ -138,9 +155,18 @@ func (r *Inputer) InputRoomEvents(
for _, e := range request.InputRoomEvents { for _, e := range request.InputRoomEvents {
inputRoomEvent := e inputRoomEvent := e
roomID := inputRoomEvent.Event.RoomID() roomID := inputRoomEvent.Event.RoomID()
index := roomID + "\000" + inputRoomEvent.Event.EventID()
if _, ok := eventsInProgress.LoadOrStore(index, struct{}{}); ok {
// We're already waiting to deal with this event, so there's no
// point in queuing it up again. We've notified NATS that we're
// working on the message still, so that will have deferred the
// redelivery by a bit.
return
}
roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc() roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc()
worker := r.workerForRoom(roomID) worker := r.workerForRoom(roomID)
worker.Act(nil, func() { worker.Act(nil, func() {
defer eventsInProgress.Delete(index)
defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec() defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec()
err := r.processRoomEvent(ctx, &inputRoomEvent) err := r.processRoomEvent(ctx, &inputRoomEvent)
if err != nil { if err != nil {