From a16db1c4085c0079f72615f0c077fa5016c4fe0f Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 7 May 2020 12:42:06 +0100 Subject: [PATCH 01/10] Improve federation sender performance, implement backoff and blacklisting, fix up invites a bit (#1007) * Improve federation sender performance and behaviour, add backoff * Tweaks * Tweaks * Tweaks * Take copies of events before passing to destination queues * Don't accidentally drop queued messages * Don't take copies again * Tidy up a bit * Break out statistics (tracked component-wide), report success and failures from Perform actions * Fix comment, use atomic add * Improve logic a bit, don't block on wakeup, move idle check * Don't retry sucessful invites, don't dispatch sendEvent, sendInvite etc * Dedupe destinations, fix other bug hopefully * Dispatch sends again * Federation sender to ignore invites that are destined locally * Loopback invite events * Remodel a bit with channels * Linter * Only loopback invite event if we know the room * We should tell other resident servers about the invite if we know about the room * Correct invite signing * Fix invite loopback * Check HTTP response codes, push new invites to front of queue * Review comments --- federationsender/consumers/roomserver.go | 21 +- federationsender/federationsender.go | 11 +- federationsender/internal/api.go | 4 + federationsender/internal/perform.go | 8 + federationsender/producers/roomserver.go | 8 +- federationsender/queue/destinationqueue.go | 257 +++++++++++++++------ federationsender/queue/queue.go | 105 ++++----- federationsender/types/statistics.go | 122 ++++++++++ roomserver/internal/input.go | 17 +- roomserver/internal/input_events.go | 63 ++++- 10 files changed, 474 insertions(+), 142 deletions(-) create mode 100644 federationsender/types/statistics.go diff --git a/federationsender/consumers/roomserver.go b/federationsender/consumers/roomserver.go index 67d08b339..901239472 100644 --- a/federationsender/consumers/roomserver.go +++ b/federationsender/consumers/roomserver.go @@ -188,11 +188,30 @@ func (s *OutputRoomEventConsumer) processInvite(oie api.OutputNewInviteEvent) er return nil } + // Ignore invites that don't have state keys - they are invalid. + if oie.Event.StateKey() == nil { + return fmt.Errorf("event %q doesn't have state key", oie.Event.EventID()) + } + + // Don't try to handle events that are actually destined for us. + stateKey := *oie.Event.StateKey() + _, destination, err := gomatrixserverlib.SplitID('@', stateKey) + if err != nil { + log.WithFields(log.Fields{ + "event_id": oie.Event.EventID(), + "state_key": stateKey, + }).Info("failed to split destination from state key") + return nil + } + if s.cfg.Matrix.ServerName == destination { + return nil + } + // Try to extract the room invite state. The roomserver will have stashed // this for us in invite_room_state if it didn't already exist. strippedState := []gomatrixserverlib.InviteV2StrippedState{} if inviteRoomState := gjson.GetBytes(oie.Event.Unsigned(), "invite_room_state"); inviteRoomState.Exists() { - if err := json.Unmarshal([]byte(inviteRoomState.Raw), &strippedState); err != nil { + if err = json.Unmarshal([]byte(inviteRoomState.Raw), &strippedState); err != nil { log.WithError(err).Warn("failed to extract invite_room_state from event unsigned") } } diff --git a/federationsender/federationsender.go b/federationsender/federationsender.go index cf4395527..8e2f256dc 100644 --- a/federationsender/federationsender.go +++ b/federationsender/federationsender.go @@ -24,6 +24,7 @@ import ( "github.com/matrix-org/dendrite/federationsender/producers" "github.com/matrix-org/dendrite/federationsender/queue" "github.com/matrix-org/dendrite/federationsender/storage" + "github.com/matrix-org/dendrite/federationsender/types" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/gomatrixserverlib" "github.com/sirupsen/logrus" @@ -42,9 +43,14 @@ func SetupFederationSenderComponent( logrus.WithError(err).Panic("failed to connect to federation sender db") } - roomserverProducer := producers.NewRoomserverProducer(rsAPI, base.Cfg.Matrix.ServerName) + roomserverProducer := producers.NewRoomserverProducer( + rsAPI, base.Cfg.Matrix.ServerName, base.Cfg.Matrix.KeyID, base.Cfg.Matrix.PrivateKey, + ) - queues := queue.NewOutgoingQueues(base.Cfg.Matrix.ServerName, federation, roomserverProducer) + statistics := &types.Statistics{} + queues := queue.NewOutgoingQueues( + base.Cfg.Matrix.ServerName, federation, roomserverProducer, statistics, + ) rsConsumer := consumers.NewOutputRoomEventConsumer( base.Cfg, base.KafkaConsumer, queues, @@ -63,6 +69,7 @@ func SetupFederationSenderComponent( queryAPI := internal.NewFederationSenderInternalAPI( federationSenderDB, base.Cfg, roomserverProducer, federation, keyRing, + statistics, ) queryAPI.SetupHTTP(http.DefaultServeMux) diff --git a/federationsender/internal/api.go b/federationsender/internal/api.go index 89a1fda40..481795220 100644 --- a/federationsender/internal/api.go +++ b/federationsender/internal/api.go @@ -9,6 +9,7 @@ import ( "github.com/matrix-org/dendrite/federationsender/api" "github.com/matrix-org/dendrite/federationsender/producers" "github.com/matrix-org/dendrite/federationsender/storage" + "github.com/matrix-org/dendrite/federationsender/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" ) @@ -18,6 +19,7 @@ type FederationSenderInternalAPI struct { api.FederationSenderInternalAPI db storage.Database cfg *config.Dendrite + statistics *types.Statistics producer *producers.RoomserverProducer federation *gomatrixserverlib.FederationClient keyRing *gomatrixserverlib.KeyRing @@ -28,6 +30,7 @@ func NewFederationSenderInternalAPI( producer *producers.RoomserverProducer, federation *gomatrixserverlib.FederationClient, keyRing *gomatrixserverlib.KeyRing, + statistics *types.Statistics, ) *FederationSenderInternalAPI { return &FederationSenderInternalAPI{ db: db, @@ -35,6 +38,7 @@ func NewFederationSenderInternalAPI( producer: producer, federation: federation, keyRing: keyRing, + statistics: statistics, } } diff --git a/federationsender/internal/perform.go b/federationsender/internal/perform.go index ff7f821c1..431b2a2d3 100644 --- a/federationsender/internal/perform.go +++ b/federationsender/internal/perform.go @@ -25,10 +25,12 @@ func (r *FederationSenderInternalAPI) PerformDirectoryLookup( request.RoomAlias, ) if err != nil { + r.statistics.ForServer(request.ServerName).Failure() return err } response.RoomID = dir.RoomID response.ServerNames = dir.Servers + r.statistics.ForServer(request.ServerName).Success() return nil } @@ -61,6 +63,7 @@ func (r *FederationSenderInternalAPI) PerformJoin( ) if err != nil { // TODO: Check if the user was not allowed to join the room. + r.statistics.ForServer(serverName).Failure() return fmt.Errorf("r.federation.MakeJoin: %w", err) } @@ -112,6 +115,7 @@ func (r *FederationSenderInternalAPI) PerformJoin( ) if err != nil { logrus.WithError(err).Warnf("r.federation.SendJoin failed") + r.statistics.ForServer(serverName).Failure() continue } @@ -137,6 +141,7 @@ func (r *FederationSenderInternalAPI) PerformJoin( } // We're all good. + r.statistics.ForServer(serverName).Success() return nil } @@ -170,6 +175,7 @@ func (r *FederationSenderInternalAPI) PerformLeave( if err != nil { // TODO: Check if the user was not allowed to leave the room. logrus.WithError(err).Warnf("r.federation.MakeLeave failed") + r.statistics.ForServer(serverName).Failure() continue } @@ -221,9 +227,11 @@ func (r *FederationSenderInternalAPI) PerformLeave( ) if err != nil { logrus.WithError(err).Warnf("r.federation.SendLeave failed") + r.statistics.ForServer(serverName).Failure() continue } + r.statistics.ForServer(serverName).Success() return nil } diff --git a/federationsender/producers/roomserver.go b/federationsender/producers/roomserver.go index 48aeed8cc..76fedf537 100644 --- a/federationsender/producers/roomserver.go +++ b/federationsender/producers/roomserver.go @@ -16,6 +16,7 @@ package producers import ( "context" + "crypto/ed25519" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/gomatrixserverlib" @@ -25,15 +26,20 @@ import ( type RoomserverProducer struct { InputAPI api.RoomserverInternalAPI serverName gomatrixserverlib.ServerName + keyID gomatrixserverlib.KeyID + privateKey ed25519.PrivateKey } // NewRoomserverProducer creates a new RoomserverProducer func NewRoomserverProducer( rsAPI api.RoomserverInternalAPI, serverName gomatrixserverlib.ServerName, + keyID gomatrixserverlib.KeyID, privateKey ed25519.PrivateKey, ) *RoomserverProducer { return &RoomserverProducer{ InputAPI: rsAPI, serverName: serverName, + keyID: keyID, + privateKey: privateKey, } } @@ -43,7 +49,7 @@ func NewRoomserverProducer( func (c *RoomserverProducer) SendInviteResponse( ctx context.Context, res gomatrixserverlib.RespInviteV2, roomVersion gomatrixserverlib.RoomVersion, ) (string, error) { - ev := res.Event.Headered(roomVersion) + ev := res.Event.Sign(string(c.serverName), c.keyID, c.privateKey).Headered(roomVersion) ire := api.InputRoomEvent{ Kind: api.KindNew, Event: ev, diff --git a/federationsender/queue/destinationqueue.go b/federationsender/queue/destinationqueue.go index 89526fcfd..45faa287c 100644 --- a/federationsender/queue/destinationqueue.go +++ b/federationsender/queue/destinationqueue.go @@ -18,12 +18,13 @@ import ( "context" "encoding/json" "fmt" - "sync" "time" "github.com/matrix-org/dendrite/federationsender/producers" + "github.com/matrix-org/dendrite/federationsender/types" + "github.com/matrix-org/gomatrix" "github.com/matrix-org/gomatrixserverlib" - "github.com/matrix-org/util" + "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus" "go.uber.org/atomic" ) @@ -33,92 +34,190 @@ import ( // ensures that only one request is in flight to a given destination // at a time. type destinationQueue struct { - rsProducer *producers.RoomserverProducer - client *gomatrixserverlib.FederationClient - origin gomatrixserverlib.ServerName - destination gomatrixserverlib.ServerName - running atomic.Bool - // The running mutex protects sentCounter, lastTransactionIDs and - // pendingEvents, pendingEDUs. - runningMutex sync.Mutex - sentCounter int - lastTransactionIDs []gomatrixserverlib.TransactionID - pendingEvents []*gomatrixserverlib.HeaderedEvent - pendingEDUs []*gomatrixserverlib.EDU - pendingInvites []*gomatrixserverlib.InviteV2Request + rsProducer *producers.RoomserverProducer // roomserver producer + client *gomatrixserverlib.FederationClient // federation client + origin gomatrixserverlib.ServerName // origin of requests + destination gomatrixserverlib.ServerName // destination of requests + running atomic.Bool // is the queue worker running? + statistics *types.ServerStatistics // statistics about this remote server + incomingPDUs chan *gomatrixserverlib.HeaderedEvent // PDUs to send + incomingEDUs chan *gomatrixserverlib.EDU // EDUs to send + incomingInvites chan *gomatrixserverlib.InviteV2Request // invites to send + lastTransactionIDs []gomatrixserverlib.TransactionID // last transaction ID + pendingPDUs []*gomatrixserverlib.HeaderedEvent // owned by backgroundSend + pendingEDUs []*gomatrixserverlib.EDU // owned by backgroundSend + pendingInvites []*gomatrixserverlib.InviteV2Request // owned by backgroundSend } // Send event adds the event to the pending queue for the destination. // If the queue is empty then it starts a background goroutine to // start sending events to that destination. func (oq *destinationQueue) sendEvent(ev *gomatrixserverlib.HeaderedEvent) { - oq.runningMutex.Lock() - defer oq.runningMutex.Unlock() - oq.pendingEvents = append(oq.pendingEvents, ev) + if oq.statistics.Blacklisted() { + // If the destination is blacklisted then drop the event. + return + } if !oq.running.Load() { go oq.backgroundSend() } + oq.incomingPDUs <- ev } // sendEDU adds the EDU event to the pending queue for the destination. // If the queue is empty then it starts a background goroutine to // start sending events to that destination. -func (oq *destinationQueue) sendEDU(e *gomatrixserverlib.EDU) { - oq.runningMutex.Lock() - defer oq.runningMutex.Unlock() - oq.pendingEDUs = append(oq.pendingEDUs, e) +func (oq *destinationQueue) sendEDU(ev *gomatrixserverlib.EDU) { + if oq.statistics.Blacklisted() { + // If the destination is blacklisted then drop the event. + return + } if !oq.running.Load() { go oq.backgroundSend() } + oq.incomingEDUs <- ev } // sendInvite adds the invite event to the pending queue for the // destination. If the queue is empty then it starts a background // goroutine to start sending events to that destination. func (oq *destinationQueue) sendInvite(ev *gomatrixserverlib.InviteV2Request) { - oq.runningMutex.Lock() - defer oq.runningMutex.Unlock() - oq.pendingInvites = append(oq.pendingInvites, ev) + if oq.statistics.Blacklisted() { + // If the destination is blacklisted then drop the event. + return + } if !oq.running.Load() { go oq.backgroundSend() } + oq.incomingInvites <- ev } // backgroundSend is the worker goroutine for sending events. +// nolint:gocyclo func (oq *destinationQueue) backgroundSend() { - oq.running.Store(true) + // Check if a worker is already running, and if it isn't, then + // mark it as started. + if !oq.running.CAS(false, true) { + return + } defer oq.running.Store(false) for { - transaction, invites := oq.nextTransaction(), oq.nextInvites() - if !transaction && !invites { - // If the queue is empty then stop processing for this destination. - // TODO: Remove this destination from the queue map. + // Wait either for incoming events, or until we hit an + // idle timeout. + select { + case pdu := <-oq.incomingPDUs: + // Ordering of PDUs is important so we add them to the end + // of the queue and they will all be added to transactions + // in order. + oq.pendingPDUs = append(oq.pendingPDUs, pdu) + case edu := <-oq.incomingEDUs: + // Likewise for EDUs, although we should probably not try + // too hard with some EDUs (like typing notifications) after + // a certain amount of time has passed. + // TODO: think about EDU expiry some more + oq.pendingEDUs = append(oq.pendingEDUs, edu) + case invite := <-oq.incomingInvites: + // There's no strict ordering requirement for invites like + // there is for transactions, so we put the invite onto the + // front of the queue. This means that if an invite that is + // stuck failing already, that it won't block our new invite + // from being sent. + oq.pendingInvites = append( + []*gomatrixserverlib.InviteV2Request{invite}, + oq.pendingInvites..., + ) + case <-time.After(time.Second * 30): + // The worker is idle so stop the goroutine. It'll + // get restarted automatically the next time we + // get an event. return } - // TODO: handle retries. - // TODO: blacklist uncooperative servers. + // If we are backing off this server then wait for the + // backoff duration to complete first. + if backoff, duration := oq.statistics.BackoffDuration(); backoff { + <-time.After(duration) + } + + // How many things do we have waiting? + numPDUs := len(oq.pendingPDUs) + numEDUs := len(oq.pendingEDUs) + numInvites := len(oq.pendingInvites) + + // If we have pending PDUs or EDUs then construct a transaction. + if numPDUs > 0 || numEDUs > 0 { + // Try sending the next transaction and see what happens. + transaction, terr := oq.nextTransaction(oq.pendingPDUs, oq.pendingEDUs, oq.statistics.SuccessCount()) + if terr != nil { + // We failed to send the transaction. + if giveUp := oq.statistics.Failure(); giveUp { + // It's been suggested that we should give up because + // the backoff has exceeded a maximum allowable value. + return + } + } else if transaction { + // If we successfully sent the transaction then clear out + // the pending events and EDUs. + oq.statistics.Success() + // Reallocate so that the underlying arrays can be GC'd, as + // opposed to growing forever. + for i := 0; i < numPDUs; i++ { + oq.pendingPDUs[i] = nil + } + for i := 0; i < numEDUs; i++ { + oq.pendingEDUs[i] = nil + } + oq.pendingPDUs = append( + []*gomatrixserverlib.HeaderedEvent{}, + oq.pendingPDUs[numPDUs:]..., + ) + oq.pendingEDUs = append( + []*gomatrixserverlib.EDU{}, + oq.pendingEDUs[numEDUs:]..., + ) + } + } + + // Try sending the next invite and see what happens. + if numInvites > 0 { + sent, ierr := oq.nextInvites(oq.pendingInvites) + if ierr != nil { + // We failed to send the transaction so increase the + // backoff and give it another go shortly. + if giveUp := oq.statistics.Failure(); giveUp { + // It's been suggested that we should give up because + // the backoff has exceeded a maximum allowable value. + return + } + } else if sent > 0 { + // If we successfully sent the invites then clear out + // the pending invites. + oq.statistics.Success() + // Reallocate so that the underlying array can be GC'd, as + // opposed to growing forever. + oq.pendingInvites = append( + []*gomatrixserverlib.InviteV2Request{}, + oq.pendingInvites[sent:]..., + ) + } + } } } // nextTransaction creates a new transaction from the pending event // queue and sends it. Returns true if a transaction was sent or // false otherwise. -func (oq *destinationQueue) nextTransaction() bool { - oq.runningMutex.Lock() - defer oq.runningMutex.Unlock() - - if len(oq.pendingEvents) == 0 && len(oq.pendingEDUs) == 0 { - return false - } - +func (oq *destinationQueue) nextTransaction( + pendingPDUs []*gomatrixserverlib.HeaderedEvent, + pendingEDUs []*gomatrixserverlib.EDU, + sentCounter uint32, +) (bool, error) { t := gomatrixserverlib.Transaction{ PDUs: []json.RawMessage{}, EDUs: []gomatrixserverlib.EDU{}, } now := gomatrixserverlib.AsTimestamp(time.Now()) - t.TransactionID = gomatrixserverlib.TransactionID(fmt.Sprintf("%d-%d", now, oq.sentCounter)) + t.TransactionID = gomatrixserverlib.TransactionID(fmt.Sprintf("%d-%d", now, sentCounter)) t.Origin = oq.origin t.Destination = oq.destination t.OriginServerTS = now @@ -129,44 +228,54 @@ func (oq *destinationQueue) nextTransaction() bool { oq.lastTransactionIDs = []gomatrixserverlib.TransactionID{t.TransactionID} - for _, pdu := range oq.pendingEvents { + for _, pdu := range pendingPDUs { // Append the JSON of the event, since this is a json.RawMessage type in the // gomatrixserverlib.Transaction struct t.PDUs = append(t.PDUs, (*pdu).JSON()) } - oq.pendingEvents = nil - oq.sentCounter += len(t.PDUs) - for _, edu := range oq.pendingEDUs { + for _, edu := range pendingEDUs { t.EDUs = append(t.EDUs, *edu) } - oq.pendingEDUs = nil - oq.sentCounter += len(t.EDUs) - util.GetLogger(context.TODO()).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs)) + logrus.WithField("server_name", oq.destination).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs)) + // TODO: we should check for 500-ish fails vs 400-ish here, + // since we shouldn't queue things indefinitely in response + // to a 400-ish error _, err := oq.client.SendTransaction(context.TODO(), t) - if err != nil { + switch e := err.(type) { + case nil: + // No error was returned so the transaction looks to have + // been successfully sent. + return true, nil + case gomatrix.HTTPError: + // We received a HTTP error back. In this instance we only + // should report an error if + if e.Code >= 400 && e.Code <= 499 { + // We tried but the remote side has sent back a client error. + // It's no use retrying because it will happen again. + return true, nil + } + // Otherwise, report that we failed to send the transaction + // and we will retry again. + return false, err + default: log.WithFields(log.Fields{ "destination": oq.destination, log.ErrorKey: err, }).Info("problem sending transaction") + return false, err } - - return true } // nextInvite takes pending invite events from the queue and sends // them. Returns true if a transaction was sent or false otherwise. -func (oq *destinationQueue) nextInvites() bool { - oq.runningMutex.Lock() - defer oq.runningMutex.Unlock() - - if len(oq.pendingInvites) == 0 { - return false - } - - for _, inviteReq := range oq.pendingInvites { +func (oq *destinationQueue) nextInvites( + pendingInvites []*gomatrixserverlib.InviteV2Request, +) (int, error) { + done := 0 + for _, inviteReq := range pendingInvites { ev, roomVersion := inviteReq.Event(), inviteReq.RoomVersion() log.WithFields(log.Fields{ @@ -180,13 +289,32 @@ func (oq *destinationQueue) nextInvites() bool { oq.destination, *inviteReq, ) - if err != nil { + switch e := err.(type) { + case nil: + done++ + case gomatrix.HTTPError: + log.WithFields(log.Fields{ + "event_id": ev.EventID(), + "state_key": ev.StateKey(), + "destination": oq.destination, + "status_code": e.Code, + }).WithError(err).Error("failed to send invite due to HTTP error") + // Check whether we should do something about the error or + // just accept it as unavoidable. + if e.Code >= 400 && e.Code <= 499 { + // We tried but the remote side has sent back a client error. + // It's no use retrying because it will happen again. + done++ + continue + } + return done, err + default: log.WithFields(log.Fields{ "event_id": ev.EventID(), "state_key": ev.StateKey(), "destination": oq.destination, }).WithError(err).Error("failed to send invite") - continue + return done, err } if _, err = oq.rsProducer.SendInviteResponse( @@ -199,10 +327,9 @@ func (oq *destinationQueue) nextInvites() bool { "state_key": ev.StateKey(), "destination": oq.destination, }).WithError(err).Error("failed to return signed invite to roomserver") + return done, err } } - oq.pendingInvites = nil - - return true + return done, nil } diff --git a/federationsender/queue/queue.go b/federationsender/queue/queue.go index 33abc8fdd..aae6c53a0 100644 --- a/federationsender/queue/queue.go +++ b/federationsender/queue/queue.go @@ -19,18 +19,20 @@ import ( "sync" "github.com/matrix-org/dendrite/federationsender/producers" + "github.com/matrix-org/dendrite/federationsender/types" "github.com/matrix-org/gomatrixserverlib" + "github.com/matrix-org/util" log "github.com/sirupsen/logrus" ) // OutgoingQueues is a collection of queues for sending transactions to other // matrix servers type OutgoingQueues struct { - rsProducer *producers.RoomserverProducer - origin gomatrixserverlib.ServerName - client *gomatrixserverlib.FederationClient - // The queuesMutex protects queues - queuesMutex sync.Mutex + rsProducer *producers.RoomserverProducer + origin gomatrixserverlib.ServerName + client *gomatrixserverlib.FederationClient + statistics *types.Statistics + queuesMutex sync.Mutex // protects the below queues map[gomatrixserverlib.ServerName]*destinationQueue } @@ -39,15 +41,37 @@ func NewOutgoingQueues( origin gomatrixserverlib.ServerName, client *gomatrixserverlib.FederationClient, rsProducer *producers.RoomserverProducer, + statistics *types.Statistics, ) *OutgoingQueues { return &OutgoingQueues{ rsProducer: rsProducer, origin: origin, client: client, + statistics: statistics, queues: map[gomatrixserverlib.ServerName]*destinationQueue{}, } } +func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *destinationQueue { + oqs.queuesMutex.Lock() + defer oqs.queuesMutex.Unlock() + oq := oqs.queues[destination] + if oq == nil { + oq = &destinationQueue{ + rsProducer: oqs.rsProducer, + origin: oqs.origin, + destination: destination, + client: oqs.client, + statistics: oqs.statistics.ForServer(destination), + incomingPDUs: make(chan *gomatrixserverlib.HeaderedEvent, 128), + incomingEDUs: make(chan *gomatrixserverlib.EDU, 128), + incomingInvites: make(chan *gomatrixserverlib.InviteV2Request, 128), + } + oqs.queues[destination] = oq + } + return oq +} + // SendEvent sends an event to the destinations func (oqs *OutgoingQueues) SendEvent( ev *gomatrixserverlib.HeaderedEvent, origin gomatrixserverlib.ServerName, @@ -62,27 +86,14 @@ func (oqs *OutgoingQueues) SendEvent( } // Remove our own server from the list of destinations. - destinations = filterDestinations(oqs.origin, destinations) + destinations = filterAndDedupeDests(oqs.origin, destinations) log.WithFields(log.Fields{ "destinations": destinations, "event": ev.EventID(), }).Info("Sending event") - oqs.queuesMutex.Lock() - defer oqs.queuesMutex.Unlock() for _, destination := range destinations { - oq := oqs.queues[destination] - if oq == nil { - oq = &destinationQueue{ - rsProducer: oqs.rsProducer, - origin: oqs.origin, - destination: destination, - client: oqs.client, - } - oqs.queues[destination] = oq - } - - oq.sendEvent(ev) + oqs.getQueue(destination).sendEvent(ev) } return nil @@ -111,23 +122,11 @@ func (oqs *OutgoingQueues) SendInvite( } log.WithFields(log.Fields{ - "event_id": ev.EventID(), + "event_id": ev.EventID(), + "server_name": destination, }).Info("Sending invite") - oqs.queuesMutex.Lock() - defer oqs.queuesMutex.Unlock() - oq := oqs.queues[destination] - if oq == nil { - oq = &destinationQueue{ - rsProducer: oqs.rsProducer, - origin: oqs.origin, - destination: destination, - client: oqs.client, - } - oqs.queues[destination] = oq - } - - oq.sendInvite(inviteReq) + oqs.getQueue(destination).sendInvite(inviteReq) return nil } @@ -146,7 +145,7 @@ func (oqs *OutgoingQueues) SendEDU( } // Remove our own server from the list of destinations. - destinations = filterDestinations(oqs.origin, destinations) + destinations = filterAndDedupeDests(oqs.origin, destinations) if len(destinations) > 0 { log.WithFields(log.Fields{ @@ -154,35 +153,27 @@ func (oqs *OutgoingQueues) SendEDU( }).Info("Sending EDU event") } - oqs.queuesMutex.Lock() - defer oqs.queuesMutex.Unlock() for _, destination := range destinations { - oq := oqs.queues[destination] - if oq == nil { - oq = &destinationQueue{ - rsProducer: oqs.rsProducer, - origin: oqs.origin, - destination: destination, - client: oqs.client, - } - oqs.queues[destination] = oq - } - - oq.sendEDU(e) + oqs.getQueue(destination).sendEDU(e) } return nil } -// filterDestinations removes our own server from the list of destinations. -// Otherwise we could end up trying to talk to ourselves. -func filterDestinations(origin gomatrixserverlib.ServerName, destinations []gomatrixserverlib.ServerName) []gomatrixserverlib.ServerName { - var result []gomatrixserverlib.ServerName - for _, destination := range destinations { - if destination == origin { +// filterAndDedupeDests removes our own server from the list of destinations +// and deduplicates any servers in the list that may appear more than once. +func filterAndDedupeDests(origin gomatrixserverlib.ServerName, destinations []gomatrixserverlib.ServerName) ( + result []gomatrixserverlib.ServerName, +) { + strs := make([]string, len(destinations)) + for i, d := range destinations { + strs[i] = string(d) + } + for _, destination := range util.UniqueStrings(strs) { + if gomatrixserverlib.ServerName(destination) == origin { continue } - result = append(result, destination) + result = append(result, gomatrixserverlib.ServerName(destination)) } return result } diff --git a/federationsender/types/statistics.go b/federationsender/types/statistics.go new file mode 100644 index 000000000..63f82756e --- /dev/null +++ b/federationsender/types/statistics.go @@ -0,0 +1,122 @@ +package types + +import ( + "math" + "sync" + "time" + + "github.com/matrix-org/gomatrixserverlib" + "go.uber.org/atomic" +) + +const ( + // How many times should we tolerate consecutive failures before we + // just blacklist the host altogether? Bear in mind that the backoff + // is exponential, so the max time here to attempt is 2**failures. + FailuresUntilBlacklist = 16 // 16 equates to roughly 18 hours. +) + +// Statistics contains information about all of the remote federated +// hosts that we have interacted with. It is basically a threadsafe +// wrapper. +type Statistics struct { + servers map[gomatrixserverlib.ServerName]*ServerStatistics + mutex sync.RWMutex +} + +// ForServer returns server statistics for the given server name. If it +// does not exist, it will create empty statistics and return those. +func (s *Statistics) ForServer(serverName gomatrixserverlib.ServerName) *ServerStatistics { + // If the map hasn't been initialised yet then do that. + if s.servers == nil { + s.mutex.Lock() + s.servers = make(map[gomatrixserverlib.ServerName]*ServerStatistics) + s.mutex.Unlock() + } + // Look up if we have statistics for this server already. + s.mutex.RLock() + server, found := s.servers[serverName] + s.mutex.RUnlock() + // If we don't, then make one. + if !found { + s.mutex.Lock() + server = &ServerStatistics{} + s.servers[serverName] = server + s.mutex.Unlock() + } + return server +} + +// ServerStatistics contains information about our interactions with a +// remote federated host, e.g. how many times we were successful, how +// many times we failed etc. It also manages the backoff time and black- +// listing a remote host if it remains uncooperative. +type ServerStatistics struct { + blacklisted atomic.Bool // is the remote side dead? + backoffUntil atomic.Value // time.Time to wait until before sending requests + failCounter atomic.Uint32 // how many times have we failed? + successCounter atomic.Uint32 // how many times have we succeeded? +} + +// Success updates the server statistics with a new successful +// attempt, which increases the sent counter and resets the idle and +// failure counters. If a host was blacklisted at this point then +// we will unblacklist it. +func (s *ServerStatistics) Success() { + s.successCounter.Add(1) + s.failCounter.Store(0) + s.blacklisted.Store(false) +} + +// Failure marks a failure and works out when to backoff until. It +// returns true if the worker should give up altogether because of +// too many consecutive failures. At this point the host is marked +// as blacklisted. +func (s *ServerStatistics) Failure() bool { + // Increase the fail counter. + failCounter := s.failCounter.Add(1) + + // Check that we haven't failed more times than is acceptable. + if failCounter >= FailuresUntilBlacklist { + // We've exceeded the maximum amount of times we're willing + // to back off, which is probably in the region of hours by + // now. Mark the host as blacklisted and tell the caller to + // give up. + s.blacklisted.Store(true) + return true + } + + // We're still under the threshold so work out the exponential + // backoff based on how many times we have failed already. The + // worker goroutine will wait until this time before processing + // anything from the queue. + backoffSeconds := time.Second * time.Duration(math.Exp2(float64(failCounter))) + s.backoffUntil.Store( + time.Now().Add(backoffSeconds), + ) + return false +} + +// BackoffDuration returns both a bool stating whether to wait, +// and then if true, a duration to wait for. +func (s *ServerStatistics) BackoffDuration() (bool, time.Duration) { + backoff, until := false, time.Second + if b, ok := s.backoffUntil.Load().(time.Time); ok { + if b.After(time.Now()) { + backoff, until = true, time.Until(b) + } + } + return backoff, until +} + +// Blacklisted returns true if the server is blacklisted and false +// otherwise. +func (s *ServerStatistics) Blacklisted() bool { + return s.blacklisted.Load() +} + +// SuccessCount returns the number of successful requests. This is +// usually useful in constructing transaction IDs. +func (s *ServerStatistics) SuccessCount() uint32 { + return s.successCounter.Load() +} diff --git a/roomserver/internal/input.go b/roomserver/internal/input.go index 19ebea660..ab3d7516b 100644 --- a/roomserver/internal/input.go +++ b/roomserver/internal/input.go @@ -58,15 +58,22 @@ func (r *RoomserverInternalAPI) InputRoomEvents( // We lock as processRoomEvent can only be called once at a time r.mutex.Lock() defer r.mutex.Unlock() + for i := range request.InputInviteEvents { + var loopback *api.InputRoomEvent + if loopback, err = processInviteEvent(ctx, r.DB, r, request.InputInviteEvents[i]); err != nil { + return err + } + // The processInviteEvent function can optionally return a + // loopback room event containing the invite, for local invites. + // If it does, we should process it with the room events below. + if loopback != nil { + request.InputRoomEvents = append(request.InputRoomEvents, *loopback) + } + } for i := range request.InputRoomEvents { if response.EventID, err = processRoomEvent(ctx, r.DB, r, request.InputRoomEvents[i]); err != nil { return err } } - for i := range request.InputInviteEvents { - if err = processInviteEvent(ctx, r.DB, r, request.InputInviteEvents[i]); err != nil { - return err - } - } return nil } diff --git a/roomserver/internal/input_events.go b/roomserver/internal/input_events.go index 6da63716c..b17076efe 100644 --- a/roomserver/internal/input_events.go +++ b/roomserver/internal/input_events.go @@ -18,6 +18,7 @@ package internal import ( "context" + "errors" "fmt" "github.com/matrix-org/dendrite/common" @@ -132,11 +133,11 @@ func calculateAndSetState( func processInviteEvent( ctx context.Context, db storage.Database, - ow OutputRoomEventWriter, + ow *RoomserverInternalAPI, input api.InputInviteEvent, -) (err error) { +) (*api.InputRoomEvent, error) { if input.Event.StateKey() == nil { - return fmt.Errorf("invite must be a state event") + return nil, fmt.Errorf("invite must be a state event") } roomID := input.Event.RoomID() @@ -151,7 +152,7 @@ func processInviteEvent( updater, err := db.MembershipUpdater(ctx, roomID, targetUserID, input.RoomVersion) if err != nil { - return err + return nil, err } succeeded := false defer func() { @@ -189,17 +190,27 @@ func processInviteEvent( // For now we will implement option 2. Since in the abesence of a retry // mechanism it will be equivalent to option 1, and we don't have a // signalling mechanism to implement option 3. - return nil + return nil, nil + } + + // Normally, with a federated invite, the federation sender would do + // the /v2/invite request (in which the remote server signs the invite) + // and then the signed event gets sent back to the roomserver as an input + // event. When the invite is local, we don't interact with the federation + // sender therefore we need to generate the loopback invite event for + // the room ourselves. + loopback, err := localInviteLoopback(ow, input) + if err != nil { + return nil, err } event := input.Event.Unwrap() - if len(input.InviteRoomState) > 0 { // If we were supplied with some invite room state already (which is // most likely to be if the event came in over federation) then use // that. if err = event.SetUnsignedField("invite_room_state", input.InviteRoomState); err != nil { - return err + return nil, err } } else { // There's no invite room state, so let's have a go at building it @@ -208,22 +219,52 @@ func processInviteEvent( // the invite room state, if we don't then we just fail quietly. if irs, ierr := buildInviteStrippedState(ctx, db, input); ierr == nil { if err = event.SetUnsignedField("invite_room_state", irs); err != nil { - return err + return nil, err } } } outputUpdates, err := updateToInviteMembership(updater, &event, nil, input.Event.RoomVersion) if err != nil { - return err + return nil, err } if err = ow.WriteOutputEvents(roomID, outputUpdates); err != nil { - return err + return nil, err } succeeded = true - return nil + return loopback, nil +} + +func localInviteLoopback( + ow *RoomserverInternalAPI, + input api.InputInviteEvent, +) (ire *api.InputRoomEvent, err error) { + if input.Event.StateKey() == nil { + return nil, errors.New("no state key on invite event") + } + ourServerName := string(ow.Cfg.Matrix.ServerName) + _, theirServerName, err := gomatrixserverlib.SplitID('@', *input.Event.StateKey()) + if err != nil { + return nil, err + } + // Check if the invite originated locally and is destined locally. + if input.Event.Origin() == ow.Cfg.Matrix.ServerName && string(theirServerName) == ourServerName { + rsEvent := input.Event.Sign( + ourServerName, + ow.Cfg.Matrix.KeyID, + ow.Cfg.Matrix.PrivateKey, + ).Headered(input.RoomVersion) + ire = &api.InputRoomEvent{ + Kind: api.KindNew, + Event: rsEvent, + AuthEventIDs: rsEvent.AuthEventIDs(), + SendAsServer: ourServerName, + TransactionID: nil, + } + } + return ire, nil } func buildInviteStrippedState( From c8e11dfe53a97ac2e207c893f3f21f1216d86343 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 7 May 2020 16:46:11 +0100 Subject: [PATCH 02/10] Direct messages (#1012) * Initial DM support, include invite event in stripped state for regular invites * Update go.mod, go.sum, test list --- are-we-synapse-yet.list | 1 + clientapi/routing/createroom.go | 45 +++++++++++++++++++++++++++++ clientapi/routing/membership.go | 6 ++-- go.mod | 2 +- go.sum | 4 +-- roomserver/internal/input_events.go | 1 + 6 files changed, 54 insertions(+), 5 deletions(-) diff --git a/are-we-synapse-yet.list b/are-we-synapse-yet.list index 71b05d1c4..cb90d6282 100644 --- a/are-we-synapse-yet.list +++ b/are-we-synapse-yet.list @@ -833,3 +833,4 @@ gst Guest user can call /events on another world_readable room (SYN-606) gst Real user can call /events on another world_readable room (SYN-606) gst Events come down the correct room pub Asking for a remote rooms list, but supplying the local server's name, returns the local rooms list +std Can send a to-device message to two users which both receive it using /sync \ No newline at end of file diff --git a/clientapi/routing/createroom.go b/clientapi/routing/createroom.go index 28e2b1514..43a16945d 100644 --- a/clientapi/routing/createroom.go +++ b/clientapi/routing/createroom.go @@ -30,6 +30,7 @@ import ( "github.com/matrix-org/dendrite/clientapi/httputil" "github.com/matrix-org/dendrite/clientapi/jsonerror" "github.com/matrix-org/dendrite/clientapi/producers" + "github.com/matrix-org/dendrite/clientapi/threepid" "github.com/matrix-org/dendrite/common" "github.com/matrix-org/dendrite/common/config" "github.com/matrix-org/gomatrixserverlib" @@ -351,6 +352,50 @@ func createRoom( } } + // If this is a direct message then we should invite the participants. + for _, invitee := range r.Invite { + // Build the membership request. + body := threepid.MembershipRequest{ + UserID: invitee, + } + // Build the invite event. + inviteEvent, err := buildMembershipEvent( + req.Context(), body, accountDB, device, gomatrixserverlib.Invite, + roomID, true, cfg, evTime, rsAPI, asAPI, + ) + if err != nil { + util.GetLogger(req.Context()).WithError(err).Error("buildMembershipEvent failed") + continue + } + // Build some stripped state for the invite. + candidates := append(gomatrixserverlib.UnwrapEventHeaders(builtEvents), *inviteEvent) + var strippedState []gomatrixserverlib.InviteV2StrippedState + for _, event := range candidates { + switch event.Type() { + // TODO: case gomatrixserverlib.MRoomEncryption: + // fallthrough + case gomatrixserverlib.MRoomMember: + fallthrough + case gomatrixserverlib.MRoomJoinRules: + strippedState = append( + strippedState, + gomatrixserverlib.NewInviteV2StrippedState(&event), + ) + } + } + // Send the invite event to the roomserver. + if err = producer.SendInvite( + req.Context(), + inviteEvent.Headered(roomVersion), + strippedState, // invite room state + cfg.Matrix.ServerName, // send as server + nil, // transaction ID + ); err != nil { + util.GetLogger(req.Context()).WithError(err).Error("producer.SendEvents failed") + return jsonerror.InternalServerError() + } + } + response := createRoomResponse{ RoomID: roomID, RoomAlias: roomAlias, diff --git a/clientapi/routing/membership.go b/clientapi/routing/membership.go index 9030f9f7e..0a56eec57 100644 --- a/clientapi/routing/membership.go +++ b/clientapi/routing/membership.go @@ -89,7 +89,8 @@ func SendMembership( } event, err := buildMembershipEvent( - req.Context(), body, accountDB, device, membership, roomID, cfg, evTime, rsAPI, asAPI, + req.Context(), body, accountDB, device, membership, + roomID, false, cfg, evTime, rsAPI, asAPI, ) if err == errMissingUserID { return util.JSONResponse{ @@ -151,7 +152,7 @@ func buildMembershipEvent( ctx context.Context, body threepid.MembershipRequest, accountDB accounts.Database, device *authtypes.Device, - membership, roomID string, + membership, roomID string, isDirect bool, cfg *config.Dendrite, evTime time.Time, rsAPI roomserverAPI.RoomserverInternalAPI, asAPI appserviceAPI.AppServiceQueryAPI, ) (*gomatrixserverlib.Event, error) { @@ -182,6 +183,7 @@ func buildMembershipEvent( DisplayName: profile.DisplayName, AvatarURL: profile.AvatarURL, Reason: reason, + IsDirect: isDirect, } if err = builder.SetContent(content); err != nil { diff --git a/go.mod b/go.mod index fd1bb3d70..3da557325 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 - github.com/matrix-org/gomatrixserverlib v0.0.0-20200505092542-ef8abbde3f6b + github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 github.com/mattn/go-sqlite3 v2.0.2+incompatible diff --git a/go.sum b/go.sum index 7ab035364..871648069 100644 --- a/go.sum +++ b/go.sum @@ -367,8 +367,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200505092542-ef8abbde3f6b h1:gxLun/noFJ7DplX7rqT8E4v4NkeDJ45tqW7LXC6k4C4= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200505092542-ef8abbde3f6b/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea h1:1qfbSjg3PwULY68AVRdZ3QIJoccNMbre0mSR7m7mqI4= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A= github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y= diff --git a/roomserver/internal/input_events.go b/roomserver/internal/input_events.go index b17076efe..a0bfaa2e3 100644 --- a/roomserver/internal/input_events.go +++ b/roomserver/internal/input_events.go @@ -308,6 +308,7 @@ func buildInviteStrippedState( inviteState := []gomatrixserverlib.InviteV2StrippedState{ gomatrixserverlib.NewInviteV2StrippedState(&input.Event.Event), } + stateEvents = append(stateEvents, types.Event{Event: input.Event.Unwrap()}) for _, event := range stateEvents { inviteState = append(inviteState, gomatrixserverlib.NewInviteV2StrippedState(&event.Event)) } From 17d27331a3329c10f1aa38a409f451402fb770aa Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 7 May 2020 17:14:32 +0100 Subject: [PATCH 03/10] Fix 'input to Unique() must be sorted' panic --- federationsender/internal/perform.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/federationsender/internal/perform.go b/federationsender/internal/perform.go index 431b2a2d3..7c5fa73fc 100644 --- a/federationsender/internal/perform.go +++ b/federationsender/internal/perform.go @@ -47,7 +47,7 @@ func (r *FederationSenderInternalAPI) PerformJoin( } // Deduplicate the server names we were provided. - util.Unique(request.ServerNames) + util.SortAndUnique(request.ServerNames) // Try each server that we were provided until we land on one that // successfully completes the make-join send-join dance. @@ -159,7 +159,7 @@ func (r *FederationSenderInternalAPI) PerformLeave( response *api.PerformLeaveResponse, ) (err error) { // Deduplicate the server names we were provided. - util.Unique(request.ServerNames) + util.SortAndUnique(request.ServerNames) // Try each server that we were provided until we land on one that // successfully completes the make-leave send-leave dance. From d6e18a33ce1dd85c922165b89aaf680fd5de8535 Mon Sep 17 00:00:00 2001 From: Will Hunt Date: Fri, 8 May 2020 12:00:32 +0100 Subject: [PATCH 04/10] Add registration_disabled to dendrite-config.yaml (#1013) Missed this when first setting up my instance, and it feels important enough that it should be part of the sample config --- dendrite-config.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/dendrite-config.yaml b/dendrite-config.yaml index 8c8fba390..536b0f42b 100644 --- a/dendrite-config.yaml +++ b/dendrite-config.yaml @@ -25,6 +25,8 @@ matrix: # public_key: Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw # - key_id: ed25519:a_RXGa # public_key: l8Hft5qXKn1vfHrg3p4+W8gELQVo8N13JkluMfmn2sQ + # Disables new users from registering (except via shared secrets) + registration_disabled: false # The media repository config media: From 4fd97df2c5ac86bf908432d29259bdda5912b89b Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 11:01:24 +0100 Subject: [PATCH 05/10] Don't return 500s from media API download requests --- mediaapi/routing/download.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/mediaapi/routing/download.go b/mediaapi/routing/download.go index 8544bd64f..9feca90e9 100644 --- a/mediaapi/routing/download.go +++ b/mediaapi/routing/download.go @@ -118,7 +118,10 @@ func Download( ) if err != nil { // TODO: Handle the fact we might have started writing the response - dReq.jsonErrorResponse(w, util.ErrorResponse(err)) + dReq.jsonErrorResponse(w, util.JSONResponse{ + Code: http.StatusNotFound, + JSON: jsonerror.NotFound("Failed to download: " + err.Error()), + }) return } @@ -138,7 +141,7 @@ func (r *downloadRequest) jsonErrorResponse(w http.ResponseWriter, res util.JSON if err != nil { r.Logger.WithError(err).Error("Failed to marshal JSONResponse") // this should never fail to be marshalled so drop err to the floor - res = util.MessageResponse(http.StatusInternalServerError, "Internal Server Error") + res = util.MessageResponse(http.StatusNotFound, "Download request failed: "+err.Error()) resBytes, _ = json.Marshal(res.JSON) } From 6e643860b12dbcf910fb6c7ba6d27f220c9fa594 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 11:50:29 +0100 Subject: [PATCH 06/10] Update sytest-whitelist --- sytest-whitelist | 1 + 1 file changed, 1 insertion(+) diff --git a/sytest-whitelist b/sytest-whitelist index c957021cb..6e1b4f631 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -264,3 +264,4 @@ User can invite local user to room with version 5 remote user can join room with version 5 User can invite remote user to room with version 5 Remote user can backfill in a room with version 5 +Alternative server names do not cause a routing loop From 615de25347bdaf5bfa1079c004757e5e1558d9cf Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 16:02:23 +0100 Subject: [PATCH 07/10] Update gomatrixserverlib for more memory-efficient state res v2 --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 3da557325..cb912be3b 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 - github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea + github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366 github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 github.com/mattn/go-sqlite3 v2.0.2+incompatible diff --git a/go.sum b/go.sum index 871648069..a6f160052 100644 --- a/go.sum +++ b/go.sum @@ -367,8 +367,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea h1:1qfbSjg3PwULY68AVRdZ3QIJoccNMbre0mSR7m7mqI4= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200507150553-025991c971ea/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366 h1:LbtCldIuE/kEgGnzKm6xebnZDfDie1vQA0mDq7ReyM4= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A= github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y= From 99e0a7dff27501df482ba929f53637f4a96c78d6 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 16:43:50 +0100 Subject: [PATCH 08/10] Update gomatrixserverlib for even more memory-efficient state res v2 --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index cb912be3b..8e6d63cd5 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 - github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366 + github.com/matrix-org/gomatrixserverlib v0.0.0-20200511154227-5cc71d36632b github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 github.com/mattn/go-sqlite3 v2.0.2+incompatible diff --git a/go.sum b/go.sum index a6f160052..53aefed37 100644 --- a/go.sum +++ b/go.sum @@ -367,8 +367,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg= github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366 h1:LbtCldIuE/kEgGnzKm6xebnZDfDie1vQA0mDq7ReyM4= -github.com/matrix-org/gomatrixserverlib v0.0.0-20200511150133-f4a8869b5366/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200511154227-5cc71d36632b h1:nAmSc1KvQOumoRTz/LD68KyrB6Q5/6q7CmQ5Bswc2nM= +github.com/matrix-org/gomatrixserverlib v0.0.0-20200511154227-5cc71d36632b/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk= github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A= github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y= From 0c892d59fa5846097647d08244059de4f73e39a6 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 18:21:25 +0100 Subject: [PATCH 09/10] Prevent panic in membership updater (#1021) --- roomserver/internal/input_membership.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/roomserver/internal/input_membership.go b/roomserver/internal/input_membership.go index cba75b4fc..19b7d8055 100644 --- a/roomserver/internal/input_membership.go +++ b/roomserver/internal/input_membership.go @@ -16,6 +16,7 @@ package internal import ( "context" + "errors" "fmt" "github.com/matrix-org/dendrite/roomserver/api" @@ -106,6 +107,13 @@ func updateMembership( return updates, nil } + if add == nil { + // This shouldn't happen. Returning an error here is better than panicking + // in the membership updater functions later on. + // TODO: Why does this happen to begin with? + return updates, errors.New("add should not be nil") + } + mu, err := updater.MembershipUpdater(targetUserNID) if err != nil { return nil, err From 32624697fd2d74d4a6e23549b647d323b210fb2a Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 11 May 2020 18:21:39 +0100 Subject: [PATCH 10/10] Add PPROFLISTEN (#1019) * Add PPROFLISTEN env var * Direct logging to more useful places * Space --- cmd/dendrite-room-server/main.go | 2 -- common/basecomponent/base.go | 3 +++ common/log.go | 12 ++++++++++++ 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/cmd/dendrite-room-server/main.go b/cmd/dendrite-room-server/main.go index 41149ad90..172468448 100644 --- a/cmd/dendrite-room-server/main.go +++ b/cmd/dendrite-room-server/main.go @@ -15,8 +15,6 @@ package main import ( - _ "net/http/pprof" - "github.com/matrix-org/dendrite/common/basecomponent" "github.com/matrix-org/dendrite/common/keydb" "github.com/matrix-org/dendrite/roomserver" diff --git a/common/basecomponent/base.go b/common/basecomponent/base.go index a7e6736a6..cb04a308e 100644 --- a/common/basecomponent/base.go +++ b/common/basecomponent/base.go @@ -42,6 +42,8 @@ import ( federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" "github.com/sirupsen/logrus" + + _ "net/http/pprof" ) // BaseDendrite is a base for creating new instances of dendrite. It parses @@ -71,6 +73,7 @@ const HTTPClientTimeout = time.Second * 30 func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite { common.SetupStdLogging() common.SetupHookLogging(cfg.Logging, componentName) + common.SetupPprof() closer, err := cfg.SetupTracing("Dendrite" + componentName) if err != nil { diff --git a/common/log.go b/common/log.go index 11339ada4..60e969650 100644 --- a/common/log.go +++ b/common/log.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "io" + "net/http" "os" "path" "path/filepath" @@ -79,6 +80,17 @@ func callerPrettyfier(f *runtime.Frame) (string, string) { return funcname, filename } +// SetupPprof starts a pprof listener. We use the DefaultServeMux here because it is +// simplest, and it gives us the freedom to run pprof on a separate port. +func SetupPprof() { + if hostPort := os.Getenv("PPROFLISTEN"); hostPort != "" { + logrus.Warn("Starting pprof on ", hostPort) + go func() { + logrus.WithError(http.ListenAndServe(hostPort, nil)).Error("Failed to setup pprof listener") + }() + } +} + // SetupStdLogging configures the logging format to standard output. Typically, it is called when the config is not yet loaded. func SetupStdLogging() { logrus.SetReportCaller(true)