Fix bugs that could wedge rooms (#2154)

* Don't flake so badly for rejected events

* Moar

* Fix panic

* Don't count rejected events as missing

* Don't treat rejected events without state as missing

* Revert "Don't count rejected events as missing"

This reverts commit 4b6139b62eb91ba059b47415b0275964b37d9b43.

* Missing events should be KindOld

* If we have state, use it, regardless of memberships which could be stale now

* Fetch missing state for KindOld too

* Tweak the condition again

* Clean up a bit

* Use room updater to get latest events in a race-free way

* Return the correct error

* Improve errors
This commit is contained in:
Neil Alexander 2022-02-07 19:10:01 +00:00 committed by GitHub
parent 908d881a6e
commit a572f4db03
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 33 additions and 31 deletions

View file

@ -297,7 +297,10 @@ func (r *Inputer) processRoomEvent(
"soft_fail": softfail, "soft_fail": softfail,
"missing_prev": missingPrev, "missing_prev": missingPrev,
}).Warn("Stored rejected event") }).Warn("Stored rejected event")
return commitTransaction, rejectionErr if rejectionErr != nil {
return commitTransaction, types.RejectedError(rejectionErr.Error())
}
return commitTransaction, nil
} }
switch input.Kind { switch input.Kind {
@ -483,16 +486,7 @@ func (r *Inputer) calculateAndSetState(
roomState := state.NewStateResolution(updater, roomInfo) roomState := state.NewStateResolution(updater, roomInfo)
if input.HasState { if input.HasState {
// Check here if we think we're in the room already.
stateAtEvent.Overwrite = true stateAtEvent.Overwrite = true
var joinEventNIDs []types.EventNID
// Request join memberships only for local users only.
if joinEventNIDs, err = updater.GetMembershipEventNIDsForRoom(ctx, roomInfo.RoomNID, true, true); err == nil {
// If we have no local users that are joined to the room then any state about
// the room that we have is quite possibly out of date. Therefore in that case
// we should overwrite it rather than merge it.
stateAtEvent.Overwrite = len(joinEventNIDs) == 0
}
// We've been told what the state at the event is so we don't need to calculate it. // We've been told what the state at the event is so we don't need to calculate it.
// Check that those state events are in the database and store the state. // Check that those state events are in the database and store the state.

View file

@ -12,6 +12,7 @@ import (
"github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/query" "github.com/matrix-org/dendrite/roomserver/internal/query"
"github.com/matrix-org/dendrite/roomserver/storage/shared" "github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util" "github.com/matrix-org/util"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
@ -75,13 +76,15 @@ func (t *missingStateReq) processEventWithMissingState(
// in the gap in the DAG // in the gap in the DAG
for _, newEvent := range newEvents { for _, newEvent := range newEvents {
_, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{ _, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{
Kind: api.KindNew, Kind: api.KindOld,
Event: newEvent.Headered(roomVersion), Event: newEvent.Headered(roomVersion),
Origin: t.origin, Origin: t.origin,
SendAsServer: api.DoNotSendToOtherServers, SendAsServer: api.DoNotSendToOtherServers,
}) })
if err != nil { if err != nil {
return fmt.Errorf("t.inputer.processRoomEvent: %w", err) if _, ok := err.(types.RejectedError); !ok {
return fmt.Errorf("t.inputer.processRoomEvent (filling gap): %w", err)
}
} }
} }
return nil return nil
@ -183,8 +186,11 @@ func (t *missingStateReq) processEventWithMissingState(
} }
// TODO: we could do this concurrently? // TODO: we could do this concurrently?
for _, ire := range outlierRoomEvents { for _, ire := range outlierRoomEvents {
if _, err = t.inputer.processRoomEvent(ctx, t.db, &ire); err != nil { _, err = t.inputer.processRoomEvent(ctx, t.db, &ire)
return fmt.Errorf("t.inputer.processRoomEvent[outlier]: %w", err) if err != nil {
if _, ok := err.(types.RejectedError); !ok {
return fmt.Errorf("t.inputer.processRoomEvent (outlier): %w", err)
}
} }
} }
@ -205,7 +211,9 @@ func (t *missingStateReq) processEventWithMissingState(
SendAsServer: api.DoNotSendToOtherServers, SendAsServer: api.DoNotSendToOtherServers,
}) })
if err != nil { if err != nil {
return fmt.Errorf("t.inputer.processRoomEvent: %w", err) if _, ok := err.(types.RejectedError); !ok {
return fmt.Errorf("t.inputer.processRoomEvent (backward extremity): %w", err)
}
} }
// Then send all of the newer backfilled events, of which will all be newer // Then send all of the newer backfilled events, of which will all be newer
@ -220,7 +228,9 @@ func (t *missingStateReq) processEventWithMissingState(
SendAsServer: api.DoNotSendToOtherServers, SendAsServer: api.DoNotSendToOtherServers,
}) })
if err != nil { if err != nil {
return fmt.Errorf("t.inputer.processRoomEvent: %w", err) if _, ok := err.(types.RejectedError); !ok {
return fmt.Errorf("t.inputer.processRoomEvent (fast forward): %w", err)
}
} }
} }
@ -395,20 +405,11 @@ retryAllowedState:
// without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events // without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events
func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled bool, err error) { func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled bool, err error) {
logger := util.GetLogger(ctx).WithField("event_id", e.EventID()).WithField("room_id", e.RoomID()) logger := util.GetLogger(ctx).WithField("event_id", e.EventID()).WithField("room_id", e.RoomID())
needed := gomatrixserverlib.StateNeededForAuth([]*gomatrixserverlib.Event{e})
// query latest events (our trusted forward extremities) latest := t.db.LatestEvents()
req := api.QueryLatestEventsAndStateRequest{ latestEvents := make([]string, len(latest))
RoomID: e.RoomID(), for i, ev := range latest {
StateToFetch: needed.Tuples(), latestEvents[i] = ev.EventID
}
var res api.QueryLatestEventsAndStateResponse
if err = t.queryer.QueryLatestEventsAndState(ctx, &req, &res); err != nil {
logger.WithError(err).Warn("Failed to query latest events")
return nil, false, err
}
latestEvents := make([]string, len(res.LatestEvents))
for i, ev := range res.LatestEvents {
latestEvents[i] = res.LatestEvents[i].EventID
t.hadEvent(ev.EventID) t.hadEvent(ev.EventID)
} }

View file

@ -149,7 +149,8 @@ func (r *Queryer) QueryMissingAuthPrevEvents(
} }
for _, prevEventID := range request.PrevEventIDs { for _, prevEventID := range request.PrevEventIDs {
if state, err := r.DB.StateAtEventIDs(ctx, []string{prevEventID}); err != nil || len(state) == 0 { state, err := r.DB.StateAtEventIDs(ctx, []string{prevEventID})
if err != nil || len(state) == 0 {
response.MissingPrevEventIDs = append(response.MissingPrevEventIDs, prevEventID) response.MissingPrevEventIDs = append(response.MissingPrevEventIDs, prevEventID)
} }
} }

View file

@ -209,6 +209,12 @@ type MissingEventError string
func (e MissingEventError) Error() string { return string(e) } func (e MissingEventError) Error() string { return string(e) }
// A RejectedError is returned when an event is stored as rejected. The error
// contains the reason why.
type RejectedError string
func (e RejectedError) Error() string { return string(e) }
// RoomInfo contains metadata about a room // RoomInfo contains metadata about a room
type RoomInfo struct { type RoomInfo struct {
RoomNID RoomNID RoomNID RoomNID