Initial Store & Forward Implementation (#2917)

This adds store & forward relays into dendrite for p2p.
A few things have changed:
- new relay api serves new http endpoints for s&f federation
- updated outbound federation queueing which will attempt to forward
using s&f if appropriate
- database entries to track s&f relays for other nodes

6 files changed, 103 insertions, 3 deletions

diff --git a/setup/base/base.go b/setup/base/base.go
index ff38209f..de8f8151 100644
--- a/setup/base/base.go
+++ b/setup/base/base.go
@@ -595,6 +595,12 @@ func (b *BaseDendrite) WaitForShutdown() {
 			logrus.Warnf("failed to flush all Sentry events!")
 		}
 	}
+	if b.Fulltext != nil {
+		err := b.Fulltext.Close()
+		if err != nil {
+			logrus.Warnf("failed to close full text search!")
+		}
+	}
 
 	logrus.Warnf("Dendrite is exiting now")
 }
diff --git a/setup/config/config.go b/setup/config/config.go
index 41d2b667..2b38cd51 100644
--- a/setup/config/config.go
+++ b/setup/config/config.go
@@ -62,6 +62,7 @@ type Dendrite struct {
 	RoomServer    RoomServer    `yaml:"room_server"`
 	SyncAPI       SyncAPI       `yaml:"sync_api"`
 	UserAPI       UserAPI       `yaml:"user_api"`
+	RelayAPI      RelayAPI      `yaml:"relay_api"`
 
 	MSCs MSCs `yaml:"mscs"`
 
@@ -349,6 +350,7 @@ func (c *Dendrite) Defaults(opts DefaultOpts) {
 	c.SyncAPI.Defaults(opts)
 	c.UserAPI.Defaults(opts)
 	c.AppServiceAPI.Defaults(opts)
+	c.RelayAPI.Defaults(opts)
 	c.MSCs.Defaults(opts)
 	c.Wiring()
 }
@@ -361,7 +363,7 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
 		&c.Global, &c.ClientAPI, &c.FederationAPI,
 		&c.KeyServer, &c.MediaAPI, &c.RoomServer,
 		&c.SyncAPI, &c.UserAPI,
-		&c.AppServiceAPI, &c.MSCs,
+		&c.AppServiceAPI, &c.RelayAPI, &c.MSCs,
 	} {
 		c.Verify(configErrs, isMonolith)
 	}
