fix: resolve connection pool concurrency issues and enhance test coverage

This commit addresses several concurrency-related issues discovered during
the mutex refactoring and improves test coverage and documentation.

Fixes:
- Initialize connection pool dead slice to empty slice instead of nil
- Resolve scheduleResurrect deadlock by passing deadSince parameter
- Fix test deadlocks in connection_internal_test.go by extracting state before lock release
- Implement rwLocker interface methods for statusConnectionPool compatibility
- Update integration test to use new mutex-embedded pool access patterns

Improvements:
- Add test coverage for metrics incrementResponse method
- Add explanatory comments for connection resurrection logic and dead connection sorting
- Document concurrency tradeoffs in connection pool management

These changes ensure robust concurrent behavior and comprehensive test coverage
for the refactored mutex patterns introduced in the previous commit.

Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>

Author: sean-

opensearchtransport/connection.go

@@ -55,6 +55,14 @@ type ConnectionPool interface {
 	URLs() []*url.URL            // URLs returns the list of URLs of available connections.
 }
 
+// rwLocker defines the interface for connection pools that support read-write locking.
+// This allows for more efficient concurrent access when only read operations are needed.
+type rwLocker interface {
+	sync.Locker // Embeds Lock() and Unlock() methods
+	RLock()
+	RUnlock()
+}
+
 // Connection represents a connection to a node.
 type Connection struct {
 	URL        *url.URL
@@ -98,8 +106,10 @@ type roundRobinSelector struct {
 
 // Compile-time checks to ensure interface compliance
 var (
-	_ ConnectionPool = (*statusConnectionPool)(nil)
 	_ ConnectionPool = (*singleConnectionPool)(nil)
+
+	_ ConnectionPool = (*statusConnectionPool)(nil)
+	_ rwLocker       = (*statusConnectionPool)(nil)
 )
 
 // NewConnectionPool creates and returns a default connection pool.
@@ -120,6 +130,7 @@ func NewConnectionPool(conns []*Connection, selector Selector) ConnectionPool {
 		resurrectTimeoutFactorCutoff: defaultResurrectTimeoutFactorCutoff,
 	}
 	pool.mu.live = conns
+	pool.mu.dead = []*Connection{}
 	return pool
 }
 
@@ -199,16 +210,26 @@ func (cp *statusConnectionPool) OnFailure(c *Connection) error {
 	}
 
 	c.markAsDeadWithLock()
-	cp.scheduleResurrect(c)
+	deadSince := c.mu.deadSince
 	c.mu.Unlock()
 
+	cp.scheduleResurrect(c, deadSince)
+
 	// Push item to dead list and sort slice by number of failures
 	cp.mu.dead = append(cp.mu.dead, c)
+
+	// Sort by failure count for resurrection prioritization.
+	// CONCURRENCY TRADEOFF: Atomic loads are used without additional locking during sort,
+	// allowing failure counts to change mid-sort and resulting in slightly inconsistent
+	// ordering. This design prioritizes common-case latency over absolute correctness
+	// during failure scenarios. While failure counts could be snapshotted before sorting,
+	// the list ordering is not guaranteed to remain perfectly sorted by failure count
+	// between operations, making "mostly correct" sorting with atomics acceptable.
+	// Any temporary misordering self-corrects on subsequent failure events.
 	sort.Slice(cp.mu.dead, func(i, j int) bool {
 		c1 := cp.mu.dead[i]
 		c2 := cp.mu.dead[j]
 
-		// Use atomic loads for failure counts - no locking needed
 		failures1 := c1.failures.Load()
 		failures2 := c2.failures.Load()
 
@@ -261,6 +282,30 @@ func (cp *statusConnectionPool) connections() []*Connection {
 	return conns
 }
 
+// RLock acquires a read lock on the connection pool.
+// Implements rwLocker interface for efficient concurrent read access.
+func (cp *statusConnectionPool) RLock() {
+	cp.mu.RLock()
+}
+
+// RUnlock releases the read lock on the connection pool.
+// Implements rwLocker interface for efficient concurrent read access.
+func (cp *statusConnectionPool) RUnlock() {
+	cp.mu.RUnlock()
+}
+
+// Lock acquires a write lock on the connection pool.
+// Implements rwLocker interface (via embedded sync.Locker).
+func (cp *statusConnectionPool) Lock() {
+	cp.mu.Lock()
+}
+
+// Unlock releases the write lock on the connection pool.
+// Implements rwLocker interface (via embedded sync.Locker).
+func (cp *statusConnectionPool) Unlock() {
+	cp.mu.Unlock()
+}
+
 // resurrect adds the connection to the list of available connections.
 // When removeDead is true, it also removes it from the dead list.
 //
@@ -294,16 +339,12 @@ func (cp *statusConnectionPool) resurrectWithLock(c *Connection, removeDead bool
 }
 
 // scheduleResurrect schedules the connection to be resurrected.
-func (cp *statusConnectionPool) scheduleResurrect(c *Connection) {
+func (cp *statusConnectionPool) scheduleResurrect(c *Connection, deadSince time.Time) {
 	failures := c.failures.Load()
 	factor := min(failures-1, int64(cp.resurrectTimeoutFactorCutoff))
 	timeout := time.Duration(cp.resurrectTimeoutInitial.Seconds() * math.Exp2(float64(factor)) * float64(time.Second))
 
 	if debugLogger != nil {
-		c.mu.RLock()
-		deadSince := c.mu.deadSince
-		c.mu.RUnlock()
-
 		debugLogger.Logf(
 			"Resurrect %s (failures=%d, factor=%d, timeout=%s) in %s\n",
 			c.URL,

opensearchtransport/connection_integration_test.go

@@ -84,7 +84,9 @@ func TestStatusConnectionPool(t *testing.T) {
 
 	transport, _ := New(cfg)
 
-	pool := transport.pool.(*statusConnectionPool)
+	transport.mu.RLock()
+	pool := transport.mu.pool.(*statusConnectionPool)
+	transport.mu.RUnlock()
 	pool.resurrectTimeoutInitial = time.Second
 
 	for i := 1; i <= 9; i++ {
@@ -98,11 +100,11 @@ func TestStatusConnectionPool(t *testing.T) {
 		}
 	}
 
-	pool.Lock()
-	if len(pool.live) != 3 {
-		t.Errorf("Unexpected number of live connections, want=3, got=%d", len(pool.live))
+	pool.mu.Lock()
+	if len(pool.mu.live) != 3 {
+		t.Errorf("Unexpected number of live connections, want=3, got=%d", len(pool.mu.live))
 	}
-	pool.Unlock()
+	pool.mu.Unlock()
 
 	server = servers[1]
 	fmt.Printf("==> Closing server: %s\n", server.Addr)
@@ -121,17 +123,17 @@ func TestStatusConnectionPool(t *testing.T) {
 		}
 	}
 
-	pool.Lock()
-	if len(pool.live) != 2 {
-		t.Errorf("Unexpected number of live connections, want=2, got=%d", len(pool.live))
+	pool.mu.Lock()
+	if len(pool.mu.live) != 2 {
+		t.Errorf("Unexpected number of live connections, want=2, got=%d", len(pool.mu.live))
 	}
-	pool.Unlock()
+	pool.mu.Unlock()
 
-	pool.Lock()
-	if len(pool.dead) != 1 {
-		t.Errorf("Unexpected number of dead connections, want=1, got=%d", len(pool.dead))
+	pool.mu.Lock()
+	if len(pool.mu.dead) != 1 {
+		t.Errorf("Unexpected number of dead connections, want=1, got=%d", len(pool.mu.dead))
 	}
-	pool.Unlock()
+	pool.mu.Unlock()
 
 	server = NewServer("localhost:10002", http.HandlerFunc(defaultHandler))
 	servers[1] = server
@@ -156,15 +158,15 @@ func TestStatusConnectionPool(t *testing.T) {
 		}
 	}
 
-	pool.Lock()
-	if len(pool.live) != 3 {
-		t.Errorf("Unexpected number of live connections, want=3, got=%d", len(pool.live))
+	pool.mu.Lock()
+	if len(pool.mu.live) != 3 {
+		t.Errorf("Unexpected number of live connections, want=3, got=%d", len(pool.mu.live))
 	}
-	pool.Unlock()
+	pool.mu.Unlock()
 
-	pool.Lock()
-	if len(pool.dead) != 0 {
-		t.Errorf("Unexpected number of dead connections, want=0, got=%d", len(pool.dead))
+	pool.mu.Lock()
+	if len(pool.mu.dead) != 0 {
+		t.Errorf("Unexpected number of dead connections, want=0, got=%d", len(pool.mu.dead))
 	}
-	pool.Unlock()
+	pool.mu.Unlock()
 }

opensearchtransport/connection_internal_test.go

@@ -259,14 +259,17 @@ func TestStatusConnectionPoolOnFailure(t *testing.T) {
 			t.Fatalf("Unexpected error: %s", err)
 		}
 		conn.mu.Lock()
-		if !conn.mu.isDead {
-			t.Errorf("Expected the connection to be dead; %s", conn)
+		isDead := conn.mu.isDead
+		deadSince := conn.mu.deadSince
+		conn.mu.Unlock()
+
+		if !isDead {
+			t.Errorf("Expected the connection to be dead")
 		}
 
-		if conn.mu.deadSince.IsZero() {
-			t.Errorf("Unexpected value for DeadSince: %s", conn.mu.deadSince)
+		if deadSince.IsZero() {
+			t.Errorf("Unexpected value for DeadSince: %s", deadSince)
 		}
-		conn.mu.Unlock()
 
 		pool.mu.Lock()
 		defer pool.mu.Unlock()
@@ -340,8 +343,8 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {
 		isDead := conn.mu.isDead
 		conn.mu.Unlock()
 
-		if conn.mu.isDead {
-			t.Errorf("Expected connection to be dead, got: %s", conn)
+		if isDead {
+			t.Errorf("Expected connection to be live, got dead=true")
 		}
 
 		if len(pool.mu.dead) != 0 {
@@ -404,7 +407,10 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {
 		}()
 
 		conn := pool.mu.dead[0]
-		pool.scheduleResurrect(conn)
+		conn.mu.RLock()
+		deadSince := conn.mu.deadSince
+		conn.mu.RUnlock()
+		pool.scheduleResurrect(conn, deadSince)
 		time.Sleep(50 * time.Millisecond)
 
 		pool.mu.Lock()
@@ -439,7 +445,7 @@ func TestConnection(t *testing.T) {
 		}
 
 		if !match {
-			t.Errorf("Unexpected output: %s", conn)
+			t.Errorf("Unexpected output: %s", conn.String())
 		}
 	})
 }

opensearchtransport/metrics.go

@@ -101,9 +101,10 @@ func (c *Client) Metrics() (Metrics, error) {
 	}
 	c.metrics.mu.RUnlock()
 
-	if lockable, ok := c.mu.pool.(sync.Locker); ok {
-		lockable.Lock()
-		defer lockable.Unlock()
+	// Acquire read lock on pool since we're only reading connection state
+	if rwLock, ok := c.mu.pool.(rwLocker); ok {
+		rwLock.RLock()
+		defer rwLock.RUnlock()
 	}
 
 	m := Metrics{

opensearchtransport/metrics_internal_test.go

@@ -33,6 +33,7 @@ import (
 	"net/http"
 	"net/url"
 	"regexp"
+	"sync"
 	"testing"
 	"time"
 )
@@ -123,4 +124,29 @@ func TestMetrics(t *testing.T) {
 			t.Errorf("Unexpected output: %s", m)
 		}
 	})
+
+	t.Run("incrementResponse method", func(t *testing.T) {
+		m := &metrics{
+			mu: struct {
+				sync.RWMutex
+				responses map[int]int
+			}{
+				responses: make(map[int]int),
+			},
+		}
+
+		// Test incrementResponse method directly
+		m.incrementResponse(200)
+		m.incrementResponse(404)
+		m.incrementResponse(200) // increment same code again
+
+		m.mu.RLock()
+		if m.mu.responses[200] != 2 {
+			t.Errorf("Expected 2 responses for status 200, got %d", m.mu.responses[200])
+		}
+		if m.mu.responses[404] != 1 {
+			t.Errorf("Expected 1 response for status 404, got %d", m.mu.responses[404])
+		}
+		m.mu.RUnlock()
+	})
 }