From e78b47de52873b27845ac8c6b96228f3706a5b12 Mon Sep 17 00:00:00 2001 From: Bohdan Siryk Date: Thu, 7 May 2026 13:09:59 +0300 Subject: [PATCH 1/9] Fix panic when using a HostFilter and keyspace is not replicated to every DC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, networkTopology.replicaMap didn't take into account the fact the driver might filter hosts of specific dcs by using HostFilter when computing amount of dcs with replicas for a keyspace. It wasn't problematic before as the tokenAwareHostPolicy was intended to work with session-level keyspace. However, it became problematic since v2.1.0 release which enabled the policy to work with all keyspaces in the cluster. This patch makes networkTopology.replicaMap dc-aware. Patch by Bohdan Siryk; reviewed by João Reis for CASSGO-122 --- CHANGELOG.md | 6 ++++++ policies.go | 4 ++-- topology.go | 35 ++++++++++++++++++++++++++--------- topology_test.go | 36 ++++++++++++++++++++++++++++++++++-- 4 files changed, 68 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b07b9bfe..cba0d801a 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). +## [2.1.2] + +### Fixed + +- Prevent panic when using a HostFilter and keyspace is not replicated to every DC (CASSGO-122) + ## [2.1.1] ### Fixed diff --git a/policies.go b/policies.go index 41f407803..3d880e977 100644 --- a/policies.go +++ b/policies.go @@ -524,7 +524,7 @@ func (t *tokenAwareHostPolicy) updateReplicas(keyspace string) { if _, ok := meta.replicas[key]; !ok { metaUpdate := t.getMetadataForUpdate() newReplicas := make(map[string]tokenRingReplicas, len(meta.replicas)) - newReplicas[key] = strat.replicaMap(metaUpdate.tokenRing) + newReplicas[key] = strat.replicaMap(metaUpdate.tokenRing, t.logger) for k, replicas := range metaUpdate.replicas { newReplicas[k] = replicas } @@ -556,7 +556,7 @@ func (t *tokenAwareHostPolicy) updateAllReplicas(meta *clusterMeta, schemaMeta * if meta != nil && meta.tokenRing != nil { key := strat.strategyKey() if _, ok := newReplicas[key]; !ok { - newReplicas[key] = strat.replicaMap(meta.tokenRing) + newReplicas[key] = strat.replicaMap(meta.tokenRing, t.logger) } } } diff --git a/topology.go b/topology.go index 3ec20b15e..0481b2ff8 100644 --- a/topology.go +++ b/topology.go @@ -66,7 +66,7 @@ func (h tokenRingReplicas) replicasFor(t token) *hostTokens { } type placementStrategy interface { - replicaMap(tokenRing *tokenRing) tokenRingReplicas + replicaMap(tokenRing *tokenRing, logger StructuredLogger) tokenRingReplicas replicationFactor(dc string) int // strategyKey returns a unique identifier string for this strategy instance. // Two strategy instances with identical configuration should return the same key. @@ -162,7 +162,7 @@ func (s *simpleStrategy) replicationFactor(dc string) int { return s.rf } -func (s *simpleStrategy) replicaMap(tokenRing *tokenRing) tokenRingReplicas { +func (s *simpleStrategy) replicaMap(tokenRing *tokenRing, _ StructuredLogger) tokenRingReplicas { tokens := tokenRing.tokens ring := make(tokenRingReplicas, len(tokens)) @@ -255,7 +255,7 @@ func (n *networkTopology) haveRF(replicaCounts map[string]int) bool { return true } -func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { +func (n *networkTopology) replicaMap(tokenRing *tokenRing, logger StructuredLogger) tokenRingReplicas { dcRacks := make(map[string]map[string]struct{}, len(n.dcs)) // skipped hosts in a dc skipped := make(map[string][]*HostInfo, len(n.dcs)) @@ -324,7 +324,11 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { continue } else if replicasInDC[dc] >= rf { if replicasInDC[dc] > rf { - panic(fmt.Sprintf("replica overflow. rf=%d have=%d in dc %q", rf, replicasInDC[dc], dc)) + logger.Warning("Replica overflow. Returning empty map.", + NewLogFieldInt("rf", rf), + NewLogFieldInt("have", replicasInDC[dc]), + NewLogFieldString("dc", dc)) + return tokenRingReplicas{} } // have enough replicas in this DC @@ -372,23 +376,36 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { } if len(replicas) == 0 { - panic(fmt.Sprintf("no replicas for token: %v", th.token)) + logger.Warning("No replicas for token. Returning empty map.", + NewLogFieldString("token", th.token.String())) + return tokenRingReplicas{} } else if !replicas[0].Equal(th.host) { - panic(fmt.Sprintf("first replica is not the primary replica for the token: expected %v got %v", replicas[0].ConnectAddress(), th.host.ConnectAddress())) + logger.Warning("First replica is not the primary replica for the token. Returning empty map.", + NewLogFieldString("token", th.token.String()), + NewLogFieldIP("expected", replicas[0].ConnectAddress()), + NewLogFieldIP("got", th.host.ConnectAddress())) + return tokenRingReplicas{} } replicaRing = append(replicaRing, hostTokens{th.token, replicas}) } dcsWithReplicas := 0 - for _, dc := range n.dcs { - if dc > 0 { + for dc, rf := range n.dcs { + // We should count only DCs that driver is aware of and have a replication factor > 0 + if _, knownDc := dcRacks[dc]; knownDc && rf > 0 { dcsWithReplicas++ } } if dcsWithReplicas == len(dcRacks) && len(replicaRing) != len(tokens) { - panic(fmt.Sprintf("token map different size to token ring: got %d expected %d", len(replicaRing), len(tokens))) + logger.Warning("Unexpected state while building replica map. Returning empty map.", + NewLogFieldString("strategy_key", n.strategyKey()), + NewLogFieldInt("dcs_with_replicas", dcsWithReplicas), + NewLogFieldInt("dcs_in_ring", len(dcRacks)), + NewLogFieldInt("token_ring_size", len(tokens)), + NewLogFieldInt("replica_ring_size", len(replicaRing))) + return tokenRingReplicas{} } return replicaRing diff --git a/topology_test.go b/topology_test.go index 10ee85e53..3e8fb9a2a 100644 --- a/topology_test.go +++ b/topology_test.go @@ -31,6 +31,8 @@ import ( "fmt" "sort" "testing" + + "github.com/stretchr/testify/require" ) func TestPlacementStrategy_SimpleStrategy(t *testing.T) { @@ -49,7 +51,7 @@ func TestPlacementStrategy_SimpleStrategy(t *testing.T) { hosts := []*HostInfo{host0, host25, host50, host75} strat := newSimpleStrategy(2) - tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}) + tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) if len(tokenReplicas) != len(tokens) { t.Fatalf("expected replica map to have %d items but has %d", len(tokens), len(tokenReplicas)) } @@ -157,7 +159,7 @@ func TestPlacementStrategy_NetworkStrategy(t *testing.T) { expReplicas += rf } - tokenReplicas := test.strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}) + tokenReplicas := test.strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) if len(tokenReplicas) != test.expectedReplicaMapSize { t.Fatalf("expected replica map to have %d items but has %d", test.expectedReplicaMapSize, len(tokenReplicas)) @@ -224,3 +226,33 @@ func TestPlacementStrategy_NetworkStrategy(t *testing.T) { }) } } + +// Regression test for CASSGO-122: +// when the token ring only contains hosts from a DC that has RF=0/unspecified for a keyspace, +// networkTopology.replicaMap should return an empty replica map. +func TestPlacementStrategy_NetworkStrategy_ReturnEmptyReplicaMapWhenNoReplicasInRing(t *testing.T) { + strat := newNetworkTopology(map[string]int{ + "dc1": 3, // replicated only in dc1 + }) + + // Hosts in ring only from dc2, so no replicas should be returned. + // hostId format: dc:rack:host which is used as a token in the token ring. + // It makes sense to use the hostId as a token in the token ring because it is unique and deterministic for test purpose. + hosts := []*HostInfo{ + {hostId: "dc2:rack1:0", dataCenter: "dc2", rack: "rack1"}, + {hostId: "dc2:rack2:1", dataCenter: "dc2", rack: "rack2"}, + {hostId: "dc2:rack3:2", dataCenter: "dc2", rack: "rack3"}, + } + + tokens := make([]hostToken, 0, len(hosts)) + for _, h := range hosts { + tokens = append(tokens, hostToken{ + token: orderedToken(h.hostId), + host: h, + }) + } + sort.Sort(&tokenRing{tokens: tokens}) + + replicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) + require.Empty(t, replicas, "expected no replicas, got %d", len(replicas)) +} From e1d69bd6c1503ec8b20f542ece1da5f65deefc50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Reis?= Date: Fri, 29 May 2026 18:20:29 +0100 Subject: [PATCH 2/9] Fix system.peers_v2 fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fallback to system.peers if system.peers_v2 doesn't exist doesn't work. This patch addresses this. Patch by João Reis; reviewed by Bohdan Siryk for CASSGO-126 --- CHANGELOG.md | 1 + conn.go | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cba0d801a..21bfbd2ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - 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) ## [2.1.1] diff --git a/conn.go b/conn.go index a615129da..9d36cbc39 100644 --- a/conn.go +++ b/conn.go @@ -1952,7 +1952,8 @@ func (c *Conn) querySystemPeers(ctx context.Context, version cassVersion) *Iter err := iter.checkErrAndNotFound() if err != nil { - if errFrame, ok := err.(errorFrame); ok && errFrame.code == ErrCodeInvalid { // system.peers_v2 not found, try system.peers + var requestErr RequestError + if errors.As(err, &requestErr) && requestErr.Code() == ErrCodeInvalid { // system.peers_v2 not found, try system.peers c.mu.Lock() c.isSchemaV2 = false c.mu.Unlock() From 590aabedfaa33398e308e253e0bda020a8dbcafe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Reis?= Date: Tue, 2 Jun 2026 19:06:44 +0100 Subject: [PATCH 3/9] Fix repeated "Pool connection error" and reconnections with small Session.Timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CHANGELOG.md | 1 + conn.go | 35 +++++++++++++++++++++++++------ conn_test.go | 25 +++++++++++++++------- control.go | 3 ++- integration_test.go | 51 +++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 101 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 21bfbd2ec..cb07d57d4 100644 --- a/CHANGELOG.md +++ b/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] diff --git a/conn.go b/conn.go index 9d36cbc39..28b015773 100644 --- a/conn.go +++ b/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 } diff --git a/conn_test.go b/conn_test.go index ad4e66e54..abc659b5d 100644 --- a/conn_test.go +++ b/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{ diff --git a/control.go b/control.go index cc21e089e..9f56958a3 100644 --- a/control.go +++ b/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() diff --git a/integration_test.go b/integration_test.go index bd6ccb5cc..8bd2e94c4 100644 --- a/integration_test.go +++ b/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) + } +} From 941f298890a66061ee9afe354eb973a6e6fc95c1 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sun, 14 Jun 2026 02:29:45 +0200 Subject: [PATCH 4/9] Add security-model discoverability (AGENTS.md -> SECURITY.md -> security model) Wires the conventional AGENTS.md -> SECURITY.md -> security model chain so automated tooling can mechanically discover the project's security model. No model content is changed. patch by Jarek Potiuk; reviewed by Stefan Miklosovic, Bret McGuire for CASSANDRA-21464 Assisted-by: Claude Opus 4.8 (1M context) --- AGENTS.md | 16 ++++++++++++++++ CHANGELOG.md | 6 ++++++ SECURITY.md | 17 +++++++++++++++++ 3 files changed, 39 insertions(+) create mode 100644 AGENTS.md create mode 100644 SECURITY.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 000000000..78d4c9cfc --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,16 @@ +# Agent guidance + +This file is read by automated agents (security scanners, code analyzers, +AI assistants) operating on this repository. It points them at the +human-authored references they should consult before producing output. + +## Security + +Security model: [SECURITY.md](./SECURITY.md), which links to the Apache +Cassandra project security model. + +This repository is part of the Apache Cassandra project. Its security model - +trust boundaries, in-scope / out-of-scope declarations, the security +properties the project provides and disclaims, and how findings are triaged - +is the umbrella Cassandra security model linked from SECURITY.md. Consult it +before reporting security issues. diff --git a/CHANGELOG.md b/CHANGELOG.md index cb07d57d4..580094c64 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). +## [2.2.0] + +### Added + +- Security-model discoverability (CASSANDRA-21464) + ## [2.1.2] ### Fixed diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 000000000..7ecb4e420 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,17 @@ +# Security Policy + +## Reporting a Vulnerability + +Please report suspected security vulnerabilities privately to the Apache +Security Team at , following the ASF process at +. Do not open public GitHub issues or pull +requests for security reports. + +## Security Model + +This repository is part of the Apache Cassandra project. The project's +security model - what is in and out of scope, the trust boundaries it +assumes, the security properties it provides and disclaims, and how findings +are triaged - is documented in the main apache/cassandra repository: + +https://github.com/apache/cassandra/blob/trunk/doc/modules/cassandra/pages/reference/security-model.adoc From 40db71c980ef56bed79636c7db03520b7dee5008 Mon Sep 17 00:00:00 2001 From: Dorian Jaminais-Grellier Date: Thu, 18 Jun 2026 15:55:21 +0200 Subject: [PATCH 5/9] Improve host_source locking and ring refresh concurrency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove ringDescriber mutex around GetHosts I/O; prevHosts/prevPartitioner were never written. Use read locks for ConnectAddressAndPort and a read-fast path for HostnameAndPort. Read peer validity under a single RLock in isValidPeer. Release errorBroadcaster mutex before sending to listeners. remove useless lock in isValidPeer Patch by Dorian Jaminais; reviewed by João Reis and Bohdan Siryk for CASSGO-121 --- CHANGELOG.md | 1 + host_source.go | 34 ++++++++++++++++++---------------- 2 files changed, 19 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 580094c64..3f132b5b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Security-model discoverability (CASSANDRA-21464) +- Improve host_source locking and ring refresh concurrency (CASSGO-121) ## [2.1.2] diff --git a/host_source.go b/host_source.go index e434b652c..a0e860f21 100644 --- a/host_source.go +++ b/host_source.go @@ -453,6 +453,14 @@ func (h *HostInfo) IsUp() bool { } func (h *HostInfo) HostnameAndPort() string { + h.mu.RLock() + if h.hostname != "" { + s := net.JoinHostPort(h.hostname, strconv.Itoa(h.port)) + h.mu.RUnlock() + return s + } + h.mu.RUnlock() + h.mu.Lock() defer h.mu.Unlock() if h.hostname == "" { @@ -463,8 +471,8 @@ func (h *HostInfo) HostnameAndPort() string { } func (h *HostInfo) ConnectAddressAndPort() string { - h.mu.Lock() - defer h.mu.Unlock() + h.mu.RLock() + defer h.mu.RUnlock() addr, _ := h.connectAddressLocked() return net.JoinHostPort(addr.String(), strconv.Itoa(h.port)) } @@ -484,10 +492,7 @@ func (h *HostInfo) String() string { // Polls system.peers at a specific interval to find new hosts type ringDescriber struct { - session *Session - mu sync.Mutex - prevHosts []*HostInfo - prevPartitioner string + session *Session } // Returns true if we are using system_schema.keyspaces instead of system.schema_keyspaces @@ -806,7 +811,7 @@ func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, er // Return true if the host is a valid peer func isValidPeer(host *HostInfo) bool { - return !(len(host.RPCAddress()) == 0 || + return !(len(host.rpcAddress) == 0 || host.hostId == "" || host.dataCenter == "" || host.missingRack || @@ -815,17 +820,14 @@ func isValidPeer(host *HostInfo) bool { // GetHosts returns a list of hosts found via queries to system.local and system.peers func (r *ringDescriber) GetHosts() ([]*HostInfo, string, error) { - r.mu.Lock() - defer r.mu.Unlock() - localHost, err := r.getLocalHostInfo() if err != nil { - return r.prevHosts, r.prevPartitioner, err + return nil, "", err } peerHosts, err := r.getClusterPeerInfo(localHost) if err != nil { - return r.prevHosts, r.prevPartitioner, err + return nil, "", err } hosts := append([]*HostInfo{localHost}, peerHosts...) @@ -1048,13 +1050,13 @@ func (b *errorBroadcaster) newListener() <-chan error { func (b *errorBroadcaster) broadcast(err error) { b.mu.Lock() - defer b.mu.Unlock() curListeners := b.listeners - if len(curListeners) > 0 { - b.listeners = nil - } else { + if len(curListeners) == 0 { + b.mu.Unlock() return } + b.listeners = nil + b.mu.Unlock() for _, listener := range curListeners { listener <- err From 9bc3b5dee806f1847db37262c5ec510d2f417140 Mon Sep 17 00:00:00 2001 From: Raj Date: Tue, 19 May 2026 17:23:45 -0700 Subject: [PATCH 6/9] Add PreparedMetadata and IsPrepared fields to ObservedQuery and ObservedBatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ObservedQuery and ObservedBatch did not expose the keyspace/table targeted by a query, requiring consumers (metrics collectors, loggers, tracers) to parse the CQL statement string. This change introduces a PreparedMetadata struct (Keyspace, Table) on ObservedQuery and a parallel []PreparedMetadata slice on ObservedBatch, populated from prepared statement metadata returned by Cassandra. An IsPrepared bool on ObservedQuery and a parallel []bool on ObservedBatch explicitly report when the metadata is valid; statements that take the unprepared path (e.g. DDL) report IsPrepared=false and zero PreparedMetadata. The information was already available internally via preparedMetadata (used for routing), so no additional round-trips or CQL parsing are needed. Updated TestObserve (including a negative case for an unprepared DDL statement) and TestBatchObserve to verify the new fields. Patch by Raj Ummadisetty; reviewed by João Reis, Bohdan Siryk for CASSGO-119 --- CHANGELOG.md | 1 + cassandra_test.go | 95 +++++++++++++++++++++++++++++++++++++++++------ conn.go | 13 +++++++ query_executor.go | 62 +++++++++++++++++++++++-------- session.go | 35 +++++++++++++++++ 5 files changed, 179 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f132b5b3..1e34ac1df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Security-model discoverability (CASSANDRA-21464) - Improve host_source locking and ring refresh concurrency (CASSGO-121) +- Add PreparedMetadata (Keyspace, Table) and IsPrepared fields to ObservedQuery, and parallel PreparedMetadata / IsPrepared slices to ObservedBatch, for statement-level observability without CQL parsing (CASSGO-119) ## [2.1.2] diff --git a/cassandra_test.go b/cassandra_test.go index 2f386c801..eda3eb9e9 100644 --- a/cassandra_test.go +++ b/cassandra_test.go @@ -172,9 +172,12 @@ func TestObserve(t *testing.T) { } var ( - observedErr error - observedKeyspace string - observedStmt string + observedErr error + observedKeyspace string + observedPreparedKeyspace string + observedStmt string + observedPreparedTable string + observedIsPrepared bool ) const keyspace = "gocql_test" @@ -182,12 +185,18 @@ func TestObserve(t *testing.T) { resetObserved := func() { observedErr = errors.New("placeholder only") // used to distinguish err=nil cases observedKeyspace = "" + observedPreparedKeyspace = "" observedStmt = "" + observedPreparedTable = "" + observedIsPrepared = false } observer := funcQueryObserver(func(ctx context.Context, o ObservedQuery) { observedKeyspace = o.Keyspace + observedPreparedKeyspace = o.PreparedMetadata.Keyspace observedStmt = o.Statement + observedPreparedTable = o.PreparedMetadata.Table + observedIsPrepared = o.IsPrepared observedErr = o.Err }) @@ -202,6 +211,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } resetObserved() @@ -213,6 +228,12 @@ func TestObserve(t *testing.T) { t.Fatal("insert: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `INSERT INTO observe (id) VALUES (?)` { t.Fatal("insert: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("insert: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("insert: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("insert: unexpected observed prepared table", observedPreparedTable) } resetObserved() @@ -227,6 +248,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } // also works from session observer @@ -240,6 +267,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } // reports errors when the query is poorly formed @@ -254,6 +287,24 @@ func TestObserve(t *testing.T) { } else if observedStmt != `SELECT id FROM unknown_table WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) } + + // statements that cannot be prepared (e.g. DDL) report IsPrepared=false and + // a zero PreparedMetadata, so the contract is explicit for observers. + resetObserved() + const ddlStmt = `CREATE TABLE IF NOT EXISTS gocql_test.observe_unprepared (id int primary key)` + if err := session.Query(ddlStmt).Observer(observer).Exec(); err != nil { + t.Fatal("ddl:", err) + } else if observedErr != nil { + t.Fatal("ddl:", observedErr) + } else if observedStmt != ddlStmt { + t.Fatal("ddl: unexpected observed stmt", observedStmt) + } else if observedIsPrepared { + t.Fatal("ddl: expected observed IsPrepared to be false for unprepared statement") + } else if observedPreparedKeyspace != "" { + t.Fatal("ddl: expected zero PreparedMetadata.Keyspace, got", observedPreparedKeyspace) + } else if observedPreparedTable != "" { + t.Fatal("ddl: expected zero PreparedMetadata.Table, got", observedPreparedTable) + } } func TestObserve_Pagination(t *testing.T) { @@ -2134,10 +2185,12 @@ func TestBatchObserve(t *testing.T) { } type observation struct { - observedErr error - observedKeyspace string - observedStmts []string - observedValues [][]interface{} + observedErr error + observedKeyspace string + observedPreparedMetadata []PreparedMetadata + observedStmts []string + observedIsPrepared []bool + observedValues [][]interface{} } var observedBatch *observation @@ -2149,10 +2202,12 @@ func TestBatchObserve(t *testing.T) { } observedBatch = &observation{ - observedKeyspace: o.Keyspace, - observedStmts: o.Statements, - observedErr: o.Err, - observedValues: o.Values, + observedKeyspace: o.Keyspace, + observedPreparedMetadata: o.PreparedMetadata, + observedStmts: o.Statements, + observedIsPrepared: o.IsPrepared, + observedErr: o.Err, + observedValues: o.Values, } })) for i := 0; i < 100; i++ { @@ -2175,11 +2230,29 @@ func TestBatchObserve(t *testing.T) { if observedBatch.observedKeyspace != "gocql_test" { t.Fatalf("expecting keyspace 'gocql_test', got %q", observedBatch.observedKeyspace) } + if len(observedBatch.observedPreparedMetadata) != 100 { + t.Fatal("expecting 100 observed prepared metadata entries, got", len(observedBatch.observedPreparedMetadata)) + } + if len(observedBatch.observedIsPrepared) != 100 { + t.Fatal("expecting 100 observed IsPrepared flags, got", len(observedBatch.observedIsPrepared)) + } for i, stmt := range observedBatch.observedStmts { if stmt != fmt.Sprintf(`INSERT INTO batch_observe_table (id,other) VALUES (?,%d)`, i) { t.Fatal("unexpected query", stmt) } + if !observedBatch.observedIsPrepared[i] { + t.Fatalf("expected observed IsPrepared at index %d to be true", i) + } + + if observedBatch.observedPreparedMetadata[i].Keyspace != "gocql_test" { + t.Fatalf("unexpected observed prepared keyspace at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Keyspace) + } + + if observedBatch.observedPreparedMetadata[i].Table != "batch_observe_table" { + t.Fatalf("unexpected observed prepared table at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Table) + } + assertDeepEqual(t, "observed value", []interface{}{i}, observedBatch.observedValues[i]) } } diff --git a/conn.go b/conn.go index 28b015773..541224457 100644 --- a/conn.go +++ b/conn.go @@ -1661,6 +1661,7 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter { q.routingInfo.keyspace = usedKeyspace } q.routingInfo.table = info.request.table + q.routingInfo.prepared = true q.routingInfo.mu.Unlock() } else { frame = &writeQueryFrame{ @@ -1856,6 +1857,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter { stmts := make(map[string]string, len(b.batchOpts.entries)) + if b.batchOpts.observer != nil { + b.entryKeyspaces = make([]string, n) + b.entryTables = make([]string, n) + b.entryPrepared = make([]bool, n) + } + for i := 0; i < n; i++ { entry := &b.batchOpts.entries[i] batchStmt := &req.statements[i] @@ -1867,6 +1874,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter { return iter } + if b.entryTables != nil { + b.entryKeyspaces[i] = info.request.keyspace + b.entryTables[i] = info.request.table + b.entryPrepared[i] = true + } + var values []interface{} if entry.binding == nil { values = entry.Args diff --git a/query_executor.go b/query_executor.go index 2d7a62335..5d848c2e8 100644 --- a/query_executor.go +++ b/query_executor.go @@ -385,15 +385,20 @@ func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Ite q.qryOpts.observer.ObserveQuery(q.qryOpts.context, ObservedQuery{ Keyspace: keyspace, Statement: q.qryOpts.stmt, - Values: q.qryOpts.values, - Start: start, - End: end, - Rows: iter.numRows, - Host: host, - Metrics: metricsForHost, - Err: iter.err, - Attempt: attempt, - Query: q.originalQuery, + PreparedMetadata: PreparedMetadata{ + Keyspace: q.routingInfo.getKeyspace(), + Table: q.routingInfo.getTable(), + }, + IsPrepared: q.routingInfo.isPrepared(), + Values: q.qryOpts.values, + Start: start, + End: end, + Rows: iter.numRows, + Host: host, + Metrics: metricsForHost, + Err: iter.err, + Attempt: attempt, + Query: q.originalQuery, }) } } @@ -432,6 +437,7 @@ func (q *internalQuery) GetRoutingKey() ([]byte, error) { q.routingInfo.mu.Lock() q.routingInfo.keyspace = meta.Keyspace q.routingInfo.table = meta.Table + q.routingInfo.prepared = true q.routingInfo.mu.Unlock() } return createRoutingKey(meta, q.qryOpts.values) @@ -562,6 +568,19 @@ type internalBatch struct { session *Session metrics *queryMetrics hostMetricsManager hostMetricsManager + + // entryKeyspaces holds the keyspace for each batch entry, + // populated from prepared statement metadata during execution. + entryKeyspaces []string + + // entryTables holds the table name for each batch entry, + // populated from prepared statement metadata during execution. + entryTables []string + + // entryPrepared[i] reports whether the i-th batch entry was prepared by + // the driver, and therefore whether entryKeyspaces[i]/entryTables[i] hold + // valid prepared statement metadata. + entryPrepared []bool } func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch { @@ -597,20 +616,30 @@ func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Ite metricsForHost := b.hostMetricsManager.attempt(latency, host) - statements := make([]string, len(b.batchOpts.entries)) - values := make([][]interface{}, len(b.batchOpts.entries)) + n := len(b.batchOpts.entries) + statements := make([]string, n) + values := make([][]interface{}, n) + preparedMetadata := make([]PreparedMetadata, n) for i, entry := range b.batchOpts.entries { statements[i] = entry.Stmt values[i] = entry.Args + if i < len(b.entryKeyspaces) { + preparedMetadata[i] = PreparedMetadata{ + Keyspace: b.entryKeyspaces[i], + Table: b.entryTables[i], + } + } } b.batchOpts.observer.ObserveBatch(b.batchOpts.context, ObservedBatch{ - Keyspace: keyspace, - Statements: statements, - Values: values, - Start: start, - End: end, + Keyspace: keyspace, + Statements: statements, + PreparedMetadata: preparedMetadata, + IsPrepared: b.entryPrepared, + Values: values, + Start: start, + End: end, // Rows not used in batch observations // TODO - might be able to support it when using BatchCAS Host: host, Metrics: metricsForHost, @@ -652,6 +681,7 @@ func (b *internalBatch) GetRoutingKey() ([]byte, error) { b.routingInfo.mu.Lock() b.routingInfo.keyspace = meta.Keyspace b.routingInfo.table = meta.Table + b.routingInfo.prepared = true b.routingInfo.mu.Unlock() } diff --git a/session.go b/session.go index fcbdd703b..dee2a8fe1 100644 --- a/session.go +++ b/session.go @@ -1086,6 +1086,8 @@ type queryRoutingInfo struct { keyspace string table string + + prepared bool } func (qr *queryRoutingInfo) getKeyspace() string { @@ -1100,6 +1102,12 @@ func (qr *queryRoutingInfo) getTable() string { return qr.table } +func (qr *queryRoutingInfo) isPrepared() bool { + qr.mu.RLock() + defer qr.mu.RUnlock() + return qr.prepared +} + func (q *Query) defaultsFromSession() { s := q.session @@ -2370,10 +2378,28 @@ func (s *Session) GetHosts() []*HostInfo { return s.ring.allHosts() } +// PreparedMetadata holds metadata extracted from a prepared statement +// (returned by Cassandra during statement preparation). It is exposed via +// ObservedQuery and ObservedBatch when the driver has prepared the statement. +type PreparedMetadata struct { + Keyspace string + Table string +} + type ObservedQuery struct { Keyspace string Statement string + // PreparedMetadata holds keyspace/table information from prepared statement + // metadata returned by Cassandra. Only valid when IsPrepared is true; the + // zero value is reported otherwise (for example, for DDL statements or + // statements not prepared by the driver). + PreparedMetadata PreparedMetadata + + // IsPrepared reports whether the statement was prepared by the driver and + // therefore whether PreparedMetadata holds valid information. + IsPrepared bool + // Values holds a slice of bound values for the query. // Do not modify the values here, they are shared with multiple goroutines. Values []interface{} @@ -2418,6 +2444,15 @@ type ObservedBatch struct { Keyspace string Statements []string + // PreparedMetadata holds prepared statement metadata for each batch statement. + // PreparedMetadata[i] corresponds to Statements[i]. Each entry is only valid + // when IsPrepared[i] is true; the zero value is reported otherwise. + PreparedMetadata []PreparedMetadata + + // IsPrepared reports, for each batch statement, whether the statement was + // prepared by the driver. IsPrepared[i] corresponds to Statements[i]. + IsPrepared []bool + // Values holds a slice of bound values for each statement. // Values[i] are bound values passed to Statements[i]. // Do not modify the values here, they are shared with multiple goroutines. From 1920205fb35b22222b468d55f0f2aa017e976188 Mon Sep 17 00:00:00 2001 From: James Hartig Date: Mon, 22 Jun 2026 17:40:20 +0000 Subject: [PATCH 7/9] fix protocol negotiation error handling In CASSGO-97 and #1920 the handling was changed to better support non-zero stream id error responses. But because of hardcoding of version 5 in the test harness there was a regression where older protocol responses were not supported anymore. By adding an Unwrap method to ErrProtocol now the existing checking in checkProtocolRelatedError will correctly catch the protocolError. Patch by James Hartig; reviewed by Bohdan Siryk for CASSGO-131 --- CHANGELOG.md | 3 +++ conn_test.go | 7 ++++++- protocol_negotiation_test.go | 11 ++++++++++- session.go | 4 ++++ 4 files changed, 23 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e34ac1df..cde64a037 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Improve host_source locking and ring refresh concurrency (CASSGO-121) - Add PreparedMetadata (Keyspace, Table) and IsPrepared fields to ObservedQuery, and parallel PreparedMetadata / IsPrepared slices to ObservedBatch, for statement-level observability without CQL parsing (CASSGO-119) +### Fixed +- Correct protocol negotiation with non-Cassandra servers (CASSGO-131) + ## [2.1.2] ### Fixed diff --git a/conn_test.go b/conn_test.go index abc659b5d..1cb5ee3f8 100644 --- a/conn_test.go +++ b/conn_test.go @@ -1259,7 +1259,12 @@ func (srv *TestServer) process(conn net.Conn, reqFrame *framer, useProtoV5, star srv.errorLocked("process frame with a nil header") return } - respFrame := newFramer(nil, byte(head.version), GlobalTypes) + // use the configured version unless it wasn't specified + version := srv.protocol + if version == 0 { + version = byte(head.version) + } + respFrame := newFramer(nil, version, GlobalTypes) if srv.customRequestHandler != nil { if err := srv.customRequestHandler(srv, reqFrame, respFrame); err != nil { diff --git a/protocol_negotiation_test.go b/protocol_negotiation_test.go index 567c74e36..e58ec1dd2 100644 --- a/protocol_negotiation_test.go +++ b/protocol_negotiation_test.go @@ -231,9 +231,18 @@ func TestProtocolNegotiation(t *testing.T) { forceZeroStreamID: tc.forceZeroStreamID, } + // use the maximum protocol supported + protocol := uint8(0) + for _, supportedVersion := range tc.supportedVersions { + supportedProto := uint8(supportedVersion) + if supportedProto > protocol { + protocol = supportedProto + } + } + srv := newTestServerOpts{ addr: "127.0.0.1:0", - protocol: 5, + protocol: protocol, customRequestHandler: handler.handle, dontFailOnProtocolMismatch: true, }.newServer(t, context.Background()) diff --git a/session.go b/session.go index dee2a8fe1..4717d95c7 100644 --- a/session.go +++ b/session.go @@ -2532,6 +2532,10 @@ var ( // ErrProtocol represents a protocol-level error. type ErrProtocol struct{ error } +func (e ErrProtocol) Unwrap() error { + return e.error +} + // NewErrProtocol creates a new protocol error with the specified format and arguments. func NewErrProtocol(format string, args ...interface{}) error { return ErrProtocol{fmt.Errorf(format, args...)} From d452f7bc9d1abb7ab07937888bf66a81fb9b1b19 Mon Sep 17 00:00:00 2001 From: Max Melentyev Date: Mon, 15 Jun 2026 11:57:41 -0400 Subject: [PATCH 8/9] Add Query.Binding() function to override binding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit As Query is supposed to be reusable since cqlbp v2, it should be possible to overwrite a binding too. Patch by Max Melentyev; reviewed by João Reis, Bohdan Siryk for CASSGO-130 --- CHANGELOG.md | 1 + session.go | 8 +++++++ session_test.go | 57 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index cde64a037..0d6a4fe19 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Security-model discoverability (CASSANDRA-21464) - Improve host_source locking and ring refresh concurrency (CASSGO-121) - Add PreparedMetadata (Keyspace, Table) and IsPrepared fields to ObservedQuery, and parallel PreparedMetadata / IsPrepared slices to ObservedBatch, for statement-level observability without CQL parsing (CASSGO-119) +- Query.Binding() method to override binding function for a query object. ### Fixed - Correct protocol negotiation with non-Cassandra servers (CASSGO-131) diff --git a/session.go b/session.go index 4717d95c7..1b4a11123 100644 --- a/session.go +++ b/session.go @@ -1326,6 +1326,14 @@ func (q *Query) Idempotent(value bool) *Query { // For supported Go to CQL type conversions for query parameters, see Session.Query documentation. func (q *Query) Bind(v ...interface{}) *Query { q.values = v + q.binding = nil + return q +} + +// Binding sets a function for dynamic generation of query arguments. +func (q *Query) Binding(binding func(q *QueryInfo) ([]interface{}, error)) *Query { + q.values = nil + q.binding = binding return q } diff --git a/session_test.go b/session_test.go index b1c7b7476..aa1f45210 100644 --- a/session_test.go +++ b/session_test.go @@ -403,3 +403,60 @@ func TestRetryType_IgnoreRethrow(t *testing.T) { resetObserved() } } + +func TestStaticQueryInfo_OverrideBindingFunction(t *testing.T) { + session := createSession(t) + defer session.Close() + + if err := createTable(session, "CREATE TABLE IF NOT EXISTS gocql_test.static_query_info_override (id int, value text, PRIMARY KEY (id))"); err != nil { + t.Fatalf("failed to create table with error '%v'", err) + } + + if err := session.Query("INSERT INTO static_query_info_override (id, value) VALUES (?, ?)", 1, "foo").Exec(); err != nil { + t.Fatalf("insert into static_query_info_override failed, err '%v'", err) + } + + if err := session.Query("INSERT INTO static_query_info_override (id, value) VALUES (?, ?)", 2, "bar").Exec(); err != nil { + t.Fatalf("insert into static_query_info_override failed, err '%v'", err) + } + + qry := session.Bind("SELECT id, value FROM static_query_info_override WHERE id = ?", func(q *QueryInfo) ([]interface{}, error) { + values := make([]interface{}, 1) + values[0] = 1 + return values, nil + }) + + iter := qry.Iter() + var id int + var value string + iter.Scan(&id, &value) + if err := iter.Close(); err != nil { + t.Fatalf("query with exposed info failed, err '%v'", err) + } + + if id != 1 { + t.Fatalf("Expected id %d, but got %d", 113, id) + } + if value != "foo" { + t.Fatalf("Expected value %s, but got %s", "foo", value) + } + + qry.Binding(func(q *QueryInfo) ([]interface{}, error) { + values := make([]interface{}, 1) + values[0] = 2 + return values, nil + }) + + iter = qry.Iter() + iter.Scan(&id, &value) + if err := iter.Close(); err != nil { + t.Fatalf("query with exposed info failed, err '%v'", err) + } + + if id != 2 { + t.Fatalf("Expected id %d, but got %d", 2, id) + } + if value != "bar" { + t.Fatalf("Expected value %s, but got %s", "bar", value) + } +} From 91a4c711c74d796ee8af3de255a1c48fa415cbaa Mon Sep 17 00:00:00 2001 From: Bohdan Siryk Date: Thu, 11 Jun 2026 11:20:39 +0300 Subject: [PATCH 9/9] Fix LZ4 test fails on arm64 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixed by bumping pierrec/lz4 lib version to v4.1.27. Additionally, added arm64 machines to the testing matrix for unit tests. Patch by Bohdan Siryk; reviwed by João Reis for CASSGO-128 --- .github/workflows/main.yml | 26 ++++++++++++++++++++++---- CHANGELOG.md | 1 + go.mod | 2 +- go.sum | 2 ++ 4 files changed, 26 insertions(+), 5 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 440aaa75a..44d7bd529 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -8,8 +8,8 @@ on: types: [ opened, synchronize, reopened ] jobs: - build: - name: Unit tests + unit-tests-amd64: + name: Unit tests AMD64 runs-on: ubuntu-latest strategy: matrix: @@ -23,11 +23,28 @@ jobs: run: make check - name: Run unit tests run: make test-unit + + unit-tests-arm64: + name: Unit tests ARM64 + runs-on: ubuntu-24.04-arm + strategy: + matrix: + go: [ '1.25', '1.26' ] + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-go@v6 + with: + go-version: ${{ matrix.go }} + - name: Run linting + run: make check + - name: Run unit tests + run: make test-unit integration-cassandra: timeout-minutes: 15 needs: - - build + - unit-tests-amd64 + - unit-tests-arm64 name: Integration Tests runs-on: ubuntu-latest strategy: @@ -93,7 +110,8 @@ jobs: integration-auth-cassandra: timeout-minutes: 15 needs: - - build + - unit-tests-amd64 + - unit-tests-arm64 name: Integration Tests with auth runs-on: ubuntu-latest strategy: diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d6a4fe19..0b6f10457 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Correct protocol negotiation with non-Cassandra servers (CASSGO-131) +- LZ4 test fails on arm64 (CASSGO-128) ## [2.1.2] diff --git a/go.mod b/go.mod index f0428c40e..23e765e95 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,7 @@ module github.com/apache/cassandra-gocql-driver/v2 require ( github.com/golang/snappy v0.0.3 github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed - github.com/pierrec/lz4/v4 v4.1.8 + github.com/pierrec/lz4/v4 v4.1.27 github.com/rs/zerolog v1.34.0 github.com/stretchr/testify v1.9.0 go.uber.org/zap v1.27.0 diff --git a/go.sum b/go.sum index c87ff1de5..5b6b743f6 100644 --- a/go.sum +++ b/go.sum @@ -22,6 +22,8 @@ github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APP github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/pierrec/lz4/v4 v4.1.8 h1:ieHkV+i2BRzngO4Wd/3HGowuZStgq6QkPsD1eolNAO4= github.com/pierrec/lz4/v4 v4.1.8/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=