Reworked connection code. Pings are now regularly sent out to every connection to check if it remains valid.

Invalid connections are moved to a second list, and eventually dropped if they don't become active.

Commands.go

@@ -8,66 +8,135 @@ package core
 
 import (
 	"fmt"
-	"net"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec"
 )
 
+// Commands between peers
+const (
+	// Peer List Management
+	CommandAnnouncement = 0 // Announcement
+	CommandResponse     = 1 // Response
+	CommandPing         = 2 // Keep-alive message (no payload).
+	CommandPong         = 3 // Response to ping (no payload).
+
+	// Blockchain
+	CommandGet = 4 // Request blocks for specified peer.
+
+	// File Discovery
+
+	// Debug
+	CommandChat = 10 // Chat message [debug]
+)
+
 // packet2 is a high-level message between peers
 type packet2 struct {
 	PacketRaw
 	SenderPublicKey *btcec.PublicKey // Sender Public Key, ECDSA (secp256k1) 257-bit
-	network         *Network         // network which received the packet
-	address         *net.UDPAddr     // address of the sender or receiver
+	connection      *Connection      // Connection that received the packet
 }
 
-// announcement handles an incoming announcement
-func (peer *PeerInfo) announcement(msg *packet2) {
+// cmdAnouncement handles an incoming announcement
+func (peer *PeerInfo) cmdAnouncement(msg *packet2) {
 	if peer == nil {
-		peer, added := PeerlistAdd(msg.SenderPublicKey, &Connection{Network: msg.network, Address: msg.address})
-		fmt.Printf("Incoming initial announcement from %s\n", msg.address.String())
+		peer, added := PeerlistAdd(msg.SenderPublicKey, msg.connection)
+		fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String())
 
 		// send the Response
 		if added {
-			peer.send(&PacketRaw{Command: 1})
+			peer.send(&PacketRaw{Command: CommandResponse})
 		}
 
 		return
 	}
-	fmt.Printf("Incoming secondary announcement from %s\n", msg.address.String())
+	fmt.Printf("Incoming secondary announcement from %s\n", msg.connection.Address.String())
 
 	// Announcement from existing peer means the peer most likely restarted
-	peer.send(&PacketRaw{Command: 1})
+	peer.send(&PacketRaw{Command: CommandResponse})
 }
 
