Merge pull request #565 from rade/515_crypto_glitch_stall

Big improvement in observed behaviour.  LGTM; closes #515

router/connection.go

@@ -298,24 +298,33 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 	}
 	conn.Log("completed handshake")
 
-	if err = conn.initHeartbeats(); err != nil { // [1]
+	// The ordering of the following is very important. [1]
+
+	if conn.remoteUDPAddr != nil {
+		if err = conn.ensureForwarders(); err != nil {
+			return
+		}
+	}
+	if err = conn.Router.Ourself.AddConnection(conn); err != nil {
 		return
 	}
-
-	conn.Router.Ourself.AddConnection(conn) // [1]
-
-	go conn.receiveTCP(dec) // [1]
-
-	err = conn.actorLoop(actionChan) // [1]
+	if err = conn.initHeartbeats(); err != nil {
+		return
+	}
+	go conn.receiveTCP(dec)
+	err = conn.actorLoop(actionChan)
 }
 
 // [1] Ordering constraints:
 //
-// (a) initHeartbeats should precede AddConnection. That way we can
-// guarantee that a) the forwarder gets initialised before the router
-// can discover the connection and attempt to invoke Forward, and b)
-// Heartbeats are the first thing that get sent on connections.
-// Neither of these are essential, but they are desirable.
+// (a) AddConnections must precede initHeartbeats. It is only after
+// the former completes that we know the connection is valid, in
+// particular is not a duplicate connection to the same peer. Sending
+// heartbeats on a duplicate connection can trip up crypto at the
+// other end, since the associated UDP packets may get decoded by the
+// other connection. It is also generally wasteful to engage in any
+// interaction with the remote on a connection that turns out to be
+// invald.
 //
 // (b) AddConnection must precede receiveTCP. In the absence of any
 // indirect connectivity to the remote peer, the first we hear about
@@ -332,7 +341,24 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 // precede shutdown, since that invokes DeleteConnection and is
 // invoked on termination of this entire function. Essentially this
 // boils down to a prohibition on running AddConnection in a separate
-// goroutine, at least not without some synchronisation.
+// goroutine, at least not without some synchronisation. Which in turn
+// requires us the launching of the receiveTCP goroutine to precede
+// actorLoop.
+//
+// (d) AddConnection should precede receiveTCP. There is no point
+// starting the latter if the former fails.
+//
+// (e) initHeartbeats should precede actorLoop. The former is setting
+// LocalConnection fields accessed by the latter. Since the latter
+// runs in a separate goroutine, we'd have to add some synchronisation
+// if initHeartbeats isn't run first.
+//
+// (f) ensureForwarders should precede AddConnection. As soon as a
+// connection has been added to LocalPeer by the latter, it becomes
+// visible to the packet routing logic, which will end up dropping
+// packets if the forwarders haven't been created yet. We cannot
+// prevent that completely, since, for example, forwarder can only be
+// created when we know the remote UDP address, but it helps to try.
 
 func (conn *LocalConnection) initHeartbeats() error {
 	conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)

router/crypto.go

@@ -271,8 +271,7 @@ type NaClDecryptorInstance struct {
 }
 
 type PacketDecodingError struct {
-	Fatal bool
-	Desc  string
+	Desc string
 }
 
 func NewNonDecryptor(conn *LocalConnection) *NonDecryptor {
@@ -335,7 +334,7 @@ func (nd *NaClDecryptor) ReceiveNonce(msg []byte) {
 func (nd *NaClDecryptor) IterateFrames(packet []byte, consumer FrameConsumer) error {
 	buf, err := nd.decrypt(packet)
 	if err != nil {
-		return PacketDecodingError{Fatal: true, Desc: fmt.Sprint("decryption failed; ", err)}
+		return PacketDecodingError{Desc: fmt.Sprint("decryption failed; ", err)}
 	}
 	return nd.NonDecryptor.IterateFrames(buf, consumer)
 }

router/local_peer.go

@@ -115,13 +115,12 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
 // ACTOR client API
 
 // Sync.
-func (peer *LocalPeer) AddConnection(conn *LocalConnection) {
-	resultChan := make(chan interface{})
+func (peer *LocalPeer) AddConnection(conn *LocalConnection) error {
+	resultChan := make(chan error)
 	peer.actionChan <- func() {
-		peer.handleAddConnection(conn)
-		resultChan <- nil
+		resultChan <- peer.handleAddConnection(conn)
 	}
-	<-resultChan
+	return <-resultChan
 }
 
 // Async.
@@ -155,7 +154,7 @@ func (peer *LocalPeer) actorLoop(actionChan <-chan LocalPeerAction) {
 	}
 }
 
-func (peer *LocalPeer) handleAddConnection(conn Connection) {
+func (peer *LocalPeer) handleAddConnection(conn Connection) error {
 	if peer.Peer != conn.Local() {
 		log.Fatal("Attempt made to add connection to peer where peer is not the source of connection")
 	}
@@ -167,26 +166,23 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) {
 	// deliberately non symmetrical
 	if dupConn, found := peer.connections[toName]; found {
 		if dupConn == conn {
-			return
+			return nil
 		}
 		switch conn.BreakTie(dupConn) {
 		case TieBreakWon:
 			dupConn.Shutdown(dupErr)
 			peer.handleDeleteConnection(dupConn)
 		case TieBreakLost:
-			conn.Shutdown(dupErr)
-			return
+			return dupErr
 		case TieBreakTied:
 			// oh good grief. Sod it, just kill both of them.
-			conn.Shutdown(dupErr)
 			dupConn.Shutdown(dupErr)
 			peer.handleDeleteConnection(dupConn)
-			return
+			return dupErr
 		}
 	}
 	if err := peer.checkConnectionLimit(); err != nil {
-		conn.Shutdown(err)
-		return
+		return err
 	}
 	_, isConnectedPeer := peer.router.Routes.Unicast(toName)
 	peer.addConnection(conn)
@@ -197,6 +193,7 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) {
 		peer.router.SendAllGossipDown(conn)
 	}
 	peer.broadcastPeerUpdate(conn.Remote())
+	return nil
 }
 
 func (peer *LocalPeer) handleConnectionEstablished(conn Connection) {

router/router.go

@@ -250,15 +250,19 @@ func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) {
 		if !ok {
 			continue
 		}
-		err = relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(dec, sender, po))
-		if pde, ok := err.(PacketDecodingError); ok {
-			if pde.Fatal {
-				relayConn.Shutdown(pde)
-			} else {
-				relayConn.Log(pde.Error())
-			}
-		} else {
-			checkWarn(err)
+		if err := relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(dec, sender, po)); err != nil {
+			// Errors during UDP packet decoding / processing are
+			// non-fatal. One common cause is that we receive and
+			// attempt to decrypt a "stray" packet. This can actually
+			// happen quite easily if there is some connection churn
+			// between two peers. After all, UDP isn't a
+			// connection-oriented protocol, yet we pretend it is.
+			//
+			// If anything really is seriously, unrecoverably amiss
+			// with a connection, that will typically result in missed
+			// heartbeats and the connection getting shut down because
+			// of that.
+			relayConn.Log(err)
 		}
 	}
 }