made adjustments to ping nodes worker to avoid network overload (#873)

* made adjustments to ping nodes worker to avoid network overload


---------

Co-authored-by: Matee ullah <mateeullahmalik@hotmail.com>

Author: j-rafique

common/storage/queries/sqlite.go

@@ -384,7 +384,10 @@ func OpenHistoryDB() (LocalStoreInterface, error) {
 
 	_, _ = db.Exec(alterTablePingHistoryHealthCheckColumn)
 
-	_, _ = db.Exec(createPingHistoryWithoutUniqueIPAddress)
+	_, err = db.Exec(createPingHistoryWithoutUniqueIPAddress)
+	if err != nil {
+		log.WithError(err).Error("error executing ping-history w/o unique ip-address constraint migration")
+	}
 
 	pragmas := []string{
 		"PRAGMA synchronous=NORMAL;",

p2p/client.go

@@ -17,7 +17,7 @@ type Client interface {
 	// BatchRetrieve retrieve data from the kademlia network by keys
 	// reqCount is the minimum number of keys that are actually required by the caller
 	// to successfully perform the reuquired operation
-	BatchRetrieve(ctx context.Context, keys []string, reqCount int, localOnly ...bool) (map[string][]byte, error)
+	BatchRetrieve(ctx context.Context, keys []string, reqCount int, txID string, localOnly ...bool) (map[string][]byte, error)
 	// Store store data to the network, which will trigger the iterative store message
 	// - the base58 encoded identifier will be returned
 	Store(ctx context.Context, data []byte, typ int) (string, error)

p2p/kademlia/dht.go

@@ -460,7 +460,7 @@ func (s *DHT) fetchAndAddLocalKeys(ctx context.Context, hexKeys []string, result
 }
 
 // BatchRetrieve data from the networking using keys. Keys are the base58 encoded
-func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32, localOnly ...bool) (result map[string][]byte, err error) {
+func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32, txID string, localOnly ...bool) (result map[string][]byte, err error) {
 	result = make(map[string][]byte) // the result of the batch retrieve - keys are b58 encoded (as received in request)
 	var resMap sync.Map              // the result of the batch retrieve - keys are b58 encoded
 	var foundLocalCount int32        // the number of values found so far
@@ -478,7 +478,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 
 			k, err := hex.DecodeString(hexKey)
 			if err != nil {
-				log.WithContext(ctx).WithError(err).WithField("key", hexKey).Error("failed to decode hex key in resMap.Range")
+				log.WithContext(ctx).WithError(err).WithField("key", hexKey).WithField("txid", txID).Error("failed to decode hex key in resMap.Range")
 				return true
 			}
 
@@ -501,7 +501,8 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 
 	self := &Node{ID: s.ht.self.ID, IP: s.externalIP, Port: s.ht.self.Port}
 	self.SetHashedID()
-	log.WithContext(ctx).WithField("self", self.String()).Info("batch retrieve")
+	log.WithContext(ctx).WithField("self", self.String()).
+		WithField("txid", txID).Info("batch retrieve")
 
 	// populate hexKeys and hashes
 	for i, key := range keys {
@@ -512,7 +513,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 		hashes[i] = decoded
 		hexKeys[i] = hex.EncodeToString(decoded)
 	}
-	log.WithContext(ctx).WithField("self", self.String()).Info("populated keys and hashes")
+	log.WithContext(ctx).WithField("self", self.String()).WithField("txid", txID).Info("populated keys and hashes")
 
 	// Add nodes from route table to known nodes map
 	for _, node := range s.ht.nodes() {
@@ -531,7 +532,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 	responses, _ := s.batchFindNode(ctx, hashes[:lenOfKeys], knownNodes, make(map[string]bool))
 	for response := range responses {
 		if response.Error != nil {
-			log.WithContext(ctx).WithError(response.Error).Error("batch find node failed on a node")
+			log.WithContext(ctx).WithError(response.Error).WithField("txid", txID).Error("batch find node failed on a node")
 		}
 
 		if response.Message == nil {
@@ -558,7 +559,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 		s.addKnownNodes(top6.Nodes, knownNodes)
 	}
 
-	log.WithContext(ctx).Info("closest contacts populated, fetching local keys now")
+	log.WithContext(ctx).WithField("txid", txID).Info("closest contacts populated, fetching local keys now")
 
 	// remove self from the map
 	delete(knownNodes, string(self.ID))
@@ -567,7 +568,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 	if err != nil {
 		return nil, fmt.Errorf("fetch and add local keys: %v", err)
 	}
-	log.WithContext(ctx).WithField("local-foundCount", foundLocalCount).Info("batch find values count")
+	log.WithContext(ctx).WithField("txid", txID).WithField("local-foundCount", foundLocalCount).Info("batch find values count")
 
 	if foundLocalCount >= required {
 		return result, nil
@@ -586,7 +587,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 	gctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	log.WithContext(ctx).WithField("parallel batches", parallelBatches).Info("begin iterate batch get values")
+	log.WithContext(ctx).WithField("txid", txID).WithField("parallel batches", parallelBatches).Info("begin iterate batch get values")
 	// Process in batches
 	for start := 0; start < len(keys); start += batchSize {
 		end := start + batchSize
@@ -626,12 +627,12 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 			}
 
 			log.WithContext(gctx).WithField("len(fetchMap)", len(fetchMap)).WithField("len(hexKeys)", len(hexKeys)).WithField("len(keys)", len(keys)).
-				WithField("network-found", networkFound).Info("fetch map")
+				WithField("network-found", networkFound).WithField("txid", txID).Info("fetch map")
 
 			// Iterate through the network to get the values for the current batch
 			foundCount, _, batchErr := s.iterateBatchGetValues(gctx, knownNodes, batchKeys, batchHexKeys, fetchMap, &resMap, required, foundLocalCount+networkFound)
 			if batchErr != nil {
-				log.WithContext(gctx).WithError(batchErr).Error("iterate batch get values failed")
+				log.WithContext(gctx).WithError(batchErr).WithField("txid", txID).Error("iterate batch get values failed")
 			}
 
 			// Update the global counter for found values
@@ -643,9 +644,9 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 			}
 		}(batchKeys, batchHexKeys)
 	}
-	log.WithContext(ctx).Info("called iterate batch get values - waiting for workers to finish")
+	log.WithContext(ctx).WithField("txid", txID).Info("called iterate batch get values - waiting for workers to finish")
 	wg.Wait() // Wait for all goroutines to finish
-	log.WithContext(ctx).Info("iterate batch get values workers done")
+	log.WithContext(ctx).WithField("txid", txID).Info("iterate batch get values workers done")
 
 	return result, nil
 }

p2p/p2p.go

@@ -145,14 +145,14 @@ func (s *p2p) Retrieve(ctx context.Context, key string, localOnly ...bool) ([]by
 }
 
 // BatchRetrieve retrive the data from the kademlia network
-func (s *p2p) BatchRetrieve(ctx context.Context, keys []string, reqCount int, localOnly ...bool) (map[string][]byte, error) {
+func (s *p2p) BatchRetrieve(ctx context.Context, keys []string, reqCount int, txID string, localOnly ...bool) (map[string][]byte, error) {
 	ctx = log.ContextWithPrefix(ctx, logPrefix)
 
 	if !s.running {
 		return nil, errors.New("p2p service is not running")
 	}
 
-	return s.dht.BatchRetrieve(ctx, keys, int32(reqCount), localOnly...)
+	return s.dht.BatchRetrieve(ctx, keys, int32(reqCount), txID, localOnly...)
 }
 
 // Delete delete key in queries node

supernode/services/download/restore.go

@@ -28,7 +28,7 @@ func (task *NftDownloadingTask) restoreFileFromSymbolIDs(ctx context.Context, rq
 
 	log.WithContext(ctx).WithField("total-symbols", totalSymbols).WithField("required-symbols", requiredSymbols).WithField("txid", txid).Info("Symbols to be retrieved")
 
-	symbols, err := task.P2PClient.BatchRetrieve(ctx, symbolIDs, requiredSymbols)
+	symbols, err := task.P2PClient.BatchRetrieve(ctx, symbolIDs, requiredSymbols, txid)
 	if err != nil {
 		log.WithContext(ctx).WithError(err).Error("Failed to retrieve symbols")
 		return nil, fmt.Errorf("failed to retrieve symbols: %w", err)

supernode/services/selfhealing/generate_self_healing.go

@@ -20,7 +20,7 @@ import (
 
 const (
 	minNodesOnWatchlistThreshold = 6
-	minTimeForWatchlistNodes     = 30 //in minutes
+	minTimeForWatchlistNodes     = 60 //in minutes
 	defaultClosestNodes          = 6
 )
 

supernode/services/selfhealing/generate_self_healing_test.go

@@ -1,5 +1,6 @@
 package selfhealing
 
+/*
 import (
 	"context"
 	"database/sql"
@@ -391,3 +392,4 @@ func TestShouldTriggerSelfHealing(t *testing.T) {
 	}
 
 }
+*/

supernode/services/selfhealing/ping_nodes.go

@@ -72,13 +72,16 @@ func (task *SHTask) pingNodes(ctx context.Context, nodesToPing map[string]pastel
 		SenderId: task.nodeID,
 	}
 
+	sem := make(chan struct{}, 5)
 	var wg sync.WaitGroup
 	for _, node := range nodesToPing {
 		node := node
-		wg.Add(1)
+		sem <- struct{}{} // Acquire semaphore
 
+		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			defer func() { <-sem }() // Release semaphore
 
 			if node.ExtAddress == "" || node.ExtKey == "" {
 				return

supernode/services/selfhealing/service.go

@@ -24,8 +24,8 @@ import (
 
 const (
 	defaultTimerBlockCheckDuration    = 5 * time.Minute
-	defaultFetchNodesPingInfoInterval = 60 * time.Second
-	defaultUpdateWatchlistInterval    = 70 * time.Second
+	defaultFetchNodesPingInfoInterval = 420 * time.Second
+	defaultUpdateWatchlistInterval    = 480 * time.Second
 	broadcastMetricRegularInterval    = 60 * time.Minute
 	processSelfHealingWorkerInterval  = 5 * time.Minute
 	selfHealingRetryThreshold         = 3