-// response handles the response to the announcement
-func (peer *PeerInfo) response(msg *packet2) {
+// cmdResponse handles the response to the announcement
+func (peer *PeerInfo) cmdResponse(msg *packet2) {
 	if peer == nil {
-		peer, _ = PeerlistAdd(msg.SenderPublicKey, &Connection{Network: msg.network, Address: msg.address})
-		fmt.Printf("Incoming initial response from %s\n", msg.address.String())
+		peer, _ = PeerlistAdd(msg.SenderPublicKey, msg.connection)
+		fmt.Printf("Incoming initial response from %s\n", msg.connection.Address.String())
 
 		return
 	}
 
-	fmt.Printf("Incoming response from %s on %s\n", msg.address.String(), msg.network.address.String())
+	fmt.Printf("Incoming response from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String())
 }
 
-// chat handles a chat message [debug]
-func (peer *PeerInfo) chat(msg *packet2) {
-	fmt.Printf("Chat from '%s': %s\n", msg.address.String(), string(msg.PacketRaw.Payload))
+// cmdPing handles an incoming ping message
+func (peer *PeerInfo) cmdPing(msg *packet2) {
+	peer.send(&PacketRaw{Command: CommandPong})
+	//fmt.Printf("Incoming ping from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String())
 }
 
-// autoPingAll automatically pings all peers. This has multiple important reasons:
-// * Keeping the UDP port open at NAT, if applicable
-// * Making sure to recognize invalid connections and drop them.
+// cmdPong handles an incoming pong message
+func (peer *PeerInfo) cmdPong(msg *packet2) {
+	//fmt.Printf("Incoming pong from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String())
+}
+
+// cmdChat handles a chat message [debug]
+func (peer *PeerInfo) cmdChat(msg *packet2) {
+	fmt.Printf("Chat from '%s': %s\n", msg.connection.Address.String(), string(msg.PacketRaw.Payload))
+}
+
+// pingTime is the time in seconds to send out ping messages
+const pingTime = 10
+
+// connectionInvalidate is the threshold in seconds to invalidate formerly active connections that no longer receive incoming packets.
+const connectionInvalidate = 20
+
+// connectionRemove is the threshold in seconds to remove inactive connections in case there is at least one active connection known.
+const connectionRemove = 2 * 60
+
+// autoPingAll sends out regular ping messages to all connections of all peers. This allows to detect invalid connections and eventually drop them.
 func autoPingAll() {
+	for {
+		time.Sleep(time.Second)
+		thresholdInvalidate := time.Now().Add(-connectionInvalidate * time.Second)
+		thresholdRemove := time.Now().Add(-(connectionRemove + connectionInvalidate) * time.Second)
+		thresholdPingOut := time.Now().Add(-pingTime * time.Second)
 
-	// TODO: If peer was previously unresponsive for X times, start pinging on all available ports.
+		for _, peer := range PeerlistGet() {
+			// first handle active connections
+			for _, connection := range peer.GetConnections(true) {
+				// Check if no incoming packet for the last X seconds. Regularly sent pings should result in incoming packets.
+				if connection.LastPacketIn.Before(thresholdInvalidate) {
+					peer.invalidateActiveConnection(connection)
+				}
+
+				// Send ping if none was sent recently and no incoming packet was received recently.
+				if connection.LastPacketIn.Before(thresholdPingOut) && connection.LastPingOut.Before(thresholdPingOut) {
+					peer.sendConnection(&PacketRaw{Command: CommandPing}, connection)
+					connection.LastPingOut = time.Now()
+				}
+			}
+
+			// handle inactive connections
+			for _, connection := range peer.GetConnections(false) {
+				// Remove connections that have been inactive for a long time; only if there is at least an active connection, or at least two other inactive ones.
+				if connection.LastPacketIn.Before(thresholdRemove) && (len(peer.connectionActive) >= 1 || len(peer.connectionInactive) > 2) {
+					peer.removeInactiveConnection(connection)
+					continue
+				}
+
+				// if no ping was sent recently, send one now
+				if connection.LastPingOut.Before(thresholdPingOut) {
+					peer.sendConnection(&PacketRaw{Command: CommandPing}, connection)
+					connection.LastPingOut = time.Now()
+				}
+			}
+		}
+	}
 }
 
 // SendChatAll sends a text message to all peers
 func SendChatAll(text string) {
 	for _, peer := range PeerlistGet() {
-		peer.send(&PacketRaw{Command: 10, Payload: []byte(text)})
+		peer.send(&PacketRaw{Command: CommandChat, Payload: []byte(text)})
 	}
 }

Connection.go

@@ -0,0 +1,239 @@
+/*
+File Name:  Connection.go
+Copyright:  2021 Peernet Foundation s.r.o.
+Author:     Peter Kleissner
+*/
+
+package core
+
+import (
+	"errors"
+	"net"
+	"sync/atomic"
+	"time"
+
+	"github.com/btcsuite/btcd/btcec"
+)
+
+// Connection is an established connection between a remote IP address and a local network adapter.
+// New connections may only be created in case of successful INCOMING packets.
+type Connection struct {
+	Network       *Network     // network which received the packet
+	Address       *net.UDPAddr // address of the sender or receiver
+	LastPacketIn  time.Time    // Last time an incoming packet was received.
+	LastPacketOut time.Time    // Last time an outgoing packet was attempted to send.
+	LastPingOut   time.Time    // Last ping out.
+}
+
+// Equal checks if the connection was established other the same network adapter using the same IP address. Port is intentionally not checked.
+func (c *Connection) Equal(other *Connection) bool {
+	return c.Address.IP.Equal(other.Address.IP) && c.Network.address.IP.Equal(other.Network.address.IP)
+}
+
+// GetConnections returns the list of connections
+func (peer *PeerInfo) GetConnections(active bool) (connections []*Connection) {
+	peer.RLock()
+	defer peer.RUnlock()
+
+	if active {
+		return peer.connectionActive
+	}
+	return peer.connectionInactive
+}
+
+// registerConnection registers an incoming connection for an existing peer. If new, it will add to the list. If previously inactive, it will elevate.
+func (peer *PeerInfo) registerConnection(incoming *Connection) (result *Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+
+	// first check if already an active connection
+	for _, connection := range peer.connectionActive {
+		if connection.Equal(incoming) {
+			// Connection already established. Verify port and update if necessary.
+			// Some NATs may rotate ports. Some mobile phone providers even rotate IPs which is not detected here.
+			if connection.Address.Port != incoming.Address.Port {
+				connection.Address.Port = incoming.Address.Port
+			}
+
+			peer.connectionLatest = connection
+			return connection
+		}
+	}
+
+	// if an inactive connection, elevate it to activated one
+	for n, connection := range peer.connectionInactive {
+		if connection.Equal(incoming) {
+			if connection.Address.Port != incoming.Address.Port {
+				connection.Address.Port = incoming.Address.Port
+			}
+
+			// elevate by adding to active and mark as latest active
+			peer.connectionLatest = connection
+			peer.connectionActive = append(peer.connectionActive, connection)
+
+			// remove from inactive
+			inactiveNew := peer.connectionInactive[:n]
+			if n < len(peer.connectionInactive)-1 {
+				inactiveNew = append(inactiveNew, peer.connectionInactive[n+1:]...)
+			}
+			peer.connectionInactive = inactiveNew
+
+			return connection
+		}
+	}
+
+	// otherwise it is a new connection!
+	peer.connectionActive = append(peer.connectionActive, incoming)
+	peer.connectionLatest = incoming
+
+	return incoming
+}
+
+// invalidateActiveConnection invalidates an active connection
+func (peer *PeerInfo) invalidateActiveConnection(input *Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+
+	// remove from connectionLatest if selected so it won't be used by standard send function
+	if peer.connectionLatest == input {
+		peer.connectionLatest = nil
+	}
+
+	for n, connection := range peer.connectionActive {
+		if connection == input {
+			// add to list of inactive connections
+			peer.connectionInactive = append(peer.connectionInactive, connection)
+
+			// remove from active
+			activeNew := peer.connectionActive[:n]
+			if n < len(peer.connectionActive)-1 {
+				activeNew = append(activeNew, peer.connectionActive[n+1:]...)
+			}
+			peer.connectionActive = activeNew
+
+			return
+		}
+	}
+}
+
+// removeInactiveConnection removes an inactive connection.
+func (peer *PeerInfo) removeInactiveConnection(input *Connection) {
+	peer.Lock()
+	defer peer.Unlock()
+
+	for n, connection := range peer.connectionInactive {
+		if connection == input {
+
+			// remove from inactive
+			inactiveNew := peer.connectionInactive[:n]
+			if n < len(peer.connectionInactive)-1 {
+				inactiveNew = append(inactiveNew, peer.connectionInactive[n+1:]...)
+			}
+			peer.connectionInactive = inactiveNew
+
+			return
+		}
+	}
+}
+
+// ---- sending code ----
+
+// send sends a raw packet to the peer. Only uses active connections.
+func (peer *PeerInfo) send(packet *PacketRaw) (err error) {
+	if len(peer.connectionActive) == 0 {
+		return errors.New("no valid connection to peer")
+	}
+
+	packet.Protocol = 0
+
+	raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet)
+	if err != nil {
+		return err
+	}
+
+	atomic.AddUint64(&peer.StatsPacketSent, 1)
+
+	// Send out the wire. Use connectionLatest if available.
+	// Failover: If sending fails and there are other connections available, try those. Automatically update connectionLatest if one is successful.
+	// Windows: This works great in case the adapter gets disabled, however, does not detect if the network cable is unplugged.
+	c := peer.connectionLatest
+	if c != nil {
+		c.LastPacketOut = time.Now()
+
+		if err = c.Network.send(c.Address.IP, c.Address.Port, raw); err == nil {
+			return nil
+		}
+
+		// Invalid connection, immediately invalidate. Fallback to broadcast to all other active ones.
+		// Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context".
+		peer.invalidateActiveConnection(c)
+	}
+
+	// If no latest connection available, broadcast on all available connections.
+	// This might be noisy, but if no latest connection is available it means the last established connection is already considered dead.
+	// The receiver is responsible for incoming deduplication of packets.
+	activeConnections := peer.GetConnections(true)
+	for _, c := range activeConnections {
+		c.LastPacketOut = time.Now()
+		c.Network.send(c.Address.IP, c.Address.Port, raw)
+	}
+
+	return nil // on broadcast no error is known and returned
+}
+
+// sendConnection sends a packet to the peer using the specific connection
+func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) (err error) {
+	packet.Protocol = 0
+	raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet)
+	if err != nil {
+		return err
+	}
+
+	atomic.AddUint64(&peer.StatsPacketSent, 1)
+	connection.LastPacketOut = time.Now()
+
+	return connection.Network.send(connection.Address.IP, connection.Address.Port, raw)
+}
+
+// sendAllNetworks sends a raw packet via all networks
+func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr) (err error) {
+	packet.Protocol = 0
+	raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet)
+	if err != nil {
+		return err
+	}
+
+	successCount := 0
+
+	if IsIPv6(remote.IP.To16()) {
+		for _, network := range networks6 {
+			// Do not mix link-local unicast targets with non link-local networks (only when iface is known, i.e. not catch all local)
+			if network.iface != nil && remote.IP.IsLinkLocalUnicast() != network.address.IP.IsLinkLocalUnicast() {
+				continue
+			}
+
+			err = network.send(remote.IP, remote.Port, raw)
+			if err == nil {
+				successCount++
+			}
+		}
+	} else {
+		for _, network := range networks4 {
+			// Do not mix link-local unicast targets with non link-local networks (only when iface is known, i.e. not catch all local)
+			if network.iface != nil && remote.IP.IsLinkLocalUnicast() != network.address.IP.IsLinkLocalUnicast() {
+				continue
+			}
+
+			err = network.send(remote.IP, remote.Port, raw)
+			if err == nil {
+				successCount++
+			}
+		}
+	}
+
+	if successCount == 0 {
+		return errors.New("no successful send")
+	}
+
+	return nil
+}

