Merge branch 'master' into kegan/redactions-4

build/gobind/monolith.go

@@ -16,6 +16,7 @@ import (
 	"github.com/matrix-org/dendrite/eduserver"
 	"github.com/matrix-org/dendrite/eduserver/cache"
 	"github.com/matrix-org/dendrite/federationsender"
+	"github.com/matrix-org/dendrite/federationsender/api"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/internal/httputil"
 	"github.com/matrix-org/dendrite/internal/setup"
@@ -23,6 +24,7 @@ import (
 	"github.com/matrix-org/dendrite/userapi"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
+	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 )
 
 type DendriteMonolith struct {
@@ -158,6 +160,32 @@ func (m *DendriteMonolith) Start() {
 		base.UseHTTPAPIs,
 	)
 
+	ygg.NotifySessionNew(func(boxPubKey crypto.BoxPubKey) {
+		serv := gomatrixserverlib.ServerName(boxPubKey.String())
+		req := &api.PerformServersAliveRequest{
+			Servers: []gomatrixserverlib.ServerName{serv},
+		}
+		res := &api.PerformServersAliveResponse{}
+		if err := fsAPI.PerformServersAlive(context.TODO(), req, res); err != nil {
+			logrus.WithError(err).Warnf("Failed to notify server %q alive due to new session", serv)
+		} else {
+			logrus.Infof("Notified server %q alive due to new session", serv)
+		}
+	})
+
+	ygg.NotifyLinkNew(func(boxPubKey crypto.BoxPubKey, linkType, remote string) {
+		serv := gomatrixserverlib.ServerName(boxPubKey.String())
+		req := &api.PerformServersAliveRequest{
+			Servers: []gomatrixserverlib.ServerName{serv},
+		}
+		res := &api.PerformServersAliveResponse{}
+		if err := fsAPI.PerformServersAlive(context.TODO(), req, res); err != nil {
+			logrus.WithError(err).Warnf("Failed to notify server %q alive due to new peer", serv)
+		} else {
+			logrus.Infof("Notified server %q alive due to new peer", serv)
+		}
+	})
+
 	// Build both ends of a HTTP multiplex.
 	httpServer := &http.Server{
 		Addr:         ":0",

cmd/dendrite-demo-yggdrasil/main.go

@@ -32,6 +32,7 @@ import (
 	"github.com/matrix-org/dendrite/eduserver"
 	"github.com/matrix-org/dendrite/eduserver/cache"
 	"github.com/matrix-org/dendrite/federationsender"
+	"github.com/matrix-org/dendrite/federationsender/api"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/internal/httputil"
@@ -39,6 +40,7 @@ import (
 	"github.com/matrix-org/dendrite/roomserver"
 	"github.com/matrix-org/dendrite/userapi"
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 
 	"github.com/sirupsen/logrus"
 )
@@ -150,6 +152,30 @@ func main() {
 		base.UseHTTPAPIs,
 	)
 
+	ygg.NotifySessionNew(func(boxPubKey crypto.BoxPubKey) {
+		req := &api.PerformServersAliveRequest{
+			Servers: []gomatrixserverlib.ServerName{
+				gomatrixserverlib.ServerName(boxPubKey.String()),
+			},
+		}
+		res := &api.PerformServersAliveResponse{}
+		if err := fsAPI.PerformServersAlive(context.TODO(), req, res); err != nil {
+			logrus.WithError(err).Warn("Failed to notify server alive due to new session")
+		}
+	})
+
+	ygg.NotifyLinkNew(func(boxPubKey crypto.BoxPubKey, linkType, remote string) {
+		req := &api.PerformServersAliveRequest{
+			Servers: []gomatrixserverlib.ServerName{
+				gomatrixserverlib.ServerName(boxPubKey.String()),
+			},
+		}
+		res := &api.PerformServersAliveResponse{}
+		if err := fsAPI.PerformServersAlive(context.TODO(), req, res); err != nil {
+			logrus.WithError(err).Warn("Failed to notify server alive due to new link")
+		}
+	})
+
 	// Build both ends of a HTTP multiplex.
 	httpServer := &http.Server{
 		Addr:         ":0",

cmd/dendrite-demo-yggdrasil/yggconn/node.go

@@ -35,6 +35,7 @@ import (
 
 	yggdrasiladmin "github.com/yggdrasil-network/yggdrasil-go/src/admin"
 	yggdrasilconfig "github.com/yggdrasil-network/yggdrasil-go/src/config"
+	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 	yggdrasilmulticast "github.com/yggdrasil-network/yggdrasil-go/src/multicast"
 	"github.com/yggdrasil-network/yggdrasil-go/src/yggdrasil"
 
@@ -136,7 +137,7 @@ func Setup(instanceName, storageDirectory string) (*Node, error) {
 		MaxIncomingStreams:    0,
 		MaxIncomingUniStreams: 0,
 		KeepAlive:             true,
-		MaxIdleTimeout:        time.Second * 900,
+		MaxIdleTimeout:        time.Second * 60,
 		HandshakeTimeout:      time.Second * 30,
 	}
 
@@ -262,3 +263,19 @@ func (n *Node) SetStaticPeer(uri string) error {
 	}
 	return nil
 }
+
+func (n *Node) NotifyLinkNew(f func(boxPubKey crypto.BoxPubKey, linkType, remote string)) {
+	n.core.NotifyLinkNew(f)
+}
+
+func (n *Node) NotifyLinkGone(f func(boxPubKey crypto.BoxPubKey, linkType, remote string)) {
+	n.core.NotifyLinkGone(f)
+}
+
+func (n *Node) NotifySessionNew(f func(boxPubKey crypto.BoxPubKey)) {
+	n.core.NotifySessionNew(f)
+}
+
+func (n *Node) NotifySessionGone(f func(boxPubKey crypto.BoxPubKey)) {
+	n.core.NotifySessionGone(f)
+}

federationsender/queue/destinationqueue.go

@@ -225,7 +225,7 @@ func (oq *destinationQueue) backgroundSend() {
 		}
 
 		// If we have pending PDUs or EDUs then construct a transaction.
-		for oq.pendingPDUs.Load() > 0 || len(oq.pendingEDUs) > 0 {
+		if oq.pendingPDUs.Load() > 0 || len(oq.pendingEDUs) > 0 {
 			// Try sending the next transaction and see what happens.
 			transaction, terr := oq.nextTransaction(oq.pendingEDUs)
 			if terr != nil {
@@ -250,8 +250,6 @@ func (oq *destinationQueue) backgroundSend() {
 				oq.statistics.Success()
 				// Clean up the in-memory buffers.
 				oq.cleanPendingEDUs()
-			} else {
-				break
 			}
 		}
 
@@ -275,6 +273,17 @@ func (oq *destinationQueue) backgroundSend() {
 				oq.cleanPendingInvites()
 			}
 		}
+
+		// If something else has come along while we were busy sending
+		// the previous transaction then we don't want the next loop
+		// iteration to sleep. Send a message if someone else hasn't
+		// already sent a wake-up.
+		if oq.pendingPDUs.Load() > 0 {
+			select {
+			case oq.notifyPDUs <- true:
+			default:
+			}
+		}
 	}
 }
 
@@ -340,7 +349,17 @@ func (oq *destinationQueue) nextTransaction(
 	// If we didn't get anything from the database and there are no
 	// pending EDUs then there's nothing to do - stop here.
 	if len(pdus) == 0 && len(pendingEDUs) == 0 {
-		log.Warnf("no pdus/edus for nextTransaction for destination %q", oq.destination)
+		log.Warnf("Expected PDUs/EDUs for destination %q but got none", oq.destination)
+		// This shouldn't really happen but since it has, let's check
+		// how many events are *really* in the database that are waiting.
+		if count, cerr := oq.db.GetPendingPDUCount(
+			context.TODO(),
+			oq.destination,
+		); cerr == nil {
+			oq.pendingPDUs.Store(count)
+		} else {
+			log.Warnf("Failed to retrieve pending PDU count for %q", oq.destination)
+		}
 		return false, nil
 	}
 
@@ -372,15 +391,17 @@ func (oq *destinationQueue) nextTransaction(
 	// TODO: we should check for 500-ish fails vs 400-ish here,
 	// since we shouldn't queue things indefinitely in response
 	// to a 400-ish error
-	_, err = oq.client.SendTransaction(context.TODO(), t)
+	ctx, cancel = context.WithTimeout(context.Background(), time.Second*15)
-	switch e := err.(type) {
+	defer cancel()
+	_, err = oq.client.SendTransaction(ctx, t)
+	switch err.(type) {
 	case nil:
 		// No error was returned so the transaction looks to have
 		// been successfully sent.
 		oq.pendingPDUs.Sub(int64(len(t.PDUs)))
 		// Clean up the transaction in the database.
 		if err = oq.db.CleanTransactionPDUs(
-			context.TODO(),
+			context.Background(),
 			t.Destination,
 			t.TransactionID,
 		); err != nil {
@@ -388,15 +409,8 @@ func (oq *destinationQueue) nextTransaction(
 		}
 		return true, nil
 	case gomatrix.HTTPError:
-		// We received a HTTP error back. In this instance we only
+		// Report that we failed to send the transaction and we
-		// should report an error if
+		// will retry again, subject to backoff.
-		if e.Code >= 400 && e.Code <= 499 {
-			// We tried but the remote side has sent back a client error.
-			// It's no use retrying because it will happen again.
-			return true, nil
-		}
-		// Otherwise, report that we failed to send the transaction
-		// and we will retry again.
 		return false, err
 	default:
 		log.WithFields(log.Fields{

federationsender/queue/queue.go

@@ -93,7 +93,7 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
 			statistics:       oqs.statistics.ForServer(destination),
 			incomingEDUs:     make(chan *gomatrixserverlib.EDU, 128),
 			incomingInvites:  make(chan *gomatrixserverlib.InviteV2Request, 128),
-			notifyPDUs:       make(chan bool, 128),
+			notifyPDUs:       make(chan bool, 1),
 			interruptBackoff: make(chan bool),
 			signing:          oqs.signing,
 		}

go.mod

@@ -36,9 +36,10 @@ require (
 	github.com/uber-go/atomic v1.3.0 // indirect
 	github.com/uber/jaeger-client-go v2.15.0+incompatible
 	github.com/uber/jaeger-lib v1.5.0
-	github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200707124509-16343a00055c
+	github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708124809-79077e271c6d
 	go.uber.org/atomic v1.4.0
 	golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5
+	golang.org/x/mobile v0.0.0-20200629153529-33b80540585f // indirect
 	gopkg.in/h2non/bimg.v1 v1.0.18
 	gopkg.in/yaml.v2 v2.2.8
 )

go.sum

@@ -12,6 +12,7 @@ github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOv
 github.com/Arceliar/phony v0.0.0-20191006174943-d0c68492aca0 h1:p3puK8Sl2xK+2FnnIvY/C0N1aqJo2kbEsdAzU+Tnv48=
 github.com/Arceliar/phony v0.0.0-20191006174943-d0c68492aca0/go.mod h1:6Lkn+/zJilRMsKmbmG1RPoamiArC6HS73xbwRyp3UyI=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/Kubuxu/go-os-helper v0.0.1/go.mod h1:N8B+I7vPCT80IcP58r50u4+gEEcsZETFUpAzWW2ep1Y=
 github.com/Shopify/sarama v1.26.1 h1:3jnfWKD7gVwbB1KSy/lE0szA9duPuSFLViK0o/d3DgA=
 github.com/Shopify/sarama v1.26.1/go.mod h1:NbSGBSSndYaIhRcBtY9V0U7AyH+x71bG668AuWys/yU=
@@ -654,6 +655,12 @@ github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:
 github.com/yggdrasil-network/yggdrasil-extras v0.0.0-20200525205615-6c8a4a2e8855/go.mod h1:xQdsh08Io6nV4WRnOVTe6gI8/2iTvfLDQ0CYa5aMt+I=
 github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200707124509-16343a00055c h1:tK1FySfDA5xVT5sAK/3XjUCE9LEoandmVnrg4Hj0fXk=
 github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200707124509-16343a00055c/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708120322-c6245869aeac h1:TMehvDn6TmWemBMxBFQoF/rQ513OmytfDY6sLPz3M64=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708120322-c6245869aeac/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708123331-4e0b0e723459 h1:B7wypB+eA3xdB9bH/rFNywP3vg0MuIoMb3SojWApt+4=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708123331-4e0b0e723459/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708124809-79077e271c6d h1:ly327dysc3r7lfG+AKJWPSAQmGf4h++fk+Y2dD8nDV4=
+github.com/yggdrasil-network/yggdrasil-go v0.3.15-0.20200708124809-79077e271c6d/go.mod h1:d+Nz6SPeG6kmeSPFL0cvfWfgwEql75fUnZiAONgvyBE=
 go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.1/go.mod h1:Ap50jQcDJrx6rB6VgeeFPtuPIf3wMRvRfrfYDO6+BmA=
@@ -679,6 +686,7 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90Pveol
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190618222545-ea8f1a30c443/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -691,10 +699,19 @@ golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5 h1:Q7tZBpemrlsc2I7IyODzhtallWRSm4Q0d09pL6XbQtU=
 golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4=
+golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
+golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
+golang.org/x/mobile v0.0.0-20200629153529-33b80540585f h1:9MxnlCHwn6IfUTinHBBzcBhmrX4OXfRmi954tWGKq+M=
+golang.org/x/mobile v0.0.0-20200629153529-33b80540585f/go.mod h1:skQtrUTUwhdJvXM/2KKJzY8pDgNr9I/FOMqDVRPBUS4=
+golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/mod v0.1.1-0.20191209134235-331c550502dd/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -781,7 +798,11 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd h1:/e+gpKk9r3dJobndpTytxS2gOy6m5uvpg+ISQoEcusQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200117012304-6edc0a871e69 h1:yBHHx+XZqXJBm6Exke3N7V9gnlsyXxoCPEb1yVenjfk=
+golang.org/x/tools v0.0.0-20200117012304-6edc0a871e69/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=