From 1d168746fcabffe77519e04ec5aa5967281eda00 Mon Sep 17 00:00:00 2001 From: Arlo Liu Date: Sun, 17 May 2026 15:48:35 +0800 Subject: [PATCH] Cap RequestErrUnprepared retry recursion on query and batch paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Conn.executeQuery and Conn.executeBatch both responded to a *RequestErrUnprepared by evicting the prepared-statement cache entry and recursing on themselves with no upper bound. When the server persistently re-reports the same statement as unprepared after re-prepare (a coordinator thrashing its prepared-statement cache, or a misbehaving proxy/fork), the recursion never terminates. The goroutine stack eventually exceeds runtime.SetMaxStack (1 GiB by default), at which point Go runtime.throw crashes the entire process with an unrecoverable stack-overflow — no recover() can intercept it. The failure mode is reachable from any prepared-statement workload; the batch path is on the hot write path. Cap recursion on both paths at maxUnprepRetries = 5 by threading an unprepAttempt counter through unexported helpers executeQueryWithUnprepRetries and executeBatchWithUnprepRetries. When the cap fires, return an Iter whose err is fmt.Errorf("...after N re-prepare attempts: %w", serverErr) The %w wrap means callers can: - Detect cap-driven failures via error message pattern. - Recover the underlying *RequestErrUnprepared with errors.As to inspect the StatementId the server kept rejecting. Behavior is a strict superset of the prior code: queries and batches that succeed within 5 attempts behave exactly as before. Only the pathological no-progress case is changed. Test fake-server (conn_test.go) gains an always-unprep opPrepare case returning id=99, an opBatch arm that replies ErrCodeUnprepared when any statement carries id=99, and case-insensitive verb-trimming in the opPrepare query-name parser (select/insert/update/delete) so DML in batches can reach the always-unprep case. Two new tests in unprep_retry_test.go drive the always-unprep path through Query.Exec and Session.ExecuteBatch respectively, assert no infinite recursion, verify the wrap is recoverable via errors.As, and check that the server received exactly maxUnprepRetries+1 prepare/execute (resp. prepare/batch) pairs. Patch by Arlo Liu --- CHANGELOG.md | 6 ++ conn.go | 39 ++++++++- conn_test.go | 89 +++++++++++++++++++- unprep_retry_test.go | 188 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 319 insertions(+), 3 deletions(-) create mode 100644 unprep_retry_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b07b9bfe..2b3f8bb09 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/conn.go b/conn.go index a615129da..2ccb8c8aa 100644 --- a/conn.go +++ b/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 diff --git a/conn_test.go b/conn_test.go index ad4e66e54..1770e6857 100644 --- a/conn_test.go +++ b/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) diff --git a/unprep_retry_test.go b/unprep_retry_test.go new file mode 100644 index 000000000..510f55497 --- /dev/null +++ b/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) + } +}