GODRIVER-2737 ConnectionId returned in heartbeats may be int64 (#1266)

Author: prestonvasquez

mongo/integration/mtest/opmsg_deployment.go

@@ -90,9 +90,15 @@ func (*connection) ID() string {
 	return "<mock_connection>"
 }
 
+// DriverConnectionID returns a fixed identifier for the driver pool connection.
+// TODO(GODRIVER-2824): replace return type with int64.
+func (*connection) DriverConnectionID() uint64 {
+	return 0
+}
+
 // ServerConnectionID returns a fixed identifier for the server connection.
-func (*connection) ServerConnectionID() *int32 {
-	serverConnectionID := int32(42)
+func (*connection) ServerConnectionID() *int64 {
+	serverConnectionID := int64(42)
 	return &serverConnectionID
 }
 

x/mongo/driver/driver.go

@@ -68,7 +68,8 @@ type Connection interface {
 	Close() error
 
 	ID() string
-	ServerConnectionID() *int32
+	ServerConnectionID() *int64
+	DriverConnectionID() uint64 // TODO(GODRIVER-2824): change type to int64.
 	Address() address.Address
 	Stale() bool
 }
@@ -178,7 +179,7 @@ type ErrorProcessor interface {
 type HandshakeInformation struct {
 	Description             description.Server
 	SpeculativeAuthenticate bsoncore.Document
-	ServerConnectionID      *int32
+	ServerConnectionID      *int64
 	SaslSupportedMechs      []string
 }
 

x/mongo/driver/drivertest/channel_conn.go

@@ -64,9 +64,15 @@ func (c *ChannelConn) ID() string {
 	return "faked"
 }
 
+// DriverConnectionID implements the driver.Connection interface.
+// TODO(GODRIVER-2824): replace return type with int64.
+func (c *ChannelConn) DriverConnectionID() uint64 {
+	return 0
+}
+
 // ServerConnectionID implements the driver.Connection interface.
-func (c *ChannelConn) ServerConnectionID() *int32 {
-	serverConnectionID := int32(42)
+func (c *ChannelConn) ServerConnectionID() *int64 {
+	serverConnectionID := int64(42)
 	return &serverConnectionID
 }
 

x/mongo/driver/operation.go

@@ -11,6 +11,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"sync"
@@ -91,22 +92,40 @@ type startedInformation struct {
 	cmdName                  string
 	documentSequenceIncluded bool
 	connID                   string
-	serverConnID             *int32
+	driverConnectionID       uint64 // TODO(GODRIVER-2824): change type to int64.
+	serverConnID             *int64
 	redacted                 bool
 	serviceID                *primitive.ObjectID
 }
 
 // finishedInformation keeps track of all of the information necessary for monitoring success and failure events.
 type finishedInformation struct {
-	cmdName      string
-	requestID    int32
-	response     bsoncore.Document
-	cmdErr       error
-	connID       string
-	serverConnID *int32
-	startTime    time.Time
-	redacted     bool
-	serviceID    *primitive.ObjectID
+	cmdName            string
+	requestID          int32
+	response           bsoncore.Document
+	cmdErr             error
+	connID             string
+	driverConnectionID uint64 // TODO(GODRIVER-2824): change type to int64.
+	serverConnID       *int64
+	startTime          time.Time
+	redacted           bool
+	serviceID          *primitive.ObjectID
+}
+
+// convertInt64PtrToInt32Ptr will convert an int64 pointer reference to an int32 pointer
+// reference. If the int64 value cannot be converted to int32 without causing
+// an overflow, then this function will return nil.
+func convertInt64PtrToInt32Ptr(i64 *int64) *int32 {
+	if i64 == nil {
+		return nil
+	}
+
+	if *i64 > math.MaxInt32 || *i64 < math.MinInt32 {
+		return nil
+	}
+
+	i32 := int32(*i64)
+	return &i32
 }
 
 // ResponseInfo contains the context required to parse a server response.
@@ -552,6 +571,7 @@ func (op Operation) Execute(ctx context.Context) error {
 
 		// set extra data and send event if possible
 		startedInfo.connID = conn.ID()
+		startedInfo.driverConnectionID = conn.DriverConnectionID()
 		startedInfo.cmdName = op.getCommandName(startedInfo.cmd)
 		op.cmdName = startedInfo.cmdName
 		startedInfo.redacted = op.redactCommand(startedInfo.cmdName, startedInfo.cmd)
@@ -574,13 +594,14 @@ func (op Operation) Execute(ctx context.Context) error {
 		}
 
 		finishedInfo := finishedInformation{
-			cmdName:      startedInfo.cmdName,
-			requestID:    startedInfo.requestID,
-			startTime:    time.Now(),
-			connID:       startedInfo.connID,
-			serverConnID: startedInfo.serverConnID,
-			redacted:     startedInfo.redacted,
-			serviceID:    startedInfo.serviceID,
+			cmdName:            startedInfo.cmdName,
+			requestID:          startedInfo.requestID,
+			startTime:          time.Now(),
+			connID:             startedInfo.connID,
+			driverConnectionID: startedInfo.driverConnectionID,
+			serverConnID:       startedInfo.serverConnID,
+			redacted:           startedInfo.redacted,
+			serviceID:          startedInfo.serviceID,
 		}
 
 		// Check for possible context error. If no context error, check if there's enough time to perform a
@@ -1704,7 +1725,7 @@ func (op Operation) publishStartedEvent(ctx context.Context, info startedInforma
 		CommandName:        info.cmdName,
 		RequestID:          int64(info.requestID),
 		ConnectionID:       info.connID,
-		ServerConnectionID: info.serverConnID,
+		ServerConnectionID: convertInt64PtrToInt32Ptr(info.serverConnID),
 		ServiceID:          info.serviceID,
 	}
 	op.CommandMonitor.Started(ctx, started)
@@ -1732,7 +1753,7 @@ func (op Operation) publishFinishedEvent(ctx context.Context, info finishedInfor
 		RequestID:          int64(info.requestID),
 		ConnectionID:       info.connID,
 		DurationNanos:      durationNanos,
-		ServerConnectionID: info.serverConnID,
+		ServerConnectionID: convertInt64PtrToInt32Ptr(info.serverConnID),
 		ServiceID:          info.serviceID,
 	}
 

x/mongo/driver/operation/hello.go

@@ -240,7 +240,7 @@ func (h *Hello) GetHandshakeInformation(ctx context.Context, _ address.Address,
 	if speculativeAuthenticate, ok := h.res.Lookup("speculativeAuthenticate").DocumentOK(); ok {
 		info.SpeculativeAuthenticate = speculativeAuthenticate
 	}
-	if serverConnectionID, ok := h.res.Lookup("connectionId").Int32OK(); ok {
+	if serverConnectionID, ok := h.res.Lookup("connectionId").AsInt64OK(); ok {
 		info.ServerConnectionID = &serverConnectionID
 	}
 	// Cast to bson.Raw to lookup saslSupportedMechs to avoid converting from bsoncore.Value to bson.RawValue for the

x/mongo/driver/operation_test.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"math"
 	"testing"
 	"time"
 
@@ -748,7 +749,7 @@ type mockConnection struct {
 	rDesc         description.Server
 	rCloseErr     error
 	rID           string
-	rServerConnID *int32
+	rServerConnID *int64
 	rAddr         address.Address
 	rCanStream    bool
 	rStreaming    bool
@@ -757,13 +758,16 @@ type mockConnection struct {
 func (m *mockConnection) Description() description.Server { return m.rDesc }
 func (m *mockConnection) Close() error                    { return m.rCloseErr }
 func (m *mockConnection) ID() string                      { return m.rID }
-func (m *mockConnection) ServerConnectionID() *int32      { return m.rServerConnID }
+func (m *mockConnection) ServerConnectionID() *int64      { return m.rServerConnID }
 func (m *mockConnection) Address() address.Address        { return m.rAddr }
 func (m *mockConnection) SupportsStreaming() bool         { return m.rCanStream }
 func (m *mockConnection) CurrentlyStreaming() bool        { return m.rStreaming }
 func (m *mockConnection) SetStreaming(streaming bool)     { m.rStreaming = streaming }
 func (m *mockConnection) Stale() bool                     { return false }
 
+// TODO:(GODRIVER-2824) replace return type with int64.
+func (m *mockConnection) DriverConnectionID() uint64 { return 0 }
+
 func (m *mockConnection) WriteWireMessage(_ context.Context, wm []byte) error {
 	m.pWriteWM = wm
 	return m.rWriteErr
@@ -836,3 +840,55 @@ func TestRetry(t *testing.T) {
 			"expected operation to complete only after the context deadline is exceeded")
 	})
 }
+
+func TestConvertI64PtrToI32Ptr(t *testing.T) {
+	t.Parallel()
+
+	newI64 := func(i64 int64) *int64 { return &i64 }
+	newI32 := func(i32 int32) *int32 { return &i32 }
+
+	tests := []struct {
+		name string
+		i64  *int64
+		want *int32
+	}{
+		{
+			name: "empty",
+			want: nil,
+		},
+		{
+			name: "in bounds",
+			i64:  newI64(1),
+			want: newI32(1),
+		},
+		{
+			name: "out of bounds negative",
+			i64:  newI64(math.MinInt32 - 1),
+		},
+		{
+			name: "out of bounds positive",
+			i64:  newI64(math.MaxInt32 + 1),
+		},
+		{
+			name: "exact min int32",
+			i64:  newI64(math.MinInt32),
+			want: newI32(math.MinInt32),
+		},
+		{
+			name: "exact max int32",
+			i64:  newI64(math.MaxInt32),
+			want: newI32(math.MaxInt32),
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := convertInt64PtrToInt32Ptr(test.i64)
+			assert.Equal(t, test.want, got, "convertInt64PtrToInt32Ptr()=%v, got %v", test.want, got)
+		})
+	}
+}

x/mongo/driver/session/client_session.go

@@ -86,7 +86,8 @@ type LoadBalancedTransactionConnection interface {
 	Description() description.Server
 	Close() error
 	ID() string
-	ServerConnectionID() *int32
+	ServerConnectionID() *int64
+	DriverConnectionID() uint64 // TODO(GODRIVER-2824): change type to int64.
 	Address() address.Address
 	Stale() bool
 

x/mongo/driver/topology/connection.go

@@ -70,12 +70,14 @@ type connection struct {
 	currentlyStreaming   bool
 	connectContextMutex  sync.Mutex
 	cancellationListener cancellationListener
-	serverConnectionID   *int32 // the server's ID for this client's connection
+	serverConnectionID   *int64 // the server's ID for this client's connection
 
 	// pool related fields
-	pool       *pool
-	poolID     uint64
-	generation uint64
+	pool *pool
+
+	// TODO(GODRIVER-2824): change driverConnectionID type to int64.
+	driverConnectionID uint64
+	generation         uint64
 }
 
 // newConnection handles the creation of a connection. It does not connect the connection.
@@ -105,6 +107,12 @@ func newConnection(addr address.Address, opts ...ConnectionOption) *connection {
 	return c
 }
 
+// DriverConnectionID returns the driver connection ID.
+// TODO(GODRIVER-2824): change return type to int64.
+func (c *connection) DriverConnectionID() uint64 {
+	return c.driverConnectionID
+}
+
 // setGenerationNumber sets the connection's generation number if a callback has been provided to do so in connection
 // configuration.
 func (c *connection) setGenerationNumber() {
@@ -527,7 +535,7 @@ func (c *connection) ID() string {
 	return c.id
 }
 
-func (c *connection) ServerConnectionID() *int32 {
+func (c *connection) ServerConnectionID() *int64 {
 	return c.serverConnectionID
 }
 
@@ -708,7 +716,7 @@ func (c *Connection) ID() string {
 }
 
 // ServerConnectionID returns the server connection ID of this connection.
-func (c *Connection) ServerConnectionID() *int32 {
+func (c *Connection) ServerConnectionID() *int64 {
 	if c.connection == nil {
 		return nil
 	}
@@ -794,6 +802,12 @@ func (c *Connection) unpin(reason string) error {
 	return nil
 }
 
+// DriverConnectionID returns the driver connection ID.
+// TODO(GODRIVER-2824): change return type to int64.
+func (c *Connection) DriverConnectionID() uint64 {
+	return c.connection.DriverConnectionID()
+}
+
 func configureTLS(ctx context.Context,
 	tlsConnSource tlsConnectionSource,
 	nc net.Conn,

x/mongo/driver/topology/connection_test.go

@@ -763,7 +763,7 @@ func TestConnection(t *testing.T) {
 				t.Errorf("LocalAddresses do not match. got %v; want %v", got, want)
 			}
 
-			want = (*int32)(nil)
+			want = (*int64)(nil)
 			got = conn.ServerConnectionID()
 			if !cmp.Equal(got, want) {
 				t.Errorf("ServerConnectionIDs do not match. got %v; want %v", got, want)