dendrite/roomserver/internal/perform/perform_inbound_peek.go
Neil Alexander a763cbb0e1
Roomserver/federation input refactor (#2104)
* Put federation client functions into their own file

* Look for missing auth events in RS input

* Remove retrieveMissingAuthEvents from federation API

* Logging

* Sorta transplanted the code over

* Use event origin failing all else

* Don't get stuck on mutexes:

* Add verifier

* Don't mark state events with zero snapshot NID as not existing

* Check missing state if not an outlier before storing the event

* Reject instead of soft-fail, don't copy roominfo so much

* Use synchronous contexts, limit time to fetch missing events

* Clean up some commented out bits

* Simplify `/send` endpoint significantly

* Submit async

* Report errors on sending to RS input

* Set max payload in NATS to 16MB

* Tweak metrics

* Add `workerForRoom` for tidiness

* Try skipping unmarshalling errors for RespMissingEvents

* Track missing prev events separately to avoid calculating state when not possible

* Tweak logic around checking missing state

* Care about state when checking missing prev events

* Don't check missing state for create events

* Try that again

* Handle create events better

* Send create room events as new

* Use given event kind when sending auth/state events

* Revert "Use given event kind when sending auth/state events"

This reverts commit 089d64d271b5fca8c104e1554711187420dbebca.

* Only search for missing prev events or state for new events

* Tweaks

* We only have missing prev if we don't supply state

* Room version tweaks

* Allow async inputs again

* Apply backpressure to consumers/synchronous requests to hopefully stop things being overwhelmed

* Set timeouts on roomserver input tasks (need to decide what timeout makes sense)

* Use work queue policy, deliver all on restart

* Reduce chance of duplicates being sent by NATS

* Limit the number of servers we attempt to reduce backpressure

* Some review comment fixes

* Tidy up a couple things

* Don't limit servers, randomise order using map

* Some context refactoring

* Update gmsl

* Don't resend create events

* Set stateIDs length correctly or else the roomserver thinks there are missing events when there aren't

* Exclude our own servername

* Try backing off servers

* Make excluding self behaviour optional

* Exclude self from g_m_e

* Update sytest-whitelist

* Update consumers for the roomserver output stream

* Remember to send outliers for state returned from /gme

* Make full HTTP tests less upsetti

* Remove 'If a device list update goes missing, the server resyncs on the next one' from the sytest blacklist

* Remove debugging test

* Fix blacklist again, remove unnecessary duplicate context

* Clearer contexts, don't use background in case there's something happening there

* Don't queue up events more than once in memory

* Correctly identify create events when checking for state

* Fill in gaps again in /gme code

* Remove `AuthEventIDs` from `InputRoomEvent`

* Remove stray field

Co-authored-by: Kegan Dougal <kegan@matrix.org>
2022-01-27 14:29:14 +00:00

130 lines
4.0 KiB
Go

// Copyright 2020 New Vector Ltd
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package perform
import (
"context"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/internal/query"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
type InboundPeeker struct {
DB storage.Database
Inputer *input.Inputer
}
// PerformInboundPeek handles peeking into matrix rooms, including over
// federation by talking to the federationapi. called when a remote server
// initiates a /peek over federation.
//
// It should atomically figure out the current state of the room (for the
// response to /peek) while adding the new inbound peek to the kafka stream so the
// fed sender can start sending peeked events without a race between the state
// snapshot and the stream of peeked events.
func (r *InboundPeeker) PerformInboundPeek(
ctx context.Context,
request *api.PerformInboundPeekRequest,
response *api.PerformInboundPeekResponse,
) error {
info, err := r.DB.RoomInfo(ctx, request.RoomID)
if err != nil {
return err
}
if info == nil || info.IsStub {
return nil
}
response.RoomExists = true
response.RoomVersion = info.RoomVersion
var stateEvents []*gomatrixserverlib.Event
var currentStateSnapshotNID types.StateSnapshotNID
latestEventRefs, currentStateSnapshotNID, _, err :=
r.DB.LatestEventIDs(ctx, info.RoomNID)
if err != nil {
return err
}
latestEvents, err := r.DB.EventsFromIDs(ctx, []string{latestEventRefs[0].EventID})
if err != nil {
return err
}
var sortedLatestEvents []*gomatrixserverlib.Event
for _, ev := range latestEvents {
sortedLatestEvents = append(sortedLatestEvents, ev.Event)
}
sortedLatestEvents = gomatrixserverlib.ReverseTopologicalOrdering(
sortedLatestEvents,
gomatrixserverlib.TopologicalOrderByPrevEvents,
)
response.LatestEvent = sortedLatestEvents[0].Headered(info.RoomVersion)
// XXX: do we actually need to do a state resolution here?
roomState := state.NewStateResolution(r.DB, info)
var stateEntries []types.StateEntry
stateEntries, err = roomState.LoadStateAtSnapshot(
ctx, currentStateSnapshotNID,
)
if err != nil {
return err
}
stateEvents, err = helpers.LoadStateEvents(ctx, r.DB, stateEntries)
if err != nil {
return err
}
// get the auth event IDs for the current state events
var authEventIDs []string
for _, se := range stateEvents {
authEventIDs = append(authEventIDs, se.AuthEventIDs()...)
}
authEventIDs = util.UniqueStrings(authEventIDs) // de-dupe
authEvents, err := query.GetAuthChain(ctx, r.DB.EventsFromIDs, authEventIDs)
if err != nil {
return err
}
for _, event := range stateEvents {
response.StateEvents = append(response.StateEvents, event.Headered(info.RoomVersion))
}
for _, event := range authEvents {
response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
}
err = r.Inputer.WriteOutputEvents(request.RoomID, []api.OutputEvent{
{
Type: api.OutputTypeNewInboundPeek,
NewInboundPeek: &api.OutputNewInboundPeek{
RoomID: request.RoomID,
PeekID: request.PeekID,
LatestEventID: latestEvents[0].EventID(),
ServerName: request.ServerName,
RenewalInterval: request.RenewalInterval,
},
},
})
return err
}