CLIENT-3891 TaskId not being properly assigned during Background Query (#545)

* CLIENT-3891 TaskId not being properly assigned during Background Query
* Added deprecation warnings and additional comments

Author: mirzakaracic

client.go

@@ -1385,7 +1385,7 @@ func (clnt *Client) QueryExecute(policy *QueryPolicy,
 
 	var errs Error
 	for i := range nodes {
-		command := newServerCommand(nodes[i], policy, writePolicy, statement, ops)
+		command := newServerCommand(nodes[i], policy, writePolicy, statement, taskId, ops)
 		if err := command.Execute(); err != nil {
 			errs = chainErrors(err, errs)
 		}
@@ -1420,7 +1420,7 @@ func (clnt *Client) ExecuteUDF(policy *QueryPolicy,
 
 	var errs Error
 	for i := range nodes {
-		command := newServerCommand(nodes[i], policy, nil, statement, nil)
+		command := newServerCommand(nodes[i], policy, nil, statement, taskId, nil)
 		if err := command.Execute(); err != nil {
 			errs = chainErrors(err, errs)
 		}
@@ -1453,7 +1453,7 @@ func (clnt *Client) ExecuteUDFNode(policy *QueryPolicy,
 
 	statement.SetAggregateFunction(packageName, functionName, functionArgs, false)
 
-	command := newServerCommand(node, policy, nil, statement, nil)
+	command := newServerCommand(node, policy, nil, statement, taskId, nil)
 	err := command.Execute()
 
 	return NewExecuteTask(clnt.cluster, statement, taskId), err

query_test.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"math"
 	"math/rand"
+	"strconv"
 
 	as "github.com/aerospike/aerospike-client-go/v8"
 	ast "github.com/aerospike/aerospike-client-go/v8/types"
@@ -684,4 +685,84 @@ var _ = gg.Describe("Query operations", func() {
 		gm.Expect(count2).To(gm.Equal(count3))
 		gm.Expect(count1).To(gm.BeNumerically(">", 0))
 	})
+
+	gg.It("must validate taskId for background query jobs using query-show info command", func() {
+		stm := as.NewStatement(ns, set)
+		stm.SetFilter(as.NewRangeFilter(bin3.Name, 0, math.MaxInt16))
+
+		updateBin := as.NewBin("Aerospike10", 888)
+
+		et, err := client.QueryExecute(queryPolicy, nil, stm, as.PutOp(updateBin))
+		gm.Expect(err).ToNot(gm.HaveOccurred())
+		gm.Expect(et).ToNot(gm.BeNil())
+
+		// Get and validate task ID
+		taskId := et.TaskId()
+		gm.Expect(taskId).To(gm.BeNumerically(">", 0))
+
+		// Query all nodes to verify the task ID is actually valid and tracked by the server
+		nodes := client.GetNodes()
+		gm.Expect(len(nodes)).To(gm.BeNumerically(">", 0))
+
+		taskIdStr := strconv.FormatUint(taskId, 10)
+		taskFoundAndValidated := false
+
+		// Check that at least one node has the task tracked with the correct task ID
+		for _, node := range nodes {
+			ip := as.NewInfoPolicy()
+
+			infoCmd := "query-show:id=" + taskIdStr
+
+			res, err := node.RequestInfo(ip, infoCmd)
+			if err == nil {
+				response := res[infoCmd]
+				// If we don't get ERROR:2, the task exists or existed
+				if !bytes.Contains([]byte(response), []byte("ERROR:2")) {
+					// Verify response contains expected job information
+					gm.Expect(response).To(gm.ContainSubstring("status="))
+
+					// Verify the response contains the task ID we're looking for
+					gm.Expect(response).To(gm.Or(
+						gm.ContainSubstring("id="+taskIdStr),
+						gm.ContainSubstring("trid="+taskIdStr),
+					))
+
+					taskFoundAndValidated = true
+				}
+			}
+		}
+
+		gm.Expect(taskFoundAndValidated).To(gm.BeTrue(),
+			"Task ID %d should be tracked by at least one server node", taskId)
+
+		gm.Expect(<-et.OnComplete()).To(gm.BeNil())
+
+		// Verify IsDone returns true after completion
+		done, err := et.IsDone()
+		gm.Expect(err).ToNot(gm.HaveOccurred())
+		gm.Expect(done).To(gm.BeTrue())
+
+		// After completion, verify that the task ID is still consistent
+		taskIdAfterCompletion := et.TaskId()
+		gm.Expect(taskIdAfterCompletion).To(gm.Equal(taskId),
+			"Task ID should remain consistent before and after completion")
+
+		// After completion, verify that records were actually updated
+		verifyStm := as.NewStatement(ns, set)
+		verifyStm.SetFilter(as.NewRangeFilter(bin3.Name, 0, math.MaxInt16))
+		recordset, err := client.Query(queryPolicy, verifyStm)
+		gm.Expect(err).ToNot(gm.HaveOccurred())
+
+		updatedCount := 0
+		for res := range recordset.Results() {
+			gm.Expect(res.Err).ToNot(gm.HaveOccurred())
+			rec := res.Record
+			if rec.Bins["Aerospike10"] == 888 {
+				updatedCount++
+			}
+		}
+
+		// All matching records should have been updated
+		gm.Expect(updatedCount).To(gm.BeNumerically(">", 0))
+	})
 })

server_command.go

@@ -22,17 +22,23 @@ import (
 )
 
 type serverCommand struct {
+	// Deprecated: should not be used and will be removed in future major release.
+	taskId uint64
 	queryCommand
 }
 
-func newServerCommand(node *Node, policy *QueryPolicy, writePolicy *WritePolicy, statement *Statement, operations []*Operation) *serverCommand {
+func newServerCommand(node *Node, policy *QueryPolicy, writePolicy *WritePolicy, statement *Statement, taskId uint64, operations []*Operation) *serverCommand {
+	// Statement does contain a taskId however we cannot rely on it since the statement might be reused.
+	// If TaskId is 0, set it to a new random value and return the same statement. Cannot modify the original
+	// statement since user might want to reuse it.
 	return &serverCommand{
 		queryCommand: *newQueryCommand(node, policy, writePolicy, statement, operations, nil),
+		taskId:       taskId,
 	}
 }
 
 func (cmd *serverCommand) writeBuffer(ifc command) (err Error) {
-	return cmd.setQuery(cmd.policy, cmd.writePolicy, cmd.statement, cmd.statement.TaskId, cmd.operations, cmd.writePolicy != nil, nil)
+	return cmd.setQuery(cmd.policy, cmd.writePolicy, cmd.statement, cmd.taskId, cmd.operations, cmd.writePolicy != nil, nil)
 }
 
 func (cmd *serverCommand) parseRecordResults(ifc command, receiveSize int) (bool, Error) {

statement.go

@@ -47,6 +47,7 @@ type Statement struct {
 
 	// TaskId determines query task id. (Optional)
 	// This value is not used anymore and will be removed later.
+	// Deprecated: Statement instances should not set TaskId directly since instances are meant to be immutable.
 	TaskId uint64
 
 	// determines if the query should return data
@@ -118,12 +119,12 @@ func (stmt *Statement) prepareTaskId() uint64 {
 	// If TaskId is 0, set it to a new random value and return the same statement.
 	// This also means that that the taskId was never set by the user.
 	// If TaskId is non-zero, it means that the user set it manually.
-	// In that case, we make a copy of the statement and set a new random taskId on the copy.
-	// This way we don't modify the original statement that the user provided.
 	// This is important because the user might want to reuse the same statement for multiple queries.
 	// However the control of the taskId is now with the client library and not the user.
 	// Important to remember is that the server will reject queries that have already been executed and the result is not
 	// available yet.
+	//
+	// The statement.TaskId is deprecated and will be removed/revamped in future major release.
 	if stmt.TaskId == 0 {
 		taskId := rand.Uint64()