a763cbb0e1
* 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>
181 lines
6.5 KiB
Go
181 lines
6.5 KiB
Go
package internal
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/matrix-org/gomatrixserverlib"
|
|
)
|
|
|
|
// Functions here are "proxying" calls to the gomatrixserverlib federation
|
|
// client.
|
|
|
|
func (a *FederationInternalAPI) GetEventAuth(
|
|
ctx context.Context, s gomatrixserverlib.ServerName,
|
|
roomVersion gomatrixserverlib.RoomVersion, roomID, eventID string,
|
|
) (res gomatrixserverlib.RespEventAuth, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.GetEventAuth(ctx, s, roomVersion, roomID, eventID)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespEventAuth{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespEventAuth), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) GetUserDevices(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, userID string,
|
|
) (gomatrixserverlib.RespUserDevices, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.GetUserDevices(ctx, s, userID)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespUserDevices{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespUserDevices), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) ClaimKeys(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string,
|
|
) (gomatrixserverlib.RespClaimKeys, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBackingOffOrBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.ClaimKeys(ctx, s, oneTimeKeys)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespClaimKeys{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespClaimKeys), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) QueryKeys(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string,
|
|
) (gomatrixserverlib.RespQueryKeys, error) {
|
|
ires, err := a.doRequestIfNotBackingOffOrBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.QueryKeys(ctx, s, keys)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespQueryKeys{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespQueryKeys), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) Backfill(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, roomID string, limit int, eventIDs []string,
|
|
) (res gomatrixserverlib.Transaction, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.Backfill(ctx, s, roomID, limit, eventIDs)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.Transaction{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.Transaction), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) LookupState(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion,
|
|
) (res gomatrixserverlib.RespState, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.LookupState(ctx, s, roomID, eventID, roomVersion)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespState{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespState), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) LookupStateIDs(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string,
|
|
) (res gomatrixserverlib.RespStateIDs, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.LookupStateIDs(ctx, s, roomID, eventID)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespStateIDs{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespStateIDs), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) LookupMissingEvents(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, roomID string,
|
|
missing gomatrixserverlib.MissingEvents, roomVersion gomatrixserverlib.RoomVersion,
|
|
) (res gomatrixserverlib.RespMissingEvents, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.LookupMissingEvents(ctx, s, roomID, missing, roomVersion)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.RespMissingEvents{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.RespMissingEvents), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) GetEvent(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, eventID string,
|
|
) (res gomatrixserverlib.Transaction, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.GetEvent(ctx, s, eventID)
|
|
})
|
|
if err != nil {
|
|
return gomatrixserverlib.Transaction{}, err
|
|
}
|
|
return ires.(gomatrixserverlib.Transaction), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) LookupServerKeys(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp,
|
|
) ([]gomatrixserverlib.ServerKeys, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Minute)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.LookupServerKeys(ctx, s, keyRequests)
|
|
})
|
|
if err != nil {
|
|
return []gomatrixserverlib.ServerKeys{}, err
|
|
}
|
|
return ires.([]gomatrixserverlib.ServerKeys), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) MSC2836EventRelationships(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest,
|
|
roomVersion gomatrixserverlib.RoomVersion,
|
|
) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Minute)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.MSC2836EventRelationships(ctx, s, r, roomVersion)
|
|
})
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
return ires.(gomatrixserverlib.MSC2836EventRelationshipsResponse), nil
|
|
}
|
|
|
|
func (a *FederationInternalAPI) MSC2946Spaces(
|
|
ctx context.Context, s gomatrixserverlib.ServerName, roomID string, r gomatrixserverlib.MSC2946SpacesRequest,
|
|
) (res gomatrixserverlib.MSC2946SpacesResponse, err error) {
|
|
ctx, cancel := context.WithTimeout(ctx, time.Minute)
|
|
defer cancel()
|
|
ires, err := a.doRequestIfNotBlacklisted(s, func() (interface{}, error) {
|
|
return a.federation.MSC2946Spaces(ctx, s, roomID, r)
|
|
})
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
return ires.(gomatrixserverlib.MSC2946SpacesResponse), nil
|
|
}
|