← back to Exo
discovery fixed
a2a37c0ebec51af92657db8e0bd35dbe6674eaba · 2025-08-15 15:23:20 +0100 · Gelu Vrabie
Co-authored-by: Gelu Vrabie <gelu@exolabs.net>
Files touched
D .DS_StoreM .gitignoreM hosts.jsonM networking/forwarder/src/event_writer.goM networking/forwarder/src/libp2p.goD networking/forwarder/src/node_id_exchange.goD networking/forwarder/src/node_id_exchange_test.goA networking/forwarder/src/tcp_agent.go
Diff
commit a2a37c0ebec51af92657db8e0bd35dbe6674eaba
Author: Gelu Vrabie <gelu.vrabie.univ@gmail.com>
Date: Fri Aug 15 15:23:20 2025 +0100
discovery fixed
Co-authored-by: Gelu Vrabie <gelu@exolabs.net>
---
.DS_Store | Bin 10244 -> 0 bytes
.gitignore | 6 +-
hosts.json | 2 +-
networking/forwarder/src/event_writer.go | 257 +++-----
networking/forwarder/src/libp2p.go | 589 ++++++++++---------
networking/forwarder/src/node_id_exchange.go | 185 ------
networking/forwarder/src/node_id_exchange_test.go | 111 ----
networking/forwarder/src/tcp_agent.go | 678 ++++++++++++++++++++++
8 files changed, 1056 insertions(+), 772 deletions(-)
diff --git a/.DS_Store b/.DS_Store
deleted file mode 100644
index 7a04f4bd..00000000
Binary files a/.DS_Store and /dev/null differ
diff --git a/.gitignore b/.gitignore
index 2b800b88..762f2302 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,5 +16,9 @@ build/
rust/target/
rust/Cargo.lock
+.DS_Store
+*/.DS_Store
+
# Says this symlink should be git-ignored https://github.com/juspay/just-flake
-just-flake.just
\ No newline at end of file
+just-flake.just
+.DS_Store
diff --git a/hosts.json b/hosts.json
index fdf160cf..e8452a99 100644
--- a/hosts.json
+++ b/hosts.json
@@ -1 +1 @@
-["s13@169.254.249.73", "s14@169.254.69.217", "s15@169.254.165.26", "s16@169.254.29.77"]
\ No newline at end of file
+["s17@169.254.17.227", "s18@169.254.27.237"]
\ No newline at end of file
diff --git a/networking/forwarder/src/event_writer.go b/networking/forwarder/src/event_writer.go
index 6465198d..34032f32 100644
--- a/networking/forwarder/src/event_writer.go
+++ b/networking/forwarder/src/event_writer.go
@@ -5,9 +5,8 @@ import (
"encoding/json"
"fmt"
"log"
- "strconv"
+ "net"
"sync"
- "time"
"github.com/google/uuid"
"github.com/libp2p/go-libp2p/core/network"
@@ -19,83 +18,69 @@ var (
eventsDBPath string
eventsDB *sql.DB
eventsDBMu sync.Mutex
-
- // Track connections to prevent duplicate events
- connectionTracker = make(map[string]bool)
- connTrackerMu sync.Mutex
)
-// SetEventsDBPath sets the path to the events database
func SetEventsDBPath(path string) {
eventsDBMu.Lock()
defer eventsDBMu.Unlock()
eventsDBPath = path
}
-// Event types matching Python's _EventType enum
const (
EventTypeTopologyEdgeCreated = "TopologyEdgeCreated"
EventTypeTopologyEdgeDeleted = "TopologyEdgeDeleted"
)
-// ConnectionProfile matches Python's ConnectionProfile (optional)
type ConnectionProfile struct {
Throughput float64 `json:"throughput"`
Latency float64 `json:"latency"`
Jitter float64 `json:"jitter"`
}
-// Multiaddr matches Python's Multiaddr structure
type Multiaddr struct {
Address string `json:"address"`
IPv4Address string `json:"ipv4_address,omitempty"`
+ IPv6Address string `json:"ipv6_address,omitempty"`
Port int `json:"port,omitempty"`
+ Transport string `json:"transport,omitempty"` // tcp/quic/ws/etc
}
-// Connection matches Python's Connection model
type Connection struct {
- LocalNodeID string `json:"local_node_id"`
- SendBackNodeID string `json:"send_back_node_id"`
- LocalMultiaddr Multiaddr `json:"local_multiaddr"`
- SendBackMultiaddr Multiaddr `json:"send_back_multiaddr"`
- ConnectionProfile *ConnectionProfile `json:"connection_profile"`
+ LocalNodeID string `json:"local_node_id"`
+ SendBackNodeID string `json:"send_back_node_id"`
+ LocalMultiaddr Multiaddr `json:"local_multiaddr"`
+ SendBackMultiaddr Multiaddr `json:"send_back_multiaddr"`
+ ConnectionProfile *ConnectionProfile `json:"connection_profile"`
}
-// TopologyEdgeCreated matches Python's TopologyEdgeCreated event
type TopologyEdgeCreated struct {
EventType string `json:"event_type"`
EventID string `json:"event_id"`
Edge Connection `json:"edge"`
}
-// TopologyEdgeDeleted matches Python's TopologyEdgeDeleted event
type TopologyEdgeDeleted struct {
EventType string `json:"event_type"`
EventID string `json:"event_id"`
Edge Connection `json:"edge"`
}
-// initEventsDB initializes the events database connection
func initEventsDB() error {
eventsDBMu.Lock()
defer eventsDBMu.Unlock()
-
if eventsDB != nil {
- return nil // Already initialized
+ return nil
}
-
if eventsDBPath == "" {
- return nil // No events DB configured
+ return nil
}
-
- var err error
- eventsDB, err = sql.Open("sqlite3", eventsDBPath)
+ db, err := sql.Open("sqlite3", eventsDBPath)
if err != nil {
return fmt.Errorf("failed to open events database: %w", err)
}
+ eventsDB = db
- // Create table if it doesn't exist (matching Python's schema)
- createTableSQL := `
+ const schema = `
CREATE TABLE IF NOT EXISTS events (
rowid INTEGER PRIMARY KEY AUTOINCREMENT,
origin TEXT NOT NULL,
@@ -108,34 +93,27 @@ func initEventsDB() error {
CREATE INDEX IF NOT EXISTS idx_events_event_type ON events(event_type);
CREATE INDEX IF NOT EXISTS idx_events_created_at ON events(created_at);
`
- _, err = eventsDB.Exec(createTableSQL)
- if err != nil {
+ if _, err := eventsDB.Exec(schema); err != nil {
eventsDB.Close()
eventsDB = nil
return fmt.Errorf("failed to create events table: %w", err)
}
-
return nil
}
-// writeEvent writes an event to the database
func writeEvent(eventType string, eventData interface{}) error {
if eventsDB == nil {
if err := initEventsDB(); err != nil {
return err
}
if eventsDB == nil {
- return nil // No events DB configured
+ return nil
}
}
-
- // Serialize event data to JSON
jsonData, err := json.Marshal(eventData)
if err != nil {
return fmt.Errorf("failed to marshal event data: %w", err)
}
-
- // Extract event ID from the event data
var eventID string
switch e := eventData.(type) {
case *TopologyEdgeCreated:
@@ -145,170 +123,97 @@ func writeEvent(eventType string, eventData interface{}) error {
default:
eventID = uuid.New().String()
}
-
- // Insert event into database
- insertSQL := `INSERT INTO events (origin, event_type, event_id, event_data) VALUES (?, ?, ?, ?)`
- _, err = eventsDB.Exec(insertSQL, GetNodeId(), eventType, eventID, string(jsonData))
- if err != nil {
- return fmt.Errorf("failed to insert event: %w", err)
- }
-
- return nil
+ const insert = `INSERT INTO events (origin, event_type, event_id, event_data) VALUES (?, ?, ?, ?)`
+ _, err = eventsDB.Exec(insert, GetNodeId(), eventType, eventID, string(jsonData))
+ return err
}
-// NotifeeHandler implements the libp2p network.Notifiee interface
-type NotifeeHandler struct{}
-
-// Listen is called when network starts listening on an addr
-func (n *NotifeeHandler) Listen(net network.Network, ma multiaddr.Multiaddr) {}
-
-// ListenClose is called when network stops listening on an addr
-func (n *NotifeeHandler) ListenClose(net network.Network, ma multiaddr.Multiaddr) {}
-
-// Connected is called when a connection is opened
-func (n *NotifeeHandler) Connected(net network.Network, conn network.Conn) {
- remotePeer := conn.RemotePeer()
- localAddr := conn.LocalMultiaddr()
- remoteAddr := conn.RemoteMultiaddr()
-
- // Check if we've already processed this connection
- connKey := fmt.Sprintf("%s-%s", conn.LocalPeer(), remotePeer)
- connTrackerMu.Lock()
- if connectionTracker[connKey] {
- connTrackerMu.Unlock()
- log.Printf("Skipping duplicate connection event for %s", remotePeer)
- return
+var WriteEdgeCreatedEvent = func(localNodeID, remoteNodeID, localIP, remoteIP, proto string) {
+ event := &TopologyEdgeCreated{
+ EventType: EventTypeTopologyEdgeCreated,
+ EventID: uuid.New().String(),
+ Edge: Connection{
+ LocalNodeID: localNodeID,
+ SendBackNodeID: remoteNodeID,
+ LocalMultiaddr: Multiaddr{
+ Address: fmt.Sprintf("/ip4/%s/tcp/7847", localIP),
+ IPv4Address: localIP,
+ Port: 7847,
+ Transport: proto,
+ },
+ SendBackMultiaddr: Multiaddr{
+ Address: fmt.Sprintf("/ip4/%s/tcp/7847", remoteIP),
+ IPv4Address: remoteIP,
+ Port: 7847,
+ Transport: proto,
+ },
+ ConnectionProfile: nil,
+ },
}
- connectionTracker[connKey] = true
- connTrackerMu.Unlock()
-
- // Get the local node ID
- localNodeID := GetNodeId()
-
- // Asynchronously exchange node IDs and write event
- go func() {
- mapper := GetNodeIDMapper()
-
- // Add a small delay to ensure both sides are ready
- time.Sleep(100 * time.Millisecond)
-
- // Exchange node IDs
- if err := mapper.ExchangeNodeID(remotePeer); err != nil {
- log.Printf("Failed to exchange node ID with %s: %v", remotePeer, err)
- // Don't write event if we can't get the node ID
- return
- }
-
- // Get the actual remote node ID
- remoteNodeID, ok := mapper.GetNodeIDForPeer(remotePeer)
- if !ok {
- log.Printf("Node ID not found for peer %s after successful exchange", remotePeer)
- return
- }
-
- // Write edge created event with correct node IDs
- writeEdgeCreatedEvent(localNodeID, remoteNodeID, localAddr, remoteAddr)
- }()
-}
-
-// Disconnected is called when a connection is closed
-func (n *NotifeeHandler) Disconnected(net network.Network, conn network.Conn) {
- remotePeer := conn.RemotePeer()
- localAddr := conn.LocalMultiaddr()
- remoteAddr := conn.RemoteMultiaddr()
-
- // Clear connection tracker
- connKey := fmt.Sprintf("%s-%s", conn.LocalPeer(), remotePeer)
- connTrackerMu.Lock()
- delete(connectionTracker, connKey)
- connTrackerMu.Unlock()
-
- // Get the actual node IDs (not peer IDs)
- localNodeID := GetNodeId()
-
- // Get the remote node ID from the mapper
- mapper := GetNodeIDMapper()
- remoteNodeID, ok := mapper.GetNodeIDForPeer(remotePeer)
- if !ok {
- // Don't write event if we don't have the node ID mapping
- log.Printf("No node ID mapping found for disconnected peer %s, skipping event", remotePeer)
- mapper.RemoveMapping(remotePeer)
- return
+ if err := writeEvent(EventTypeTopologyEdgeCreated, event); err != nil {
+ log.Printf("Failed to write edge created event: %v", err)
+ } else {
+ log.Printf("Wrote TCP edge created event: %s -> %s (%s:%s)", localNodeID, remoteNodeID, remoteIP, proto)
}
-
- // Clean up the mapping
- mapper.RemoveMapping(remotePeer)
+}
- // Create disconnection event
+var WriteEdgeDeletedEvent = func(localNodeID, remoteNodeID, localIP, remoteIP, proto string) {
event := &TopologyEdgeDeleted{
EventType: EventTypeTopologyEdgeDeleted,
EventID: uuid.New().String(),
Edge: Connection{
- LocalNodeID: localNodeID,
- SendBackNodeID: remoteNodeID,
- LocalMultiaddr: parseMultiaddr(localAddr),
- SendBackMultiaddr: parseMultiaddr(remoteAddr),
+ LocalNodeID: localNodeID,
+ SendBackNodeID: remoteNodeID,
+ LocalMultiaddr: Multiaddr{
+ Address: fmt.Sprintf("/ip4/%s/tcp/7847", localIP),
+ IPv4Address: localIP,
+ Port: 7847,
+ Transport: proto,
+ },
+ SendBackMultiaddr: Multiaddr{
+ Address: fmt.Sprintf("/ip4/%s/tcp/7847", remoteIP),
+ IPv4Address: remoteIP,
+ Port: 7847,
+ Transport: proto,
+ },
ConnectionProfile: nil,
},
}
-
- // Write event to database
if err := writeEvent(EventTypeTopologyEdgeDeleted, event); err != nil {
log.Printf("Failed to write edge deleted event: %v", err)
} else {
- log.Printf("Wrote edge deleted event: %s -> %s", localNodeID, remoteNodeID)
+ log.Printf("Wrote TCP edge deleted event: %s -> %s (%s:%s)", localNodeID, remoteNodeID, remoteIP, proto)
}
}
-// OpenedStream is called when a stream is opened
-func (n *NotifeeHandler) OpenedStream(net network.Network, str network.Stream) {}
+type NotifeeHandler struct{}
-// ClosedStream is called when a stream is closed
-func (n *NotifeeHandler) ClosedStream(net network.Network, str network.Stream) {}
+func (n *NotifeeHandler) Listen(net network.Network, ma multiaddr.Multiaddr) {}
+func (n *NotifeeHandler) ListenClose(net network.Network, ma multiaddr.Multiaddr) {}
+func (n *NotifeeHandler) Connected(netw network.Network, conn network.Conn) {
+ pid := conn.RemotePeer()
+ rawR := conn.RemoteMultiaddr()
-// parseMultiaddr converts a libp2p multiaddr to our Multiaddr struct
-func parseMultiaddr(ma multiaddr.Multiaddr) Multiaddr {
- result := Multiaddr{
- Address: ma.String(),
- }
-
- // Extract IPv4 address if present
- if ipStr, err := ma.ValueForProtocol(multiaddr.P_IP4); err == nil {
- result.IPv4Address = ipStr
+ if node != nil && node.ConnManager() != nil {
+ node.ConnManager().Protect(pid, "multipath-"+hostTransportKey(rawR))
}
-
- // Extract port if present
- if portStr, err := ma.ValueForProtocol(multiaddr.P_TCP); err == nil {
- if port, err := strconv.Atoi(portStr); err == nil {
- result.Port = port
+
+ if ipStr, err := rawR.ValueForProtocol(multiaddr.P_IP4); err == nil && ipStr != "" {
+ if ip := net.ParseIP(ipStr); ip != nil {
+ GetTCPAgent().UpdateDiscoveredIPs(pid, []net.IP{ip})
}
}
-
- return result
}
+func (n *NotifeeHandler) Disconnected(net network.Network, conn network.Conn) {
+ pid := conn.RemotePeer()
+ rawR := conn.RemoteMultiaddr()
-// writeEdgeCreatedEvent writes a topology edge created event
-func writeEdgeCreatedEvent(localNodeID, remoteNodeID string, localAddr, remoteAddr multiaddr.Multiaddr) {
- event := &TopologyEdgeCreated{
- EventType: EventTypeTopologyEdgeCreated,
- EventID: uuid.New().String(),
- Edge: Connection{
- LocalNodeID: localNodeID,
- SendBackNodeID: remoteNodeID,
- LocalMultiaddr: parseMultiaddr(localAddr),
- SendBackMultiaddr: parseMultiaddr(remoteAddr),
- ConnectionProfile: nil,
- },
- }
-
- if err := writeEvent(EventTypeTopologyEdgeCreated, event); err != nil {
- log.Printf("Failed to write edge created event: %v", err)
- } else {
- log.Printf("Wrote edge created event: %s -> %s", localNodeID, remoteNodeID)
+ if node != nil && node.ConnManager() != nil {
+ tag := "multipath-" + hostTransportKey(rawR)
+ node.ConnManager().Unprotect(pid, tag)
}
}
+func (n *NotifeeHandler) OpenedStream(net network.Network, str network.Stream) {}
+func (n *NotifeeHandler) ClosedStream(net network.Network, str network.Stream) {}
-// GetNotifee returns a singleton instance of the notifee handler
-func GetNotifee() network.Notifiee {
- return &NotifeeHandler{}
-}
\ No newline at end of file
+func GetNotifee() network.Notifiee { return &NotifeeHandler{} }
diff --git a/networking/forwarder/src/libp2p.go b/networking/forwarder/src/libp2p.go
index 798cfcbd..2b802707 100644
--- a/networking/forwarder/src/libp2p.go
+++ b/networking/forwarder/src/libp2p.go
@@ -22,6 +22,7 @@ import (
"github.com/libp2p/go-libp2p/core/peerstore"
"github.com/libp2p/go-libp2p/core/pnet"
mdns "github.com/libp2p/go-libp2p/p2p/discovery/mdns"
+ connmgr "github.com/libp2p/go-libp2p/p2p/net/connmgr"
"github.com/libp2p/go-libp2p/p2p/security/noise"
"github.com/multiformats/go-multiaddr"
)
@@ -29,37 +30,41 @@ import (
var node host.Host
var ps *pubsub.PubSub
var mdnsSer mdns.Service
+
var once sync.Once
var mu sync.Mutex
var refCount int
var topicsMap = make(map[string]*pubsub.Topic)
-// Connection retry state tracking
type peerConnState struct {
retryCount int
lastAttempt time.Time
}
-var peerLastAddrs = make(map[peer.ID][]multiaddr.Multiaddr)
-var addrsMu sync.Mutex
+type peerAddrKey struct {
+ id peer.ID
+ addr string // host+transport key (IP|transport)
+}
-var connecting = make(map[peer.ID]bool)
-var connMu sync.Mutex
-var peerRetryState = make(map[peer.ID]*peerConnState)
-var retryMu sync.Mutex
+var (
+ peerRetryState = make(map[peerAddrKey]*peerConnState)
+ retryMu sync.Mutex
-const (
- maxRetries = 5 // Increased for more tolerance to transient failures
- initialBackoff = 2 * time.Second
- maxBackoff = 33 * time.Second
- retryResetTime = 1 * time.Minute // Reduced for faster recovery after max retries
+ connecting = make(map[peerAddrKey]bool)
+ connMu sync.Mutex
+
+ mdnsRestartMu sync.Mutex
+ lastMdnsRestart time.Time
+ restartPending bool
+ minRestartSpacing = 2 * time.Second
)
-type discoveryNotifee struct {
- h host.Host
-}
+const (
+ connectTimeout = 25 * time.Second
+ mdnsFastInterval = 1 * time.Second
+ mdnsSlowInterval = 30 * time.Second
+)
-// sortAddrs returns a sorted copy of addresses for comparison
func sortAddrs(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
s := make([]multiaddr.Multiaddr, len(addrs))
copy(s, addrs)
@@ -69,7 +74,6 @@ func sortAddrs(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
return s
}
-// addrsChanged checks if two address sets differ
func addrsChanged(a, b []multiaddr.Multiaddr) bool {
if len(a) != len(b) {
return true
@@ -84,46 +88,73 @@ func addrsChanged(a, b []multiaddr.Multiaddr) bool {
return false
}
-// isAddressValid checks if an address should be used for connections
+func canonicalAddr(a multiaddr.Multiaddr) string {
+ cs := multiaddr.Split(a)
+ out := make([]multiaddr.Multiaddrer, 0, len(cs))
+ for _, c := range cs {
+ for _, p := range c.Protocols() {
+ if p.Code == multiaddr.P_P2P {
+ goto NEXT
+ }
+ }
+ out = append(out, c.Multiaddr())
+ NEXT:
+ }
+ return multiaddr.Join(out...).String()
+}
+
+func ipString(a multiaddr.Multiaddr) string {
+ if v, err := a.ValueForProtocol(multiaddr.P_IP4); err == nil {
+ return v
+ }
+ if v, err := a.ValueForProtocol(multiaddr.P_IP6); err == nil {
+ return v
+ }
+ return ""
+}
+
+func hostTransportKey(a multiaddr.Multiaddr) string {
+ ip := ipString(a)
+ t := "tcp"
+ if _, err := a.ValueForProtocol(multiaddr.P_QUIC_V1); err == nil {
+ t = "quic"
+ }
+ if _, err := a.ValueForProtocol(multiaddr.P_WS); err == nil {
+ t = "ws"
+ }
+ return ip + "|" + t
+}
+
func isAddressValid(addr multiaddr.Multiaddr) bool {
- // Allow loopback for testing if env var is set
allowLoopback := os.Getenv("FORWARDER_ALLOW_LOOPBACK") == "true"
- // Check IPv4 addresses
- ipStr, err := addr.ValueForProtocol(multiaddr.P_IP4)
- if err == nil && ipStr != "" {
+ if ipStr, err := addr.ValueForProtocol(multiaddr.P_IP4); err == nil && ipStr != "" {
ip := net.ParseIP(ipStr)
if ip == nil {
return false
}
- // Filter out loopback, unspecified addresses (unless testing)
if !allowLoopback && (ip.IsLoopback() || ip.IsUnspecified()) {
return false
}
if ip.IsUnspecified() {
return false
}
- // Filter out common VPN ranges (Tailscale uses 100.64.0.0/10)
- if ip.To4() != nil && ip.To4()[0] == 100 && ip.To4()[1] >= 64 && ip.To4()[1] <= 127 {
+ if b := ip.To4(); b != nil && b[0] == 100 && b[1] >= 64 && b[1] <= 127 {
return false
}
}
- // Check IPv6 addresses
- ipStr, err = addr.ValueForProtocol(multiaddr.P_IP6)
- if err == nil && ipStr != "" {
+ if ipStr, err := addr.ValueForProtocol(multiaddr.P_IP6); err == nil && ipStr != "" {
ip := net.ParseIP(ipStr)
if ip == nil {
return false
}
- // Filter out loopback, unspecified addresses (unless testing)
if !allowLoopback && (ip.IsLoopback() || ip.IsUnspecified()) {
return false
}
if ip.IsUnspecified() {
return false
}
- // Filter out Tailscale IPv6 (fd7a:115c:a1e0::/48)
if strings.HasPrefix(strings.ToLower(ipStr), "fd7a:115c:a1e0:") {
return false
}
@@ -132,7 +163,6 @@ func isAddressValid(addr multiaddr.Multiaddr) bool {
return true
}
-// customInterfaceAddresses returns IPs only from interfaces that are up and running (has link)
func customInterfaceAddresses() ([]net.IP, error) {
var ips []net.IP
ifaces, err := net.Interfaces()
@@ -140,15 +170,15 @@ func customInterfaceAddresses() ([]net.IP, error) {
return nil, err
}
for _, ifi := range ifaces {
- if ifi.Flags&net.FlagUp == 0 || ifi.Flags&net.FlagRunning == 0 {
+ if ifi.Flags&net.FlagUp == 0 {
continue
}
addrs, err := ifi.Addrs()
if err != nil {
return nil, err
}
- for _, addr := range addrs {
- if ipnet, ok := addr.(*net.IPNet); ok && ipnet.IP != nil {
+ for _, a := range addrs {
+ if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP != nil {
ips = append(ips, ipnet.IP)
}
}
@@ -156,7 +186,6 @@ func customInterfaceAddresses() ([]net.IP, error) {
return ips, nil
}
-// customAddrsFactory expands wildcard listen addrs to actual IPs on up+running interfaces, then filters
func customAddrsFactory(listenAddrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
ips, err := customInterfaceAddresses()
if err != nil {
@@ -177,22 +206,19 @@ func customAddrsFactory(listenAddrs []multiaddr.Multiaddr) []multiaddr.Multiaddr
}
code := protos[0].Code
val, err := first.ValueForProtocol(code)
- var isWildcard bool
- if err == nil && ((code == multiaddr.P_IP4 && val == "0.0.0.0") || (code == multiaddr.P_IP6 && val == "::")) {
- isWildcard = true
- }
+ isWildcard := (err == nil &&
+ ((code == multiaddr.P_IP4 && val == "0.0.0.0") ||
+ (code == multiaddr.P_IP6 && val == "::")))
if isWildcard {
- // Expand to each valid IP
for _, ip := range ips {
- var pcodeStr string
+ var pcode string
if ip.To4() != nil {
- pcodeStr = "4"
+ pcode = "4"
} else {
- pcodeStr = "6"
+ pcode = "6"
}
- newIPStr := "/ip" + pcodeStr + "/" + ip.String()
- newIPMA, err := multiaddr.NewMultiaddr(newIPStr)
+ newIPMA, err := multiaddr.NewMultiaddr("/ip" + pcode + "/" + ip.String())
if err != nil {
continue
}
@@ -201,9 +227,9 @@ func customAddrsFactory(listenAddrs []multiaddr.Multiaddr) []multiaddr.Multiaddr
for _, c := range comps[1:] {
newComps = append(newComps, c.Multiaddr())
}
- newa := multiaddr.Join(newComps...)
- if isAddressValid(newa) {
- advAddrs = append(advAddrs, newa)
+ newAddr := multiaddr.Join(newComps...)
+ if isAddressValid(newAddr) {
+ advAddrs = append(advAddrs, newAddr)
}
}
} else if isAddressValid(la) {
@@ -213,159 +239,128 @@ func customAddrsFactory(listenAddrs []multiaddr.Multiaddr) []multiaddr.Multiaddr
return advAddrs
}
+type discoveryNotifee struct{ h host.Host }
+
func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
log.Printf("mDNS discovered peer %s with %d addresses", pi.ID, len(pi.Addrs))
- // Check if already connected first
- if n.h.Network().Connectedness(pi.ID) == network.Connected {
- log.Printf("Already connected to peer %s", pi.ID)
- return
+ var ipList []string
+ for _, a := range pi.Addrs {
+ if v := ipString(a); v != "" {
+ ipList = append(ipList, v)
+ }
+ }
+ if len(ipList) > 0 {
+ log.Printf("mDNS %s IPs: %s", pi.ID, strings.Join(ipList, ", "))
}
- // Clear any existing addresses for this peer to ensure we use only fresh ones from mDNS
- ps := n.h.Peerstore()
- ps.ClearAddrs(pi.ID)
- log.Printf("Cleared old addresses for peer %s", pi.ID)
-
- // During normal operation, only higher ID connects to avoid double connections
- // But if we have retry state for this peer, both sides should attempt
- // Also, if we have no connections at all, both sides should attempt
- retryMu.Lock()
- _, hasRetryState := peerRetryState[pi.ID]
- retryMu.Unlock()
+ var filtered []multiaddr.Multiaddr
+ var ips []net.IP
+ for _, a := range pi.Addrs {
+ if isAddressValid(a) {
+ filtered = append(filtered, a)
- // Check if we should skip based on ID comparison
- // Skip only if we have a higher ID, no retry state, and we already have connections
- if n.h.ID() >= pi.ID && !hasRetryState && len(n.h.Network().Peers()) > 0 {
- log.Printf("Skipping initial connection to peer %s (lower ID)", pi.ID)
+ if ipStr := ipString(a); ipStr != "" {
+ if ip := net.ParseIP(ipStr); ip != nil {
+ ips = append(ips, ip)
+ }
+ }
+ }
+ }
+ if len(filtered) == 0 {
+ log.Printf("No valid addrs for %s", pi.ID)
return
}
- // Filter addresses before attempting connection
- var filteredAddrs []multiaddr.Multiaddr
- for _, addr := range pi.Addrs {
- if isAddressValid(addr) {
- filteredAddrs = append(filteredAddrs, addr)
- log.Printf("Valid address for %s: %s", pi.ID, addr)
- } else {
- log.Printf("Filtered out address for %s: %s", pi.ID, addr)
- }
- }
+ ps := n.h.Peerstore()
+ ps.AddAddrs(pi.ID, filtered, peerstore.TempAddrTTL)
- if len(filteredAddrs) == 0 {
- log.Printf("No valid addresses for peer %s after filtering, skipping connection attempt", pi.ID)
- return
+ tcpAgent := GetTCPAgent()
+ if len(ips) > 0 {
+ tcpAgent.UpdateDiscoveredIPs(pi.ID, ips)
}
- // Check for address changes and reset retries if changed
- addrsMu.Lock()
- lastAddrs := peerLastAddrs[pi.ID]
- addrsMu.Unlock()
- if addrsChanged(lastAddrs, filteredAddrs) {
- log.Printf("Detected address change for peer %s, resetting retry count", pi.ID)
- retryMu.Lock()
- if state, ok := peerRetryState[pi.ID]; ok {
- state.retryCount = 0
+ existing := make(map[string]struct{})
+ for _, c := range n.h.Network().ConnsToPeer(pi.ID) {
+ if cm, ok := c.(network.ConnMultiaddrs); ok {
+ existing[hostTransportKey(cm.RemoteMultiaddr())] = struct{}{}
}
- retryMu.Unlock()
- // Update last known addresses
- addrsMu.Lock()
- peerLastAddrs[pi.ID] = append([]multiaddr.Multiaddr(nil), filteredAddrs...) // Copy
- addrsMu.Unlock()
}
- pi.Addrs = filteredAddrs
-
- // Add the filtered addresses to the peerstore with a reasonable TTL
- ps.AddAddrs(pi.ID, filteredAddrs, peerstore.TempAddrTTL)
-
- // Attempt connection with retry logic
- go n.connectWithRetry(pi)
+ for _, a := range filtered {
+ if _, seen := existing[hostTransportKey(a)]; seen {
+ continue
+ }
+ go n.connectWithRetryToAddr(pi.ID, a)
+ }
}
-func (n *discoveryNotifee) connectWithRetry(pi peer.AddrInfo) {
- // Serialize connection attempts per peer
+func (n *discoveryNotifee) connectWithRetryToAddr(pid peer.ID, addr multiaddr.Multiaddr) {
+ key := peerAddrKey{pid, hostTransportKey(addr)}
+
connMu.Lock()
- if connecting[pi.ID] {
+ if connecting[key] {
connMu.Unlock()
- log.Printf("Already connecting to peer %s, skipping duplicate attempt", pi.ID)
return
}
- connecting[pi.ID] = true
+ connecting[key] = true
connMu.Unlock()
defer func() {
connMu.Lock()
- delete(connecting, pi.ID)
+ delete(connecting, key)
connMu.Unlock()
}()
retryMu.Lock()
- state, exists := peerRetryState[pi.ID]
- if !exists {
+ state, ok := peerRetryState[key]
+ if !ok {
state = &peerConnState{}
- peerRetryState[pi.ID] = state
+ peerRetryState[key] = state
}
-
- // Check if we've exceeded max retries
- if state.retryCount >= maxRetries {
- // Check if enough time has passed to reset retry count
- if time.Since(state.lastAttempt) > retryResetTime {
- state.retryCount = 0
- log.Printf("Reset retry count for peer %s due to time elapsed", pi.ID)
- } else {
- retryMu.Unlock()
- log.Printf("Max retries reached for peer %s, skipping", pi.ID)
- return
- }
- }
-
- // Calculate backoff duration
- backoffDuration := time.Duration(1<<uint(state.retryCount)) * initialBackoff
- if backoffDuration > maxBackoff {
- backoffDuration = maxBackoff
+ backoff := time.Duration(1<<uint(state.retryCount)) * initialBackoff
+ if backoff > maxBackoff {
+ backoff = maxBackoff
}
-
- // Check if we need to wait before retrying
- if state.retryCount > 0 && time.Since(state.lastAttempt) < backoffDuration {
+ if state.retryCount > 0 && time.Since(state.lastAttempt) < backoff {
retryMu.Unlock()
- log.Printf("Backoff active for peer %s, skipping attempt", pi.ID)
return
}
-
state.lastAttempt = time.Now()
retryMu.Unlock()
- ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ ai := peer.AddrInfo{ID: pid, Addrs: []multiaddr.Multiaddr{addr}}
+
+ ctx, cancel := context.WithTimeout(network.WithForceDirectDial(context.Background(), "ensure-multipath"), connectTimeout)
defer cancel()
- if err := n.h.Connect(ctx, pi); err != nil {
- log.Printf("Failed to connect to %s (attempt %d/%d): %v", pi.ID, state.retryCount+1, maxRetries, err)
+ n.h.Peerstore().AddAddrs(pid, []multiaddr.Multiaddr{addr}, peerstore.TempAddrTTL)
+ if err := n.h.Connect(ctx, ai); err != nil {
+ log.Printf("Dial %s@%s failed (attempt %d): %v", pid, addr, state.retryCount+1, err)
retryMu.Lock()
state.retryCount++
retryMu.Unlock()
- // Schedule retry if we haven't exceeded max attempts
- if state.retryCount < maxRetries {
- time.AfterFunc(backoffDuration, func() {
- // Check if we're still not connected before retrying
- if n.h.Network().Connectedness(pi.ID) != network.Connected {
- n.connectWithRetry(pi)
+ time.AfterFunc(backoff, func() {
+ pathStillMissing := true
+ for _, c := range n.h.Network().ConnsToPeer(pid) {
+ if cm, ok := c.(network.ConnMultiaddrs); ok &&
+ hostTransportKey(cm.RemoteMultiaddr()) == key.addr {
+ pathStillMissing = false
+ break
}
- })
- }
- } else {
- log.Printf("Successfully connected to %s", pi.ID)
-
- // Reset retry state on successful connection
- retryMu.Lock()
- delete(peerRetryState, pi.ID)
- retryMu.Unlock()
- addrsMu.Lock()
- delete(peerLastAddrs, pi.ID)
- addrsMu.Unlock()
- log.Printf("Cleared last addresses for disconnected peer %s", pi.ID)
+ }
+ if pathStillMissing {
+ n.connectWithRetryToAddr(pid, addr)
+ }
+ })
+ return
}
+
+ log.Printf("Connected to %s via %s", pid, addr)
+ retryMu.Lock()
+ delete(peerRetryState, key)
+ retryMu.Unlock()
}
func getPrivKey(nodeId string) (crypto.PrivKey, error) {
@@ -380,7 +375,9 @@ func getPrivKey(nodeId string) (crypto.PrivKey, error) {
func getNode(ctx context.Context) {
once.Do(func() {
nodeId := GetNodeId()
+
var opts []libp2p.Option
+
priv, err := getPrivKey(nodeId)
if err != nil {
log.Fatalf("failed to generate key: %v", err)
@@ -392,31 +389,30 @@ func getNode(ctx context.Context) {
psk := pnet.PSK(pskHash[:])
opts = append(opts, libp2p.PrivateNetwork(psk))
- // Performance optimizations
- opts = append(opts, libp2p.ConnectionManager(nil)) // No connection limits
- opts = append(opts, libp2p.EnableHolePunching()) // Better NAT traversal
- opts = append(opts, libp2p.EnableRelay()) // Allow relaying
+ opts = append(opts, libp2p.EnableHolePunching())
+ opts = append(opts, libp2p.EnableRelay())
- // Custom address factory to avoid advertising down interfaces
opts = append(opts, libp2p.AddrsFactory(customAddrsFactory))
- node, err = libp2p.New(opts...)
- if err != nil {
- log.Fatalf("failed to create host: %v", err)
+ cm, _ := connmgr.NewConnManager(100, 1000, connmgr.WithGracePeriod(2*time.Minute))
+ opts = append(opts, libp2p.ConnectionManager(cm))
+
+ var errNode error
+ node, errNode = libp2p.New(opts...)
+ if errNode != nil {
+ log.Fatalf("failed to create host: %v", errNode)
}
- // Configure GossipSub for better performance
gossipOpts := []pubsub.Option{
- pubsub.WithMessageSigning(false), // Disable message signing for speed
- pubsub.WithStrictSignatureVerification(false), // Disable signature verification
- pubsub.WithMaxMessageSize(1024 * 1024), // 1MB max message size for batches
- pubsub.WithValidateQueueSize(1000), // Larger validation queue
- pubsub.WithPeerOutboundQueueSize(1000), // Larger peer queues
+ pubsub.WithMessageSigning(false),
+ pubsub.WithStrictSignatureVerification(false),
+ pubsub.WithMaxMessageSize(1024 * 1024),
+ pubsub.WithValidateQueueSize(1000),
+ pubsub.WithPeerOutboundQueueSize(1000),
}
-
ps, err = pubsub.NewGossipSub(ctx, node, gossipOpts...)
if err != nil {
- node.Close()
+ _ = node.Close()
log.Fatalf("failed to create pubsub: %v", err)
}
@@ -424,117 +420,144 @@ func getNode(ctx context.Context) {
notifee := &discoveryNotifee{h: node}
mdnsSer = mdns.NewMdnsService(node, rendezvous, notifee)
if err := mdnsSer.Start(); err != nil {
- node.Close()
+ _ = node.Close()
log.Fatalf("failed to start mdns service: %v", err)
}
- // Register disconnect notifiee to clear stale addresses
node.Network().Notify(&disconnectNotifee{})
-
- // Register event notifiee to track topology changes
node.Network().Notify(GetNotifee())
-
- // Set up node ID mapper
- GetNodeIDMapper().SetHost(node)
- // Start a goroutine to periodically trigger mDNS discovery
+ tcpAgent := GetTCPAgent()
+ if err := tcpAgent.Start(ctx, node.ID()); err != nil {
+ log.Printf("Failed to start TCP agent: %v", err)
+ }
+
go periodicMDNSDiscovery()
+ go watchInterfacesAndKickMDNS()
})
}
-// periodicMDNSDiscovery ensures mDNS continues to work after network changes
func periodicMDNSDiscovery() {
- // Start with faster checks, then slow down
- fastCheckDuration := 5 * time.Second
- slowCheckDuration := 30 * time.Second
- currentDuration := fastCheckDuration
- noConnectionCount := 0
+ current := mdnsSlowInterval
+ t := time.NewTicker(current)
+ defer t.Stop()
- ticker := time.NewTicker(currentDuration)
- defer ticker.Stop()
+ lastNoPeerRestart := time.Time{}
- for range ticker.C {
+ for range t.C {
if mdnsSer == nil || node == nil {
return
}
-
- // Log current connection status
- peers := node.Network().Peers()
- if len(peers) == 0 {
- noConnectionCount++
- log.Printf("No connected peers (check #%d), mDNS service running: %v", noConnectionCount, mdnsSer != nil)
-
- // Force mDNS to re-announce when we have no peers
- // This helps recovery after network interface changes
- if noConnectionCount > 1 { // Skip first check to avoid unnecessary restart
- forceRestartMDNS()
+ n := len(node.Network().Peers())
+ if n == 0 {
+ if current != mdnsFastInterval {
+ current = mdnsFastInterval
+ t.Reset(current)
}
-
- // Keep fast checking when disconnected
- if currentDuration != fastCheckDuration {
- currentDuration = fastCheckDuration
- ticker.Reset(currentDuration)
- log.Printf("Switching to fast mDNS checks (every %v)", currentDuration)
+ if time.Since(lastNoPeerRestart) > 5*time.Second {
+ forceRestartMDNS("no-peers")
+ lastNoPeerRestart = time.Now()
}
} else {
- log.Printf("Currently connected to %d peers", len(peers))
- noConnectionCount = 0
-
- // Switch to slow checking when connected
- if currentDuration != slowCheckDuration {
- currentDuration = slowCheckDuration
- ticker.Reset(currentDuration)
- log.Printf("Switching to slow mDNS checks (every %v)", currentDuration)
+ if current != mdnsSlowInterval {
+ current = mdnsSlowInterval
+ t.Reset(current)
}
}
}
}
-// forceRestartMDNS restarts the mDNS service to force re-announcement
-func forceRestartMDNS() {
+func watchInterfacesAndKickMDNS() {
+ snap := interfacesSignature()
+ t := time.NewTicker(1 * time.Second)
+ defer t.Stop()
+
+ for range t.C {
+ next := interfacesSignature()
+ if next != snap {
+ snap = next
+ kickMDNSBurst("iface-change")
+ }
+ }
+}
+
+func kickMDNSBurst(reason string) {
+ forceRestartMDNS(reason)
+ time.AfterFunc(2*time.Second, func() { forceRestartMDNS(reason + "-stabilize-2s") })
+ time.AfterFunc(6*time.Second, func() { forceRestartMDNS(reason + "-stabilize-6s") })
+}
+
+func interfacesSignature() string {
+ ifaces, _ := net.Interfaces()
+ var b strings.Builder
+ for _, ifi := range ifaces {
+ if ifi.Flags&net.FlagUp == 0 {
+ continue
+ }
+ addrs, _ := ifi.Addrs()
+ b.WriteString(ifi.Name)
+ b.WriteByte('|')
+ b.WriteString(ifi.Flags.String())
+ for _, a := range addrs {
+ b.WriteByte('|')
+ b.WriteString(a.String())
+ }
+ b.WriteByte(';')
+ }
+ return b.String()
+}
+
+func forceRestartMDNS(reason string) {
+ mdnsRestartMu.Lock()
+ defer mdnsRestartMu.Unlock()
+
+ now := time.Now()
+ if restartPending || now.Sub(lastMdnsRestart) < minRestartSpacing {
+ if !restartPending {
+ restartPending = true
+ wait := minRestartSpacing - now.Sub(lastMdnsRestart)
+ if wait < 0 {
+ wait = minRestartSpacing
+ }
+ time.AfterFunc(wait, func() {
+ forceRestartMDNS("coalesced")
+ })
+ }
+ return
+ }
+ restartPending = false
+ lastMdnsRestart = now
+
mu.Lock()
defer mu.Unlock()
if mdnsSer != nil && node != nil {
- log.Printf("Force restarting mDNS service for re-announcement")
- oldMdns := mdnsSer
+ log.Printf("Restarting mDNS (%s)", reason)
+ old := mdnsSer
rendezvous := "forwarder_network"
notifee := &discoveryNotifee{h: node}
newMdns := mdns.NewMdnsService(node, rendezvous, notifee)
-
if err := newMdns.Start(); err != nil {
- log.Printf("Failed to restart mDNS service: %v", err)
- } else {
- oldMdns.Close()
- mdnsSer = newMdns
- log.Printf("Successfully restarted mDNS service")
+ log.Printf("Failed to restart mDNS: %v", err)
+ return
}
+ _ = old.Close()
+ mdnsSer = newMdns
+ GetTCPAgent().OnInterfaceChange()
+
+ retryMu.Lock()
+ peerRetryState = make(map[peerAddrKey]*peerConnState)
+ retryMu.Unlock()
}
}
-// disconnectNotifee clears stale peer addresses on disconnect
type disconnectNotifee struct{}
func (d *disconnectNotifee) Connected(network.Network, network.Conn) {}
func (d *disconnectNotifee) Disconnected(n network.Network, c network.Conn) {
- p := c.RemotePeer()
- ps := n.Peerstore()
-
- // Clear all addresses from peerstore to force fresh discovery on reconnect
- ps.ClearAddrs(p)
-
- // Also clear retry state for this peer
- retryMu.Lock()
- delete(peerRetryState, p)
- retryMu.Unlock()
-
- log.Printf("Cleared stale addresses and retry state for disconnected peer %s", p)
-
- // Try to restart mDNS discovery after a short delay to handle network interface changes
go func() {
- time.Sleep(2 * time.Second)
- log.Printf("Triggering mDNS re-discovery after disconnect")
- forceRestartMDNS()
+ time.Sleep(400 * time.Millisecond)
+ forceRestartMDNS("disconnect")
}()
}
func (d *disconnectNotifee) OpenedStream(network.Network, network.Stream) {}
@@ -551,7 +574,6 @@ type libP2PConnector struct {
ctx context.Context
cancel context.CancelFunc
- // Async publishing
writeChan chan RecordData
batchSize int
batchTimeout time.Duration
@@ -571,7 +593,6 @@ func newLibP2PConnector(topic string, ctx context.Context, cancel context.Cancel
}
topicsMap[topic] = t
}
-
t2, okResend := topicsMap[topic+"/resend"]
if !okResend {
t2, err = ps.Join(topic + "/resend")
@@ -581,11 +602,10 @@ func newLibP2PConnector(topic string, ctx context.Context, cancel context.Cancel
}
topicsMap[topic+"/resend"] = t2
}
-
refCount++
mu.Unlock()
- connector := &libP2PConnector{
+ conn := &libP2PConnector{
topic: topic,
top: t,
topResend: t2,
@@ -596,10 +616,8 @@ func newLibP2PConnector(topic string, ctx context.Context, cancel context.Cancel
batchTimeout: 10 * time.Millisecond,
workerPool: 5,
}
-
- connector.startAsyncPublishers()
-
- return connector
+ conn.startAsyncPublishers()
+ return conn
}
func (c *libP2PConnector) tail(handler func(record RecordData) error) {
@@ -631,8 +649,7 @@ func handleSub[T any](sub *pubsub.Subscription, ctx context.Context, handler fun
return
}
var rec T
- err = json.Unmarshal(msg.Data, &rec)
- if err != nil {
+ if err := json.Unmarshal(msg.Data, &rec); err != nil {
log.Printf("unmarshal error for topic %s: %v", sub.Topic(), err)
continue
}
@@ -654,38 +671,31 @@ func handleRecordSub(sub *pubsub.Subscription, ctx context.Context, handler func
log.Printf("subscription error for topic %s: %v", sub.Topic(), err)
return
}
-
- // Try to unmarshal as batch first
var batch BatchRecord
if err := json.Unmarshal(msg.Data, &batch); err == nil && len(batch.Records) > 0 {
- // Handle batched records
- for _, record := range batch.Records {
+ for _, r := range batch.Records {
if handler != nil {
- if err := handler(record); err != nil {
+ if err := handler(r); err != nil {
log.Printf("handler error for batched record: %v", err)
}
}
}
continue
}
-
- // Try to unmarshal as single record (backwards compatibility)
- var record RecordData
- if err := json.Unmarshal(msg.Data, &record); err == nil {
+ var single RecordData
+ if err := json.Unmarshal(msg.Data, &single); err == nil {
if handler != nil {
- if err := handler(record); err != nil {
+ if err := handler(single); err != nil {
log.Printf("handler error for single record: %v", err)
}
}
continue
}
-
- log.Printf("failed to unmarshal message as batch or single record for topic %s", sub.Topic())
+ log.Printf("failed to unmarshal message for topic %s", sub.Topic())
}
}
func (c *libP2PConnector) startAsyncPublishers() {
- // Start worker pool for batched async publishing
for i := 0; i < c.workerPool; i++ {
go c.publishWorker()
}
@@ -699,36 +709,28 @@ func (c *libP2PConnector) publishWorker() {
for {
select {
case <-c.ctx.Done():
- // Flush final batch
if len(batch) > 0 {
- err := c.publishBatch(batch)
- if err != nil {
+ if err := c.publishBatch(batch); err != nil {
log.Printf("Error publishing batch: %v", err)
}
}
return
- case record := <-c.writeChan:
- batch = append(batch, record)
-
- // Check if we should flush
+ case r := <-c.writeChan:
+ batch = append(batch, r)
if len(batch) >= c.batchSize {
- err := c.publishBatch(batch)
- if err != nil {
+ if err := c.publishBatch(batch); err != nil {
log.Printf("Error publishing batch: %v", err)
}
batch = batch[:0]
timer.Stop()
} else if len(batch) == 1 {
- // First record in batch, start timer
timer.Reset(c.batchTimeout)
}
case <-timer.C:
- // Timer expired, flush whatever we have
if len(batch) > 0 {
- err := c.publishBatch(batch)
- if err != nil {
+ if err := c.publishBatch(batch); err != nil {
log.Printf("Error publishing batch: %v", err)
}
batch = batch[:0]
@@ -741,24 +743,15 @@ func (c *libP2PConnector) publishBatch(records []RecordData) error {
if len(records) == 0 {
return nil
}
-
- // Create batch record
- batchRecord := BatchRecord{Records: records}
-
- data, err := json.Marshal(batchRecord)
+ data, err := json.Marshal(BatchRecord{Records: records})
if err != nil {
return err
}
-
- // Publish with timeout to prevent blocking
go func() {
- pubCtx, pubCancel := context.WithTimeout(c.ctx, 100*time.Millisecond)
- defer pubCancel()
-
- if err := c.top.Publish(pubCtx, data); err != nil {
- if err != context.DeadlineExceeded {
- log.Printf("Error publishing batch of %d records: %v", len(records), err)
- }
+ pubCtx, cancel := context.WithTimeout(c.ctx, 100*time.Millisecond)
+ defer cancel()
+ if err := c.top.Publish(pubCtx, data); err != nil && err != context.DeadlineExceeded {
+ log.Printf("Error publishing batch of %d: %v", len(records), err)
}
}()
return nil
@@ -771,7 +764,6 @@ func (c *libP2PConnector) write(record RecordData) error {
case <-c.ctx.Done():
return c.ctx.Err()
default:
- // Channel full, try to publish directly
return c.publishSingle(record)
}
}
@@ -813,8 +805,8 @@ func (c *libP2PConnector) close() error {
if c.subResend != nil {
c.subResend.Cancel()
}
+
if closeHost {
- // close all topics when shutting down host
for _, top := range topicsMap {
_ = top.Close()
}
@@ -832,19 +824,20 @@ func (c *libP2PConnector) close() error {
mdnsSer = nil
}
+ tcpAgent := GetTCPAgent()
+ if err := tcpAgent.Stop(); err != nil {
+ log.Printf("Error stopping TCP agent: %v", err)
+ }
+
var err error
if node != nil {
err = node.Close()
}
-
node = nil
ps = nil
refCount = 0
once = sync.Once{}
-
return err
}
-func (c *libP2PConnector) getType() string {
- return "libp2p"
-}
+func (c *libP2PConnector) getType() string { return "libp2p" }
diff --git a/networking/forwarder/src/node_id_exchange.go b/networking/forwarder/src/node_id_exchange.go
deleted file mode 100644
index e584f83a..00000000
--- a/networking/forwarder/src/node_id_exchange.go
+++ /dev/null
@@ -1,185 +0,0 @@
-package forwarder
-
-import (
- "bufio"
- "context"
- "encoding/json"
- "fmt"
- "log"
- "sync"
- "time"
-
- "github.com/libp2p/go-libp2p/core/host"
- "github.com/libp2p/go-libp2p/core/network"
- "github.com/libp2p/go-libp2p/core/peer"
-)
-
-const (
- // NodeIDExchangeProtocol is the protocol ID for node ID exchange
- NodeIDExchangeProtocol = "/forwarder/nodeid/1.0.0"
-
- // Exchange timeout - balanced for reliability
- exchangeTimeout = 5 * time.Second
-)
-
-// NodeIDMessage is the message format for node ID exchange
-type NodeIDMessage struct {
- NodeID string `json:"node_id"`
-}
-
-// NodeIDMapper manages the mapping between peer IDs and node IDs
-type NodeIDMapper struct {
- mu sync.RWMutex
- peerToNode map[peer.ID]string
- nodeToPeer map[string]peer.ID
- host host.Host
-}
-
-var (
- nodeIDMapper *NodeIDMapper
- mapperOnce sync.Once
-)
-
-// GetNodeIDMapper returns the singleton NodeIDMapper instance
-func GetNodeIDMapper() *NodeIDMapper {
- mapperOnce.Do(func() {
- nodeIDMapper = &NodeIDMapper{
- peerToNode: make(map[peer.ID]string),
- nodeToPeer: make(map[string]peer.ID),
- }
- })
- return nodeIDMapper
-}
-
-// SetHost sets the libp2p host for the mapper
-func (m *NodeIDMapper) SetHost(h host.Host) {
- m.mu.Lock()
- defer m.mu.Unlock()
- m.host = h
-
- // Set up the stream handler for incoming node ID exchanges
- h.SetStreamHandler(NodeIDExchangeProtocol, m.handleNodeIDStream)
-}
-
-// GetNodeIDForPeer returns the node ID for a given peer ID
-func (m *NodeIDMapper) GetNodeIDForPeer(peerID peer.ID) (string, bool) {
- m.mu.RLock()
- defer m.mu.RUnlock()
- nodeID, ok := m.peerToNode[peerID]
- return nodeID, ok
-}
-
-// GetPeerIDForNode returns the peer ID for a given node ID
-func (m *NodeIDMapper) GetPeerIDForNode(nodeID string) (peer.ID, bool) {
- m.mu.RLock()
- defer m.mu.RUnlock()
- peerID, ok := m.nodeToPeer[nodeID]
- return peerID, ok
-}
-
-// SetMapping sets the mapping between a peer ID and node ID
-func (m *NodeIDMapper) SetMapping(peerID peer.ID, nodeID string) {
- m.mu.Lock()
- defer m.mu.Unlock()
- m.peerToNode[peerID] = nodeID
- m.nodeToPeer[nodeID] = peerID
- log.Printf("Mapped peer %s to node %s", peerID, nodeID)
-}
-
-// RemoveMapping removes the mapping for a peer
-func (m *NodeIDMapper) RemoveMapping(peerID peer.ID) {
- m.mu.Lock()
- defer m.mu.Unlock()
- if nodeID, ok := m.peerToNode[peerID]; ok {
- delete(m.peerToNode, peerID)
- delete(m.nodeToPeer, nodeID)
- log.Printf("Removed mapping for peer %s (was node %s)", peerID, nodeID)
- }
-}
-
-// ExchangeNodeID initiates a node ID exchange with a peer
-func (m *NodeIDMapper) ExchangeNodeID(peerID peer.ID) error {
- if m.host == nil {
- return fmt.Errorf("host not set")
- }
-
- // Check if we already have the mapping
- if _, ok := m.GetNodeIDForPeer(peerID); ok {
- return nil // Already have the mapping
- }
-
- // Try up to 3 times with exponential backoff
- var lastErr error
- for attempt := 0; attempt < 3; attempt++ {
- if attempt > 0 {
- // Exponential backoff: 100ms, 200ms, 400ms
- time.Sleep(time.Duration(100<<uint(attempt-1)) * time.Millisecond)
- }
-
- ctx, cancel := context.WithTimeout(context.Background(), exchangeTimeout)
-
- // Open a stream to the peer
- stream, err := m.host.NewStream(ctx, peerID, NodeIDExchangeProtocol)
- if err != nil {
- cancel()
- lastErr = fmt.Errorf("failed to open stream: %w", err)
- continue
- }
-
- // Send our node ID
- msg := NodeIDMessage{NodeID: GetNodeId()}
- encoder := json.NewEncoder(stream)
- if err := encoder.Encode(&msg); err != nil {
- stream.Close()
- cancel()
- lastErr = fmt.Errorf("failed to send node ID: %w", err)
- continue
- }
-
- // Read their node ID
- decoder := json.NewDecoder(bufio.NewReader(stream))
- var response NodeIDMessage
- if err := decoder.Decode(&response); err != nil {
- stream.Close()
- cancel()
- lastErr = fmt.Errorf("failed to read node ID: %w", err)
- continue
- }
-
- stream.Close()
- cancel()
-
- // Store the mapping
- m.SetMapping(peerID, response.NodeID)
-
- return nil
- }
-
- return lastErr
-}
-
-// handleNodeIDStream handles incoming node ID exchange requests
-func (m *NodeIDMapper) handleNodeIDStream(stream network.Stream) {
- defer stream.Close()
-
- peerID := stream.Conn().RemotePeer()
-
- // Read their node ID
- decoder := json.NewDecoder(bufio.NewReader(stream))
- var msg NodeIDMessage
- if err := decoder.Decode(&msg); err != nil {
- log.Printf("Failed to read node ID from %s: %v", peerID, err)
- return
- }
-
- // Store the mapping
- m.SetMapping(peerID, msg.NodeID)
-
- // Send our node ID back
- response := NodeIDMessage{NodeID: GetNodeId()}
- encoder := json.NewEncoder(stream)
- if err := encoder.Encode(&response); err != nil {
- log.Printf("Failed to send node ID to %s: %v", peerID, err)
- return
- }
-}
\ No newline at end of file
diff --git a/networking/forwarder/src/node_id_exchange_test.go b/networking/forwarder/src/node_id_exchange_test.go
deleted file mode 100644
index 8803e991..00000000
--- a/networking/forwarder/src/node_id_exchange_test.go
+++ /dev/null
@@ -1,111 +0,0 @@
-package forwarder
-
-import (
- "bufio"
- "context"
- "encoding/json"
- "log"
- "testing"
- "time"
-
- "github.com/libp2p/go-libp2p"
- "github.com/libp2p/go-libp2p/core/network"
- "github.com/libp2p/go-libp2p/core/peer"
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-// mockNodeIDStreamHandler creates a stream handler that responds with a specific node ID
-func mockNodeIDStreamHandler(nodeID string) func(stream network.Stream) {
- return func(stream network.Stream) {
- defer stream.Close()
-
- peerID := stream.Conn().RemotePeer()
-
- // Read their node ID
- decoder := json.NewDecoder(bufio.NewReader(stream))
- var msg NodeIDMessage
- if err := decoder.Decode(&msg); err != nil {
- log.Printf("Failed to read node ID from %s: %v", peerID, err)
- return
- }
-
- // Send our node ID back
- response := NodeIDMessage{NodeID: nodeID}
- encoder := json.NewEncoder(stream)
- if err := encoder.Encode(&response); err != nil {
- log.Printf("Failed to send node ID to %s: %v", peerID, err)
- return
- }
- }
-}
-
-func TestNodeIDExchange(t *testing.T) {
- // Create two test hosts
- h1, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"))
- require.NoError(t, err)
- defer h1.Close()
-
- h2, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"))
- require.NoError(t, err)
- defer h2.Close()
-
- // Set up node ID for host 1
- SetNodeId("node-1")
- mapper1 := GetNodeIDMapper()
- mapper1.SetHost(h1)
-
- // Set up host 2 with a mock handler that responds with "node-2"
- h2.SetStreamHandler(NodeIDExchangeProtocol, mockNodeIDStreamHandler("node-2"))
-
- // Connect the hosts
- h1.Peerstore().AddAddrs(h2.ID(), h2.Addrs(), 3600)
- ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
- defer cancel()
-
- err = h1.Connect(ctx, peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
- require.NoError(t, err)
-
- // Exchange node IDs
- err = mapper1.ExchangeNodeID(h2.ID())
- require.NoError(t, err)
-
- // Verify the mapping on host 1
- nodeID, ok := mapper1.GetNodeIDForPeer(h2.ID())
- assert.True(t, ok)
- assert.Equal(t, "node-2", nodeID)
-}
-
-func TestNodeIDMapperOperations(t *testing.T) {
- mapper := &NodeIDMapper{
- peerToNode: make(map[peer.ID]string),
- nodeToPeer: make(map[string]peer.ID),
- }
-
- // Test peer ID (simulated)
- peerID := peer.ID("test-peer-id")
- nodeID := "test-node-id"
-
- // Set mapping
- mapper.SetMapping(peerID, nodeID)
-
- // Verify forward mapping
- gotNodeID, ok := mapper.GetNodeIDForPeer(peerID)
- assert.True(t, ok)
- assert.Equal(t, nodeID, gotNodeID)
-
- // Verify reverse mapping
- gotPeerID, ok := mapper.GetPeerIDForNode(nodeID)
- assert.True(t, ok)
- assert.Equal(t, peerID, gotPeerID)
-
- // Remove mapping
- mapper.RemoveMapping(peerID)
-
- // Verify removal
- _, ok = mapper.GetNodeIDForPeer(peerID)
- assert.False(t, ok)
-
- _, ok = mapper.GetPeerIDForNode(nodeID)
- assert.False(t, ok)
-}
\ No newline at end of file
diff --git a/networking/forwarder/src/tcp_agent.go b/networking/forwarder/src/tcp_agent.go
new file mode 100644
index 00000000..b4b1a3e9
--- /dev/null
+++ b/networking/forwarder/src/tcp_agent.go
@@ -0,0 +1,678 @@
+package forwarder
+
+import (
+ "bufio"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "net"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/libp2p/go-libp2p/core/peer"
+)
+
+const (
+ AgentPort = 7847
+
+ HandshakeTimeout = 5 * time.Second
+ HeartbeatInterval = 1 * time.Second
+ HeartbeatReadGrace = 4 * time.Second
+ HeartbeatWriteGrace = 3 * time.Second
+
+ tbGraceWindow = 90 * time.Second
+
+ dialTimeoutDefault = 6 * time.Second
+ dialTimeoutLinkLocal = 12 * time.Second
+ initialBackoff = 500 * time.Millisecond
+ maxBackoff = 60 * time.Second
+
+ scheduleTick = 300 * time.Millisecond
+ maxConcurrentDials = 32
+
+ ttlDiscovered = 5 * time.Minute
+ ttlObserved = 20 * time.Minute
+)
+
+type HandshakeMessage struct {
+ NodeID string `json:"node_id"`
+ AgentVer string `json:"agent_version"`
+ PeerID string `json:"peer_id"`
+ IPv4s []string `json:"ipv4s,omitempty"`
+ Timestamp int64 `json:"timestamp"`
+}
+
+type Edge struct {
+ LocalNodeID string
+ RemoteNodeID string
+ LocalIP string
+ RemoteIP string
+ Proto string
+}
+
+func (e Edge) Key() string {
+ return fmt.Sprintf("%s|%s|%s|%s|%s", e.LocalNodeID, e.RemoteNodeID, e.LocalIP, e.RemoteIP, e.Proto)
+}
+
+type connTrack struct {
+ tc *net.TCPConn
+ edge Edge
+ dialer bool
+ closed chan struct{}
+ closeMx sync.Once
+}
+
+type ipStamp struct {
+ seenAt time.Time
+ ttl time.Duration
+}
+
+type dialState struct {
+ backoff time.Duration
+ nextAttempt time.Time
+ connecting bool
+}
+
+type TCPAgent struct {
+ nodeID string
+ myPeerID peer.ID
+
+ listener *net.TCPListener
+ ctx context.Context
+ cancel context.CancelFunc
+
+ edgesMu sync.RWMutex
+ activeEdges map[string]*connTrack
+
+ activeByRemoteIPMu sync.RWMutex
+ activeByRemoteIP map[string]bool
+
+ ipDBMu sync.RWMutex
+ ipDB map[peer.ID]map[string]ipStamp
+
+ dialStatesMu sync.Mutex
+ dialStates map[string]*dialState
+ stopScheduler chan struct{}
+ schedulerOnce sync.Once
+ schedulerWG sync.WaitGroup
+
+ dialSem chan struct{}
+
+ ifaceGraceUntilMu sync.RWMutex
+ ifaceGraceUntil time.Time
+}
+
+var (
+ TCPAgentInstance *TCPAgent
+ TCPAgentOnce sync.Once
+)
+
+func GetTCPAgent() *TCPAgent {
+ TCPAgentOnce.Do(func() {
+ TCPAgentInstance = &TCPAgent{
+ nodeID: GetNodeId(),
+ activeEdges: make(map[string]*connTrack),
+ activeByRemoteIP: make(map[string]bool),
+ ipDB: make(map[peer.ID]map[string]ipStamp),
+ dialStates: make(map[string]*dialState),
+ stopScheduler: make(chan struct{}),
+ dialSem: make(chan struct{}, maxConcurrentDials),
+ }
+ })
+ return TCPAgentInstance
+}
+
+func (a *TCPAgent) Start(ctx context.Context, myPeerID peer.ID) error {
+ a.nodeID = GetNodeId()
+ a.myPeerID = myPeerID
+
+ ctx2, cancel := context.WithCancel(ctx)
+ a.ctx, a.cancel = ctx2, cancel
+
+ ln, err := net.ListenTCP("tcp", &net.TCPAddr{Port: AgentPort})
+ if err != nil {
+ return fmt.Errorf("failed to start TCP agent listener: %w", err)
+ }
+ a.listener = ln
+ log.Printf("TCP path agent listening on :%d", AgentPort)
+
+ a.schedulerOnce.Do(func() {
+ a.schedulerWG.Add(1)
+ go a.dialSchedulerLoop()
+ })
+
+ go a.acceptLoop()
+ return nil
+}
+
+func (a *TCPAgent) Stop() error {
+ if a.cancel != nil {
+ a.cancel()
+ }
+ close(a.stopScheduler)
+ a.schedulerWG.Wait()
+
+ if a.listener != nil {
+ _ = a.listener.Close()
+ }
+
+ a.edgesMu.Lock()
+ for _, ct := range a.activeEdges {
+ a.closeConn(ct, "agent_stop")
+ }
+ a.activeEdges = make(map[string]*connTrack)
+ a.edgesMu.Unlock()
+ return nil
+}
+
+func (a *TCPAgent) UpdateDiscoveredIPs(peerID peer.ID, ips []net.IP) {
+ now := time.Now()
+ add := make(map[string]ipStamp)
+ for _, ip := range ips {
+ if ip == nil {
+ continue
+ }
+ if v4 := ip.To4(); v4 != nil {
+ add[v4.String()] = ipStamp{seenAt: now, ttl: ttlDiscovered}
+ }
+ }
+ if len(add) == 0 {
+ return
+ }
+
+ a.ipDBMu.Lock()
+ a.ipDB[peerID] = mergeStamps(a.ipDB[peerID], add)
+ a.ipDBMu.Unlock()
+
+ a.dialStatesMu.Lock()
+ for ipStr := range add {
+ key := peerID.String() + "|" + ipStr
+ if _, ok := a.dialStates[key]; !ok {
+ a.dialStates[key] = &dialState{backoff: 0, nextAttempt: time.Now()}
+ }
+ }
+ a.dialStatesMu.Unlock()
+}
+
+func (a *TCPAgent) OnInterfaceChange() {
+ now := time.Now()
+ a.ifaceGraceUntilMu.Lock()
+ a.ifaceGraceUntil = now.Add(tbGraceWindow)
+ a.ifaceGraceUntilMu.Unlock()
+
+ a.dialStatesMu.Lock()
+ for _, ds := range a.dialStates {
+ ds.backoff = 0
+ ds.nextAttempt = now
+ }
+ a.dialStatesMu.Unlock()
+}
+
+func (a *TCPAgent) acceptLoop() {
+ for {
+ conn, err := a.listener.AcceptTCP()
+ if err != nil {
+ select {
+ case <-a.ctx.Done():
+ return
+ default:
+ }
+ log.Printf("TCP accept error: %v", err)
+ continue
+ }
+ a.setTCPOptions(conn)
+ go a.handleIncoming(conn)
+ }
+}
+
+func (a *TCPAgent) dialSchedulerLoop() {
+ defer a.schedulerWG.Done()
+ t := time.NewTicker(scheduleTick)
+ defer t.Stop()
+
+ for {
+ select {
+ case <-a.stopScheduler:
+ return
+ case <-a.ctx.Done():
+ return
+ case <-t.C:
+ a.expireIPs(false)
+
+ type want struct {
+ pid peer.ID
+ ip string
+ }
+ var wants []want
+
+ a.ipDBMu.RLock()
+ for pid, set := range a.ipDB {
+ if a.myPeerID.String() <= pid.String() {
+ continue
+ }
+ for ipStr, stamp := range set {
+ if time.Since(stamp.seenAt) > stamp.ttl {
+ continue
+ }
+ if a.hasActiveToRemoteIP(ipStr) {
+ continue
+ }
+ wants = append(wants, want{pid: pid, ip: ipStr})
+ }
+ }
+ a.ipDBMu.RUnlock()
+
+ sort.Slice(wants, func(i, j int) bool {
+ if wants[i].pid == wants[j].pid {
+ return wants[i].ip < wants[j].ip
+ }
+ return wants[i].pid.String() < wants[j].pid.String()
+ })
+
+ now := time.Now()
+ for _, w := range wants {
+ key := w.pid.String() + "|" + w.ip
+ a.dialStatesMu.Lock()
+ ds, ok := a.dialStates[key]
+ if !ok {
+ ds = &dialState{}
+ a.dialStates[key] = ds
+ }
+ if ds.connecting || now.Before(ds.nextAttempt) {
+ a.dialStatesMu.Unlock()
+ continue
+ }
+ ds.connecting = true
+ a.dialStatesMu.Unlock()
+
+ select {
+ case a.dialSem <- struct{}{}:
+ case <-a.ctx.Done():
+ return
+ }
+
+ go func(pid peer.ID, ip string) {
+ defer func() {
+ <-a.dialSem
+ a.dialStatesMu.Lock()
+ if ds := a.dialStates[pid.String()+"|"+ip]; ds != nil {
+ ds.connecting = false
+ }
+ a.dialStatesMu.Unlock()
+ }()
+ a.dialAndMaintain(pid, ip)
+ }(w.pid, w.ip)
+ }
+ }
+ }
+}
+
+func (a *TCPAgent) dialAndMaintain(pid peer.ID, remoteIP string) {
+ remoteAddr := fmt.Sprintf("%s:%d", remoteIP, AgentPort)
+ d := net.Dialer{Timeout: dialTimeoutForIP(remoteIP)}
+ rawConn, err := d.DialContext(a.ctx, "tcp", remoteAddr)
+ if err != nil {
+ a.bumpDialBackoff(pid, remoteIP, err)
+ return
+ }
+ tc := rawConn.(*net.TCPConn)
+ a.setTCPOptions(tc)
+
+ remoteNodeID, remotePeerID, observedIPv4s, err := a.performHandshake(tc, true)
+ if err != nil {
+ _ = tc.Close()
+ a.bumpDialBackoff(pid, remoteIP, err)
+ return
+ }
+
+ finalPID := pid
+ if remotePeerID != "" {
+ if parsed, perr := peer.Decode(remotePeerID); perr == nil {
+ finalPID = parsed
+ }
+ }
+
+ a.updateObservedIPv4s(finalPID, observedIPv4s)
+
+ localIP := tc.LocalAddr().(*net.TCPAddr).IP.String()
+ ct := &connTrack{
+ tc: tc,
+ dialer: true,
+ edge: Edge{
+ LocalNodeID: a.nodeID,
+ RemoteNodeID: remoteNodeID,
+ LocalIP: localIP,
+ RemoteIP: remoteIP,
+ Proto: "tcp",
+ },
+ closed: make(chan struct{}),
+ }
+
+ if !a.registerConn(ct) {
+ _ = tc.Close()
+ a.bumpDialBackoff(finalPID, remoteIP, errors.New("duplicate edge"))
+ return
+ }
+
+ a.dialStatesMu.Lock()
+ if ds := a.dialStates[finalPID.String()+"|"+remoteIP]; ds != nil {
+ ds.backoff = 0
+ ds.nextAttempt = time.Now().Add(HeartbeatInterval)
+ }
+ a.dialStatesMu.Unlock()
+
+ a.runHeartbeatLoops(ct)
+}
+
+func (a *TCPAgent) handleIncoming(tc *net.TCPConn) {
+ remoteNodeID, remotePeerID, observedIPv4s, err := a.performHandshake(tc, false)
+ if err != nil {
+ _ = tc.Close()
+ return
+ }
+ if remotePeerID != "" {
+ if pid, perr := peer.Decode(remotePeerID); perr == nil {
+ a.updateObservedIPv4s(pid, observedIPv4s)
+ }
+ }
+
+ localIP := tc.LocalAddr().(*net.TCPAddr).IP.String()
+ remoteIP := tc.RemoteAddr().(*net.TCPAddr).IP.String()
+
+ ct := &connTrack{
+ tc: tc,
+ dialer: false,
+ edge: Edge{
+ LocalNodeID: a.nodeID,
+ RemoteNodeID: remoteNodeID,
+ LocalIP: localIP,
+ RemoteIP: remoteIP,
+ Proto: "tcp",
+ },
+ closed: make(chan struct{}),
+ }
+
+ if !a.registerConn(ct) {
+ _ = tc.Close()
+ return
+ }
+ a.runHeartbeatLoops(ct)
+}
+
+func (a *TCPAgent) setTCPOptions(tc *net.TCPConn) {
+ _ = tc.SetNoDelay(true)
+ _ = tc.SetKeepAlive(true)
+ _ = tc.SetKeepAlivePeriod(5 * time.Second)
+}
+
+func (a *TCPAgent) performHandshake(tc *net.TCPConn, isDialer bool) (remoteNodeID, remotePeerID string, observedIPv4s []string, err error) {
+ _ = tc.SetDeadline(time.Now().Add(HandshakeTimeout))
+ defer tc.SetDeadline(time.Time{})
+
+ self := HandshakeMessage{
+ NodeID: a.nodeID,
+ AgentVer: "2.2.0",
+ PeerID: a.myPeerID.String(),
+ IPv4s: currentLocalIPv4s(),
+ Timestamp: time.Now().UnixNano(),
+ }
+ var remote HandshakeMessage
+
+ if isDialer {
+ if err = json.NewEncoder(tc).Encode(&self); err != nil {
+ return "", "", nil, fmt.Errorf("send handshake: %w", err)
+ }
+ if err = json.NewDecoder(tc).Decode(&remote); err != nil {
+ return "", "", nil, fmt.Errorf("read handshake: %w", err)
+ }
+ } else {
+ if err = json.NewDecoder(tc).Decode(&remote); err != nil {
+ return "", "", nil, fmt.Errorf("read handshake: %w", err)
+ }
+ if err = json.NewEncoder(tc).Encode(&self); err != nil {
+ return "", "", nil, fmt.Errorf("send handshake: %w", err)
+ }
+ }
+
+ if remote.NodeID == "" {
+ return "", "", nil, errors.New("empty remote node id")
+ }
+ for _, ip := range remote.IPv4s {
+ if ip != "" && strings.Count(ip, ":") == 0 {
+ observedIPv4s = append(observedIPv4s, ip)
+ }
+ }
+ return remote.NodeID, remote.PeerID, observedIPv4s, nil
+}
+
+func (a *TCPAgent) registerConn(ct *connTrack) bool {
+ key := ct.edge.Key()
+
+ a.edgesMu.Lock()
+ if _, exists := a.activeEdges[key]; exists {
+ a.edgesMu.Unlock()
+ return false
+ }
+ a.activeEdges[key] = ct
+
+ a.activeByRemoteIPMu.Lock()
+ a.activeByRemoteIP[ct.edge.RemoteIP] = true
+ a.activeByRemoteIPMu.Unlock()
+ a.edgesMu.Unlock()
+
+ WriteEdgeCreatedEvent(ct.edge.LocalNodeID, ct.edge.RemoteNodeID, ct.edge.LocalIP, ct.edge.RemoteIP, ct.edge.Proto)
+ return true
+}
+
+func (a *TCPAgent) hasActiveToRemoteIP(remoteIP string) bool {
+ a.activeByRemoteIPMu.RLock()
+ ok := a.activeByRemoteIP[remoteIP]
+ a.activeByRemoteIPMu.RUnlock()
+ return ok
+}
+
+func (a *TCPAgent) recalcRemoteIPActive(remoteIP string) {
+ a.edgesMu.RLock()
+ active := false
+ for _, ct := range a.activeEdges {
+ if ct.edge.RemoteIP == remoteIP {
+ active = true
+ break
+ }
+ }
+ a.edgesMu.RUnlock()
+
+ a.activeByRemoteIPMu.Lock()
+ if active {
+ a.activeByRemoteIP[remoteIP] = true
+ } else {
+ delete(a.activeByRemoteIP, remoteIP)
+ }
+ a.activeByRemoteIPMu.Unlock()
+}
+
+func (a *TCPAgent) closeConn(ct *connTrack, _ string) {
+ ct.closeMx.Do(func() {
+ _ = ct.tc.Close()
+ key := ct.edge.Key()
+
+ a.edgesMu.Lock()
+ delete(a.activeEdges, key)
+ a.edgesMu.Unlock()
+
+ a.recalcRemoteIPActive(ct.edge.RemoteIP)
+
+ WriteEdgeDeletedEvent(ct.edge.LocalNodeID, ct.edge.RemoteNodeID, ct.edge.LocalIP, ct.edge.RemoteIP, ct.edge.Proto)
+ })
+}
+
+func (a *TCPAgent) runHeartbeatLoops(ct *connTrack) {
+ go func() {
+ r := bufio.NewReader(ct.tc)
+ for {
+ _ = ct.tc.SetReadDeadline(time.Now().Add(HeartbeatReadGrace))
+ if _, err := r.ReadByte(); err != nil {
+ a.closeConn(ct, "read_error")
+ return
+ }
+ }
+ }()
+
+ go func() {
+ t := time.NewTicker(HeartbeatInterval)
+ defer t.Stop()
+ for {
+ select {
+ case <-t.C:
+ _ = ct.tc.SetWriteDeadline(time.Now().Add(HeartbeatWriteGrace))
+ if _, err := ct.tc.Write([]byte{0x01}); err != nil {
+ a.closeConn(ct, "write_error")
+ return
+ }
+ case <-a.ctx.Done():
+ a.closeConn(ct, "agent_ctx_done")
+ return
+ }
+ }
+ }()
+}
+
+func (a *TCPAgent) bumpDialBackoff(pid peer.ID, ip string, err error) {
+ key := pid.String() + "|" + ip
+ a.dialStatesMu.Lock()
+ ds, ok := a.dialStates[key]
+ if !ok {
+ ds = &dialState{}
+ a.dialStates[key] = ds
+ }
+ if ds.backoff == 0 {
+ ds.backoff = initialBackoff
+ } else {
+ ds.backoff *= 2
+ if ds.backoff > maxBackoff {
+ ds.backoff = maxBackoff
+ }
+ }
+ ds.nextAttempt = time.Now().Add(ds.backoff)
+ a.dialStatesMu.Unlock()
+
+ log.Printf("dial %s@%s failed: %v; next in %s", pid, ip, err, ds.backoff)
+}
+
+func mergeStamps(dst map[string]ipStamp, src map[string]ipStamp) map[string]ipStamp {
+ if dst == nil {
+ dst = make(map[string]ipStamp, len(src))
+ }
+ for ip, s := range src {
+ prev, ok := dst[ip]
+ if !ok || s.seenAt.After(prev.seenAt) {
+ dst[ip] = s
+ }
+ }
+ return dst
+}
+
+func (a *TCPAgent) updateObservedIPv4s(pid peer.ID, ipv4s []string) {
+ if len(ipv4s) == 0 {
+ return
+ }
+ now := time.Now()
+ add := make(map[string]ipStamp, len(ipv4s))
+ for _, ip := range ipv4s {
+ if ip != "" && strings.Count(ip, ":") == 0 {
+ add[ip] = ipStamp{seenAt: now, ttl: ttlObserved}
+ }
+ }
+
+ a.ipDBMu.Lock()
+ a.ipDB[pid] = mergeStamps(a.ipDB[pid], add)
+ a.ipDBMu.Unlock()
+
+ a.dialStatesMu.Lock()
+ for ip := range add {
+ key := pid.String() + "|" + ip
+ if _, ok := a.dialStates[key]; !ok {
+ a.dialStates[key] = &dialState{backoff: 0, nextAttempt: time.Now()}
+ }
+ }
+ a.dialStatesMu.Unlock()
+}
+
+func (a *TCPAgent) expireIPs(_ bool) {
+ a.ifaceGraceUntilMu.RLock()
+ graceUntil := a.ifaceGraceUntil
+ a.ifaceGraceUntilMu.RUnlock()
+ if time.Now().Before(graceUntil) {
+ return
+ }
+
+ now := time.Now()
+ a.ipDBMu.Lock()
+ for pid, set := range a.ipDB {
+ for ip, stamp := range set {
+ if now.Sub(stamp.seenAt) > stamp.ttl {
+ delete(set, ip)
+
+ a.dialStatesMu.Lock()
+ delete(a.dialStates, pid.String()+"|"+ip)
+ a.dialStatesMu.Unlock()
+
+ log.Printf("TCP agent: expired ip %s for %s", ip, pid)
+ }
+ }
+ if len(set) == 0 {
+ delete(a.ipDB, pid)
+ }
+ }
+ a.ipDBMu.Unlock()
+}
+
+func currentLocalIPv4s() []string {
+ var out []string
+ ifaces, err := net.Interfaces()
+ if err != nil {
+ return out
+ }
+ for _, ifi := range ifaces {
+ if ifi.Flags&net.FlagUp == 0 {
+ continue
+ }
+ addrs, _ := ifi.Addrs()
+ for _, a := range addrs {
+ if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP != nil {
+ if v4 := ipnet.IP.To4(); v4 != nil && !v4.IsLoopback() && !v4.IsUnspecified() {
+ out = append(out, v4.String())
+ }
+ }
+ }
+ }
+ sort.Strings(out)
+ return dedupeStrings(out)
+}
+
+func dedupeStrings(xs []string) []string {
+ if len(xs) < 2 {
+ return xs
+ }
+ out := xs[:0]
+ last := ""
+ for _, s := range xs {
+ if s == last {
+ continue
+ }
+ out = append(out, s)
+ last = s
+ }
+ return out
+}
+
+func dialTimeoutForIP(ip string) time.Duration {
+ if strings.HasPrefix(ip, "169.254.") {
+ return dialTimeoutLinkLocal
+ }
+ return dialTimeoutDefault
+}
← 57073f35 collection of fixes for Shanghai demo
·
back to Exo
·
improved go caching with nix ea3eeea8 →