Reset backoff on incoming federation (#1080)

* Reset backoffs in response to incoming federation requests

* Federation wakeups no more than once per minute per origin

2 files changed, 54 insertions, 19 deletions

diff --git a/federationapi/routing/routing.go b/federationapi/routing/routing.go
index 86d3192a..b75a7941 100644
--- a/federationapi/routing/routing.go
+++ b/federationapi/routing/routing.go
@@ -48,7 +48,7 @@ func Setup(
 	asAPI appserviceAPI.AppServiceQueryAPI,
 	producer *producers.RoomserverProducer,
 	eduProducer *producers.EDUServerProducer,
-	federationSenderAPI federationSenderAPI.FederationSenderInternalAPI,
+	fsAPI federationSenderAPI.FederationSenderInternalAPI,
 	keys gomatrixserverlib.KeyRing,
 	federation *gomatrixserverlib.FederationClient,
 	accountDB accounts.Database,
@@ -58,6 +58,10 @@ func Setup(
 	v1fedmux := publicAPIMux.PathPrefix(pathPrefixV1Federation).Subrouter()
 	v2fedmux := publicAPIMux.PathPrefix(pathPrefixV2Federation).Subrouter()
 
+	wakeup := &internal.FederationWakeups{
+		FsAPI: fsAPI,
+	}
+
 	localKeys := internal.MakeExternalAPI("localkeys", func(req *http.Request) util.JSONResponse {
 		return LocalKeys(cfg)
 	})
@@ -71,7 +75,7 @@ func Setup(
 	v2keysmux.Handle("/server", localKeys).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/send/{txnID}", internal.MakeFedAPI(
-		"federation_send", cfg.Matrix.ServerName, keys,
+		"federation_send", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -85,7 +89,7 @@ func Setup(
 	)).Methods(http.MethodPut, http.MethodOptions)
 
 	v2fedmux.Handle("/invite/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_invite", cfg.Matrix.ServerName, keys,
+		"federation_invite", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -105,7 +109,7 @@ func Setup(
 	)).Methods(http.MethodPost, http.MethodOptions)
 
 	v1fedmux.Handle("/exchange_third_party_invite/{roomID}", internal.MakeFedAPI(
-		"exchange_third_party_invite", cfg.Matrix.ServerName, keys,
+		"exchange_third_party_invite", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -118,7 +122,7 @@ func Setup(
 	)).Methods(http.MethodPut, http.MethodOptions)
 
 	v1fedmux.Handle("/event/{eventID}", internal.MakeFedAPI(
-		"federation_get_event", cfg.Matrix.ServerName, keys,
+		"federation_get_event", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -131,7 +135,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/state/{roomID}", internal.MakeFedAPI(
-		"federation_get_state", cfg.Matrix.ServerName, keys,
+		"federation_get_state", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -144,7 +148,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/state_ids/{roomID}", internal.MakeFedAPI(
-		"federation_get_state_ids", cfg.Matrix.ServerName, keys,
+		"federation_get_state_ids", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -157,7 +161,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/event_auth/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_get_event_auth", cfg.Matrix.ServerName, keys,
+		"federation_get_event_auth", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars := mux.Vars(httpReq)
 			return GetEventAuth(
@@ -167,16 +171,16 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/query/directory", internal.MakeFedAPI(
-		"federation_query_room_alias", cfg.Matrix.ServerName, keys,
+		"federation_query_room_alias", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			return RoomAliasToID(
-				httpReq, federation, cfg, rsAPI, federationSenderAPI,
+				httpReq, federation, cfg, rsAPI, fsAPI,
 			)
 		},
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/query/profile", internal.MakeFedAPI(
-		"federation_query_profile", cfg.Matrix.ServerName, keys,
+		"federation_query_profile", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			return GetProfile(
 				httpReq, accountDB, cfg, asAPI,
@@ -185,7 +189,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/user/devices/{userID}", internal.MakeFedAPI(
-		"federation_user_devices", cfg.Matrix.ServerName, keys,
+		"federation_user_devices", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -198,7 +202,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/make_join/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_make_join", cfg.Matrix.ServerName, keys,
+		"federation_make_join", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -227,7 +231,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/send_join/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_send_join", cfg.Matrix.ServerName, keys,
+		"federation_send_join", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -249,7 +253,7 @@ func Setup(
 	)).Methods(http.MethodPut)
 
 	v2fedmux.Handle("/send_join/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_send_join", cfg.Matrix.ServerName, keys,
+		"federation_send_join", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -264,7 +268,7 @@ func Setup(
 	)).Methods(http.MethodPut)
 
 	v1fedmux.Handle("/make_leave/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_make_leave", cfg.Matrix.ServerName, keys,
+		"federation_make_leave", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -279,7 +283,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v2fedmux.Handle("/send_leave/{roomID}/{eventID}", internal.MakeFedAPI(
-		"federation_send_leave", cfg.Matrix.ServerName, keys,
+		"federation_send_leave", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -301,7 +305,7 @@ func Setup(
 	)).Methods(http.MethodGet)
 
 	v1fedmux.Handle("/get_missing_events/{roomID}", internal.MakeFedAPI(
-		"federation_get_missing_events", cfg.Matrix.ServerName, keys,
+		"federation_get_missing_events", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
@@ -312,7 +316,7 @@ func Setup(
 	)).Methods(http.MethodPost)
 
 	v1fedmux.Handle("/backfill/{roomID}", internal.MakeFedAPI(
-		"federation_backfill", cfg.Matrix.ServerName, keys,
+		"federation_backfill", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
 			vars, err := internal.URLDecodeMapValues(mux.Vars(httpReq))
 			if err != nil {
diff --git a/internal/httpapi.go b/internal/httpapi.go
index 07bbacdd..bacd1768 100644
--- a/internal/httpapi.go
+++ b/internal/httpapi.go
@@ -1,17 +1,20 @@
 package internal
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/clientapi/auth"
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	federationsenderAPI "github.com/matrix-org/dendrite/federationsender/api"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/internal/httpapis"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -170,6 +173,7 @@ func MakeFedAPI(
 	metricsName string,
 	serverName gomatrixserverlib.ServerName,
 	keyRing gomatrixserverlib.KeyRing,
+	wakeup *FederationWakeups,
 	f func(*http.Request, *gomatrixserverlib.FederationRequest) util.JSONResponse,
 ) http.Handler {
 	h := func(req *http.Request) util.JSONResponse {
@@ -179,11 +183,38 @@ func MakeFedAPI(
 		if fedReq == nil {
 			return errResp
 		}
+		go wakeup.Wakeup(req.Context(), fedReq.Origin())
 		return f(req, fedReq)
 	}
 	return MakeExternalAPI(metricsName, h)
 }
 
+type FederationWakeups struct {
+	FsAPI   federationsenderAPI.FederationSenderInternalAPI
+	origins sync.Map
+}
+
+func (f *FederationWakeups) Wakeup(ctx context.Context, origin gomatrixserverlib.ServerName) {
+	key, keyok := f.origins.Load(origin)
+	if keyok {
+		lastTime, ok := key.(time.Time)
+		if ok && time.Since(lastTime) < time.Minute {
+			return
+		}
+	}
+	aliveReq := federationsenderAPI.PerformServersAliveRequest{
+		Servers: []gomatrixserverlib.ServerName{origin},
+	}
+	aliveRes := federationsenderAPI.PerformServersAliveResponse{}
+	if err := f.FsAPI.PerformServersAlive(ctx, &aliveReq, &aliveRes); err != nil {
+		util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
+			"origin": origin,
+		}).Warn("incoming federation request failed to notify server alive")
+	} else {
+		f.origins.Store(origin, time.Now())
+	}
+}
+
 // SetupHTTPAPI registers an HTTP API mux under /api and sets up a metrics
 // listener.
 func SetupHTTPAPI(servMux *http.ServeMux, publicApiMux *mux.Router, internalApiMux *mux.Router, cfg *config.Dendrite, enableHTTPAPIs bool) {