Merge pull request #480 from aerospike/stage

Go Client v8.2.1

Author: khaf

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Change History
 
+## April 14 2025: v8.2.1
+
+- **Improvements**
+  - [CLIENT-3399] Update dependencies due to snyk security reports.
+
+- **Fixes**
+  - [CLIENT-3384] Panic in BatchGet with using MRT and Filter Expression together.
+  - [CLIENT-3397] Retries fail on PutPayload command.
+  - [CLIENT-3398] Fix Backward compatibility broken in Connection.SetTimeout API
+
 ## March 18 2025: v8.2.0
 
 - **Fixes**

admin_command.go

@@ -396,7 +396,7 @@ func (acmd *AdminCommand) executeCommand(conn *Connection, policy *AdminPolicy)
 		timeout = policy.Timeout
 	}
 
-	if err := conn.SetTimeout(timeout, timeout); err != nil {
+	if err := conn.setTimeout(timeout, timeout); err != nil {
 		return err
 	}
 
@@ -426,7 +426,7 @@ func (acmd *AdminCommand) readUsers(conn *Connection, policy *AdminPolicy) ([]*U
 		timeout = policy.Timeout
 	}
 
-	if err := conn.SetTimeout(timeout, timeout); err != nil {
+	if err := conn.setTimeout(timeout, timeout); err != nil {
 		return nil, err
 	}
 
@@ -579,7 +579,7 @@ func (acmd *AdminCommand) readRoles(conn *Connection, policy *AdminPolicy) ([]*R
 		timeout = policy.Timeout
 	}
 
-	if err := conn.SetTimeout(timeout, timeout); err != nil {
+	if err := conn.setTimeout(timeout, timeout); err != nil {
 		return nil, err
 	}
 

batch_test.go

@@ -427,7 +427,7 @@ var _ = gg.Describe("Aerospike", func() {
 				// gm.Expect(err.Matches(types.BATCH_MAX_REQUESTS_EXCEEDED)).To(gm.BeTrue())
 			})
 
-			gg.It("XXXShould return the error for invalid namespace", func() {
+			gg.It("Should return the error for invalid namespace", func() {
 				var brs []as.BatchRecordIfc
 
 				for i := 0; i < 1; i++ {

cdt_context.go

@@ -198,6 +198,6 @@ func CtxMapKeyCreate(key Value, order mapOrderType) *CDTContext {
 }
 
 // CtxMapValue defines Lookup map by value.
-func CtxMapValue(key Value) *CDTContext {
-	return &CDTContext{ctxTypeMapValue, key}
+func CtxMapValue(value Value) *CDTContext {
+	return &CDTContext{ctxTypeMapValue, value}
 }

client_policy.go

@@ -31,12 +31,12 @@ type ClientPolicy struct {
 	// in hashed format. Leave empty for clusters running without restricted access.
 	Password string
 
-	// ClusterName sets the expected cluster ID.  If not nil, server nodes must return this cluster ID in order to
+	// ClusterName sets the expected cluster ID. If not nil, server nodes must return this cluster ID in order to
 	// join the client's view of the cluster. Should only be set when connecting to servers that
 	// support the "cluster-name" info command. (v3.10+)
 	ClusterName string //=""
 
-	// Initial host connection timeout duration.  The timeout when opening a connection
+	// Initial host connection timeout duration. The timeout when opening a connection
 	// to the server host for the first time.
 	Timeout time.Duration //= 30 seconds
 
@@ -115,7 +115,7 @@ type ClientPolicy struct {
 	TendInterval time.Duration //= 1 second
 
 	// A IP translation table is used in cases where different clients
-	// use different server IP addresses.  This may be necessary when
+	// use different server IP addresses. This may be necessary when
 	// using clients from both inside and outside a local area
 	// network. Default is no translation.
 	// The key is the IP address returned from friend info requests to other servers.
@@ -125,7 +125,7 @@ type ClientPolicy struct {
 	// UseServicesAlternate determines if the client should use "services-alternate" instead of "services"
 	// in info request during cluster tending.
 	//"services-alternate" returns server configured external IP addresses that client
-	// uses to talk to nodes.  "services-alternate" can be used in place of providing a client "ipMap".
+	// uses to talk to nodes. "services-alternate" can be used in place of providing a client "ipMap".
 	// This feature is recommended instead of using the client-side IpMap above.
 	//
 	// "services-alternate" is available with Aerospike Server versions >= 3.7.1.

client_test.go

@@ -472,7 +472,7 @@ var _ = gg.Describe("Aerospike", func() {
 
 				binMap := as.BinMap{
 					"Aerospike":  "value",
-					"Aerospike1": "value2",
+					"Aerospike1": strings.Repeat("a", 1e5+1e4),
 				}
 
 				wcmd, err := as.NewWriteCommand(nil, wpolicy, key, nil, binMap)

connection.go

@@ -156,7 +156,7 @@ func newConnection(address string, timeout time.Duration) (*Connection, Error) {
 	newConn.limitReader = &io.LimitedReader{R: conn, N: 0}
 
 	// set timeout at the last possible moment
-	if err := newConn.SetTimeout(timeout, timeout); err != nil {
+	if err := newConn.setTimeout(timeout, timeout); err != nil {
 		newConn.Close()
 		return nil, err
 	}
@@ -333,7 +333,16 @@ func (ctn *Connection) updateDeadline() Error {
 }
 
 // SetTimeout sets connection timeout for both read and write operations.
-func (ctn *Connection) SetTimeout(totalTimeout, socketTimeout time.Duration) Error {
+func (ctn *Connection) setTimeout(totalTimeout, socketTimeout time.Duration) Error {
+	var deadline time.Time
+	if totalTimeout > 0 {
+		deadline = time.Now().Add(totalTimeout)
+	}
+	return ctn.SetTimeout(deadline, socketTimeout)
+}
+
+// SetTimeout sets connection timeout for both read and write operations.
+func (ctn *Connection) SetTimeout(deadline time.Time, socketTimeout time.Duration) Error {
 	now := time.Now()
 	ctn.socketTimeout = _DEFAULT_TIMEOUT
 	ctn.deadline = time.Time{}
@@ -343,10 +352,10 @@ func (ctn *Connection) SetTimeout(totalTimeout, socketTimeout time.Duration) Err
 	}
 
 	// keep the deadline.IsZero() == true if totalTimeout is not set
-	if totalTimeout > 0 {
-		ctn.deadline = now.Add(totalTimeout)
+	if !deadline.IsZero() {
+		ctn.deadline = deadline
 		if socketTimeout <= 0 {
-			ctn.socketTimeout = totalTimeout
+			ctn.socketTimeout = deadline.Sub(now)
 		}
 	}
 	return nil

connection_heap.go

@@ -165,7 +165,7 @@ func (h *singleConnectionHeap) RefreshIdleTail(tendInterval time.Duration) bool
 
 			// refresh in a goroutine asynchronously
 			go func() {
-				conn.SetTimeout(time.Second, time.Second)
+				conn.setTimeout(time.Second, time.Second)
 				conn.refresh()
 				if _, err := conn.RequestInfo("build"); err == nil {
 					// return to the pool

connection_test.go

@@ -43,6 +43,13 @@ var _ = gg.Describe("Connection Test", func() {
 	})
 
 	gg.It("Dealines should be calculated correctly", func() {
+		deadline := func(timeout time.Duration) (res time.Time) {
+			if timeout > 0 {
+				res = time.Now().Add(timeout)
+			}
+			return res
+		}
+
 		testMatrix := []testExpectations{
 			{0, 0, time.Time{}, time.Now().Add(as.DefaultTimeout()), as.DefaultTimeout()},
 			{0, time.Second, time.Time{}, time.Now().Add(time.Second), time.Second},
@@ -52,7 +59,7 @@ var _ = gg.Describe("Connection Test", func() {
 
 		for _, matrix := range testMatrix {
 			gg.By(fmt.Sprintf("TotalTimeout: %v, SocketTimeout: %v", matrix.totalTimeout, matrix.socketTimeout))
-			err := conn.SetTimeout(matrix.totalTimeout, matrix.socketTimeout)
+			err := conn.SetTimeout(deadline(matrix.totalTimeout), matrix.socketTimeout)
 			gm.Expect(err).ToNot(gm.HaveOccurred())
 
 			expTotalDeadline, expSocketDeadline, expSocketTimeout, err := conn.UpdateDeadline()
@@ -62,7 +69,7 @@ var _ = gg.Describe("Connection Test", func() {
 
 			gm.Expect(expTotalDeadline).To(gm.BeTemporally("~", matrix.expTotalDeadline, time.Millisecond))
 			gm.Expect(expSocketDeadline).To(gm.BeTemporally("~", matrix.expSocketDeadline, time.Millisecond))
-			gm.Expect(expSocketTimeout).To(gm.Equal(matrix.expSocketTimeout))
+			gm.Expect(expSocketTimeout).To(gm.BeNumerically("~", matrix.expSocketTimeout, time.Millisecond))
 		}
 	})
 

go.mod

@@ -1,13 +1,13 @@
 module github.com/aerospike/aerospike-client-go/v8
 
-go 1.23
+go 1.23.0
 
 require (
 	github.com/onsi/ginkgo/v2 v2.22.2
 	github.com/onsi/gomega v1.36.2
 	github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
 	github.com/yuin/gopher-lua v1.1.1
-	golang.org/x/sync v0.10.0
+	golang.org/x/sync v0.12.0
 )
 
 require (
@@ -17,10 +17,10 @@ require (
 	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/stretchr/testify v1.10.0 // indirect
-	golang.org/x/net v0.33.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
-	golang.org/x/tools v0.28.0 // indirect
+	golang.org/x/net v0.37.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/tools v0.31.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -31,16 +31,16 @@ github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad h1:W0LEBv82YCGEtc
 github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad/go.mod h1:Hy8o65+MXnS6EwGElrSRjUzQDLXreJlzYLlWiHtt8hM=
 github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
 github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
-golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
-golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
-golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
-golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
-golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
+golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
+golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
+golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
+golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+golang.org/x/tools v0.31.0 h1:0EedkvKDbh+qistFTd0Bcwe/YLh4vHwWEkiI0toFIBU=
+golang.org/x/tools v0.31.0/go.mod h1:naFTU+Cev749tSJRXJlna0T3WxKvb1kWEx15xA4SdmQ=
 google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk=
 google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

login_command.go

@@ -93,7 +93,7 @@ func (lcmd *loginCommand) login(policy *ClientPolicy, conn *Connection, hashedPa
 
 	lcmd.writeSize()
 
-	conn.SetTimeout(policy.LoginTimeout, policy.LoginTimeout)
+	conn.setTimeout(policy.LoginTimeout, policy.LoginTimeout)
 
 	if _, err := conn.Write(lcmd.dataBuffer[:lcmd.dataOffset]); err != nil {
 		return err

multi_command.go

@@ -247,7 +247,7 @@ func (cmd *baseMultiCommand) parseVersion(fieldCount int) (*uint64, Error) {
 		size := fieldlen - 1
 
 		if fieldType == RECORD_VERSION && size == 7 {
-			version = Buffer.VersionBytesToUint64(cmd.dataBuffer, cmd.dataOffset)
+			version = Buffer.VersionBytesToUint64(cmd.dataBuffer, 1)
 		}
 	}
 	return version, nil

node.go

@@ -538,7 +538,7 @@ func (nd *Node) getConnectionWithHint(totalTimeout, socketTimeout time.Duration,
 		return nil, ErrConnectionPoolEmpty.err()
 	}
 
-	if err = conn.SetTimeout(totalTimeout, socketTimeout); err != nil {
+	if err = conn.setTimeout(totalTimeout, socketTimeout); err != nil {
 		nd.stats.ConnectionsFailed.IncrementAndGet()
 
 		// Do not put back into pool.
@@ -718,7 +718,7 @@ func (nd *Node) usingTendConn(timeout time.Duration, f func(conn *Connection)) (
 		}
 
 		// Set timeout for tend conn
-		if err = (*conn).SetTimeout(timeout, timeout); err != nil {
+		if err = (*conn).setTimeout(timeout, timeout); err != nil {
 			return
 		}
 

txn_test.go

@@ -793,5 +793,52 @@ var _ = gg.Describe("Aerospike", func() {
 			}
 		})
 
+		gg.It("must succeed BatchGet with expressions and transaction", func() {
+			client.Truncate(nil, "test", "", nil)
+
+			startKey := 0
+			endKey := 999
+			num := 300 // example value, change to test others like 0, 600, 1010
+			// Generate keys
+			keys := make([]*as.Key, 0, endKey-startKey+1)
+			for i := startKey; i <= endKey; i++ {
+				key, err := as.NewKey(ns, set, i)
+				gm.Expect(err).ToNot(gm.HaveOccurred())
+
+				keys = append(keys, key)
+			}
+
+			// Create the expression: bin_i > num
+			expr := as.ExpGreater(
+				as.ExpIntBin("bin_i"),
+				as.ExpIntVal(int64(num)),
+			)
+
+			// Create MRT transaction
+			txn := as.NewTxn()
+
+			// Create policy with expression and txn
+			policy := as.NewBatchPolicy()
+			policy.BasePolicy = *as.NewPolicy()
+			policy.FilterExpression = expr
+			policy.Txn = txn
+
+			// Perform BatchGet
+			records, err := client.BatchGet(policy, keys)
+			gm.Expect(err).ToNot(gm.HaveOccurred())
+
+			// Check filtered-in records
+			for _, rec := range records {
+				if rec != nil {
+					val := rec.Bins["bin_i"]
+					gm.Expect(val).NotTo(gm.BeNil())
+					gm.Expect(val.(int)).To(gm.BeNumerically(">", num))
+				}
+			}
+
+			_, err = client.Commit(txn)
+			gm.Expect(err).ToNot(gm.HaveOccurred())
+		})
+
 	}) // describe
 })

value.go

@@ -782,6 +782,11 @@ func (vl FloatValue) String() string {
 // Supported by Aerospike server v5.6+ only.
 type BoolValue bool
 
+// NewBoolValue generates a BoolValue instance.
+func NewBoolValue(b bool) BoolValue {
+	return BoolValue(b)
+}
+
 // EstimateSize returns the size of the BoolValue in wire protocol.
 func (vb BoolValue) EstimateSize() (int, Error) {
 	return PackBool(nil, bool(vb))

write_payload_command.go

@@ -60,8 +60,10 @@ func (cmd *writePayloadCommand) getPolicy(ifc command) Policy {
 }
 
 func (cmd *writePayloadCommand) writeBuffer(ifc command) Error {
-	cmd.dataBuffer = cmd.payload
-	cmd.dataOffset = len(cmd.payload)
+	if err := cmd.sizeBufferSz(len(cmd.payload), false); err != nil {
+		return err
+	}
+	cmd.dataOffset = copy(cmd.dataBuffer, cmd.payload)
 	return nil
 }
 
@@ -75,12 +77,6 @@ func (cmd *writePayloadCommand) prepareRetry(ifc command, isTimeout bool) bool {
 }
 
 func (cmd *writePayloadCommand) parseResult(ifc command, conn *Connection) Error {
-	// make sure the payload is not put back in the buffer pool
-	defer func() {
-		cmd.dataBuffer = cmd.conn.origDataBuffer
-		cmd.dataOffset = 0
-	}()
-
 	// Read header.
 	if _, err := conn.Read(cmd.dataBuffer, int(_MSG_TOTAL_HEADER_SIZE)); err != nil {
 		return err