fix(lavasession): add mutex to Endpoint to prevent race conditions

Add a sync.RWMutex to the Endpoint struct to protect concurrent access
to Connections, ConnectionRefusals, and Enabled fields. This fixes a
pre-existing race condition that occurred when multiple goroutines
(from probeProviders) accessed the same Endpoint object simultaneously.

The race was detected in fetchEndpointConnectionFromConsumerSessionWithProvider
where multiple providers could share the same Endpoint object and
concurrently modify its fields without synchronization.

The fix:
- Added `mu sync.RWMutex` field to Endpoint struct
- Wrapped modifications to Connections, ConnectionRefusals, and Enabled
  with Lock/Unlock
- Used RLock/RUnlock for read-only access to Enabled field
- Carefully release lock before blocking network calls and re-acquire after

Author: NadavLevi

protocol/lavasession/consumer_types.go

@@ -141,6 +141,7 @@ type Endpoint struct {
 	Addons             map[string]struct{}
 	Extensions         map[string]struct{}
 	Geolocation        planstypes.Geolocation
+	mu                 sync.RWMutex // Protects Connections, ConnectionRefusals, and Enabled fields
 }
 
 func (e *Endpoint) CheckSupportForServices(addon string, extensions []string) (supported bool) {
@@ -507,7 +508,10 @@ func (cswp *ConsumerSessionsWithProvider) fetchEndpointConnectionFromConsumerSes
 		for idx, endpoint := range cswp.Endpoints {
 			// retryDisabledEndpoints will attempt to reconnect to the provider even though we have disabled the endpoint
 			// this is used on a routine that tries to reconnect to a provider that has been disabled due to being unable to connect to it.
-			if !retryDisabledEndpoints && !endpoint.Enabled {
+			endpoint.mu.RLock()
+			enabled := endpoint.Enabled
+			endpoint.mu.RUnlock()
+			if !retryDisabledEndpoints && !enabled {
 				continue
 			}
 			if retryDisabledEndpoints {
@@ -521,6 +525,10 @@ func (cswp *ConsumerSessionsWithProvider) fetchEndpointConnectionFromConsumerSes
 			}
 			// return
 			connectEndpoint := func(cswp *ConsumerSessionsWithProvider, ctx context.Context, endpoint *Endpoint) (endpointConnection_ *EndpointConnection, connected_ bool) {
+				// Lock the endpoint to protect concurrent access to Connections, ConnectionRefusals, and Enabled
+				endpoint.mu.Lock()
+				defer endpoint.mu.Unlock()
+
 				// Clean up dead connections before iterating to prevent accumulation
 				cleanedConnections := make([]*EndpointConnection, 0, len(endpoint.Connections))
 				deadConnectionCount := 0
@@ -589,7 +597,13 @@ func (cswp *ConsumerSessionsWithProvider) fetchEndpointConnectionFromConsumerSes
 						}
 					}
 				}
+
+				// Release lock before making network call to avoid blocking other goroutines
+				endpoint.mu.Unlock()
 				client, conn, err := cswp.ConnectRawClientWithTimeout(ctx, endpoint.NetworkAddress)
+				// Re-acquire lock to update endpoint state
+				endpoint.mu.Lock()
+
 				if err != nil {
 					endpoint.ConnectionRefusals++
 					utils.LavaFormatInfo("error connecting to provider",
@@ -621,7 +635,9 @@ func (cswp *ConsumerSessionsWithProvider) fetchEndpointConnectionFromConsumerSes
 			if !connected_ {
 				continue
 			}
+			endpoint.mu.Lock()
 			cswp.Endpoints[idx].Enabled = true // return enabled once we successfully reconnect
+			endpoint.mu.Unlock()
 			// successful new connection add to endpoints list
 			endpoints = append(endpoints, &EndpointAndChosenConnection{endpoint: endpoint, chosenEndpointConnection: endpointConnection})
 			if !getAllEndpoints {
@@ -638,7 +654,10 @@ func (cswp *ConsumerSessionsWithProvider) fetchEndpointConnectionFromConsumerSes
 		// before verifying all are Disabled.
 		allDisabled = true
 		for _, endpoint := range cswp.Endpoints {
-			if !endpoint.Enabled {
+			endpoint.mu.RLock()
+			enabled := endpoint.Enabled
+			endpoint.mu.RUnlock()
+			if !enabled {
 				continue
 			}
 			// even one endpoint is enough for us to not purge.