feat(explorer): make possible to run sync in a separate process (#3224)

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/explorer/database.go

@@ -7,58 +7,83 @@ import (
 	"os"
 	"sort"
 	"sync"
+
+	"github.com/gofrs/flock"
 )
 
 // Database is a simple JSON database for storing and retrieving p2p network tokens and a name and description.
 type Database struct {
-	sync.RWMutex
-	path string
-	data map[string]TokenData
+	path  string
+	data  map[string]TokenData
+	flock *flock.Flock
+	sync.Mutex
 }
 
 // TokenData is a p2p network token with a name and description.
 type TokenData struct {
 	Name        string `json:"name"`
 	Description string `json:"description"`
+	Clusters    []ClusterData
+	Failures    int
+}
+
+type ClusterData struct {
+	Workers   []string
+	Type      string
+	NetworkID string
 }
 
 // NewDatabase creates a new Database with the given path.
 func NewDatabase(path string) (*Database, error) {
+	fileLock := flock.New(path + ".lock")
 	db := &Database{
-		data: make(map[string]TokenData),
-		path: path,
+		data:  make(map[string]TokenData),
+		path:  path,
+		flock: fileLock,
 	}
 	return db, db.load()
 }
 
 // Get retrieves a Token from the Database by its token.
 func (db *Database) Get(token string) (TokenData, bool) {
-	db.RLock()
-	defer db.RUnlock()
+	db.flock.Lock() // we are making sure that the file is not being written to
+	defer db.flock.Unlock()
+	db.Lock() // we are making sure that is safe if called by another instance in the same process
+	defer db.Unlock()
+	db.load()
 	t, ok := db.data[token]
 	return t, ok
 }
 
 // Set stores a Token in the Database by its token.
 func (db *Database) Set(token string, t TokenData) error {
+	db.flock.Lock()
+	defer db.flock.Unlock()
 	db.Lock()
+	defer db.Unlock()
+	db.load()
 	db.data[token] = t
-	db.Unlock()
 
-	return db.Save()
+	return db.save()
 }
 
 // Delete removes a Token from the Database by its token.
 func (db *Database) Delete(token string) error {
+	db.flock.Lock()
+	defer db.flock.Unlock()
 	db.Lock()
+	defer db.Unlock()
+	db.load()
 	delete(db.data, token)
-	db.Unlock()
-	return db.Save()
+	return db.save()
 }
 
 func (db *Database) TokenList() []string {
-	db.RLock()
-	defer db.RUnlock()
+	db.flock.Lock()
+	defer db.flock.Unlock()
+	db.Lock()
+	defer db.Unlock()
+	db.load()
 	tokens := []string{}
 	for k := range db.data {
 		tokens = append(tokens, k)
@@ -74,9 +99,6 @@ func (db *Database) TokenList() []string {
 
 // load reads the Database from disk.
 func (db *Database) load() error {
-	db.Lock()
-	defer db.Unlock()
-
 	if _, err := os.Stat(db.path); os.IsNotExist(err) {
 		return nil
 	}
@@ -91,10 +113,7 @@ func (db *Database) load() error {
 }
 
 // Save writes the Database to disk.
-func (db *Database) Save() error {
-	db.RLock()
-	defer db.RUnlock()
-
+func (db *Database) save() error {
 	// Marshal db.data into JSON
 	// Write the JSON to the file
 	f, err := os.Create(db.path)

core/explorer/discovery.go

@@ -16,22 +16,10 @@ import (
 type DiscoveryServer struct {
 	sync.Mutex
 	database       *Database
-	networkState   *NetworkState
 	connectionTime time.Duration
-	failures       map[string]int
 	errorThreshold int
 }
 
-type NetworkState struct {
-	Networks map[string]Network
-}
-
-func (s *DiscoveryServer) NetworkState() *NetworkState {
-	s.Lock()
-	defer s.Unlock()
-	return s.networkState
-}
-
 // NewDiscoveryServer creates a new DiscoveryServer with the given Database.
 // it keeps the db state in sync with the network state
 func NewDiscoveryServer(db *Database, dur time.Duration, failureThreshold int) *DiscoveryServer {
@@ -44,11 +32,7 @@ func NewDiscoveryServer(db *Database, dur time.Duration, failureThreshold int) *
 	return &DiscoveryServer{
 		database:       db,
 		connectionTime: dur,
-		networkState: &NetworkState{
-			Networks: map[string]Network{},
-		},
 		errorThreshold: failureThreshold,
-		failures:       make(map[string]int),
 	}
 }
 
@@ -116,10 +100,10 @@ func (s *DiscoveryServer) runBackground() {
 
 		if hasWorkers {
 			s.Lock()
-			s.networkState.Networks[token] = Network{
-				Clusters: ledgerK,
-			}
-			delete(s.failures, token)
+			data, _ := s.database.Get(token)
+			(&data).Clusters = ledgerK
+			(&data).Failures = 0
+			s.database.Set(token, data)
 			s.Unlock()
 		} else {
 			s.failedToken(token)
@@ -132,27 +116,23 @@ func (s *DiscoveryServer) runBackground() {
 func (s *DiscoveryServer) failedToken(token string) {
 	s.Lock()
 	defer s.Unlock()
-	s.failures[token]++
+	data, _ := s.database.Get(token)
+	(&data).Failures++
+	s.database.Set(token, data)
 }
 
 func (s *DiscoveryServer) deleteFailedConnections() {
 	s.Lock()
 	defer s.Unlock()
-	for k, v := range s.failures {
-		if v > s.errorThreshold {
-			log.Info().Any("network", k).Msg("Network has been removed from the database")
-			s.database.Delete(k)
-			delete(s.failures, k)
+	for _, t := range s.database.TokenList() {
+		data, _ := s.database.Get(t)
+		if data.Failures > s.errorThreshold {
+			log.Info().Any("token", t).Msg("Token has been removed from the database")
+			s.database.Delete(t)
 		}
 	}
 }
 
-type ClusterData struct {
-	Workers   []string
-	Type      string
-	NetworkID string
-}
-
 func (s *DiscoveryServer) retrieveNetworkData(c context.Context, ledger *blockchain.Ledger, networkData chan ClusterData) {
 	clusters := map[string]ClusterData{}
 
@@ -217,7 +197,7 @@ func (s *DiscoveryServer) retrieveNetworkData(c context.Context, ledger *blockch
 }
 
 // Start the discovery server. This is meant to be run in to a goroutine.
-func (s *DiscoveryServer) Start(ctx context.Context) error {
+func (s *DiscoveryServer) Start(ctx context.Context, keepRunning bool) error {
 	for {
 		select {
 		case <-ctx.Done():
@@ -225,6 +205,9 @@ func (s *DiscoveryServer) Start(ctx context.Context) error {
 		default:
 			// Collect data
 			s.runBackground()
+			if !keepRunning {
+				return nil
+			}
 		}
 	}
 }