mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2024-11-26 16:21:55 -06:00
Code Issues Packages Projects Releases Wiki Activity

Use synchronous contexts, limit time to fetch missing events

Browse source
This commit is contained in:
Neil Alexander 2022-01-07 11:57:11 +00:00
parent af34b4abe3
commit 10e1d347e2
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
3 changed files with 29 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
roomserver/internal/input/input.go
View file

@ -126,7 +126,7 @@ func (r *Inputer) InputRoomEvents(
inputRoomEvent := e
inbox, _ := r.workers.LoadOrStore(inputRoomEvent.Event.RoomID(), &phony.Inbox{})
inbox.(*phony.Inbox).Act(nil, func() {
err := r.processRoomEvent(context.TODO(), &inputRoomEvent)
err := r.processRoomEvent(ctx, &inputRoomEvent)
if err != nil {
sentry.CaptureException(err)
} else {
@ -142,6 +142,7 @@ func (r *Inputer) InputRoomEvents(
for i := 0; i < len(request.InputRoomEvents); i++ {
select {
case <-ctx.Done():
response.ErrMsg = context.DeadlineExceeded.Error()
return
case err := <-responses:
if err != nil {

40
roomserver/internal/input/input_events.go
View file

@ -84,7 +84,7 @@ func (r *Inputer) processRoomEvent(
})
if input.Origin == "" {
input.Origin = event.Origin()
// input.Origin = event.Origin()
}
logger.Println("XXX: Processing event")
@ -131,17 +131,18 @@ func (r *Inputer) processRoomEvent(
return fmt.Errorf("r.FSAPI.QueryJoinedHostServerNamesInRoom: %w", err)
}
}
if input.Origin != "" {
serverRes.ServerNames = append([]gomatrixserverlib.ServerName{input.Origin}, serverRes.ServerNames...)
}
// First of all, check that the auth events of the event are known.
// If they aren't then we will ask the federation API for them.
isRejected := false
authEvents := gomatrixserverlib.NewAuthEvents(nil)
knownEvents := map[string]*types.Event{}
logger.Println("Starting to check for missing auth events")
if err = r.checkForMissingAuthEvents(ctx, logger, input.Event, &authEvents, knownEvents, serverRes.ServerNames); err != nil {
return fmt.Errorf("r.checkForMissingAuthEvents: %w", err)
}
logger.Println("Checked for missing auth events")
// Check if the event is allowed by its auth events. If it isn't then
// we consider the event to be "rejected" — it will still be persisted.
@ -172,21 +173,26 @@ func (r *Inputer) processRoomEvent(
}
if input.Kind != api.KindOutlier && len(missingRes.MissingPrevEventIDs) > 0 {
missingState := missingStateReq{
origin: input.Origin,
inputer: r,
queryer: r.Queryer,
db: r.DB,
federation: r.FSAPI,
keys: r.KeyRing,
roomsMu: internal.NewMutexByRoom(),
servers: []gomatrixserverlib.ServerName{input.Origin},
hadEvents: map[string]bool{},
haveEvents: map[string]*gomatrixserverlib.HeaderedEvent{},
}
if err = missingState.processEventWithMissingState(ctx, input.Event.Unwrap(), input.Event.RoomVersion); err != nil {
if len(serverRes.ServerNames) > 0 {
missingState := missingStateReq{
origin: input.Origin,
inputer: r,
queryer: r.Queryer,
db: r.DB,
federation: r.FSAPI,
keys: r.KeyRing,
roomsMu: internal.NewMutexByRoom(),
servers: serverRes.ServerNames,
hadEvents: map[string]bool{},
haveEvents: map[string]*gomatrixserverlib.HeaderedEvent{},
}
if err = missingState.processEventWithMissingState(ctx, input.Event.Unwrap(), input.Event.RoomVersion); err != nil {
isRejected = true
rejectionErr = fmt.Errorf("missingState.processEventWithMissingState: %w", err)
}
} else {
isRejected = true
rejectionErr = fmt.Errorf("missingState.processEventWithMissingState: %w", err)
rejectionErr = fmt.Errorf("missing prev events and no other servers to ask")
}
}

6
roomserver/internal/input/input_missing.go
View file

@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"sync"
"time"
fedapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal"
@ -369,9 +370,10 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve
var missingResp *gomatrixserverlib.RespMissingEvents
for _, server := range t.servers {
logger.Infof("Trying server %q for missing events", server)
var m gomatrixserverlib.RespMissingEvents
if m, err = t.federation.LookupMissingEvents(ctx, server, e.RoomID(), gomatrixserverlib.MissingEvents{
rctx, cancel := context.WithTimeout(ctx, time.Second*30)
defer cancel()
if m, err = t.federation.LookupMissingEvents(rctx, server, e.RoomID(), gomatrixserverlib.MissingEvents{
Limit: 20,
// The latest event IDs that the sender already has. These are skipped when retrieving the previous events of latest_events.
EarliestEvents: latestEvents,