Yggdrasil demo updates (#1241)

* PerformServersAlive in PerformBroadcastEDU

* Don't double-pointer

* More reliable QUIC session handling

* Direct peer lookup, other tweaks

* Tweaks

* Try to wake up queues on incoming QUIC session

* Set session callbak on gobind build

* Fix incoming session storage

* Stateless reset, other tweaks

* Reset sessions when coordinates change

* Disable HTTP connection reuse, tweak timeouts

1 files changed, 21 insertions, 8 deletions

diff --git a/build/gobind/monolith.go b/build/gobind/monolith.go
index 84103f38..c0c0ccb4 100644
--- a/build/gobind/monolith.go
+++ b/build/gobind/monolith.go
@@ -25,7 +25,6 @@ import (
 	"github.com/matrix-org/dendrite/userapi"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
-	"go.uber.org/atomic"
 )
 
 type DendriteMonolith struct {
@@ -34,8 +33,6 @@ type DendriteMonolith struct {
 	StorageDirectory string
 	listener         net.Listener
 	httpServer       *http.Server
-	httpListening    atomic.Bool
-	yggListening     atomic.Bool
 }
 
 func (m *DendriteMonolith) BaseURL() string {
@@ -46,6 +43,10 @@ func (m *DendriteMonolith) PeerCount() int {
 	return m.YggdrasilNode.PeerCount()
 }
 
+func (m *DendriteMonolith) SessionCount() int {
+	return m.YggdrasilNode.SessionCount()
+}
+
 func (m *DendriteMonolith) SetMulticastEnabled(enabled bool) {
 	m.YggdrasilNode.SetMulticastEnabled(enabled)
 }
@@ -86,7 +87,7 @@ func (m *DendriteMonolith) Start() {
 	cfg.Matrix.ServerName = gomatrixserverlib.ServerName(ygg.DerivedServerName())
 	cfg.Matrix.PrivateKey = ygg.SigningPrivateKey()
 	cfg.Matrix.KeyID = gomatrixserverlib.KeyID(signing.KeyID)
-	cfg.Matrix.FederationMaxRetries = 6
+	cfg.Matrix.FederationMaxRetries = 8
 	cfg.Kafka.UseNaffka = true
 	cfg.Kafka.Topics.OutputRoomEvent = "roomserverOutput"
 	cfg.Kafka.Topics.OutputClientData = "clientapiOutput"
@@ -98,7 +99,7 @@ func (m *DendriteMonolith) Start() {
 	cfg.Database.SyncAPI = config.DataSource(fmt.Sprintf("file:%s/dendrite-syncapi.db", m.StorageDirectory))
 	cfg.Database.RoomServer = config.DataSource(fmt.Sprintf("file:%s/dendrite-roomserver.db", m.StorageDirectory))
 	cfg.Database.ServerKey = config.DataSource(fmt.Sprintf("file:%s/dendrite-serverkey.db", m.StorageDirectory))
-	cfg.Database.E2EKey = config.DataSource(fmt.Sprintf("file:%s/dendrite-e2ekey.db", m.StorageDirectory))
+	cfg.Database.E2EKey = config.DataSource(fmt.Sprintf("file:%s/dendrite-keyserver.db", m.StorageDirectory))
 	cfg.Database.FederationSender = config.DataSource(fmt.Sprintf("file:%s/dendrite-federationsender.db", m.StorageDirectory))
 	cfg.Database.AppService = config.DataSource(fmt.Sprintf("file:%s/dendrite-appservice.db", m.StorageDirectory))
 	cfg.Database.CurrentState = config.DataSource(fmt.Sprintf("file:%s/dendrite-currentstate.db", m.StorageDirectory))
@@ -136,6 +137,18 @@ func (m *DendriteMonolith) Start() {
 		base, federation, rsAPI, stateAPI, keyRing,
 	)
 
+	ygg.SetSessionFunc(func(address string) {
+		req := &api.PerformServersAliveRequest{
+			Servers: []gomatrixserverlib.ServerName{
+				gomatrixserverlib.ServerName(address),
+			},
+		}
+		res := &api.PerformServersAliveResponse{}
+		if err := fsAPI.PerformServersAlive(context.TODO(), req, res); err != nil {
+			logrus.WithError(err).Error("Failed to send wake-up message to newly connected node")
+		}
+	})
+
 	// The underlying roomserver implementation needs to be able to call the fedsender.
 	// This is different to rsAPI which can be the http client which doesn't need this dependency
 	rsAPI.SetFederationSenderAPI(fsAPI)
@@ -175,9 +188,9 @@ func (m *DendriteMonolith) Start() {
 	m.httpServer = &http.Server{
 		Addr:         ":0",
 		TLSNextProto: map[string]func(*http.Server, *tls.Conn, http.Handler){},
-		ReadTimeout:  30 * time.Second,
-		WriteTimeout: 30 * time.Second,
-		IdleTimeout:  60 * time.Second,
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+		IdleTimeout:  30 * time.Second,
 		BaseContext: func(_ net.Listener) context.Context {
 			return context.Background()
 		},