prevent premature gc of peers from topology

Topology gossip can be received on a connection before that connection
has been marked as 'established'. If we do not include unestablished
connections in topology gossip, then parts of the topology can be
removed by gc, since they are deemed unreachable, which can cause
subsequent incremental topology gossips to be rejected due to them
containing references to peers that have been removed by the gc.

Including unestablished connections in topology gossip requires:

- removing the check for 'established' connections in the encoding of
  peers for topology gossip,
- adding the 'established' flag to the internal representation of
  remote connections,
- adding the 'established' flag to to encoding of connections in
  topology gossip,
- treating the *addition* of a connection as an update to the local
  peer.

The latter entails

- bumping the peer version on connection addition (as well as on
  marking 'established' as previously),
- bumping the peer version *unconditionally* on connection deletion
  (rather than only when the connection was 'established', as
  previously),
- moving the initial sending of complete topology on the connection,
  and the broadcasting of a topology update containing the two peers
  at the ends of the connection, from the connection 'established'
  logic to the connection addition logic,
- broadcasting just an topology update for the local peer when a
  connection gets marked as 'established',
- broadcasting a topology update for the local peer *unconditionally*
  when a connection is removed (rather than only doing this when the
  connection was 'established', as previously).

router/connection.go

@@ -33,6 +33,7 @@ type RemoteConnection struct {
 	local         *Peer
 	remote        *Peer
 	remoteTCPAddr string
+	established   bool
 }
 
 type LocalConnection struct {
@@ -41,7 +42,6 @@ type LocalConnection struct {
 	TCPConn            *net.TCPConn
 	tcpSender          TCPSender
 	remoteUDPAddr      *net.UDPAddr
-	established        bool
 	receivedHeartbeat  bool
 	stackFrag          bool
 	effectivePMTU      int
@@ -66,11 +66,12 @@ type ConnectionInteraction struct {
 	payload interface{}
 }
 
-func NewRemoteConnection(from, to *Peer, tcpAddr string) *RemoteConnection {
+func NewRemoteConnection(from, to *Peer, tcpAddr string, established bool) *RemoteConnection {
 	return &RemoteConnection{
 		local:         from,
 		remote:        to,
-		remoteTCPAddr: tcpAddr}
+		remoteTCPAddr: tcpAddr,
+		established:   established}
 }
 
 func (conn *RemoteConnection) Local() *Peer {
@@ -90,7 +91,7 @@ func (conn *RemoteConnection) RemoteTCPAddr() string {
 }
 
 func (conn *RemoteConnection) Established() bool {
-	return true
+	return conn.established
 }
 
 func (conn *RemoteConnection) Shutdown(error) {

router/gossip_test.go

@@ -41,7 +41,7 @@ func (router *Router) AddTestChannelConnection(r *Router) {
 	r.Peers.FetchWithDefault(fromPeer)    // Has side-effect of incrementing refcount
 	router.Peers.FetchWithDefault(toPeer) //
 
-	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, ""}, r}
+	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false}, r}
 	router.Ourself.handleAddConnection(conn)
 	router.Ourself.handleConnectionEstablished(conn)
 }

router/local_peer.go

@@ -108,7 +108,7 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
 	if err != nil {
 		return err
 	}
-	connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String())
+	connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), false)
 	connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.Router)
 	connLocal.Start(acceptNewPeer)
 	return nil
@@ -226,6 +226,8 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) bool {
 		return false
 	}
 	peer.addConnection(conn)
+	peer.Router.SendAllGossipDown(conn)
+	peer.broadcastPeerUpdate(conn.Remote())
 	return true
 }
 
@@ -238,8 +240,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) bool {
 		return false
 	}
 	peer.connectionEstablished(conn)
-	peer.Router.SendAllGossipDown(conn)
-	peer.broadcastPeerUpdate(conn.Remote())
+	peer.broadcastPeerUpdate()
 	return true
 }
 
@@ -258,9 +259,7 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) bool {
 	// Must do garbage collection first to ensure we don't send out an
 	// update with unreachable peers (can cause looping)
 	peer.Router.Peers.GarbageCollect()
-	if conn.Established() {
-		peer.broadcastPeerUpdate()
-	}
+	peer.broadcastPeerUpdate()
 	return true
 }
 

router/mocks_test.go

@@ -27,7 +27,7 @@ func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
 	toName := p2.Name
 	toPeer := NewPeer(toName, p2.UID, 0)
 	toPeer = peers.FetchWithDefault(toPeer)
