fix race in tests

Author: ndyakov

auth/conn_reauth_credentials_listener.go

@@ -37,12 +37,7 @@ func (c *ConnReAuthCredentialsListener) OnNext(credentials Credentials) {
 	// this is important because the connection pool may be in the process of reconnecting the connection
 	// and we don't want to interfere with that process
 	// but we also don't want to block for too long, so incorporate a timeout
-	for {
-		// we were able to mark the connection as unusable
-		if c.conn.Usable.CompareAndSwap(true, false) {
-			break
-		}
-
+	for !c.conn.Usable.CompareAndSwap(true, false) {
 		select {
 		case <-timeout:
 			err = pool.ErrConnUnusableTimeout

internal/pool/buffer_size_test.go

@@ -3,7 +3,6 @@ package pool_test
 import (
 	"bufio"
 	"context"
-	"net"
 	"unsafe"
 
 	. "github.com/bsm/ginkgo/v2"
@@ -124,20 +123,27 @@ var _ = Describe("Buffer Size Configuration", func() {
 })
 
 // Helper functions to extract buffer sizes using unsafe pointers
+// The struct layout must match pool.Conn exactly to avoid checkptr violations
 func getWriterBufSizeUnsafe(cn *pool.Conn) int {
+	// Import required for atomic types
+	type atomicBool struct{ _ uint32 }
+	type atomicInt64 struct{ _ int64 }
+
 	cnPtr := (*struct {
-		usedAt  int64
-		netConn net.Conn
-		rd      *proto.Reader
-		bw      *bufio.Writer
-		wr      *proto.Writer
-		// ... other fields
+		id            uint64        // First field in pool.Conn
+		usedAt        int64         // Second field (atomic)
+		netConnAtomic interface{}   // atomic.Value (interface{} has same size)
+		rd            *proto.Reader
+		bw            *bufio.Writer
+		wr            *proto.Writer
+		// We only need fields up to bw, so we can stop here
 	})(unsafe.Pointer(cn))
 
 	if cnPtr.bw == nil {
 		return -1
 	}
 
+	// bufio.Writer internal structure
 	bwPtr := (*struct {
 		err error
 		buf []byte
@@ -150,18 +156,20 @@ func getWriterBufSizeUnsafe(cn *pool.Conn) int {
 
 func getReaderBufSizeUnsafe(cn *pool.Conn) int {
 	cnPtr := (*struct {
-		usedAt  int64
-		netConn net.Conn
-		rd      *proto.Reader
-		bw      *bufio.Writer
-		wr      *proto.Writer
-		// ... other fields
+		id            uint64        // First field in pool.Conn
+		usedAt        int64         // Second field (atomic)
+		netConnAtomic interface{}   // atomic.Value (interface{} has same size)
+		rd            *proto.Reader
+		bw            *bufio.Writer
+		wr            *proto.Writer
+		// We only need fields up to rd, so we can stop here
 	})(unsafe.Pointer(cn))
 
 	if cnPtr.rd == nil {
 		return -1
 	}
 
+	// proto.Reader internal structure
 	rdPtr := (*struct {
 		rd *bufio.Reader
 	})(unsafe.Pointer(cnPtr.rd))
@@ -170,6 +178,7 @@ func getReaderBufSizeUnsafe(cn *pool.Conn) int {
 		return -1
 	}
 
+	// bufio.Reader internal structure
 	bufReaderPtr := (*struct {
 		buf          []byte
 		rd           interface{}

redis.go

@@ -308,7 +308,7 @@ func (c *baseClient) connReAuthCredentialsListener(poolCn *pool.Conn) (auth.Cred
 	credListener, ok := c.credListeners[poolCn]
 	c.credListenersLock.RUnlock()
 	if ok {
-		return credListener.(auth.CredentialsListener), func() {
+		return credListener, func() {
 			c.removeCredListener(poolCn)
 		}
 	}

redis_test.go

@@ -872,6 +872,7 @@ var _ = Describe("Credentials Provider Priority", func() {
 })
 
 type mockStreamingProvider struct {
+	mu          sync.RWMutex
 	credentials auth.Credentials
 	err         error
 	updates     chan auth.Credentials
@@ -885,12 +886,18 @@ func (m *mockStreamingProvider) Subscribe(listener auth.CredentialsListener) (au
 	// Start goroutine to handle updates
 	go func() {
 		for creds := range m.updates {
+			m.mu.Lock()
 			m.credentials = creds
+			m.mu.Unlock()
 			listener.OnNext(creds)
 		}
 	}()
 
-	return m.credentials, func() (err error) {
+	m.mu.RLock()
+	currentCreds := m.credentials
+	m.mu.RUnlock()
+
+	return currentCreds, func() (err error) {
 		defer func() {
 			if r := recover(); r != nil {
 				// this is just a mock: