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

Actually read the DeviceChangeID not the Offset in synapi

Browse source
This commit is contained in:
Kegan Dougal 2022-01-20 16:41:10 +00:00
parent 2884058b01
commit b700594c49
3 changed files with 25 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
keyserver/storage/postgres/key_changes_table.go
View file

@ -43,10 +43,8 @@ const upsertKeyChangeSQL = "" +
" DO UPDATE SET change_id = nextval('keyserver_key_changes_seq')" + " DO UPDATE SET change_id = nextval('keyserver_key_changes_seq')" +
" RETURNING change_id" " RETURNING change_id"
// select the highest offset for each user in the range. The grouping by user gives distinct entries and then we just
// take the max offset value as the latest offset.
const selectKeyChangesSQL = "" + const selectKeyChangesSQL = "" +
"SELECT user_id, MAX(change_id) FROM keyserver_key_changes WHERE change_id > $1 AND change_id <= $2 GROUP BY user_id" "SELECT user_id, change_id FROM keyserver_key_changes WHERE change_id > $1 AND change_id <= $2"
type keyChangesStatements struct { type keyChangesStatements struct {
db *sql.DB db *sql.DB

4
keyserver/storage/sqlite3/key_changes_table.go
View file

@ -44,10 +44,8 @@ const upsertKeyChangeSQL = "" +
" DO UPDATE SET change_id = change_id + 1" + " DO UPDATE SET change_id = change_id + 1" +
" RETURNING change_id" " RETURNING change_id"
// select the highest offset for each user in the range. The grouping by user gives distinct entries and then we just
// take the max offset value as the latest offset.
const selectKeyChangesSQL = "" + const selectKeyChangesSQL = "" +
"SELECT user_id, MAX(change_id) FROM keyserver_key_changes WHERE change_id > $1 AND change_id <= $2 GROUP BY user_id" "SELECT user_id, change_id FROM keyserver_key_changes WHERE change_id > $1 AND change_id <= $2"
type keyChangesStatements struct { type keyChangesStatements struct {
db *sql.DB db *sql.DB

65
syncapi/consumers/keychange.go
View file

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