-	peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, ""})
+	peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false})
 }
 
 func (peers *Peers) DeleteTestConnection(p *Peer) {
@@ -44,7 +44,7 @@ func (peers *Peers) DeleteTestConnection(p *Peer) {
 // from what is created by the real code.
 func newMockConnection(from, to *Peer) Connection {
 	type mockConnection struct{ RemoteConnection }
-	return &mockConnection{RemoteConnection{from, to, ""}}
+	return &mockConnection{RemoteConnection{from, to, "", false}}
 }
 
 func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Connection) {

router/peer.go

@@ -90,16 +90,14 @@ func (peer *Peer) addConnection(conn Connection) {
 	peer.Lock()
 	defer peer.Unlock()
 	peer.connections[conn.Remote().Name] = conn
+	peer.version += 1
 }
 
 func (peer *Peer) deleteConnection(conn Connection) {
-	established := conn.Established()
 	peer.Lock()
 	defer peer.Unlock()
 	delete(peer.connections, conn.Remote().Name)
-	if established {
-		peer.version += 1
-	}
+	peer.version += 1
 }
 
 func (peer *Peer) connectionEstablished(conn Connection) {

router/peers.go

@@ -108,7 +108,7 @@ func EncodePeers(peers ...*Peer) []byte {
 	buf := new(bytes.Buffer)
 	enc := gob.NewEncoder(buf)
 	for _, peer := range peers {
-		peer.encodePeer(enc)
+		peer.encode(enc)
 	}
 	return buf.Bytes()
 }
@@ -163,7 +163,7 @@ func encodePeersMap(peers map[PeerName]*Peer) []byte {
 	buf := new(bytes.Buffer)
 	enc := gob.NewEncoder(buf)
 	for _, peer := range peers {
-		peer.encodePeer(enc)
+		peer.encode(enc)
 	}
 	return buf.Bytes()
 }
@@ -199,7 +199,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
 	}
 
 	for _, connsBuf := range decodedConns {
-		decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string) {
+		decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string, _ bool) {
 			remoteName := PeerNameFromBin(remoteNameByte)
 			if _, found := newPeers[remoteName]; found {
 				return
@@ -252,7 +252,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte) ma
 	return newUpdate
 }
 
-func (peer *Peer) encodePeer(enc *gob.Encoder) {
+func (peer *Peer) encode(enc *gob.Encoder) {
 	peer.RLock()
 	defer peer.RUnlock()
 
@@ -264,11 +264,10 @@ func (peer *Peer) encodePeer(enc *gob.Encoder) {
 	connsEnc := gob.NewEncoder(connsBuf)
 	for _, conn := range peer.connections {
 		// DANGER holding rlock on peer, going to take rlock on conn
-		if !conn.Established() {
-			continue
-		}
 		checkFatal(connsEnc.Encode(conn.Remote().NameByte))
 		checkFatal(connsEnc.Encode(conn.RemoteTCPAddr()))
+		// DANGER holding rlock on peer, going to take rlock on conn
+		checkFatal(connsEnc.Encode(conn.Established()))
 	}
 	checkFatal(enc.Encode(connsBuf.Bytes()))
 }
@@ -289,7 +288,7 @@ func decodePeerNoConns(dec *gob.Decoder) (nameByte []byte, uid uint64, version u
 	return
 }
 
-func connsIterator(input []byte, fun func([]byte, string)) error {
+func connsIterator(input []byte, fun func([]byte, string, bool)) error {
 	buf := new(bytes.Buffer)
 	buf.Write(input)
 	dec := gob.NewDecoder(buf)
@@ -302,16 +301,20 @@ func connsIterator(input []byte, fun func([]byte, string)) error {
 		if err := dec.Decode(&foundAt); err != nil {
 			return err
 		}
-		fun(nameByte, string(foundAt))
+		var established bool
+		if err := dec.Decode(&established); err != nil {
+			return err
+		}
+		fun(nameByte, foundAt, established)
 	}
 }
 
 func readConnsMap(peer *Peer, buf []byte, table map[PeerName]*Peer) map[PeerName]Connection {
 	conns := make(map[PeerName]Connection)
-	if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string) {
+	if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string, established bool) {
 		name := PeerNameFromBin(nameByte)
 		remotePeer := table[name]
-		conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr)
+		conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr, established)
 		conns[name] = conn
 	}); err != io.EOF {
 		// this should never happen since we've already successfully

router/protocol.go

@@ -2,7 +2,7 @@ package router
 
 const (
 	Protocol        = "weave"
-	ProtocolVersion = 11
+	ProtocolVersion = 12
 )
 
 type ProtocolTag byte

router/router.go

@@ -192,7 +192,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
 	// start.
 	remoteAddrStr := tcpConn.RemoteAddr().String()
 	log.Printf("->[%s] connection accepted\n", remoteAddrStr)
-	connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr)
+	connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false)
 	connLocal := NewLocalConnection(connRemote, tcpConn, nil, router)
 	connLocal.Start(true)
 }

site/how-it-works.md

@@ -108,12 +108,16 @@ network knows about any change.
 Topology is communicated over the TCP links between peers, using a
 Gossip mechanism.  Topology messages are sent by a peer...
 
-- when a connection has been established; the entire topology is sent
-  to the remote peer, and an incremental update, containing
-  information on just the two peers at the ends of the connection, is
-  sent to all neighbours
+- when a connection has been added; the entire topology is sent to the
+  remote peer, and an incremental update, containing information on
+  just the two peers at the ends of the connection, is sent to all
+  neighbours,
+- when a connection has been marked as 'established', indicating that
+  the remote peer can receive UDP traffic from the peer; an update
+  containing just information about the local peer is sent to all
+  neighbours,
 - when a connection has been torn down; an update containing just
-  information about the local peer is sent to all neighbours
+  information about the local peer is sent to all neighbours,
 - periodically, on a timer, in case someone has missed an update.
 
 The receiver of a topology update merges that update with its own