Protocol version negotiation doesn't work if server replies with stream id different than 0

Previously, protocol negotiation didn't work properly when C* was responding with stream id different from 0.

This patch changes the way protocol negotiation works. Instead of parsing a supported protocol version from C* error response, the driver tries to connect with each supported protocol starting from the latest.

Patch by Bohdan Siryk; Reviewed by João Reis for CASSGO-98

Author: worryg0d

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Prevent panic with queries during session init (CASSGO-92)
 - Return correct values from RowData (CASSGO-95)
 - Prevent setting a compression flag in a frame header when native proto v5 is being used (CASSGO-98)
+- Use protocol downgrading approach during protocol negotiation (CASSGO-97)
 
 ## [2.0.0]
 

Makefile

@@ -100,7 +100,7 @@ test-integration-auth: .prepare-cassandra-cluster
 test-unit:
 	@echo "Run unit tests"
 	@go clean -testcache
-	go test -tags unit -timeout=5m -race ./...
+	go test -v -tags unit -timeout=5m -race ./...
 
 check: .prepare-golangci
 	@echo "Build"

conn.go

@@ -378,6 +378,13 @@ func (s *startupCoordinator) setupConn(ctx context.Context) error {
 	select {
 	case err := <-startupErr:
 		if err != nil {
+			if s.checkProtocolRelatedError(err) {
+				return &unsupportedProtocolVersionError{
+					err:      err,
+					hostInfo: s.conn.host,
+					version:  protoVersion(s.conn.version),
+				}
+			}
 			return err
 		}
 	case <-ctx.Done():
@@ -387,6 +394,38 @@ func (s *startupCoordinator) setupConn(ctx context.Context) error {
 	return nil
 }
 
+// Checks if the error is protocol related and should be retried during startup.
+// It returns the frame that caused the error and whether the error should be retried.
+func (s *startupCoordinator) checkProtocolRelatedError(err error) bool {
+	var unwrappedFrame frame
+
+	var protocolErr *protocolError
+	if !errors.As(err, &protocolErr) {
+		var errFrame errorFrame
+		if !errors.As(err, &errFrame) {
+			return false
+		} else {
+			unwrappedFrame = errFrame
+		}
+	} else {
+		unwrappedFrame = protocolErr.frame
+	}
+
+	switch frame := unwrappedFrame.(type) {
+	case *supportedFrame:
+		// We can receive a supportedFrame wrapped in protocolError from Conn.recv if the host responds to a 0 stream id.
+		// If we receive a supportedFrame then we know that the host is not compatible with the protocol version, but it is reachable, so we can retry
+		return true
+	case errorFrame:
+		// If we receive an errorFrame with codes ErrCodeProtocol or ErrCodeServer,
+		// then we should try to downgrade a protocol version, so do not skip the host
+		return frame.code == ErrCodeProtocol || frame.code == ErrCodeServer
+	default:
+		// In any other case we should not retry as it means the host is not reachable or some other error happened
+		return false
+	}
+}
+
 func (s *startupCoordinator) write(ctx context.Context, frame frameBuilder, startupCompleted *atomic.Bool) (frame, error) {
 	select {
 	case s.frameTicker <- struct{}{}:
@@ -408,12 +447,14 @@ func (s *startupCoordinator) options(ctx context.Context, startupCompleted *atom
 		return err
 	}
 
-	supported, ok := frame.(*supportedFrame)
-	if !ok {
-		return NewErrProtocol("Unknown type of response to startup frame: %T", frame)
+	switch frame := frame.(type) {
+	case *supportedFrame:
+		return s.startup(ctx, frame.supported, startupCompleted)
+	case error:
+		return frame
+	default:
+		return NewErrProtocol("Unknown type of response to startup frame: %T (frame=%s)", frame, frame.String())
 	}
-
-	return s.startup(ctx, supported.supported, startupCompleted)
 }
 
 func (s *startupCoordinator) startup(ctx context.Context, supported map[string][]string, startupCompleted *atomic.Bool) error {

conn_test.go

@@ -1054,6 +1054,9 @@ type newTestServerOpts struct {
 	addr     string
 	protocol uint8
 	recvHook func(*framer)
+
+	customRequestHandler       func(srv *TestServer, reqFrame, respFrame *framer) error
+	dontFailOnProtocolMismatch bool
 }
 
 func (nts newTestServerOpts) newServer(t testing.TB, ctx context.Context) *TestServer {
@@ -1078,6 +1081,9 @@ func (nts newTestServerOpts) newServer(t testing.TB, ctx context.Context) *TestS
 		cancel:     cancel,
 
 		onRecv: nts.recvHook,
+
+		customRequestHandler:       nts.customRequestHandler,
+		dontFailOnProtocolMismatch: nts.dontFailOnProtocolMismatch,
 	}
 
 	go srv.closeWatch()
@@ -1142,6 +1148,10 @@ type TestServer struct {
 
 	// onRecv is a hook point for tests, called in receive loop.
 	onRecv func(*framer)
+
+	// customRequestHandler allows overriding the default request handling for testing purposes.
+	customRequestHandler       func(srv *TestServer, reqFrame, respFrame *framer) error
+	dontFailOnProtocolMismatch bool
 }
 
 func (srv *TestServer) closeWatch() {
@@ -1162,9 +1172,26 @@ func (srv *TestServer) serve() {
 		}
 
 		go func(conn net.Conn) {
+			var startupCompleted bool
+			var useProtoV5 bool
+
 			defer conn.Close()
 			for !srv.isClosed() {
-				framer, err := srv.readFrame(conn)
+				var reader io.Reader = conn
+
+				if useProtoV5 && startupCompleted {
+					frame, _, err := readUncompressedSegment(conn)
+					if err != nil {
+						if errors.Is(err, io.EOF) {
+							return
+						}
+						srv.errorLocked(err)
+						return
+					}
+					reader = bytes.NewReader(frame)
+				}
+
+				framer, err := srv.readFrame(reader)
 				if err != nil {
 					if err == io.EOF {
 						return
@@ -1177,7 +1204,7 @@ func (srv *TestServer) serve() {
 					srv.onRecv(framer)
 				}
 
-				go srv.process(conn, framer)
+				srv.process(conn, framer, &useProtoV5, &startupCompleted)
 			}
 		}(conn)
 	}
@@ -1215,13 +1242,22 @@ func (srv *TestServer) errorLocked(err interface{}) {
 	srv.t.Error(err)
 }
 
-func (srv *TestServer) process(conn net.Conn, reqFrame *framer) {
+func (srv *TestServer) process(conn net.Conn, reqFrame *framer, useProtoV5, startupCompleted *bool) {
 	head := reqFrame.header
 	if head == nil {
 		srv.errorLocked("process frame with a nil header")
 		return
 	}
-	respFrame := newFramer(nil, reqFrame.proto, GlobalTypes)
+	respFrame := newFramer(nil, byte(head.version), GlobalTypes)
+
+	if srv.customRequestHandler != nil {
+		if err := srv.customRequestHandler(srv, reqFrame, respFrame); err != nil {
+			srv.errorLocked(err)
+			return
+		}
+		// Dont like this but...
+		goto finish
+	}
 
 	switch head.op {
 	case opStartup:
@@ -1412,34 +1448,54 @@ func (srv *TestServer) process(conn net.Conn, reqFrame *framer) {
 		respFrame.writeString("not supported")
 	}
 
-	respFrame.buf[0] = srv.protocol | 0x80
+finish:
+
+	respFrame.buf[0] |= 0x80
 
 	if err := respFrame.finish(); err != nil {
 		srv.errorLocked(err)
 	}
 
-	if err := respFrame.writeTo(conn); err != nil {
-		srv.errorLocked(err)
+	if *useProtoV5 && *startupCompleted {
+		segment, err := newUncompressedSegment(respFrame.buf, true)
+		if err == nil {
+			_, err = conn.Write(segment)
+		}
+		if err != nil {
+			srv.errorLocked(err)
+			return
+		}
+	} else {
+		if err := respFrame.writeTo(conn); err != nil {
+			srv.errorLocked(err)
+		}
+
+		if reqFrame.header.op == opStartup {
+			*startupCompleted = true
+			if head.version == protoVersion5 {
+				*useProtoV5 = true
+			}
+		}
 	}
 }
 
-func (srv *TestServer) readFrame(conn net.Conn) (*framer, error) {
+func (srv *TestServer) readFrame(reader io.Reader) (*framer, error) {
 	buf := make([]byte, srv.headerSize)
-	head, err := readHeader(conn, buf)
+	head, err := readHeader(reader, buf)
 	if err != nil {
 		return nil, err
 	}
 	framer := newFramer(nil, srv.protocol, GlobalTypes)
 
-	err = framer.readFrame(conn, &head)
+	err = framer.readFrame(reader, &head)
 	if err != nil {
 		return nil, err
 	}
 
 	// should be a request frame
 	if head.version.response() {
 		return nil, fmt.Errorf("expected to read a request frame got version: %v", head.version)
-	} else if head.version.version() != srv.protocol {
+	} else if !srv.dontFailOnProtocolMismatch && head.version.version() != srv.protocol {
 		return nil, fmt.Errorf("expected to read protocol version 0x%x got 0x%x", srv.protocol, head.version.version())
 	}
 

control.go

@@ -32,7 +32,6 @@ import (
 	"math/rand"
 	"net"
 	"os"
-	"regexp"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -202,56 +201,9 @@ func shuffleHosts(hosts []*HostInfo) []*HostInfo {
 	return shuffled
 }
 
-// this is going to be version dependant and a nightmare to maintain :(
-var protocolSupportRe = regexp.MustCompile(`the lowest supported version is \d+ and the greatest is (\d+)$`)
-var betaProtocolRe = regexp.MustCompile(`Beta version of the protocol used \(.*\), but USE_BETA flag is unset`)
-
-func parseProtocolFromError(err error) int {
-	errStr := err.Error()
-
-	var errProtocol ErrProtocol
-	if errors.As(err, &errProtocol) {
-		err = errProtocol.error
-	}
-
-	// I really wish this had the actual info in the error frame...
-	matches := betaProtocolRe.FindAllStringSubmatch(errStr, -1)
-	if len(matches) == 1 {
-		var protoErr *protocolError
-		if errors.As(err, &protoErr) {
-			version := protoErr.frame.Header().version.version()
-			if version > 0 {
-				return int(version - 1)
-			}
-		}
-		return 0
-	}
-
-	matches = protocolSupportRe.FindAllStringSubmatch(errStr, -1)
-	if len(matches) != 1 || len(matches[0]) != 2 {
-		var protoErr *protocolError
-		if errors.As(err, &protoErr) {
-			return int(protoErr.frame.Header().version.version())
-		}
-		return 0
-	}
-
-	max, err := strconv.Atoi(matches[0][1])
-	if err != nil {
-		return 0
-	}
-
-	return max
-}
-
-const highestProtocolVersionSupported = 5
-
 func (c *controlConn) discoverProtocol(hosts []*HostInfo) (int, error) {
 	hosts = shuffleHosts(hosts)
 
-	connCfg := *c.session.connCfg
-	connCfg.ProtoVersion = highestProtocolVersionSupported
-
 	handler := connErrorHandlerFn(func(c *Conn, err error, closed bool) {
 		// we should never get here, but if we do it means we connected to a
 		// host successfully which means our attempted protocol version worked
@@ -261,30 +213,56 @@ func (c *controlConn) discoverProtocol(hosts []*HostInfo) (int, error) {
 	})
 
 	var err error
+	var proto int
 	for _, host := range hosts {
-		var conn *Conn
-		conn, err = c.session.dial(c.session.ctx, host, &connCfg, handler)
+		proto, err = c.tryProtocolVersionsForHost(host, handler)
+		if err == nil {
+			return proto, nil
+		}
+
+		c.session.logger.Debug("Failed to discover protocol version for host.",
+			NewLogFieldIP("host_addr", host.ConnectAddress()),
+			NewLogFieldError("err", err))
+	}
+
+	return 0, err
+}
+
+func (c *controlConn) tryProtocolVersionsForHost(host *HostInfo, handler ConnErrorHandler) (int, error) {
+	connCfg := *c.session.connCfg
+
+	var triedVersions []int
+
+	for proto := highestProtocolVersionSupported; proto >= lowestProtocolVersionSupported; proto-- {
+		connCfg.ProtoVersion = proto
+
+		conn, err := c.session.dial(c.session.ctx, host, &connCfg, handler)
 		if conn != nil {
 			conn.Close()
 		}
 
 		if err == nil {
-			c.session.logger.Debug("Discovered protocol version using host.",
-				NewLogFieldInt("protocol_version", connCfg.ProtoVersion), NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()))
-			return connCfg.ProtoVersion, nil
+			return proto, nil
 		}
 
-		if proto := parseProtocolFromError(err); proto > 0 {
-			c.session.logger.Debug("Discovered protocol version using host after parsing protocol error.",
-				NewLogFieldInt("protocol_version", proto), NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()))
-			return proto, nil
+		var unsupportedErr *unsupportedProtocolVersionError
+		if errors.As(err, &unsupportedErr) {
+			// the host does not support this protocol version, try a lower version
+			c.session.logger.Debug("Failed to connect to host during protocol negotiation.",
+				NewLogFieldIP("host_addr", host.ConnectAddress()),
+				NewLogFieldInt("proto_version", proto),
+				NewLogFieldError("err", err))
+			triedVersions = append(triedVersions, connCfg.ProtoVersion)
+			continue
 		}
 
-		c.session.logger.Debug("Failed to discover protocol version using host.",
-			NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()), NewLogFieldError("err", err))
+		c.session.logger.Debug("Error connecting to host during protocol negotiation.",
+			NewLogFieldIP("host_addr", host.ConnectAddress()),
+			NewLogFieldError("err", err))
+		return 0, err
 	}
 
-	return 0, err
+	return 0, fmt.Errorf("gocql: failed to discover protocol version for host %s, tried versions: %v", host.ConnectAddress(), triedVersions)
 }
 
 func (c *controlConn) connect(hosts []*HostInfo, sessionInit bool) error {