Network.go

@@ -129,25 +129,35 @@ func packetWorker(packets <-chan networkWire) {
 			continue
 		}
 
+		connection := &Connection{Network: packet.network, Address: packet.sender}
+
 		peer := PeerlistLookup(senderPublicKey)
 		if peer != nil {
 			// Existing peers: Update statistics and network address if new
 			atomic.AddUint64(&peer.StatsPacketReceived, 1)
-			peer.registerConnection(packet.sender, packet.network)
+			connection = peer.registerConnection(connection)
 		}
 
+		connection.LastPacketIn = time.Now()
+
 		// process the packet
-		message := &packet2{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, network: packet.network, address: packet.sender}
+		message := &packet2{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, connection: connection}
 
 		switch decoded.Command {
-		case 0: // Announce
-			peer.announcement(message)
+		case CommandAnnouncement: // Announce
+			peer.cmdAnouncement(message)
 
-		case 1: // Response
-			peer.response(message)
+		case CommandResponse: // Response
+			peer.cmdResponse(message)
 
-		case 10: // Chat [debug]
-			peer.chat(message)
+		case CommandPing: // Ping
+			peer.cmdPing(message)
+
+		case CommandPong: // Ping
+			peer.cmdPong(message)
+
+		case CommandChat: // Chat [debug]
+			peer.cmdChat(message)
 
 		default: // Unknown command
 
@@ -171,3 +181,11 @@ func GetNetworks(networkType int) (networks []*Network) {
 func (network *Network) GetListen() (listen *net.UDPAddr, multicastIPv6 net.IP, broadcastIPv4 []net.IP) {
 	return network.address, network.multicastIP, network.broadcastIPv4
 }
+
+// GetAdapterName returns the adapter name, if available
+func (network *Network) GetAdapterName() string {
+	if network.iface != nil {
+		return network.iface.Name
+	}
+	return "[unknown adapter]"
+}

Peer ID.go

@@ -8,12 +8,9 @@ package core
 
 import (
 	"encoding/hex"
-	"errors"
 	"log"
-	"net"
 	"os"
 	"sync"
-	"sync/atomic"
 
 	"github.com/btcsuite/btcd/btcec"
 )
@@ -66,18 +63,13 @@ func ExportPrivateKey() (privateKey *btcec.PrivateKey, publicKey *btcec.PublicKe
 	return peerPrivateKey, peerPublicKey
 }
 
-// Connection is an established connection between a remote IP address and a local network adapter.
-// New connections may only be created in case of successful INCOMING packets.
-type Connection struct {
-	Network *Network     // network which received the packet
-	Address *net.UDPAddr // address of the sender or receiver
-}
-
 // PeerInfo stores information about a single remote peer
 type PeerInfo struct {
-	PublicKey        *btcec.PublicKey // Public key
-	Connections      []*Connection    // list of established connections to the peer
-	connectionLatest *Connection      // Latest valid connection. Allows to switch on the fly between connections; if one becomes invalid, try others.
+	PublicKey          *btcec.PublicKey // Public key
+	connectionActive   []*Connection    // List of active established connections to the peer.
+	connectionInactive []*Connection    // List of former connections that are no longer valid. They may be removed after a while.
+	connectionLatest   *Connection      // Latest valid connection.
+	sync.RWMutex                        // Mutex for access to list of connections.
 
 	// statistics
 	StatsPacketSent     uint64 // Count of packets sent
@@ -87,7 +79,7 @@ type PeerInfo struct {
 var peerList map[[btcec.PubKeyBytesLenCompressed]byte]*PeerInfo
 var peerlistMutex sync.RWMutex
 
-// PeerlistAdd adds a new peer to the peer list. It does not validate the peer info. If the peer is already added, it does nothing.
+// PeerlistAdd adds a new peer to the peer list. It does not validate the peer info. If the peer is already added, it does nothing. Connections must be live.
 func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer *PeerInfo, added bool) {
 	if len(connections) == 0 {
 		return nil, false
@@ -101,7 +93,7 @@ func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer *
 		return peer, false
 	}
 
-	peer = &PeerInfo{PublicKey: PublicKey, Connections: connections, connectionLatest: connections[0]}
+	peer = &PeerInfo{PublicKey: PublicKey, connectionActive: connections, connectionLatest: connections[0]}
 	peerList[publicKey2Compressed(peer.PublicKey)] = peer
 
 	return peer, true
@@ -149,115 +141,3 @@ func publicKey2Compressed(publicKey *btcec.PublicKey) [btcec.PubKeyBytesLenCompr
 	copy(key[:], publicKey.SerializeCompressed())
 	return key
 }
-
-// send sends a raw packet to the peer
-func (peer *PeerInfo) send(packet *PacketRaw) (err error) {
-	if len(peer.Connections) == 0 {
-		return errors.New("no valid connection to peer")
-	}
-
-	packet.Protocol = 0
-
-	raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet)
-	if err != nil {
-		return err
-	}
-
-	atomic.AddUint64(&peer.StatsPacketSent, 1)
-
-	// Send out the wire. Use connectionLatest if available.
-	// Failover: If sending fails and there are other connections available, try those. Automatically update connectionLatest if one is successful.
-	// Windows: This works great on if the adapter is disabled, however, does not detect if the network cable is unplugged.
-	if peer.connectionLatest != nil {
-		useConnection := peer.connectionLatest
-		if err = useConnection.Network.send(useConnection.Address.IP, useConnection.Address.Port, raw); err != nil && len(peer.Connections) > 1 {
-			err = peer.sendFailover(useConnection, raw)
-		}
-		return err
-	}
-
-	if err = peer.Connections[0].Network.send(peer.Connections[0].Address.IP, peer.Connections[0].Address.Port, raw); err != nil && len(peer.Connections) > 1 {
-		err = peer.sendFailover(peer.Connections[0], raw)
-	}
-	return err
-}
-
-// sendFailover tries to send the packet over any other different (than the failed) connection.
-// If send works, it will switch over connectionLatest.
-// Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context".
-func (peer *PeerInfo) sendFailover(failed *Connection, raw []byte) (err error) {
-	for _, connection := range peer.Connections {
-		if connection != failed {
-			err = connection.Network.send(connection.Address.IP, connection.Address.Port, raw)
-			if err == nil {
-				peer.connectionLatest = connection
-				return nil
-			}
-		}
-	}
-	return err
-}
-
-// sendAllNetworks sends a raw packet via all networks
-func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr) (err error) {
-	packet.Protocol = 0
-	raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet)
-	if err != nil {
-		return err
-	}
-
-	successCount := 0
-
-	if IsIPv6(remote.IP.To16()) {
-		for _, network := range networks6 {
-			// Do not mix link-local unicast targets with non link-local networks (only when iface is known, i.e. not catch all local)
-			if network.iface != nil && remote.IP.IsLinkLocalUnicast() != network.address.IP.IsLinkLocalUnicast() {
-				continue
-			}
-
-			err = network.send(remote.IP, remote.Port, raw)
-			if err == nil {
-				successCount++
-			}
-		}
-	} else {
-		for _, network := range networks4 {
-			// Do not mix link-local unicast targets with non link-local networks (only when iface is known, i.e. not catch all local)
-			if network.iface != nil && remote.IP.IsLinkLocalUnicast() != network.address.IP.IsLinkLocalUnicast() {
-				continue
-			}
-
-			err = network.send(remote.IP, remote.Port, raw)
-			if err == nil {
-				successCount++
-			}
-		}
-	}
-
-	if successCount == 0 {
-		return errors.New("no successful send")
-	}
-
-	return nil
-}
-
-// registerConnection registers an incoming connection for an existing peer. If new, it will add to the list.
-// Currently there is only a connectionLatest indicating the latest valid connection. In the future other priority such as last packet time, speed, local vs remote IP might be used.
-func (peer *PeerInfo) registerConnection(remote *net.UDPAddr, network *Network) {
-	for n := range peer.Connections {
-		if peer.Connections[n].Address.IP.Equal(remote.IP) && peer.Connections[n].Network.address.IP.Equal(network.address.IP) {
-			// Connection already established. Verify port and update if necessary.
-			// Some NATs may rotate ports. Some mobile phone providers even rotate IPs which is not detected here.
-			if peer.Connections[n].Address.Port != remote.Port {
-				peer.Connections[n].Address.Port = remote.Port
-			}
-
-			peer.connectionLatest = peer.Connections[n]
-			return
-		}
-	}
-
-	peer.Connections = append(peer.Connections, &Connection{Address: remote, Network: network})
-
-	peer.connectionLatest = peer.Connections[len(peer.Connections)-1]
-}

Peernet.go

@@ -19,4 +19,5 @@ func Init() {
 func Connect() {
 	go bootstrap()
 	go autoMulticastBroadcast()
+	go autoPingAll()
 }