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

document the peek send stream race better

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-23 00:11:58 +01:00
parent 3202c7e76f
commit 0ab4bc9e8e
2 changed files with 9 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
federationsender/consumers/roomserver.go
View file

@ -118,7 +118,15 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
// processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any) // processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)
// causing the federationsender to start sending messages to the peeking server // causing the federationsender to start sending messages to the peeking server
func (s *OutputRoomEventConsumer) processInboundPeek(orp api.OutputNewInboundPeek) error { func (s *OutputRoomEventConsumer) processInboundPeek(orp api.OutputNewInboundPeek) error {
// FIXME: do something with orp.LatestEventID to prevent races
// FIXME: there's a race here - we should start /sending new peeked events
// atomically after the orp.LatestEventID to ensure there are no gaps between
// the peek beginning and the send stream beginning.
//
// We probably need to track orp.LatestEventID on the inbound peek, but it's
// unclear how we then use that to prevent the race when we start the send
// stream.
return s.db.AddInboundPeek(context.TODO(), orp.ServerName, orp.RoomID, orp.PeekID, orp.RenewalInterval) return s.db.AddInboundPeek(context.TODO(), orp.ServerName, orp.RoomID, orp.PeekID, orp.RenewalInterval)
} }

4
roomserver/internal/perform/perform_inbound_peek.go
View file

@ -106,10 +106,6 @@ func (r *InboundPeeker) PerformInboundPeek(
response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion)) response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
} }
// FIXME: there's a race here - we really should be atomically telling the
// federationsender to start sending peek events alongside having captured
// the current state, but it's unclear if/how we can do that.
err = r.Inputer.WriteOutputEvents(request.RoomID, []api.OutputEvent{ err = r.Inputer.WriteOutputEvents(request.RoomID, []api.OutputEvent{
{ {
Type: api.OutputTypeNewInboundPeek, Type: api.OutputTypeNewInboundPeek,