feat: remote worker selection with local worker fallback & configuration and fine tuning

pkg/api/config.go

@@ -0,0 +1,22 @@
+package api
+
+import "time"
+
+// APIConfig contains configuration settings for the API
+type APIConfig struct {
+	WorkerResponseTimeout time.Duration
+	// Add other API-specific configuration fields here
+}
+
+var DefaultConfig = APIConfig{
+	WorkerResponseTimeout: 60 * time.Second,
+	// Set default values for other fields here
+}
+
+// LoadConfig loads the API configuration
+// This can be expanded later to load from environment variables or a file
+func LoadConfig() (*APIConfig, error) {
+	// For now, we'll just return the default config
+	config := DefaultConfig
+	return &config, nil
+}

pkg/api/handlers_data.go

@@ -75,6 +75,13 @@ func publishWorkRequest(api *API, requestID string, request workers.WorkerType,
 // - c: The gin.Context object, which provides the context for the HTTP request.
 // - responseCh: A channel that receives the worker's response as a byte slice.
 func handleWorkResponse(c *gin.Context, responseCh chan []byte) {
+	config, err := LoadConfig()
+	if err != nil {
+		logrus.Errorf("Failed to load API config: %v", err)
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "Internal server error"})
+		return
+	}
+
 	for {
 		select {
 		case response := <-responseCh:
@@ -101,8 +108,7 @@ func handleWorkResponse(c *gin.Context, responseCh chan []byte) {
 
 			c.JSON(http.StatusOK, result)
 			return
-		// teslashibe: adjust to timeout after 10 seconds for performance testing
-		case <-time.After(10 * time.Second):
+		case <-time.After(config.WorkerResponseTimeout):
 			c.JSON(http.StatusGatewayTimeout, gin.H{"error": "Request timed out, check that port 4001 TCP inbound is open."})
 			return
 		case <-c.Done():

pkg/config/app.go

@@ -112,8 +112,8 @@ func GetInstance() *AppConfig {
 	once.Do(func() {
 		instance = &AppConfig{}
 
-		instance.setEnvVariableConfig()
 		instance.setDefaultConfig()
+		instance.setEnvVariableConfig()
 		instance.setFileConfig(viper.GetString("FILE_PATH"))
 		err := instance.setCommandLineConfig()
 		if err != nil {
@@ -194,6 +194,7 @@ func (c *AppConfig) setCommandLineConfig() error {
 	pflag.StringVar(&c.StakeAmount, "stake", viper.GetString(StakeAmount), "Amount of tokens to stake")
 	pflag.BoolVar(&c.Debug, "debug", viper.GetBool(Debug), "Override some protections for debugging (temporary)")
 	pflag.StringVar(&c.Environment, "env", viper.GetString(Environment), "Environment to connect to")
+	pflag.StringVar(&c.Version, "version", viper.GetString("VERSION"), "application version")
 	pflag.BoolVar(&c.AllowedPeer, "allowedPeer", viper.GetBool(AllowedPeer), "Set to true to allow setting this node as the allowed peer")
 	pflag.StringVar(&c.PrivateKey, "privateKey", viper.GetString(PrivateKey), "The private key")
 	pflag.StringVar(&c.PrivateKeyFile, "privKeyFile", viper.GetString(PrivKeyFile), "The private key file")
@@ -222,6 +223,7 @@ func (c *AppConfig) setCommandLineConfig() error {
 	pflag.BoolVar(&c.WebScraper, "webScraper", viper.GetBool(WebScraper), "WebScraper")
 	pflag.BoolVar(&c.LlmServer, "llmServer", viper.GetBool(LlmServer), "Can service LLM requests")
 	pflag.BoolVar(&c.Faucet, "faucet", viper.GetBool(Faucet), "Faucet")
+
 	pflag.Parse()
 
 	// Bind command line flags to Viper (optional, if you want to use Viper for additional configuration)

pkg/network/discover.go

@@ -8,9 +8,10 @@ import (
 	"github.com/cenkalti/backoff/v4"
 
 	"github.com/libp2p/go-libp2p/core/peer"
-	"github.com/masa-finance/masa-oracle/pkg/config"
 	"github.com/multiformats/go-multiaddr"
 
+	"github.com/masa-finance/masa-oracle/pkg/config"
+
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
@@ -39,7 +40,7 @@ func Discover(ctx context.Context, host host.Host, dht *dht.IpfsDHT, protocol pr
 		logrus.Infof("[+] Successfully advertised protocol %s", protocolString)
 	}
 
-	ticker := time.NewTicker(time.Second * 10)
+	ticker := time.NewTicker(time.Minute * 1)
 	defer ticker.Stop()
 
 	var peerChan <-chan peer.AddrInfo

pkg/network/kdht.go

@@ -30,7 +30,7 @@ func (dbValidator) Validate(_ string, _ []byte) error        { return nil }
 func (dbValidator) Select(_ string, _ [][]byte) (int, error) { return 0, nil }
 
 func WithDht(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.Multiaddr,
-	protocolId, prefix protocol.ID, peerChan chan PeerEvent, isStaked bool, removePeerCallback func(peer.ID)) (*dht.IpfsDHT, error) {
+	protocolId, prefix protocol.ID, peerChan chan PeerEvent, isStaked bool) (*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
@@ -46,8 +46,6 @@ func WithDht(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.Mul
 	go monitorRoutingTable(ctx, kademliaDHT, time.Minute)
 
 	kademliaDHT.RoutingTable().PeerAdded = func(p peer.ID) {
-		logrus.Infof("[+] Peer added to DHT: %s", p.String())
-
 		pe := PeerEvent{
 			AddrInfo: peer.AddrInfo{ID: p},
 			Action:   PeerAdded,
@@ -57,16 +55,12 @@ func WithDht(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.Mul
 	}
 
 	kademliaDHT.RoutingTable().PeerRemoved = func(p peer.ID) {
-		logrus.Infof("[-] Peer removed from DHT: %s", p)
 		pe := PeerEvent{
 			AddrInfo: peer.AddrInfo{ID: p},
 			Action:   PeerRemoved,
 			Source:   "kdht",
 		}
 		peerChan <- pe
-		if removePeerCallback != nil {
-			removePeerCallback(p)
-		}
 	}
 
 	if err = kademliaDHT.Bootstrap(ctx); err != nil {
@@ -153,10 +147,6 @@ func monitorRoutingTable(ctx context.Context, dht *dht.IpfsDHT, interval time.Du
 			routingTable := dht.RoutingTable()
 			// Log the size of the routing table
 			logrus.Infof("[+] Routing table size: %d", routingTable.Size())
-			// Log the peer IDs in the routing table
-			for _, p := range routingTable.ListPeers() {
-				logrus.Debugf("[+] Peer in routing table: %s", p.String())
-			}
 		case <-ctx.Done():
 			// If the context is cancelled, stop the goroutine
 			return

pkg/oracle_node.go

@@ -24,7 +24,6 @@ import (
 	dht "github.com/libp2p/go-libp2p-kad-dht"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
-	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
 	"github.com/libp2p/go-libp2p/p2p/muxer/yamux"
@@ -37,7 +36,8 @@ import (
 
 	shell "github.com/ipfs/go-ipfs-api"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	chain "github.com/masa-finance/masa-oracle/pkg/chain"
+
+	"github.com/masa-finance/masa-oracle/pkg/chain"
 	"github.com/masa-finance/masa-oracle/pkg/config"
 	"github.com/masa-finance/masa-oracle/pkg/masacrypto"
 	myNetwork "github.com/masa-finance/masa-oracle/pkg/network"
@@ -216,11 +216,7 @@ func (node *OracleNode) Start() (err error) {
 	go node.ListenToNodeTracker()
 	go node.handleDiscoveredPeers()
 
-	removePeerCallback := func(p peer.ID) {
-		node.NodeTracker.RemoveNodeData(p.String())
-	}
-
-	node.DHT, err = myNetwork.WithDht(node.Context, node.Host, bootNodeAddrs, node.Protocol, config.MasaPrefix, node.PeerChan, node.IsStaked, removePeerCallback)
+	node.DHT, err = myNetwork.WithDht(node.Context, node.Host, bootNodeAddrs, node.Protocol, config.MasaPrefix, node.PeerChan, node.IsStaked)
 	if err != nil {
 		return err
 	}

pkg/pubsub/node_data.go

@@ -91,6 +91,10 @@ func NewNodeData(addr multiaddr.Multiaddr, peerId peer.ID, publicKey string, act
 // and peer ID in the format "/ip4/127.0.0.1/tcp/4001/p2p/QmcgpsyWgH8Y8ajJz1Cu72KnS5uo2Aa2LpzU7kinSupNKC".
 // This can be used by other nodes to connect to this node.
 func (n *NodeData) Address() string {
+	// Add a check for empty addresses
+	if len(n.Multiaddrs) == 0 {
+		return ""
+	}
 	return fmt.Sprintf("%s/p2p/%s", n.Multiaddrs[0].String(), n.PeerId.String())
 }
 
@@ -117,16 +121,18 @@ func (n *NodeData) CanDoWork(workerType WorkerCategory) bool {
 		logrus.Infof("[+] Skipping worker %s due to timeout", n.PeerId)
 		return false
 	}
-
+	if !(n.IsStaked && n.IsActive) {
+		return false
+	}
 	switch workerType {
 	case CategoryTwitter:
-		return n.IsActive && n.IsTwitterScraper
+		return n.IsTwitterScraper
 	case CategoryDiscord:
-		return n.IsActive && n.IsDiscordScraper
+		return n.IsDiscordScraper
 	case CategoryTelegram:
-		return n.IsActive && n.IsTelegramScraper
+		return n.IsTelegramScraper
 	case CategoryWeb:
-		return n.IsActive && n.IsWebScraper
+		return n.IsWebScraper
 	default:
 		return false
 	}

pkg/pubsub/node_event_tracker.go

@@ -249,7 +249,7 @@ func (net *NodeEventTracker) GetUpdatedNodes(since time.Time) []NodeData {
 }
 
 // GetEthAddress returns the Ethereum address for the given remote peer.
-// It gets the peer's public key from the network's peerstore, converts
+// It gets the peer's public key from the network's peer store, converts
 // it to a hex string, and converts that to an Ethereum address.
 // Returns an empty string if there is no public key for the peer.
 func GetEthAddress(remotePeer peer.ID, n network.Network) string {
@@ -273,6 +273,18 @@ func GetEthAddress(remotePeer peer.ID, n network.Network) string {
 	return publicKeyHex
 }
 
+// GetEligibleWorkerNodes returns a slice of NodeData for nodes that are eligible to perform a specific category of work.
+func (net *NodeEventTracker) GetEligibleWorkerNodes(category WorkerCategory) []NodeData {
+	logrus.Debugf("Getting eligible worker nodes for category: %s", category)
+	result := make([]NodeData, 0)
+	for _, nodeData := range net.GetAllNodeData() {
+		if nodeData.CanDoWork(category) {
+			result = append(result, nodeData)
+		}
+	}
+	return result
+}
+
 // IsStaked returns whether the node with the given peerID is marked as staked in the node data tracker.
 // Returns false if no node data is found for the given peerID.
 func (net *NodeEventTracker) IsStaked(peerID string) bool {
@@ -366,7 +378,7 @@ func (net *NodeEventTracker) AddOrUpdateNodeData(nodeData *NodeData, forceGossip
 // entry, and if expired, processes the connect and removes the entry.
 func (net *NodeEventTracker) ClearExpiredBufferEntries() {
 	for {
-		time.Sleep(30 * time.Second) // E.g., every 5 seconds
+		time.Sleep(1 * time.Minute)
 		now := time.Now()
 		for peerID, entry := range net.ConnectBuffer {
 			if now.Sub(entry.ConnectTime) > time.Minute*1 {
@@ -388,11 +400,13 @@ func (net *NodeEventTracker) ClearExpiredBufferEntries() {
 //
 // Parameters:
 //   - peerID: A string representing the ID of the peer to be removed.
-func (net *NodeEventTracker) RemoveNodeData(peerID string) {
-	net.nodeData.Delete(peerID)
-	delete(net.ConnectBuffer, peerID)
-	logrus.Infof("[+] Removed peer %s from NodeTracker", peerID)
-}
+//
+// TODO: we should never remove node data from the internal map. Otherwise we lose all tracking of activity.
+//func (net *NodeEventTracker) RemoveNodeData(peerID string) {
+//	net.nodeData.Delete(peerID)
+//	delete(net.ConnectBuffer, peerID)
+//	logrus.Infof("[+] Removed peer %s from NodeTracker", peerID)
+//}
 
 // ClearExpiredWorkerTimeouts periodically checks and clears expired worker timeouts.
 // It runs in an infinite loop, sleeping for 5 minutes between each iteration.
@@ -445,7 +459,6 @@ func (net *NodeEventTracker) cleanupStalePeers(hostId string) {
 		if now.Sub(time.Unix(nodeData.LastUpdatedUnix, 0)) > maxDisconnectionTime {
 			if nodeData.PeerId.String() != hostId {
 				logrus.Infof("Removing stale peer: %s", nodeData.PeerId)
-				net.RemoveNodeData(nodeData.PeerId.String())
 				delete(net.ConnectBuffer, nodeData.PeerId.String())
 
 				// Notify about peer removal

pkg/workers/config.go

@@ -0,0 +1,31 @@
+package workers
+
+import (
+	"time"
+)
+
+type WorkerConfig struct {
+	WorkerTimeout         time.Duration
+	WorkerResponseTimeout time.Duration
+	ConnectionTimeout     time.Duration
+	MaxRetries            int
+	MaxSpawnAttempts      int
+	WorkerBufferSize      int
+	MaxRemoteWorkers      int
+}
+
+var DefaultConfig = WorkerConfig{
+	WorkerTimeout:         55 * time.Second,
+	WorkerResponseTimeout: 30 * time.Second,
+	ConnectionTimeout:     1 * time.Second,
+	MaxRetries:            1,
+	MaxSpawnAttempts:      1,
+	WorkerBufferSize:      100,
+	MaxRemoteWorkers:      1,
+}
+
+func LoadConfig() (*WorkerConfig, error) {
+	// For now, we'll just return the default config
+	config := DefaultConfig
+	return &config, nil
+}

pkg/workers/handler.go

@@ -9,15 +9,16 @@ import (
 
 	"github.com/asynkron/protoactor-go/actor"
 	pubsub2 "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/multiformats/go-multiaddr"
+	"github.com/sirupsen/logrus"
+
 	masa "github.com/masa-finance/masa-oracle/pkg"
 	"github.com/masa-finance/masa-oracle/pkg/config"
 	"github.com/masa-finance/masa-oracle/pkg/scrapers/discord"
 	"github.com/masa-finance/masa-oracle/pkg/scrapers/telegram"
 	"github.com/masa-finance/masa-oracle/pkg/scrapers/twitter"
 	"github.com/masa-finance/masa-oracle/pkg/scrapers/web"
 	"github.com/masa-finance/masa-oracle/pkg/workers/messages"
-	"github.com/multiformats/go-multiaddr"
-	"github.com/sirupsen/logrus"
 )
 
 type LLMChatBody struct {
@@ -158,7 +159,10 @@ func (a *Worker) HandleWork(ctx actor.Context, m *messages.Work, node *masa.Orac
 	}
 
 	if err != nil {
-		host, _, err := net.SplitHostPort(m.Sender.Address)
+		host, _, err2 := net.SplitHostPort(m.Sender.Address)
+		if err2 != nil {
+			logrus.Errorf("[-] Error splitting host and port: %v", err2)
+		}
 		addrs := node.Host.Addrs()
 		isLocalHost := false
 		for _, addr := range addrs {