Fix repeated "Pool connection error" and reconnections with small Session.Timeout

In GoCQL v1.x, the read deadline is set to 0 when reading a new response frame to prevent the connection from closing when it idles.

In v2.0.0 a regression was introduced causing the read deadline to be set to clusterCfg.Timeout for every read operation which leads to connections constantly erroring and reconnecting if they idle.

This patch fixes this regression by implementing the behavior of 1.x. A follow up ticket (CASSGO-127) will rework the timeout configuration so users can tune the read and write timeouts independently from the request timeout.

Patch by João Reis; reviewed by Bohdan Siryk for CASSGO-125

Author: joao-r-reis

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Prevent panic when using a HostFilter and keyspace is not replicated to every DC (CASSGO-122)
 - system.peers fallback doesn't work in some scenarios (CASSGO-126)
+- Many "Pool connection error" with small Session.Timeout (CASSGO-125)
 
 ## [2.1.1]
 

conn.go

@@ -171,6 +171,7 @@ type Conn struct {
 	w contextWriter
 
 	writeTimeout   time.Duration
+	requestTimeout time.Duration // Request timeout, used for setting up request timers
 	cfg            *ConnConfig
 	frameObserver  FrameHeaderObserver
 	streamObserver StreamObserver
@@ -279,6 +280,7 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		logger:         logger,
 		streamObserver: s.streamObserver,
 		writeTimeout:   writeTimeout,
+		requestTimeout: cfg.ConnectTimeout,
 	}
 
 	if err := c.init(ctx, dialedHost); err != nil {
@@ -312,6 +314,7 @@ func (c *Conn) init(ctx context.Context, dialedHost *DialedHost) error {
 	}
 
 	c.r.SetTimeout(c.cfg.Timeout)
+	c.requestTimeout = c.cfg.Timeout
 
 	// dont coalesce startup frames
 	if c.session.cfg.WriteCoalesceWaitTime > 0 && !c.cfg.disableCoalesce && !dialedHost.DisableCoalesce {
@@ -335,8 +338,8 @@ type startupCoordinator struct {
 
 func (s *startupCoordinator) setupConn(ctx context.Context) error {
 	var cancel context.CancelFunc
-	if s.conn.r.GetTimeout() > 0 {
-		ctx, cancel = context.WithTimeout(ctx, s.conn.r.GetTimeout())
+	if s.conn.requestTimeout > 0 {
+		ctx, cancel = context.WithTimeout(ctx, s.conn.requestTimeout)
 	} else {
 		ctx, cancel = context.WithCancel(ctx)
 	}
@@ -688,8 +691,9 @@ func (c *Conn) processFrame(ctx context.Context, r io.Reader) error {
 
 	// read a full header, ignore timeouts, as this is being ran in a loop
 	// TODO: TCP level deadlines? or just query level deadlines?
-	if c.r.GetTimeout() > 0 {
-		c.r.SetReadDeadline(time.Time{})
+	readTimeout := c.r.GetTimeout()
+	if readTimeout > 0 {
+		c.r.SetTimeout(0)
 	}
 
 	headStartTime := time.Now()
@@ -700,6 +704,11 @@ func (c *Conn) processFrame(ctx context.Context, r io.Reader) error {
 		return err
 	}
 
+	// Set timeout back for reading frame body
+	if readTimeout > 0 {
+		c.r.SetTimeout(readTimeout)
+	}
+
 	if c.frameObserver != nil {
 		c.frameObserver.ObserveFrameHeader(context.Background(), ObservedFrameHeader{
 			Version: protoVersion(head.version),
@@ -802,12 +811,24 @@ func (c *Conn) recvSegment(ctx context.Context) error {
 		err             error
 	)
 
+	// Read segment without timeout, as this is being run in a loop waiting for the next segment
+	readTimeout := c.r.GetTimeout()
+	if readTimeout > 0 {
+		c.r.SetTimeout(0)
+	}
+
 	// Read frame based on compression
 	if c.compressor != nil {
 		frame, isSelfContained, err = readCompressedSegment(c.r, c.compressor)
 	} else {
 		frame, isSelfContained, err = readUncompressedSegment(c.r)
 	}
+
+	// Restore timeout for subsequent segment reads in multi-segment frames
+	if readTimeout > 0 {
+		c.r.SetTimeout(readTimeout)
+	}
+
 	if err != nil {
 		return err
 	}
@@ -908,6 +929,8 @@ func (c *connReader) Read(p []byte) (n int, err error) {
 		var nn int
 		if c.timeout > 0 {
 			c.conn.SetReadDeadline(time.Now().Add(c.timeout))
+		} else if c.timeout == 0 {
+			c.conn.SetReadDeadline(time.Time{})
 		}
 
 		nn, err = io.ReadFull(c.r, p[n:])
@@ -1309,7 +1332,7 @@ func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer Tracer
 	}
 
 	var timeoutCh <-chan time.Time
-	if timeout := c.r.GetTimeout(); timeout > 0 {
+	if c.requestTimeout > 0 {
 		if call.timer == nil {
 			call.timer = time.NewTimer(0)
 			<-call.timer.C
@@ -1322,7 +1345,7 @@ func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer Tracer
 			}
 		}
 
-		call.timer.Reset(timeout)
+		call.timer.Reset(c.requestTimeout)
 		timeoutCh = call.timer.C
 	}
 

conn_test.go

@@ -287,7 +287,9 @@ func TestTimeout(t *testing.T) {
 	srv := NewTestServer(t, defaultProto, ctx)
 	defer srv.Stop()
 
-	db, err := newTestSession(defaultProto, srv.Address)
+	cluster := testCluster(defaultProto, srv.Address)
+	cluster.Timeout = 2 * time.Second
+	db, err := cluster.CreateSession()
 	if err != nil {
 		t.Fatalf("NewCluster: %v", err)
 	}
@@ -306,12 +308,18 @@ func TestTimeout(t *testing.T) {
 		}
 	}()
 
-	if err := db.Query("kill").WithContext(ctx).Exec(); err == nil {
+	now := time.Now()
+	err = db.Query("timeout").ExecContext(ctx)
+	if err == nil {
 		t.Fatal("expected error got nil")
 	}
 	cancel()
-
 	wg.Wait()
+
+	elapsed := time.Since(now)
+	if elapsed < 1*time.Second || elapsed > 4*time.Second {
+		t.Fatalf("timeout is not respected (took %v)", elapsed.String())
+	}
 }
 
 func TestCancel(t *testing.T) {
@@ -329,13 +337,11 @@ func TestCancel(t *testing.T) {
 	}
 	defer db.Close()
 
-	qry := db.Query("timeout").WithContext(ctx)
-
 	// Make sure we finish the query without leftovers
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
-		err = qry.Exec()
+		err = db.Query("timeout").ExecContext(ctx)
 		wg.Done()
 	}()
 
@@ -720,9 +726,14 @@ func TestStream0(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	clientConn, serverConn := net.Pipe()
+	defer clientConn.Close()
+	defer serverConn.Close()
+
 	conn := &Conn{
 		r: &connReader{
-			r: bufio.NewReader(&buf),
+			r:    bufio.NewReader(&buf),
+			conn: clientConn,
 		},
 		streams: streams.New(protoVersion4),
 		session: &Session{

control.go

@@ -361,7 +361,8 @@ func (c *controlConn) setupConn(conn *Conn, sessionInit bool) error {
 	c.conn.Store(ch)
 
 	c.session.logger.Info("Control connection connected to host.",
-		NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()))
+		NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()),
+		NewLogFieldInt("protocol_version", c.session.cfg.ProtoVersion))
 
 	if c.session.initialized() {
 		refreshErr := c.session.schemaDescriber.refreshSchemaMetadata()

integration_test.go

@@ -977,3 +977,54 @@ func TestSliceMapMapScanCollectionTypes(t *testing.T) {
 		})
 	}
 }
+
+// TestSmallTimeoutNoPoolErrors verifies that small Session.Timeout values
+// don't cause connections to timeout and reconnect constantly. This is a
+// regression test for https://github.com/apache/cassandra-gocql-driver/issues/1919
+//
+// The issue was that the timeout was being applied to frame header reads,
+// causing connections to timeout while waiting for the next frame. The fix
+// ensures frame headers are read without timeout, while frame bodies are
+// read with timeout.
+func TestSmallTimeoutNoPoolErrors(t *testing.T) {
+	// Create a test logger to capture log messages
+	logger := newTestLogger(LogLevelDebug)
+	defer func() {
+		t.Log(logger.String())
+	}()
+
+	cluster := createCluster()
+	cluster.ConnectTimeout = 10 * time.Second
+	cluster.Timeout = 750 * time.Millisecond
+	cluster.NumConns = 1
+	cluster.Logger = logger
+
+	db, err := cluster.CreateSession()
+	if err != nil {
+		t.Fatalf("CreateSession: %v", err)
+	}
+	defer db.Close()
+
+	// Wait for connections to sit idle
+	// If the bug exists, connections will timeout while waiting for frame headers
+	// and "Pool connection error" messages will be logged repeatedly
+	time.Sleep(5 * time.Second)
+
+	// Get log output for analysis
+	logOutput := strings.ToLower(logger.String())
+
+	// Count successful connection messages - should be exactly NumConns * number of nodes
+	connectedCount := strings.Count(logOutput, "pool connected to node")
+	if connectedCount != *clusterSize*cluster.NumConns {
+		t.Fatalf("Expected exactly %d 'Pool connected to node' messages, got %d:\n%s",
+			*clusterSize*cluster.NumConns, connectedCount, logOutput)
+	}
+
+	// Count error messages - should be zero
+	// With the bug, we'd see many errors as connections constantly timeout and reconnect
+	errorCount := strings.Count(logOutput, "pool connection error")
+	if errorCount > 0 {
+		t.Fatalf("Found %d 'Pool connection error' messages - connections are timing out and reconnecting:\n%s",
+			errorCount, logOutput)
+	}
+}