Merge pull request #746 from pastelnetwork/PSL-1090_BugFixes

[PSL-1090] fix bugs and edge cases handling

common/storage/local.go

@@ -21,6 +21,6 @@ type LocalStoreInterface interface {
 	GetPingInfoBySupernodeID(supernodeID string) (*types.PingInfo, error)
 	GetAllPingInfos() (types.PingInfos, error)
 	GetWatchlistPingInfo() ([]types.PingInfo, error)
-	UpdatePingInfo(supernodeID string) error
+	UpdatePingInfo(supernodeID string, isOnWatchlist, isAdjusted bool) error
 	CloseHistoryDB(ctx context.Context)
 }

common/storage/local/sqlite.go

@@ -267,15 +267,15 @@ func (s *SQLiteStore) GetWatchlistPingInfo() ([]types.PingInfo, error) {
 }
 
 // UpdatePingInfo updates the ping info
-func (s *SQLiteStore) UpdatePingInfo(supernodeID string) error {
+func (s *SQLiteStore) UpdatePingInfo(supernodeID string, isOnWatchlist, isAdjusted bool) error {
 	// Update query
 	const updateQuery = `
 UPDATE ping_history
-SET is_adjusted = true, is_on_watchlist = false
+SET is_adjusted = ?, is_on_watchlist = ?
 WHERE supernode_id = ?;`
 
 	// Execute the update query
-	_, err := s.db.Exec(updateQuery, supernodeID)
+	_, err := s.db.Exec(updateQuery, isAdjusted, isOnWatchlist, supernodeID)
 	if err != nil {
 		return err
 	}

supernode/services/download/restore.go

@@ -179,7 +179,8 @@ func (task *NftDownloadingTask) getSymbolIDsFromMetadataFile(ctx context.Context
 	return symbolIDs, nil
 }
 
-func (task *NftDownloadingTask) restoreFile(ctx context.Context, rqID []string, rqOti []byte, dataHash []byte, txid string) ([]byte, error) {
+// RestoreFile restores the file using the available rq-ids
+func (task *NftDownloadingTask) RestoreFile(ctx context.Context, rqID []string, rqOti []byte, dataHash []byte, txid string) ([]byte, error) {
 	var file []byte
 	var lastErr error
 	var err error

supernode/services/download/task.go

@@ -382,7 +382,7 @@ func (task *NftDownloadingTask) Download(ctx context.Context, txid, timestamp, s
 		// Get the list of "symbols/chunks" from Kademlia by using symbol identifiers from file
 		// Pass all symbols/chunks to the raptorq service to decode (also passing encoder parameters: rq_oti)
 		// Validate hash of the restored image matches the image hash in the Art Reistration ticket (data_hash)
-		file, err = task.restoreFile(ctx, info.rqIDs, info.rqOti, info.dataHash, txid)
+		file, err = task.RestoreFile(ctx, info.rqIDs, info.rqOti, info.dataHash, txid)
 		if err != nil {
 			log.WithContext(ctx).WithField("txid", txid).Error("restore file failed")
 			err = errors.Errorf("restore file: %w", err)

supernode/services/selfhealing/self_healing_worker.go → supernode/services/selfhealing/generate_self_healing.go

@@ -17,8 +17,9 @@ import (
 )
 
 const (
-	watchlistThreshold  = 6
-	defaultClosestNodes = 6
+	minNodesOnWatchlistThreshold = 6
+	minTimeForWatchlistNodes     = 30 //in minutes
+	defaultClosestNodes          = 6
 )
 
 // TicketType will identify the type of ticket
@@ -38,8 +39,8 @@ type SymbolFileKeyDetails struct {
 	TicketType TicketType
 }
 
-// SelfHealingWorker checks the ping info and decide self-healing files
-func (task *SHTask) SelfHealingWorker(ctx context.Context) error {
+// GenerateSelfHealingChallenge worker checks the ping info and identify self-healing tickets and their recipients
+func (task *SHTask) GenerateSelfHealingChallenge(ctx context.Context) error {
 	log.WithContext(ctx).Infoln("Self Healing Worker has been invoked")
 
 	watchlistPingInfos, err := task.retrieveWatchlistPingInfo(ctx)
@@ -49,11 +50,12 @@ func (task *SHTask) SelfHealingWorker(ctx context.Context) error {
 	}
 	log.WithContext(ctx).Info("watchlist ping history has been retrieved")
 
-	if len(watchlistPingInfos) < watchlistThreshold {
+	shouldTrigger, watchlistPingInfos := task.shouldTriggerSelfHealing(watchlistPingInfos)
+	if !shouldTrigger {
 		log.WithContext(ctx).WithField("no_of_nodes_on_watchlist", len(watchlistPingInfos)).Info("not enough nodes on the watchlist, skipping further processing")
 		return nil
 	}
-	log.WithContext(ctx).Info("watchlist threshold has been reached, proceeding forward")
+	log.WithContext(ctx).Info("self-healing has been triggered, proceeding with the identification of files & recipients")
 
 	keys, symbolFileKeyMap, err := task.ListSymbolFileKeysFromNFTAndActionTickets(ctx)
 	if err != nil {
@@ -117,6 +119,23 @@ func (task *SHTask) retrieveWatchlistPingInfo(ctx context.Context) (types.PingIn
 	return infos, nil
 }
 
+func (task *SHTask) shouldTriggerSelfHealing(infos types.PingInfos) (bool, types.PingInfos) {
+	var filteredPings types.PingInfos
+	for _, ping := range infos {
+		if ping.LastSeen.Valid {
+			if time.Since(ping.LastSeen.Time).Minutes() <= minTimeForWatchlistNodes {
+				filteredPings = append(filteredPings, ping)
+			}
+		}
+	}
+
+	if len(filteredPings) < minNodesOnWatchlistThreshold {
+		return false, filteredPings
+	}
+
+	return true, filteredPings
+}
+
 // ListSymbolFileKeysFromNFTAndActionTickets : Get an NFT and Action Ticket's associated raptor q ticket file id's.
 func (service *SHService) ListSymbolFileKeysFromNFTAndActionTickets(ctx context.Context) ([]string, map[string]SymbolFileKeyDetails, error) {
 	var keys = make([]string, 0)
@@ -285,7 +304,12 @@ func (task *SHTask) identifyChallengeRecipients(ctx context.Context, selfHealing
 			continue
 		}
 
-		challengeRecipients := task.SHService.GetNClosestSupernodeIDsToComparisonString(ctx, 1, string(dataHash), nil)
+		listOfSupernodes, err := task.getListOfSupernode(ctx)
+		if err != nil {
+			logger.WithError(err).Error("unable to retrieve list of supernodes")
+		}
+
+		challengeRecipients := task.SHService.GetNClosestSupernodeIDsToComparisonString(ctx, 1, string(dataHash), listOfSupernodes)
 		if len(challengeRecipients) < 1 {
 			log.WithContext(ctx).WithField("file_hash", dataHash).Info("no closest nodes have found against the file")
 			continue
@@ -571,7 +595,7 @@ func (task *SHTask) updateWatchlist(ctx context.Context, watchlistPingInfos type
 		defer store.CloseHistoryDB(ctx)
 
 		for _, info := range watchlistPingInfos {
-			err = store.UpdatePingInfo(info.SupernodeID)
+			err = store.UpdatePingInfo(info.SupernodeID, false, true)
 			if err != nil {
 				log.WithContext(ctx).WithField("supernode_id", info.SupernodeID).
 					Error("error updating watchlist ping info")
@@ -583,3 +607,17 @@ func (task *SHTask) updateWatchlist(ctx context.Context, watchlistPingInfos type
 
 	return nil
 }
+
+func (service *SHService) getListOfSupernode(ctx context.Context) ([]string, error) {
+	var ret = make([]string, 0)
+	listMN, err := service.SuperNodeService.PastelClient.MasterNodesExtra(ctx)
+	if err != nil {
+		return ret, err
+	}
+
+	for _, node := range listMN {
+		ret = append(ret, node.ExtKey)
+	}
+
+	return ret, nil
+}

supernode/services/selfhealing/self_healing_worker_test.go → supernode/services/selfhealing/generate_self_healing_test.go

@@ -2,12 +2,13 @@ package selfhealing
 
 import (
 	"context"
-	json "github.com/json-iterator/go"
-	"github.com/pastelnetwork/gonode/common/types"
-	"golang.org/x/crypto/sha3"
+	"database/sql"
 	"testing"
+	"time"
 
 	fuzz "github.com/google/gofuzz"
+	json "github.com/json-iterator/go"
+	"github.com/pastelnetwork/gonode/common/types"
 	p2pMock "github.com/pastelnetwork/gonode/p2p/test"
 	"github.com/pastelnetwork/gonode/pastel"
 	pastelMock "github.com/pastelnetwork/gonode/pastel/test"
@@ -16,6 +17,7 @@ import (
 	rqmock "github.com/pastelnetwork/gonode/raptorq/node/test"
 	shtest "github.com/pastelnetwork/gonode/supernode/node/test/self_healing"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/crypto/sha3"
 )
 
 func TestListSymbolFileKeysForNFTAndActionTickets(t *testing.T) {
@@ -302,3 +304,75 @@ func TestCreateSelfHealingTicketsMap(t *testing.T) {
 	}
 
 }
+
+func TestShouldTriggerSelfHealing(t *testing.T) {
+	t.Parallel()
+
+	config := NewConfig()
+	pastelClient := pastelMock.NewMockClient(t)
+	p2pClient := p2pMock.NewMockClient(t)
+	raptorQClient := rqmock.NewMockClient(t)
+	var nodeClient *shtest.Client
+
+	tests := []struct {
+		testcase string
+		infos    types.PingInfos
+		expect   func(*testing.T, bool, types.PingInfos)
+	}{
+		{
+			testcase: "when more than 6 nodes that are on watchlist, have last_seen within 30 min time span",
+			infos: []types.PingInfo{
+				types.PingInfo{SupernodeID: "A", LastSeen: sql.NullTime{Time: time.Now().Add(-5 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "B", LastSeen: sql.NullTime{Time: time.Now().Add(-10 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "C", LastSeen: sql.NullTime{Time: time.Now().Add(-15 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "D", LastSeen: sql.NullTime{Time: time.Now().Add(-45 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "E", LastSeen: sql.NullTime{Time: time.Now().Add(-50 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "F", LastSeen: sql.NullTime{Time: time.Now().Add(-75 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "G", LastSeen: sql.NullTime{Time: time.Now().Add(-100 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "H", LastSeen: sql.NullTime{Time: time.Now().Add(-25 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "I", LastSeen: sql.NullTime{Time: time.Now().Add(-29 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "J", LastSeen: sql.NullTime{Time: time.Now().Add(-22 * time.Minute), Valid: true}},
+			},
+			expect: func(t *testing.T, shouldTrigger bool, infos types.PingInfos) {
+				require.Equal(t, len(infos), 6)
+				require.Equal(t, shouldTrigger, true)
+			},
+		},
+		{
+			testcase: "when less than 6 nodes that are on watchlist, have last_seen within 30 min time span",
+			infos: []types.PingInfo{
+				types.PingInfo{SupernodeID: "A", LastSeen: sql.NullTime{Time: time.Now().Add(-55 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "B", LastSeen: sql.NullTime{Time: time.Now().Add(-90 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "C", LastSeen: sql.NullTime{Time: time.Now().Add(-45 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "D", LastSeen: sql.NullTime{Time: time.Now().Add(-45 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "E", LastSeen: sql.NullTime{Time: time.Now().Add(-50 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "F", LastSeen: sql.NullTime{Time: time.Now().Add(-75 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "G", LastSeen: sql.NullTime{Time: time.Now().Add(-100 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "H", LastSeen: sql.NullTime{Time: time.Now().Add(-75 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "I", LastSeen: sql.NullTime{Time: time.Now().Add(-29 * time.Minute), Valid: true}},
+				types.PingInfo{SupernodeID: "J", LastSeen: sql.NullTime{Time: time.Now().Add(-22 * time.Minute), Valid: true}},
+			},
+			expect: func(t *testing.T, shouldTrigger bool, infos types.PingInfos) {
+				require.Equal(t, len(infos), 2)
+				require.Equal(t, shouldTrigger, false)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.testcase, func(t *testing.T) {
+			// Run the setup for the testcase
+			service := NewService(config, nil, pastelClient, nodeClient,
+				p2pClient, nil)
+			task := NewSHTask(service)
+			task.StorageHandler.RqClient = raptorQClient
+			// call the function to get return values
+			shouldTrigger, filteredPings := task.shouldTriggerSelfHealing(tt.infos)
+			// handle the test case's assertions with the provided func
+			tt.expect(t, shouldTrigger, filteredPings)
+		})
+	}
+
+}

supernode/services/selfhealing/fetch_and_maintain_nodes_ping_info.go → supernode/services/selfhealing/ping_nodes.go

@@ -87,7 +87,6 @@ func (task *SHTask) pingNodes(ctx context.Context, nodesToPing pastel.MasterNode
 					SupernodeID:      node.ExtKey,
 					IPAddress:        node.ExtAddress,
 					IsOnline:         false,
-					IsAdjusted:       false,
 					LastResponseTime: 0.0,
 				}
 
@@ -107,7 +106,6 @@ func (task *SHTask) pingNodes(ctx context.Context, nodesToPing pastel.MasterNode
 				SupernodeID:      node.ExtKey,
 				IPAddress:        node.ExtAddress,
 				IsOnline:         res.IsOnline,
-				IsAdjusted:       false,
 				LastResponseTime: respondedAt.Sub(timeBeforePing).Seconds(),
 				LastSeen:         sql.NullTime{Time: respondedAt, Valid: true},
 			}
@@ -233,11 +231,15 @@ func GetPingInfoToInsert(existedInfo, info *types.PingInfo) *types.PingInfo {
 
 	if info.IsOnline {
 		info.TotalSuccessfulPings = existedInfo.TotalSuccessfulPings + 1
+		info.IsOnWatchlist = false
+		info.IsAdjusted = false
 	}
 
 	if !info.IsOnline {
 		info.TotalSuccessfulPings = existedInfo.TotalSuccessfulPings
 		info.LastSeen = existedInfo.LastSeen
+		info.IsOnWatchlist = existedInfo.IsOnWatchlist
+		info.IsAdjusted = existedInfo.IsAdjusted
 	}
 
 	info.CumulativeResponseTime = existedInfo.CumulativeResponseTime + info.LastResponseTime