fix: set network operations timeouts for miners discovery

Author: gmlexx

miners/avalon.go

@@ -367,7 +367,7 @@ func getAddresses(ctx context.Context, network string) iter.Seq[netip.Addr] {
 				return
 			default:
 			}
-			
+
 			if !yield(next) {
 				return
 			}
@@ -454,7 +454,9 @@ func version(ctx context.Context, address string, port int) (*AvalonQVersion, er
 }
 
 func send[T any](ctx context.Context, address string, port int, sender Sender, receiver Receiver[T]) (T, error) {
-	var d net.Dialer
+	d := net.Dialer{
+		Timeout: time.Second,
+	}
 	ctx, cancel := context.WithTimeout(ctx, time.Second)
 	defer cancel()
 
@@ -465,6 +467,12 @@ func send[T any](ctx context.Context, address string, port int, sender Sender, r
 	}
 	defer conn.Close()
 
+	// Set deadline for all operations on this connection
+	if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		var zero T
+		return zero, err
+	}
+
 	if err := sender(conn); err != nil {
 		var zero T
 		return zero, err

scheduler/miners.go

@@ -21,21 +21,22 @@ func (s *MinerScheduler) discoverMiners(ctx context.Context) error {
 		newlyDiscoveredMiners = miners.Discover(ctx, s.config.Network)
 	}
 
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	// Add only new miners that don't already exist
 	newMinersCount := 0
 	for _, newMiner := range newlyDiscoveredMiners {
 		key := fmt.Sprintf("%s:%d", newMiner.Address, newMiner.Port)
-		if _, exists := s.discoveredMiners[key]; !exists {
-			s.discoveredMiners[key] = newMiner
+		if _, exists := s.discoveredMiners.LoadOrStore(key, newMiner); !exists {
 			newMinersCount++
 			s.logger.Printf("  New miner discovered: %s:%d", newMiner.Address, newMiner.Port)
 		}
 	}
 
-	totalMiners := len(s.discoveredMiners)
+	// Count total miners
+	totalMiners := 0
+	s.discoveredMiners.Range(func(_, _ any) bool {
+		totalMiners++
+		return true
+	})
 	s.logger.Printf("Discovery complete: %d total miners (%d newly discovered)", totalMiners, newMinersCount)
 
 	return nil

scheduler/scheduler.go

@@ -85,7 +85,7 @@ type MinerScheduler struct {
 	config *Config
 
 	// State
-	discoveredMiners       map[string]*miners.AvalonQHost
+	discoveredMiners       sync.Map // map[string]*miners.AvalonQHost
 	pricesMarketData       *entsoe.PublicationMarketData
 	pricesMarketDataExpiry time.Time
 	isRunning              bool
@@ -119,10 +119,9 @@ func NewMinerScheduler(config *Config, logger *log.Logger) *MinerScheduler {
 	}
 
 	scheduler := &MinerScheduler{
-		config:           config,
-		discoveredMiners: make(map[string]*miners.AvalonQHost),
-		stopChan:         make(chan struct{}),
-		logger:           logger,
+		config:   config,
+		stopChan: make(chan struct{}),
+		logger:   logger,
 		weatherCache: WeatherForecastCache{
 			cacheDuration: 2 * time.Hour,
 		},
@@ -154,14 +153,14 @@ func (s *MinerScheduler) GetConfig() *Config {
 
 // GetDiscoveredMiners returns a copy of the currently discovered miners
 func (s *MinerScheduler) GetDiscoveredMiners() []*miners.AvalonQHost {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	// Convert map to slice and return a copy
-	minersCopy := make([]*miners.AvalonQHost, 0, len(s.discoveredMiners))
-	for _, miner := range s.discoveredMiners {
-		minersCopy = append(minersCopy, miner)
-	}
+	// Convert sync.Map to slice
+	minersCopy := make([]*miners.AvalonQHost, 0)
+	s.discoveredMiners.Range(func(_, value any) bool {
+		if miner, ok := value.(*miners.AvalonQHost); ok {
+			minersCopy = append(minersCopy, miner)
+		}
+		return true
+	})
 	return minersCopy
 }
 
@@ -366,9 +365,16 @@ func (s *MinerScheduler) GetStatus() Status {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
+	// Count miners in sync.Map
+	minersCount := 0
+	s.discoveredMiners.Range(func(_, _ any) bool {
+		minersCount++
+		return true
+	})
+
 	return Status{
 		IsRunning:     s.isRunning,
-		MinersCount:   len(s.discoveredMiners),
+		MinersCount:   minersCount,
 		HasMarketData: s.pricesMarketData != nil,
 	}
 }

scheduler/scheduler_test.go

@@ -315,13 +315,10 @@ func TestGetDiscoveredMiners(t *testing.T) {
 		{Address: "192.168.1.101", Port: 4028},
 	}
 
-	scheduler.mu.Lock()
-	scheduler.discoveredMiners = make(map[string]*miners.AvalonQHost)
 	for _, miner := range mockMiners {
 		key := fmt.Sprintf("%s:%d", miner.Address, miner.Port)
-		scheduler.discoveredMiners[key] = miner
+		scheduler.discoveredMiners.Store(key, miner)
 	}
-	scheduler.mu.Unlock()
 
 	// Test getting miners
 	retrievedMiners := scheduler.GetDiscoveredMiners()
@@ -352,13 +349,10 @@ func TestDiscoverMinersPreservesExisting(t *testing.T) {
 		{Address: "192.168.1.101", Port: 4028},
 	}
 
-	scheduler.mu.Lock()
-	scheduler.discoveredMiners = make(map[string]*miners.AvalonQHost)
 	for _, miner := range initialMiners {
 		key := fmt.Sprintf("%s:%d", miner.Address, miner.Port)
-		scheduler.discoveredMiners[key] = miner
+		scheduler.discoveredMiners.Store(key, miner)
 	}
-	scheduler.mu.Unlock()
 
 	// Run discovery (should not find anything on 127.0.0.1/32, but existing miners should be preserved)
 	err := scheduler.discoverMiners(context.Background())
@@ -422,12 +416,11 @@ func TestSchedulerStatus_WithData(t *testing.T) {
 		CreatedDateTime: "2024-01-15T10:30:00Z",
 	}
 
-	scheduler.mu.Lock()
-	scheduler.discoveredMiners = make(map[string]*miners.AvalonQHost)
 	for _, miner := range mockMiners {
 		key := fmt.Sprintf("%s:%d", miner.Address, miner.Port)
-		scheduler.discoveredMiners[key] = miner
+		scheduler.discoveredMiners.Store(key, miner)
 	}
+	scheduler.mu.Lock()
 	scheduler.pricesMarketData = mockDoc
 	scheduler.mu.Unlock()
 
@@ -495,10 +488,9 @@ func BenchmarkSchedulerGetStatus(b *testing.B) {
 			Port:    4028,
 		}
 	}
-	scheduler.discoveredMiners = make(map[string]*miners.AvalonQHost)
 	for _, miner := range mockMiners {
 		key := fmt.Sprintf("%s:%d", miner.Address, miner.Port)
-		scheduler.discoveredMiners[key] = miner
+		scheduler.discoveredMiners.Store(key, miner)
 	}
 
 	for b.Loop() {
@@ -654,10 +646,9 @@ func BenchmarkSchedulerGetDiscoveredMiners(b *testing.B) {
 			Port:    4028,
 		}
 	}
-	scheduler.discoveredMiners = make(map[string]*miners.AvalonQHost)
 	for _, miner := range mockMiners {
 		key := fmt.Sprintf("%s:%d", miner.Address, miner.Port)
-		scheduler.discoveredMiners[key] = miner
+		scheduler.discoveredMiners.Store(key, miner)
 	}
 
 	for b.Loop() {