@@ -377,6 +379,7 @@ func (c *Dendrite) Wiring() {
 	c.SyncAPI.Matrix = &c.Global
 	c.UserAPI.Matrix = &c.Global
 	c.AppServiceAPI.Matrix = &c.Global
+	c.RelayAPI.Matrix = &c.Global
 	c.MSCs.Matrix = &c.Global
 
 	c.ClientAPI.Derived = &c.Derived
diff --git a/setup/config/config_federationapi.go b/setup/config/config_federationapi.go
index 0f853865..6c198018 100644
--- a/setup/config/config_federationapi.go
+++ b/setup/config/config_federationapi.go
@@ -18,6 +18,12 @@ type FederationAPI struct {
 	// The default value is 16 if not specified, which is circa 18 hours.
 	FederationMaxRetries uint32 `yaml:"send_max_retries"`
 
+	// P2P Feature: How many consecutive failures that we should tolerate when
+	// sending federation requests to a specific server until we should assume they
+	// are offline. If we assume they are offline then we will attempt to send
+	// messages to their relay server if we know of one that is appropriate.
+	P2PFederationRetriesUntilAssumedOffline uint32 `yaml:"p2p_retries_until_assumed_offline"`
+
 	// FederationDisableTLSValidation disables the validation of X.509 TLS certs
 	// on remote federation endpoints. This is not recommended in production!
 	DisableTLSValidation bool `yaml:"disable_tls_validation"`
@@ -43,6 +49,7 @@ func (c *FederationAPI) Defaults(opts DefaultOpts) {
 		c.Database.Defaults(10)
 	}
 	c.FederationMaxRetries = 16
+	c.P2PFederationRetriesUntilAssumedOffline = 2
 	c.DisableTLSValidation = false
 	c.DisableHTTPKeepalives = false
 	if opts.Generate {
diff --git a/setup/config/config_relayapi.go b/setup/config/config_relayapi.go
new file mode 100644
index 00000000..5a6b093d
--- /dev/null
+++ b/setup/config/config_relayapi.go
@@ -0,0 +1,52 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// 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 config
+
+type RelayAPI struct {
+	Matrix *Global `yaml:"-"`
+
+	InternalAPI InternalAPIOptions `yaml:"internal_api,omitempty"`
+	ExternalAPI ExternalAPIOptions `yaml:"external_api,omitempty"`
+
+	// The database stores information used by the relay queue to
+	// forward transactions to remote servers.
+	Database DatabaseOptions `yaml:"database,omitempty"`
+}
+
+func (c *RelayAPI) Defaults(opts DefaultOpts) {
+	if !opts.Monolithic {
+		c.InternalAPI.Listen = "http://localhost:7775"
+		c.InternalAPI.Connect = "http://localhost:7775"
+		c.ExternalAPI.Listen = "http://[::]:8075"
+		c.Database.Defaults(10)
+	}
+	if opts.Generate {
+		if !opts.Monolithic {
+			c.Database.ConnectionString = "file:relayapi.db"
+		}
+	}
+}
+
+func (c *RelayAPI) Verify(configErrs *ConfigErrors, isMonolith bool) {
+	if isMonolith { // polylith required configs below
+		return
+	}
+	if c.Matrix.DatabaseOptions.ConnectionString == "" {
+		checkNotEmpty(configErrs, "relay_api.database.connection_string", string(c.Database.ConnectionString))
+	}
+	checkURL(configErrs, "relay_api.external_api.listen", string(c.ExternalAPI.Listen))
+	checkURL(configErrs, "relay_api.internal_api.listen", string(c.InternalAPI.Listen))
+	checkURL(configErrs, "relay_api.internal_api.connect", string(c.InternalAPI.Connect))
+}
diff --git a/setup/config/config_test.go b/setup/config/config_test.go
index 3408bf46..ffbf4c3c 100644
--- a/setup/config/config_test.go
+++ b/setup/config/config_test.go
@@ -20,11 +20,12 @@ import (
 	"testing"
 
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 )
 
 func TestLoadConfigRelative(t *testing.T) {
-	_, err := loadConfig("/my/config/dir", []byte(testConfig),
+	cfg, err := loadConfig("/my/config/dir", []byte(testConfig),
 		mockReadFile{
 			"/my/config/dir/matrix_key.pem": testKey,
 			"/my/config/dir/tls_cert.pem":   testCert,
@@ -34,6 +35,15 @@ func TestLoadConfigRelative(t *testing.T) {
 	if err != nil {
 		t.Error("failed to load config:", err)
 	}
+
+	configErrors := &ConfigErrors{}
+	cfg.Verify(configErrors, false)
+	if len(*configErrors) > 0 {
+		for _, err := range *configErrors {
+			logrus.Errorf("Configuration error: %s", err)
+		}
+		t.Error("configuration verification failed")
+	}
 }
 
 const testConfig = `
@@ -68,6 +78,8 @@ global:
     display_name: "Server alerts"
     avatar: ""
     room_name: "Server Alerts"	
+  jetstream:
+    addresses: ["test"]
 app_service_api:
   internal_api:
     listen: http://localhost:7777
@@ -84,7 +96,7 @@ client_api:
     connect: http://localhost:7771
   external_api:
     listen: http://[::]:8071
-  registration_disabled: false
+  registration_disabled: true
   registration_shared_secret: ""
   enable_registration_captcha: false
   recaptcha_public_key: ""
@@ -112,6 +124,8 @@ federation_api:
     connect: http://localhost:7772
   external_api:
     listen: http://[::]:8072
+  database:
+    connection_string: file:federationapi.db
 key_server:
   internal_api:
     listen: http://localhost:7779
@@ -194,6 +208,17 @@ user_api:
     max_open_conns: 100
     max_idle_conns: 2
     conn_max_lifetime: -1
+relay_api:
+  internal_api:
+    listen: http://localhost:7775
+    connect: http://localhost:7775
+  external_api:
+    listen: http://[::]:8075
+  database:
+    connection_string: file:relayapi.db
+mscs:
+  database:
+    connection_string: file:mscs.db
 tracing:
   enabled: false
   jaeger:
diff --git a/setup/monolith.go b/setup/monolith.go
index 41a89702..5bbe4019 100644
--- a/setup/monolith.go
+++ b/setup/monolith.go
@@ -23,6 +23,8 @@ import (
 	"github.com/matrix-org/dendrite/internal/transactions"
 	keyAPI "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/mediaapi"
+	"github.com/matrix-org/dendrite/relayapi"
+	relayAPI "github.com/matrix-org/dendrite/relayapi/api"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
@@ -44,6 +46,7 @@ type Monolith struct {
 	RoomserverAPI roomserverAPI.RoomserverInternalAPI
 	UserAPI       userapi.UserInternalAPI
 	KeyAPI        keyAPI.KeyInternalAPI
+	RelayAPI      relayAPI.RelayInternalAPI
 
 	// Optional
 	ExtPublicRoomsProvider   api.ExtraPublicRoomsProvider
@@ -71,4 +74,8 @@ func (m *Monolith) AddAllPublicRoutes(base *base.BaseDendrite) {
 	syncapi.AddPublicRoutes(
 		base, m.UserAPI, m.RoomserverAPI, m.KeyAPI,
 	)
+
+	if m.RelayAPI != nil {
+		relayapi.AddPublicRoutes(base, m.KeyRing, m.RelayAPI)
+	}
 }