Merge branch 'main' into neilalexander/syncsnapshot

build/gobind-pinecone/monolith.go

@@ -30,6 +30,8 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/atomic"
+
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/appservice"
 	"github.com/matrix-org/dendrite/clientapi/userutil"
@@ -66,6 +68,7 @@ const (
 	PeerTypeRemote    = pineconeRouter.PeerTypeRemote
 	PeerTypeMulticast = pineconeRouter.PeerTypeMulticast
 	PeerTypeBluetooth = pineconeRouter.PeerTypeBluetooth
+	PeerTypeBonjour   = pineconeRouter.PeerTypeBonjour
 )
 
 type DendriteMonolith struct {
@@ -82,6 +85,10 @@ type DendriteMonolith struct {
 	userAPI           userapiAPI.UserInternalAPI
 }
 
+func (m *DendriteMonolith) PublicKey() string {
+	return m.PineconeRouter.PublicKey().String()
+}
+
 func (m *DendriteMonolith) BaseURL() string {
 	return fmt.Sprintf("http://%s", m.listener.Addr().String())
 }
@@ -94,6 +101,20 @@ func (m *DendriteMonolith) SessionCount() int {
 	return len(m.PineconeQUIC.Protocol("matrix").Sessions())
 }
 
+func (m *DendriteMonolith) RegisterNetworkInterface(name string, index int, mtu int, up bool, broadcast bool, loopback bool, pointToPoint bool, multicast bool, addrs string) {
+	m.PineconeMulticast.RegisterInterface(pineconeMulticast.InterfaceInfo{
+		Name: name,
+		Index: index,
+		Mtu: mtu,
+		Up: up,
+		Broadcast: broadcast,
+		Loopback: loopback,
+		PointToPoint: pointToPoint,
+		Multicast: multicast,
+		Addrs: addrs,
+	})
+}
+
 func (m *DendriteMonolith) SetMulticastEnabled(enabled bool) {
 	if enabled {
 		m.PineconeMulticast.Start()
@@ -134,32 +155,21 @@ func (m *DendriteMonolith) Conduit(zone string, peertype int) (*Conduit, error)
 	go func() {
 		conduit.portMutex.Lock()
 		defer conduit.portMutex.Unlock()
-	loop:
+
-		for i := 1; i <= 10; i++ {
+		logrus.Errorf("Attempting authenticated connect")
-			logrus.Errorf("Attempting authenticated connect (attempt %d)", i)
+		var err error
-			var err error
+		if conduit.port, err = m.PineconeRouter.Connect(
-			conduit.port, err = m.PineconeRouter.Connect(
+			l,
-				l,
+			pineconeRouter.ConnectionZone(zone),
-				pineconeRouter.ConnectionZone(zone),
+			pineconeRouter.ConnectionPeerType(peertype),
-				pineconeRouter.ConnectionPeerType(peertype),
+		); err != nil {
-			)
+			logrus.Errorf("Authenticated connect failed: %s", err)
-			switch err {
+			_ = l.Close()
-			case io.ErrClosedPipe:
+			_ = r.Close()
-				logrus.Errorf("Authenticated connect failed due to closed pipe (attempt %d)", i)
+			_ = conduit.Close()
-				return
+			return
-			case io.EOF:
-				logrus.Errorf("Authenticated connect failed due to EOF (attempt %d)", i)
-				break loop
-			case nil:
-				logrus.Errorf("Authenticated connect succeeded, connected to port %d (attempt %d)", conduit.port, i)
-				return
-			default:
-				logrus.WithError(err).Errorf("Authenticated connect failed (attempt %d)", i)
-				time.Sleep(time.Second)
-			}
 		}
-		_ = l.Close()
+		logrus.Infof("Authenticated connect succeeded (port %d)", conduit.port)
-		_ = r.Close()
 	}()
 	return conduit, nil
 }
@@ -395,6 +405,7 @@ func (m *DendriteMonolith) Stop() {
 const MaxFrameSize = types.MaxFrameSize
 
 type Conduit struct {
+	closed    atomic.Bool
 	conn      net.Conn
 	port      types.SwitchPortID
 	portMutex sync.Mutex
@@ -407,10 +418,16 @@ func (c *Conduit) Port() int {
 }
 
 func (c *Conduit) Read(b []byte) (int, error) {
+	if c.closed.Load() {
+		return 0, io.EOF
+	}
 	return c.conn.Read(b)
 }
 
 func (c *Conduit) ReadCopy() ([]byte, error) {
+	if c.closed.Load() {
+		return nil, io.EOF
+	}
 	var buf [65535 * 2]byte
 	n, err := c.conn.Read(buf[:])
 	if err != nil {
@@ -420,9 +437,16 @@ func (c *Conduit) ReadCopy() ([]byte, error) {
 }
 
 func (c *Conduit) Write(b []byte) (int, error) {
+	if c.closed.Load() {
+		return 0, io.EOF
+	}
 	return c.conn.Write(b)
 }
 
 func (c *Conduit) Close() error {
+	if c.closed.Load() {
+		return io.ErrClosedPipe
+	}
+	c.closed.Store(true)
 	return c.conn.Close()
 }

federationapi/consumers/sendtodevice.go

@@ -80,7 +80,6 @@ func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msgs []*nats
 		return true
 	}
 	if originServerName != t.ServerName {
-		log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
 		return true
 	}
 	// Extract the send-to-device event from msg.

go.mod

@@ -22,8 +22,8 @@ require (
 	github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
 	github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20220926161602-759a8ee7c4d5
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20220929155203-377b320ad5b1
-	github.com/matrix-org/pinecone v0.0.0-20220927101513-d0beb180f44d
+	github.com/matrix-org/pinecone v0.0.0-20220929155234-2ce51dd4a42c
 	github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
 	github.com/mattn/go-sqlite3 v1.14.15
 	github.com/nats-io/nats-server/v2 v2.9.1-0.20220920152220-52d7b481c4b5

go.sum

@@ -384,10 +384,10 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91 h1:s7fexw
 github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo=
 github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5dLDCud4r0r55eP4j9FuUNpl60Gmntcop4=
 github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20220926161602-759a8ee7c4d5 h1:cQMA9hip0WSp6cv7CUfButa9Jl/9E6kqWmQyOjx5A5s=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20220929155203-377b320ad5b1 h1:kS4jgi/DshFGJArjF14b1u42fVWC+8LMVdSVLdQYq3E=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20220926161602-759a8ee7c4d5/go.mod h1:Mtifyr8q8htcBeugvlDnkBcNUy5LO8OzUoplAf1+mb4=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20220929155203-377b320ad5b1/go.mod h1:Mtifyr8q8htcBeugvlDnkBcNUy5LO8OzUoplAf1+mb4=
-github.com/matrix-org/pinecone v0.0.0-20220927101513-d0beb180f44d h1:kGPJ6Rg8nn5an2CbCZrRiuTNyNzE0rRMiqm4UXJYrRs=
+github.com/matrix-org/pinecone v0.0.0-20220929155234-2ce51dd4a42c h1:iCHLYwwlPsf4TYFrvhKdhQoAM2lXzcmDZYqwBNWcnVk=
-github.com/matrix-org/pinecone v0.0.0-20220927101513-d0beb180f44d/go.mod h1:K0N1ixHQxXoCyqolDqVxPM3ArrDtcMs8yegOx2Lfv9k=
+github.com/matrix-org/pinecone v0.0.0-20220929155234-2ce51dd4a42c/go.mod h1:K0N1ixHQxXoCyqolDqVxPM3ArrDtcMs8yegOx2Lfv9k=
 github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4 h1:eCEHXWDv9Rm335MSuB49mFUK44bwZPFSDde3ORE3syk=
 github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=