Fixed ws connection issues and added concurrent pin fetching to fetchPins

Author: nathansenn

ipfs/ipfs.go

@@ -256,10 +256,7 @@ func SyncNode(allPins map[string]ipfs.PinInfo, name string) {
 		go func(i int, key string) {
 			defer wg.Done()
 			size, _ := FileSize(key)
-			fmt.Println("Size: ", size)
-			fmt.Println("Name: ", name)
 			lock.Lock()
-			fmt.Println("PeerSize: ", localdata.PeerSize[name])
 			peersize = peersize + size
 			lock.Unlock()
 			// Check if the key exists in Pins
@@ -297,6 +294,7 @@ func SyncNode(allPins map[string]ipfs.PinInfo, name string) {
 			lock.Lock()
 			localdata.PeerSize[name] = peersize
 			fmt.Println("Synced: ", name)
+			fmt.Println("PeerSize: ", peersize)
 			localdata.NodesStatus[name] = "Synced"
 			lock.Unlock()
 			return

main.go

@@ -12,6 +12,7 @@ import (
 	"proofofaccess/localdata"
 	"proofofaccess/proofcrypto"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -64,7 +65,15 @@ func initialize(ctx context.Context) {
 	} else {
 		go fetchPins(ctx)
 	}
+	if *useWS && *nodeType == 2 {
+		localdata.UseWS = *useWS
+		fmt.Println()
+		go messaging.StartWsClient()
 
+	} else {
+		go pubsubHandler(ctx)
+		go connectToValidators(ctx, nodeType)
+	}
 	go api.StartAPI(ctx)
 
 }
@@ -126,12 +135,18 @@ func pubsubHandler(ctx context.Context) {
 	}
 }
 func fetchPins(ctx context.Context) {
+	var lock sync.Mutex // Mutex lock
+	newPins := false    // Assuming this is a boolean based on your usage
+
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		default:
+			lock.Lock()
 			ipfs.Pins = ipfs.NewPins
+			lock.Unlock()
+
 			fmt.Println("Fetching pins...")
 			allPins, err := ipfs.Shell.Pins()
 			for _, cid := range messaging.PinFileCids {
@@ -143,10 +158,16 @@ func fetchPins(ctx context.Context) {
 				fmt.Println("Error fetching pins:", err)
 				continue
 			}
+
+			lock.Lock()
 			ipfs.NewPins = make(map[string]interface{})
+			lock.Unlock()
+
 			for key, pinInfo := range allPins {
 				if pinInfo.Type == "recursive" {
+					lock.Lock()
 					ipfs.NewPins[key] = key
+					lock.Unlock()
 				}
 			}
 
@@ -155,44 +176,58 @@ func fetchPins(ctx context.Context) {
 
 			keysNotFound := 0
 
+			// Create a WaitGroup to wait for the function to finish
+			var wg sync.WaitGroup
+
 			// Iterate through the keys in NewPins
 			for key := range ipfs.NewPins {
-				// Check if the key exists in Pins
+				wg.Add(1)
+				go func(key string) {
+					defer wg.Done()
+					// Check if the key exists in Pins
+					lock.Lock()
+					_, exists := ipfs.Pins[key]
+					lock.Unlock()
 
-				if _, exists := ipfs.Pins[key]; !exists {
-					size, _ := ipfs.FileSize(key)
-					localdata.PeerSize[localdata.NodeName] = localdata.PeerSize[localdata.NodeName] + size
-					newPins = true
-					// If the key doesn't exist in Pins, add it to the pinsNotIncluded map
-					localdata.SavedRefs[key], _ = ipfs.Refs(key)
-					if localdata.Synced == false {
+					if !exists {
+						size, _ := ipfs.FileSize(key)
+						lock.Lock()
+						localdata.PeerSize[localdata.NodeName] += size
+						newPins = true
+						lock.Unlock()
 
+						// If the key doesn't exist in Pins, add it to the pinsNotIncluded map
+						savedRefs, _ := ipfs.Refs(key)
+						lock.Lock()
+						localdata.SavedRefs[key] = savedRefs
+						lock.Unlock()
+						lock.Lock()
 						localdata.SyncedPercentage = float32(keysNotFound) / float32(mapLength) * 100
 						fmt.Println("Synced: ", localdata.SyncedPercentage, "%")
+						lock.Unlock()
+						keysNotFound++
 					}
-					keysNotFound++
-				}
+				}(key)
 			}
-			if localdata.Synced == false {
-				fmt.Println("Synced: ", 100)
-				localdata.SyncedPercentage = 100
-				if *useWS && *nodeType == 2 {
-					localdata.UseWS = *useWS
-					go messaging.StartWsClient()
-				} else {
-					go pubsubHandler(ctx)
-					go connectToValidators(ctx, nodeType)
-				}
-			}
-			if newPins == true {
+
+			wg.Wait()
+
+			fmt.Println("Synced: ", 100)
+			lock.Lock()
+			localdata.SyncedPercentage = 100
+			lock.Unlock()
+
+			if newPins {
 				fmt.Println("New pins found")
 				messaging.SendCIDS("validator1")
+				newPins = false
 			}
-			newPins = false
+
 			time.Sleep(60 * time.Second)
 		}
 	}
 }
+
 func checkSynced(ctx context.Context) {
 	for {
 		select {

messaging/ws.go

@@ -29,9 +29,10 @@ func StartWsClient() {
 	go func() {
 		for {
 			if isConnected {
-				_, message, err := c.ReadMessage()
+				_, message, err := localdata.WsValidators["Validator1"].ReadMessage()
 				if err != nil {
 					log.Println("read:", err)
+					fmt.Println("Connection lost. Reconnecting...")
 					isConnected = false
 					continue
 				}
@@ -46,27 +47,51 @@ func StartWsClient() {
 
 	// Connection or reconnection loop
 	for {
-		if !isConnected {
-			c, _, err = websocket.DefaultDialer.Dial("ws://spk.tv/messaging", nil)
-			if err != nil {
-				log.Println("dial:", err)
-				time.Sleep(1 * time.Second) // Sleep for a second before next reconnection attempt
-				continue
+		for {
+			if !isConnected {
+				c, _, err = websocket.DefaultDialer.Dial("ws://spk.tv/messaging", nil)
+				if err != nil {
+					log.Println("dial:", err)
+					time.Sleep(1 * time.Second)
+					continue
+				}
+				isConnected = true
+				log.Println("Connected to the server")
+				salt, _ := proofcrypto.CreateRandomHash()
+				localdata.WsValidators["Validator1"] = c
+				fmt.Println("Connected to validator1")
+				wsPing(salt)
+			} else {
+				// Ping the server to check if still connected
+				err = localdata.WsValidators["Validator1"].WriteMessage(websocket.PingMessage, nil)
+				if err != nil {
+					log.Println("write:", err)
+					fmt.Println("Connection lost. Reconnecting...")
+					isConnected = false
+				}
 			}
-			isConnected = true
-			log.Println("Connected to the server")
-			salt, _ := proofcrypto.CreateRandomHash()
-			localdata.WsValidators["Validator1"] = c
-			fmt.Println("Connected to validator1")
-			wsPing(salt, c)
 
+			select {
+			case <-interrupt:
+				log.Println("interrupt")
+				if isConnected {
+					err = localdata.WsValidators["Validator1"].WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+					if err != nil {
+						log.Println("write close:", err)
+						return
+					}
+				}
+				return
+			default:
+				time.Sleep(1 * time.Second)
+			}
 		}
 
 		select {
 		case <-interrupt:
 			log.Println("interrupt")
 			if isConnected {
-				err = c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+				err = localdata.WsValidators["Validator1"].WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
 				if err != nil {
 					log.Println("write close:", err)
 					return
@@ -80,7 +105,8 @@ func StartWsClient() {
 	}
 }
 
-func wsPing(hash string, c *websocket.Conn) {
+func wsPing(hash string) {
+	localdata.Synced = true
 	data := map[string]string{
 		"type": TypePingPongPing,
 		"hash": hash,
@@ -92,5 +118,5 @@ func wsPing(hash string, c *websocket.Conn) {
 		return
 	}
 	fmt.Println("Client send: ", string(jsonData))
-	err = c.WriteMessage(websocket.TextMessage, jsonData)
+	err = localdata.WsValidators["Validator1"].WriteMessage(websocket.TextMessage, jsonData)
 }

validation/validation.go

@@ -35,7 +35,6 @@ func CreatProofHash(hash string, CID string) string {
 	// Get all the file blocks CIDs from the Target Files CID
 	fmt.Println("CID: ", CID)
 	cids := localdata.SavedRefs[CID]
-	fmt.Println("cids: ", cids)
 	// Get the length of the CIDs
 	length := len(cids)
 	fmt.Println("length", length)