From df9527fcc0c6a4f07fc6ff6c9d7d8deecf50d1e5 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 24 Jul 2020 12:16:29 +0100 Subject: [PATCH] Flesh out structure for handling device list updates for room membership changes --- internal/consumers.go | 14 +++- syncapi/consumers/keychange.go | 145 ++++++++++++++++++++++++++++++--- syncapi/types/types.go | 4 + 3 files changed, 148 insertions(+), 15 deletions(-) diff --git a/internal/consumers.go b/internal/consumers.go index d7917f235..c000c1719 100644 --- a/internal/consumers.go +++ b/internal/consumers.go @@ -58,11 +58,17 @@ var ErrShutdown = fmt.Errorf("shutdown") // Returns nil once all the goroutines are started. // Returns an error if it can't start consuming for any of the partitions. func (c *ContinualConsumer) Start() error { + _, err := c.StartOffsets() + return err +} + +// StartOffsets is the same as Start but returns the loaded offsets as well. +func (c *ContinualConsumer) StartOffsets() ([]sqlutil.PartitionOffset, error) { offsets := map[int32]int64{} partitions, err := c.Consumer.Partitions(c.Topic) if err != nil { - return err + return nil, err } for _, partition := range partitions { // Default all the offsets to the beginning of the stream. @@ -71,7 +77,7 @@ func (c *ContinualConsumer) Start() error { storedOffsets, err := c.PartitionStore.PartitionOffsets(context.TODO(), c.Topic) if err != nil { - return err + return nil, err } for _, offset := range storedOffsets { // We've already processed events from this partition so advance the offset to where we got to. @@ -87,7 +93,7 @@ func (c *ContinualConsumer) Start() error { for _, p := range partitionConsumers { p.Close() // nolint: errcheck } - return err + return nil, err } partitionConsumers = append(partitionConsumers, pc) } @@ -95,7 +101,7 @@ func (c *ContinualConsumer) Start() error { go c.consumePartition(pc) } - return nil + return storedOffsets, nil } // consumePartition consumes the room events for a single partition of the kafkaesque stream. diff --git a/syncapi/consumers/keychange.go b/syncapi/consumers/keychange.go index 23961452d..8d7e1e1f5 100644 --- a/syncapi/consumers/keychange.go +++ b/syncapi/consumers/keychange.go @@ -17,6 +17,7 @@ package consumers import ( "context" "encoding/json" + "sync" "github.com/Shopify/sarama" currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api" @@ -24,6 +25,7 @@ import ( "github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/keyserver/api" "github.com/matrix-org/dendrite/syncapi/storage" + "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" log "github.com/sirupsen/logrus" ) @@ -35,6 +37,8 @@ type OutputKeyChangeEventConsumer struct { serverName gomatrixserverlib.ServerName // our server name currentStateAPI currentstateAPI.CurrentStateInternalAPI // keyAPI api.KeyInternalAPI + partitionToOffset map[int32]int64 + partitionToOffsetMu sync.Mutex } // NewOutputKeyChangeEventConsumer creates a new OutputKeyChangeEventConsumer. @@ -53,10 +57,12 @@ func NewOutputKeyChangeEventConsumer( } s := &OutputKeyChangeEventConsumer{ - keyChangeConsumer: &consumer, - db: store, - serverName: cfg.Matrix.ServerName, - currentStateAPI: currentStateAPI, + keyChangeConsumer: &consumer, + db: store, + serverName: cfg.Matrix.ServerName, + currentStateAPI: currentStateAPI, + partitionToOffset: make(map[int32]int64), + partitionToOffsetMu: sync.Mutex{}, } consumer.ProcessMessage = s.onMessage @@ -66,10 +72,25 @@ func NewOutputKeyChangeEventConsumer( // Start consuming from the key server func (s *OutputKeyChangeEventConsumer) Start() error { - return s.keyChangeConsumer.Start() + 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 } func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { + defer func() { + s.updateOffset(msg) + }() var output api.DeviceKeys if err := json.Unmarshal(msg.Value, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream @@ -78,18 +99,120 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er } // work out who we need to notify about the new key var queryRes currentstateAPI.QuerySharedUsersResponse - err := s.currentStateAPI.QuerySharedUsers(context.Background(), ¤tstateAPI.QuerySharedUsersRequest{}, &queryRes) + err := s.currentStateAPI.QuerySharedUsers(context.Background(), ¤tstateAPI.QuerySharedUsersRequest{ + UserID: output.UserID, + }, &queryRes) if err != nil { log.WithError(err).Error("syncapi: failed to QuerySharedUsers for key change event from key server") return err } - // TODO: notify users by waking up streams + // TODO: f.e queryRes.UserIDsToCount : notify users by waking up streams return nil } -// Catchup returns a list of user IDs of users who have changed their device keys between the partition|offset given and now. -// Returns the new offset for this partition. -func (s *OutputKeyChangeEventConsumer) Catchup(parition int32, offset int64) (userIDs []string, newOffset int, err error) { - //return s.keyAPI.QueryKeyChangeCatchup(ctx, partition, offset) +// Catchup fills in the given response for the given user ID to bring it up-to-date with device lists. hasNew=true if the response +// was filled in, else false if there are no new device list changes because there is nothing to catch up on. The response MUST +// be already filled in with join/leave information. +func (s *OutputKeyChangeEventConsumer) Catchup( + ctx context.Context, userID string, res *types.Response, tok types.StreamingToken, +) (hasNew bool, err error) { + // Track users who we didn't track before but now do by virtue of sharing a room with them, or not. + newlyJoinedRooms := joinedRooms(res, userID) + newlyLeftRooms := leftRooms(res) + if len(newlyJoinedRooms) > 0 || len(newlyLeftRooms) > 0 { + changed, left, err := s.trackChangedUsers(userID, newlyJoinedRooms, newlyLeftRooms) + if err != nil { + return false, err + } + res.DeviceLists.Changed = changed + res.DeviceLists.Left = left + } + + // TODO: now also track users who we already share rooms with but who have updated their devices between the two tokens return } + +func (s *OutputKeyChangeEventConsumer) OnJoinEvent(ev *gomatrixserverlib.HeaderedEvent) { + // work out who we are now sharing rooms with which we previously were not and notify them about the joining + // users keys: + changed, _, err := s.trackChangedUsers(*ev.StateKey(), []string{ev.RoomID()}, nil) + if err != nil { + log.WithError(err).Error("OnJoinEvent: failed to work out changed users") + return + } + // TODO: f.e changed, wake up stream + for _, userID := range changed { + log.Infof("OnJoinEvent:Notify %s that %s should have device lists tracked", userID, *ev.StateKey()) + } +} + +func (s *OutputKeyChangeEventConsumer) OnLeaveEvent(ev *gomatrixserverlib.HeaderedEvent) { + // work out who we are no longer sharing any rooms with and notify them about the leaving user + _, left, err := s.trackChangedUsers(*ev.StateKey(), nil, []string{ev.RoomID()}) + if err != nil { + log.WithError(err).Error("OnLeaveEvent: failed to work out left users") + return + } + // TODO: f.e left, wake up stream + for _, userID := range left { + log.Infof("OnLeaveEvent:Notify %s that %s should no longer track device lists", userID, *ev.StateKey()) + } + +} + +func (s *OutputKeyChangeEventConsumer) trackChangedUsers(userID string, newlyJoinedRooms, newlyLeftRooms []string) (changed, left []string, err error) { + // process leaves first, then joins afterwards so if we join/leave/join/leave we err on the side of including users. + + // Leave algorithm: + // - Get set of users and number of times they appear in room prior to leave. - QuerySharedUsersRequest with 'IncludeRoomID'. + // - Get users in newly left room. - QueryCurrentState + // - Loop set of users and decrement by 1 for each user in newly left room. + // - If count=0 then they share no more rooms so inform BOTH parties of this via 'left'=[...] in /sync. + + // Join algorithm: + // - Get the set of all joined users prior to joining room - QuerySharedUsersRequest with 'ExcludeRoomID'. + // - Get users in newly joined room - QueryCurrentState + // - Loop set of users in newly joined room, do they appear in the set of users prior to joining? + // - If yes: then they already shared a room in common, do nothing. + // - If no: then they are a brand new user so inform BOTH parties of this via 'changed=[...]' + return +} + +func joinedRooms(res *types.Response, userID string) []string { + var roomIDs []string + for roomID, join := range res.Rooms.Join { + // we would expect to see our join event somewhere if we newly joined the room. + // Normal events get put in the join section so it's not enough to know the room ID is present in 'join'. + newlyJoined := membershipEventPresent(join.State.Events, userID) + if newlyJoined { + roomIDs = append(roomIDs, roomID) + continue + } + newlyJoined = membershipEventPresent(join.Timeline.Events, userID) + if newlyJoined { + roomIDs = append(roomIDs, roomID) + } + } + return roomIDs +} + +func leftRooms(res *types.Response) []string { + roomIDs := make([]string, len(res.Rooms.Leave)) + i := 0 + for roomID := range res.Rooms.Leave { + roomIDs[i] = roomID + i++ + } + return roomIDs +} + +func membershipEventPresent(events []gomatrixserverlib.ClientEvent, userID string) bool { + for _, ev := range events { + // it's enough to know that we have our member event here, don't need to check membership content + // as it's implied by being in the respective section of the sync response. + if ev.Type == gomatrixserverlib.MRoomMember && ev.StateKey != nil && *ev.StateKey == userID { + return true + } + } + return false +} diff --git a/syncapi/types/types.go b/syncapi/types/types.go index 019f2e69b..7dc022811 100644 --- a/syncapi/types/types.go +++ b/syncapi/types/types.go @@ -302,6 +302,10 @@ type Response struct { ToDevice struct { Events []gomatrixserverlib.SendToDeviceEvent `json:"events"` } `json:"to_device"` + DeviceLists struct { + Changed []string `json:"changed,omitempty"` + Left []string `json:"left,omitempty"` + } `json:"device_lists,omitempty"` } // NewResponse creates an empty response with initialised maps.