Cap RequestErrUnprepared retry recursion on query and batch paths

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Cap `RequestErrUnprepared` retry recursion on `Conn.executeQuery` and `Conn.executeBatch` to prevent process-crashing stack overflow under server-side prepared-statement cache thrash.
+
 ## [2.1.1]
 
 ### Fixed

conn.go

@@ -1546,7 +1546,20 @@ func marshalQueryValue(typ TypeInfo, value interface{}, dst *queryValues) error
 	return nil
 }
 
+// maxUnprepRetries caps the number of times executeQuery and executeBatch
+// will respond to a RequestErrUnprepared by evicting the prepared-statement
+// cache entry and retrying. Without this cap, a server-side cache thrash
+// (Cassandra evicting our prepared statement between every attempt) would
+// recurse unboundedly and stack-overflow the goroutine. Five retries is
+// generous — a single legitimate eviction is the realistic case — while
+// ruling out pathological recursion.
+const maxUnprepRetries = 5
+
 func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
+	return c.executeQueryWithUnprepRetries(ctx, q, 0)
+}
+
+func (c *Conn) executeQueryWithUnprepRetries(ctx context.Context, q *internalQuery, unprepAttempt int) *Iter {
 	qryOpts := q.qryOpts
 	params := queryParams{
 		consistency: q.GetConsistency(),
@@ -1746,9 +1759,18 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
 		// is not consistent with regards to its schema.
 		return iter
 	case *RequestErrUnprepared:
+		if unprepAttempt >= maxUnprepRetries {
+			// Pathological re-prepare loop (server-side cache evicting
+			// our prepared statement between every attempt). Bail with
+			// the underlying server error rather than recursing
+			// indefinitely.
+			iter.err = fmt.Errorf("gocql: failed to execute prepared statement after %d re-prepare attempts: %w",
+				unprepAttempt+1, x)
+			return iter
+		}
 		stmtCacheKey := c.session.stmtsLRU.keyFor(c.host.HostID(), usedKeyspace, qryOpts.stmt)
 		c.session.stmtsLRU.evictPreparedID(stmtCacheKey, x.StatementId)
-		return c.executeQuery(ctx, q)
+		return c.executeQueryWithUnprepRetries(ctx, q, unprepAttempt+1)
 	case error:
 		iter.err = x
 		iter.framer = framer
@@ -1809,6 +1831,10 @@ func (c *Conn) UseKeyspace(keyspace string) error {
 }
 
 func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
+	return c.executeBatchWithUnprepRetries(ctx, b, 0)
+}
+
+func (c *Conn) executeBatchWithUnprepRetries(ctx context.Context, b *internalBatch, unprepAttempt int) *Iter {
 	iter := newIter(b.metrics, b.Keyspace(), b.routingInfo, nil)
 	n := len(b.batchOpts.entries)
 	req := &writeBatchFrame{
@@ -1905,12 +1931,21 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
 	case *resultVoidFrame:
 		return iter
 	case *RequestErrUnprepared:
+		if unprepAttempt >= maxUnprepRetries {
+			// Pathological re-prepare loop on the batch path (server-side
+			// cache evicting our prepared statement between every attempt).
+			// Bail with the underlying server error rather than recursing
+			// indefinitely.
+			iter.err = fmt.Errorf("gocql: failed to execute batch after %d re-prepare attempts: %w",
+				unprepAttempt+1, x)
+			return iter
+		}
 		stmt, found := stmts[string(x.StatementId)]
 		if found {
 			key := c.session.stmtsLRU.keyFor(c.host.HostID(), usedKeyspace, stmt)
 			c.session.stmtsLRU.evictPreparedID(key, x.StatementId)
 		}
-		return c.executeBatch(ctx, b)
+		return c.executeBatchWithUnprepRetries(ctx, b, unprepAttempt+1)
 	case *resultRowsFrame:
 		iter.meta = x.meta
 		iter.framer = framer

conn_test.go

@@ -1338,11 +1338,32 @@ func (srv *TestServer) process(conn net.Conn, reqFrame *framer, useProtoV5, star
 			srv.errorLocked(err)
 			return
 		}
-		name := strings.TrimPrefix(query, "select ")
+		name := query
+		for _, verb := range []string{"select ", "insert ", "update ", "delete "} {
+			if strings.HasPrefix(strings.ToLower(name), verb) {
+				name = name[len(verb):]
+				break
+			}
+		}
 		if n := strings.Index(name, " "); n > 0 {
 			name = name[:n]
 		}
 		switch strings.ToLower(name) {
+		case "always-unprep":
+			// Returns id=99 which is intentionally NOT 1 or 2, so the
+			// opExecute default branch will respond with
+			// ErrCodeUnprepared. This drives the re-prepare retry loop
+			// in Conn.executeQuery — used by the recursion-cap test.
+			respFrame.writeHeader(0, opResult, head.stream)
+			respFrame.writeInt(resultKindPrepared)
+			respFrame.writeShortBytes(binary.BigEndian.AppendUint64(nil, 99))
+			respFrame.writeInt(0)
+			respFrame.writeInt(0)
+			if srv.protocol >= protoVersion4 {
+				respFrame.writeInt(0)
+			}
+			respFrame.writeInt(int32(flagNoMetaData))
+			respFrame.writeInt(0)
 		case "nometadata":
 			respFrame.writeHeader(0, opResult, head.stream)
 			respFrame.writeInt(resultKindPrepared)
@@ -1442,6 +1463,72 @@ func (srv *TestServer) process(conn net.Conn, reqFrame *framer, useProtoV5, star
 			respFrame.writeString("unprepared")
 			respFrame.writeShortBytes(binary.BigEndian.AppendUint64(nil, id))
 		}
+	case opBatch:
+		// Walk the batch frame far enough to extract any prepared statement
+		// IDs. If any equals the always-unprep id (99), respond with
+		// ErrCodeUnprepared to drive the executeBatch re-prepare loop in
+		// the recursion-cap test. Otherwise respond with a void result.
+		if _, err := reqFrame.readByte(); err != nil { // batch type
+			srv.errorLocked(err)
+			return
+		}
+		nStmt, err := reqFrame.readShort()
+		if err != nil {
+			srv.errorLocked(err)
+			return
+		}
+		var unprepID []byte
+		for i := uint16(0); i < nStmt; i++ {
+			kind, err := reqFrame.readByte()
+			if err != nil {
+				srv.errorLocked(err)
+				return
+			}
+			var id []byte
+			if kind == 1 {
+				id, err = reqFrame.readShortBytes()
+				if err != nil {
+					srv.errorLocked(err)
+					return
+				}
+			} else {
+				if _, err := reqFrame.readLongString(); err != nil {
+					srv.errorLocked(err)
+					return
+				}
+			}
+			nVal, err := reqFrame.readShort()
+			if err != nil {
+				srv.errorLocked(err)
+				return
+			}
+			for j := uint16(0); j < nVal; j++ {
+				sz, err := reqFrame.readInt()
+				if err != nil {
+					srv.errorLocked(err)
+					return
+				}
+				if sz > 0 {
+					if len(reqFrame.buf) < sz {
+						srv.errorLocked(fmt.Errorf("opBatch: short read on value"))
+						return
+					}
+					reqFrame.buf = reqFrame.buf[sz:]
+				}
+			}
+			if kind == 1 && len(id) == 8 && binary.BigEndian.Uint64(id) == 99 {
+				unprepID = id
+			}
+		}
+		if unprepID != nil {
+			respFrame.writeHeader(0, opError, head.stream)
+			respFrame.writeInt(ErrCodeUnprepared)
+			respFrame.writeString("unprepared")
+			respFrame.writeShortBytes(unprepID)
+		} else {
+			respFrame.writeHeader(0, opResult, head.stream)
+			respFrame.writeInt(resultKindVoid)
+		}
 	default:
 		respFrame.writeHeader(0, opError, head.stream)
 		respFrame.writeInt(0)

unprep_retry_test.go

@@ -0,0 +1,188 @@
+//go:build all || unit
+// +build all unit
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package gocql
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestExecuteBatch_UnprepRetryIsCapped verifies that Conn.executeBatch
+// stops re-preparing after maxUnprepRetries when the server returns
+// ErrCodeUnprepared on every batch attempt.
+//
+// Mirrors TestExecuteQuery_UnprepRetryIsCapped for the batch path. Without
+// the cap, executeBatch would recurse indefinitely on a server-side cache
+// thrash (Cassandra evicting our prepared statement between attempts) and
+// stack-overflow the goroutine.
+//
+// The fake server's opPrepare handler returns id=99 for "always-unprep".
+// Its opBatch handler returns ErrCodeUnprepared with id=99 whenever any
+// statement in the batch carries that id. Each driver retry: evict cache,
+// re-prepare (server gives 99 again), send batch (server says unprepared)
+// — loops forever absent the cap.
+func TestExecuteBatch_UnprepRetryIsCapped(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var prepCount, batchCount uint64
+	srv := newTestServerOpts{
+		addr:     "127.0.0.1:0",
+		protocol: defaultProto,
+		recvHook: func(f *framer) {
+			switch f.header.op {
+			case opPrepare:
+				atomic.AddUint64(&prepCount, 1)
+			case opBatch:
+				atomic.AddUint64(&batchCount, 1)
+			}
+		},
+	}.newServer(t, ctx)
+	defer srv.Stop()
+
+	cluster := testCluster(defaultProto, srv.Address)
+	cluster.Timeout = 5 * time.Second
+	db, err := cluster.CreateSession()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	b := db.NewBatch(LoggedBatch)
+	// Use Bind with an empty values slice so the batch entry goes through
+	// prepareStatement (binding != nil) but the post-prepare arity check
+	// matches the fake server's always-unprep prepared response, which
+	// declares 0 request columns.
+	b.Bind("insert always-unprep into x (k) values (?)", func(*QueryInfo) ([]interface{}, error) {
+		return nil, nil
+	})
+	err = db.ExecuteBatch(b)
+	if err == nil {
+		t.Fatalf("expected re-prepare cap error, got nil")
+	}
+
+	if !strings.Contains(err.Error(), "re-prepare attempts") {
+		t.Errorf("error %q does not mention re-prepare attempts; cap behavior may be missing", err)
+	}
+
+	var serverErr *RequestErrUnprepared
+	if !errors.As(err, &serverErr) {
+		t.Errorf("errors.As(err, *RequestErrUnprepared) = false; %%w not in effect")
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	wantPairs := uint64(maxUnprepRetries + 1)
+	gotPrep := atomic.LoadUint64(&prepCount)
+	gotBatch := atomic.LoadUint64(&batchCount)
+	if gotPrep != wantPairs {
+		t.Errorf("prepare count = %d, want %d (cap=%d allows %d retries plus initial)",
+			gotPrep, wantPairs, maxUnprepRetries, maxUnprepRetries)
+	}
+	if gotBatch != wantPairs {
+		t.Errorf("batch count = %d, want %d", gotBatch, wantPairs)
+	}
+}
+
+// TestExecuteQuery_UnprepRetryIsCapped verifies that Conn.executeQuery
+// stops re-preparing after maxUnprepRetries when the server returns
+// ErrCodeUnprepared on every Execute attempt.
+//
+// Without the cap, a server-side prepared-statement cache thrash
+// (evicting the driver's statement between every attempt) would cause
+// unbounded recursion in executeQuery and stack-overflow the goroutine.
+// This test drives that exact scenario via the fake test server.
+//
+// The server's opPrepare handler for "always-unprep" returns id=99 each
+// time. The opExecute default branch returns ErrCodeUnprepared for any
+// id != 1 and != 2. Each driver retry: evict, re-prepare (server gives
+// 99 again), execute (server says unprepared) — loops forever absent
+// the cap.
+func TestExecuteQuery_UnprepRetryIsCapped(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var prepCount, execCount uint64
+	srv := newTestServerOpts{
+		addr:     "127.0.0.1:0",
+		protocol: defaultProto,
+		recvHook: func(f *framer) {
+			switch f.header.op {
+			case opPrepare:
+				atomic.AddUint64(&prepCount, 1)
+			case opExecute:
+				atomic.AddUint64(&execCount, 1)
+			}
+		},
+	}.newServer(t, ctx)
+	defer srv.Stop()
+
+	cluster := testCluster(defaultProto, srv.Address)
+	cluster.Timeout = 5 * time.Second
+	db, err := cluster.CreateSession()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	// "select always-unprep ..." routes through the fake server's
+	// opPrepare always-unprep case. shouldPrepare requires a SELECT/INSERT
+	// keyword prefix so we name the query accordingly.
+	err = db.Query("select always-unprep from x").Exec()
+	if err == nil {
+		t.Fatalf("expected re-prepare cap error, got nil")
+	}
+
+	// The wrap message must mention the attempt count so operators can
+	// diagnose the server-side cache thrash.
+	if !strings.Contains(err.Error(), "re-prepare attempts") {
+		t.Errorf("error %q does not mention re-prepare attempts; cap behavior may be missing", err)
+	}
+
+	// errors.As must recover the underlying server error.
+	var serverErr *RequestErrUnprepared
+	if !errors.As(err, &serverErr) {
+		t.Errorf("errors.As(err, *RequestErrUnprepared) = false; %%w not in effect")
+	}
+
+	// Wait for any in-flight goroutines to settle. The exec/prep counters
+	// can lag the test if the server logged the request just before we
+	// read the counter.
+	time.Sleep(50 * time.Millisecond)
+
+	// We should have seen exactly maxUnprepRetries+1 prepare-execute
+	// pairs: the initial attempt, plus N retries.
+	wantPairs := uint64(maxUnprepRetries + 1)
+	gotPrep := atomic.LoadUint64(&prepCount)
+	gotExec := atomic.LoadUint64(&execCount)
+	if gotPrep != wantPairs {
+		t.Errorf("prepare count = %d, want %d (cap=%d allows %d retries plus initial)",
+			gotPrep, wantPairs, maxUnprepRetries, maxUnprepRetries)
+	}
+	if gotExec != wantPairs {
+		t.Errorf("execute count = %d, want %d", gotExec, wantPairs)
+	}
+}