mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-07 06:53:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Some tweaks for the device list updater

Browse source
This commit is contained in:
Till Faelligen 2023-10-31 10:06:49 +01:00
parent 32f7c4b166
commit f2dcdf6213
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
5 changed files with 30 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
setup/jetstream/streams.go
View file

@ -54,6 +54,7 @@ var streams = []*nats.StreamConfig{
Name: InputDeviceListUpdate,
Retention: nats.InterestPolicy,
Storage: nats.FileStorage,
MaxAge: time.Hour * 24,
},
{
Name: InputSigningKeyUpdate,

6
userapi/consumers/devicelistupdate.go
View file

@ -17,6 +17,7 @@ package consumers
import (
"context"
"encoding/json"
"time"
"github.com/matrix-org/dendrite/userapi/internal"
"github.com/matrix-org/gomatrixserverlib"
@ -82,7 +83,10 @@ func (t *DeviceListUpdateConsumer) onMessage(ctx context.Context, msgs []*nats.M
return true
}
err := t.updater.Update(ctx, m)
timeoutCtx, cancel := context.WithTimeout(ctx, time.Second*30)
defer cancel()
err := t.updater.Update(timeoutCtx, m)
if err != nil {
logrus.WithFields(logrus.Fields{
"user_id": m.UserID,

17
userapi/internal/device_list_update.go
View file

@ -21,6 +21,7 @@ import (
"fmt"
"hash/fnv"
"net"
"strconv"
"sync"
"time"
@ -142,13 +143,27 @@ type KeyChangeProducer interface {
ProduceKeyChanges(keys []api.DeviceMessage) error
}
var deviceListUpdaterBackpressure = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "dendrite",
Subsystem: "keyserver",
Name: "worker_backpressure",
Help: "How many device list updater requests are queued",
},
[]string{"worker_id"},
)
// NewDeviceListUpdater creates a new updater which fetches fresh device lists when they go stale.
func NewDeviceListUpdater(
process *process.ProcessContext, db DeviceListUpdaterDatabase,
api DeviceListUpdaterAPI, producer KeyChangeProducer,
fedClient fedsenderapi.KeyserverFederationAPI, numWorkers int,
rsAPI rsapi.KeyserverRoomserverAPI, thisServer spec.ServerName,
enableMetrics bool,
) *DeviceListUpdater {
if enableMetrics {
prometheus.MustRegister(deviceListUpdaterBackpressure)
}
return &DeviceListUpdater{
process: process,
userIDToMutex: make(map[string]*sync.Mutex),
@ -343,6 +358,8 @@ func (u *DeviceListUpdater) notifyWorkers(userID string) {
index := int(int64(hash.Sum32()) % int64(len(u.workerChans)))
ch := u.assignChannel(userID)
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(index)}).Inc()
defer deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(index)}).Dec()
u.workerChans[index] <- remoteServer
select {
case <-ch:

9
userapi/internal/device_list_update_test.go
View file

@ -27,6 +27,7 @@ import (
"testing"
"time"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/fclient"
@ -161,7 +162,7 @@ func TestUpdateHavePrevID(t *testing.T) {
}
ap := &mockDeviceListUpdaterAPI{}
producer := &mockKeyChangeProducer{}
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, nil, 1, nil, "localhost")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, nil, 1, nil, "localhost", caching.DisableMetrics)
event := gomatrixserverlib.DeviceListUpdateEvent{
DeviceDisplayName: "Foo Bar",
Deleted: false,
@ -233,7 +234,7 @@ func TestUpdateNoPrevID(t *testing.T) {
`)),
}, nil
})
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 2, nil, "example.test")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 2, nil, "example.test", caching.DisableMetrics)
if err := updater.Start(); err != nil {
t.Fatalf("failed to start updater: %s", err)
}
@ -303,7 +304,7 @@ func TestDebounce(t *testing.T) {
close(incomingFedReq)
return <-fedCh, nil
})
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 1, nil, "localhost")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 1, nil, "localhost", caching.DisableMetrics)
if err := updater.Start(); err != nil {
t.Fatalf("failed to start updater: %s", err)
}
@ -406,7 +407,7 @@ func TestDeviceListUpdater_CleanUp(t *testing.T) {
updater := NewDeviceListUpdater(processCtx, db, nil,
nil, nil,
0, rsAPI, "test")
0, rsAPI, "test", caching.DisableMetrics)
if err := updater.CleanUp(); err != nil {
t.Error(err)
}

3
userapi/userapi.go
View file

@ -18,6 +18,7 @@ import (
"time"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/pushgateway"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
@ -99,7 +100,7 @@ func NewInternalAPI(
FedClient: fedClient,
}
updater := internal.NewDeviceListUpdater(processContext, keyDB, userAPI, keyChangeProducer, fedClient, 8, rsAPI, dendriteCfg.Global.ServerName) // 8 workers TODO: configurable
updater := internal.NewDeviceListUpdater(processContext, keyDB, userAPI, keyChangeProducer, fedClient, 8, rsAPI, dendriteCfg.Global.ServerName, caching.EnableMetrics) // 8 workers TODO: configurable
userAPI.Updater = updater
// Remove users which we don't share a room with anymore
if err := updater.CleanUp(); err != nil {