Add mutex locks for thread safety in network operations

Introduce mutex locks in the peer connection count and node data update functions to ensure thread safety. These changes prevent potential race conditions when multiple goroutines access shared resources. Note that some outdated comments have been updated to reflect the need for careful cleanup.

pkg/network/kdht.go

@@ -30,14 +30,16 @@ type dbValidator struct{}
 func (dbValidator) Validate(_ string, _ []byte) error        { return nil }
 func (dbValidator) Select(_ string, _ [][]byte) (int, error) { return 0, nil }
 
+var peerCountMutex sync.Mutex
+
 func EnableDHT(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.Multiaddr, protocolId, prefix protocol.ID, peerChan chan PeerEvent, nodeData *pubsub.NodeData) (*dht.IpfsDHT, error) {
 	options := make([]dht.Option, 0)
 	options = append(options, dht.BucketSize(100))                          // Adjust bucket size
 	options = append(options, dht.Concurrency(100))                         // Increase concurrency
 	options = append(options, dht.RoutingTableRefreshPeriod(time.Minute*5)) // Set refresh interval
 	options = append(options, dht.Mode(dht.ModeAutoServer))
 	options = append(options, dht.ProtocolPrefix(prefix))
-	// WTF: Why?
+	// TODO: If we are no longer using this it should be cleaned up, but carefully. the Node Listener assumes this is there
 	options = append(options, dht.NamespacedValidator("db", dbValidator{}))
 
 	kademliaDHT, err := dht.New(ctx, host, options...)
@@ -115,7 +117,9 @@ func EnableDHT(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.M
 					}
 					return
 				}
+				peerCountMutex.Lock()
 				peerConnectionCount++
+				peerCountMutex.Unlock()
 				defer func(stream network.Stream) {
 					err := stream.Close()
 					if err != nil {

pkg/pubsub/node_event_tracker.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"sort"
+	"sync"
 	"time"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
@@ -18,10 +19,9 @@ import (
 
 type NodeEventTracker struct {
 	NodeDataChan chan *NodeData
-	// WTF: Do we really need this? Can't we store it in the libp2p PeerStore metadata?
-	nodeData *SafeMap
-	// WTF: Unused?
-	nodeDataFile  string
+	// TODO: Do we really need this? Can't we store it in the libp2p PeerStore metadata?
+	nodeData      *SafeMap
+	NodeDataMutex sync.Mutex
 	ConnectBuffer map[string]ConnectBufferEntry
 	nodeVersion   string
 }
@@ -99,7 +99,6 @@ func NewNodeEventTracker(version, environment, hostId string) *NodeEventTracker
 		nodeData:      NewSafeMap(),
 		nodeVersion:   version,
 		NodeDataChan:  make(chan *NodeData),
-		nodeDataFile:  fmt.Sprintf("%s_%s_node_data.json", version, environment),
 		ConnectBuffer: make(map[string]ConnectBufferEntry),
 	}
 	go net.ClearExpiredBufferEntries()
@@ -377,6 +376,8 @@ func (net *NodeEventTracker) IsStaked(peerID string) bool {
 // It also sends the updated node data to the NodeDataChan if the data changed or forceGossip is true.
 func (net *NodeEventTracker) AddOrUpdateNodeData(nodeData *NodeData, forceGossip bool) error {
 	logrus.Debugf("Handling node data for: %s", nodeData.PeerId)
+	net.NodeDataMutex.Lock()
+	defer net.NodeDataMutex.Unlock()
 	dataChanged := false
 
 	nd, ok := net.nodeData.Get(nodeData.PeerId.String())