hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-07-29 12:42:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Remodel how device list change IDs are created (#2098)

Browse source 
* Remodel how device list change IDs are created

Previously we made them using the offset Kafka supplied.
We don't run Kafka anymore, so now we make the SQL table assign
the change ID via an AUTOINCREMENTing ID. Redesign the
`keyserver_key_changes` table to have `UNIQUE(user_id)` so we
don't accumulate key changes forevermore, we now have at most 1
row per user which contains the highest change ID.

This needs a SQL migration.

* Ensure we bump the change ID on sqlite

* Actually read the DeviceChangeID not the Offset in synapi

* Add SQL migrations

* Prepare after migration; fixup dendrite-upgrade-test logging

* Use higher version numbers; fix sqlite query to increment better

* Default 0 on postgres

* fixup postgres migration on fresh dendrite instances
This commit is contained in:
kegsay 2022-01-21 09:56:06 +00:00 • committed by GitHub
parent db7d9cba8a
commit 2c581377a5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 336 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

65
syncapi/consumers/keychange.go
View file

@ -17,7 +17,6 @@ package consumers
import (
"context"
"encoding/json"
"sync"
"github.com/Shopify/sarama"
"github.com/getsentry/sentry-go"
@ -34,16 +33,14 @@ import (
// OutputKeyChangeEventConsumer consumes events that originated in the key server.
type OutputKeyChangeEventConsumer struct {
ctx context.Context
keyChangeConsumer *internal.ContinualConsumer
db storage.Database
notifier *notifier.Notifier
stream types.StreamProvider
serverName gomatrixserverlib.ServerName // our server name
rsAPI roomserverAPI.RoomserverInternalAPI
keyAPI api.KeyInternalAPI
partitionToOffset map[int32]int64
partitionToOffsetMu sync.Mutex
ctx context.Context
keyChangeConsumer *internal.ContinualConsumer
db storage.Database
notifier *notifier.Notifier
stream types.StreamProvider
serverName gomatrixserverlib.ServerName // our server name
rsAPI roomserverAPI.RoomserverInternalAPI
keyAPI api.KeyInternalAPI
}
// NewOutputKeyChangeEventConsumer creates a new OutputKeyChangeEventConsumer.
@ -69,16 +66,14 @@ func NewOutputKeyChangeEventConsumer(
}
s := &OutputKeyChangeEventConsumer{
ctx: process.Context(),
keyChangeConsumer: &consumer,
db: store,
serverName: serverName,
keyAPI: keyAPI,
rsAPI: rsAPI,
partitionToOffset: make(map[int32]int64),
partitionToOffsetMu: sync.Mutex{},
notifier: notifier,
stream: stream,
ctx: process.Context(),
keyChangeConsumer: &consumer,
db: store,
serverName: serverName,
keyAPI: keyAPI,
rsAPI: rsAPI,
notifier: notifier,
stream: stream,
}
consumer.ProcessMessage = s.onMessage
@ -88,24 +83,10 @@ func NewOutputKeyChangeEventConsumer(
// Start consuming from the key server
func (s *OutputKeyChangeEventConsumer) Start() error {
offsets, err := s.keyChangeConsumer.StartOffsets()
s.partitionToOffsetMu.Lock()
for _, o := range offsets {
s.partitionToOffset[o.Partition] = o.Offset
}
s.partitionToOffsetMu.Unlock()
return err
}
func (s *OutputKeyChangeEventConsumer) updateOffset(msg *sarama.ConsumerMessage) {
s.partitionToOffsetMu.Lock()
defer s.partitionToOffsetMu.Unlock()
s.partitionToOffset[msg.Partition] = msg.Offset
return s.keyChangeConsumer.Start()
}
func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
defer s.updateOffset(msg)
var m api.DeviceMessage
if err := json.Unmarshal(msg.Value, &m); err != nil {
logrus.WithError(err).Errorf("failed to read device message from key change topic")
@ -118,15 +99,15 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er
}
switch m.Type {
case api.TypeCrossSigningUpdate:
return s.onCrossSigningMessage(m, msg.Offset)
return s.onCrossSigningMessage(m, m.DeviceChangeID)
case api.TypeDeviceKeyUpdate:
fallthrough
default:
return s.onDeviceKeyMessage(m, msg.Offset)
return s.onDeviceKeyMessage(m, m.DeviceChangeID)
}
}
func (s *OutputKeyChangeEventConsumer) onDeviceKeyMessage(m api.DeviceMessage, offset int64) error {
func (s *OutputKeyChangeEventConsumer) onDeviceKeyMessage(m api.DeviceMessage, deviceChangeID int64) error {
if m.DeviceKeys == nil {
return nil
}
@ -143,7 +124,7 @@ func (s *OutputKeyChangeEventConsumer) onDeviceKeyMessage(m api.DeviceMessage, o
}
// make sure we get our own key updates too!
queryRes.UserIDsToCount[output.UserID] = 1
posUpdate := types.StreamPosition(offset)
posUpdate := types.StreamPosition(deviceChangeID)
s.stream.Advance(posUpdate)
for userID := range queryRes.UserIDsToCount {
@ -153,7 +134,7 @@ func (s *OutputKeyChangeEventConsumer) onDeviceKeyMessage(m api.DeviceMessage, o
return nil
}
func (s *OutputKeyChangeEventConsumer) onCrossSigningMessage(m api.DeviceMessage, offset int64) error {
func (s *OutputKeyChangeEventConsumer) onCrossSigningMessage(m api.DeviceMessage, deviceChangeID int64) error {
output := m.CrossSigningKeyUpdate
// work out who we need to notify about the new key
var queryRes roomserverAPI.QuerySharedUsersResponse
@ -167,7 +148,7 @@ func (s *OutputKeyChangeEventConsumer) onCrossSigningMessage(m api.DeviceMessage
}
// make sure we get our own key updates too!
queryRes.UserIDsToCount[output.UserID] = 1
posUpdate := types.StreamPosition(offset)
posUpdate := types.StreamPosition(deviceChangeID)
s.stream.Advance(posUpdate)
for userID := range queryRes.UserIDsToCount {