datastore: Parallel Transaction on same kind with unique keys giving concurrent transaction error

[mehulparmariitr]

Client

datastore

Environment

Go executable on Linux

Go Environment

$ go version - go version go1.21.0 darwin/amd64
$ go env -

GO111MODULE='on'
GOARCH='amd64'
GOBIN=''
GOCACHE='/Users/mehparmar/Library/Caches/go-build'
GOENV='/Users/mehparmar/Library/Application Support/go/env'
GOEXE=''
GOEXPERIMENT=''
GOFLAGS=''
GOHOSTARCH='amd64'
GOHOSTOS='darwin'
GOINSECURE=''
GOMODCACHE='/Users/mehparmar/go/pkg/mod'
GONOPROXY=''
GONOSUMDB=''
GOOS='darwin'
GOPATH='/Users/mehparmar/go'
GOPRIVATE=''
GOPROXY='https://proxy.golang.org,direct'
GOROOT='/usr/local/go'
GOSUMDB='sum.golang.org'
GOTMPDIR=''
GOTOOLCHAIN='auto'
GOTOOLDIR='/usr/local/go/pkg/tool/darwin_amd64'
GOVCS=''
GOVERSION='go1.21.0'
GCCGO='gccgo'
GOAMD64='v1'
AR='ar'
CC='clang'
CXX='clang++'
CGO_ENABLED='1'
GOMOD='/Users/mehparmar/Repos/mehparmar/abcserv/go.mod'
GOWORK=''
CGO_CFLAGS='-O2 -g'
CGO_CPPFLAGS=''
CGO_CXXFLAGS='-O2 -g'
CGO_FFLAGS='-O2 -g'
CGO_LDFLAGS='-O2 -g'
PKG_CONFIG='pkg-config'
GOGCCFLAGS='-fPIC -arch x86_64 -m64 -pthread -fno-caret-diagnostics -Qunused-arguments -fmessage-length=0 -ffile-prefix-map=/var/folders/6n/kcp2mfmx0s9_lqw9y30yqq500000gq/T/go-build1900003096=/tmp/go-build -gno-record-gcc-switches -fno-common'

Code

e.g.

func processBatch(ctx context.Context, logger *l.Logger, desQSp *gds.QuerySpecs, tableName string, batch []dataanalytics.AggregationEntity, wg *sync.WaitGroup) {
	defer wg.Done() // Signal that this goroutine is done

        //Calling GetMulti on keys passed here, then checking if its insert or update call.
	err, _, _, _ := dataanalytics.Upsert(ctx, logger, (*gds2.QuerySpecs)(desQSp), tableName, batch)
	if err != nil {
		// Handle the error appropriately (logging, retries, etc.)
		log.Printf("Error during Upsert: %v", err)
	}
}

// code in main function
// go routines to write data in parallel for unique keys
var wg sync.WaitGroup // To wait for goroutines to finish
		//At max 1000 keys can be queried from Google Datastore, Hence creating batches of 1000
		for i := 0; i < len(msgs); i += 1000 {
			end := i + 1000
			if end > len(msgs) {
				end = len(msgs)
			}
			batch := msgs[i:end]
			// Process the batch here
			//wg.Add(1)
			go processBatch(ctx, l.New("abc-services"), (*gds.QuerySpecs)(desQSp), preAggregationRequest.DestinationKind, batch, &wg)
		}

		wg.Wait() // Wait for all goroutines to complete

Expected behavior

All entities should be written. There are around 275,000

Actual behavior

2024/04/30 05:06:42 Error during Upsert: datastore: concurrent transaction
UUID-NOT-COLLECTED API-NOT-COLLECTED class.Upsert tx.Commit: datastore: concurrent transaction
2024/04/30 05:06:43 Error during Upsert: datastore: concurrent transaction
UUID-NOT-COLLECTED API-NOT-COLLECTED class.Upsert tx.Commit: rpc error: code = InvalidArgument desc = Invalid transaction.
2024/04/30 05:06:43 Error during Upsert: rpc error: code = InvalidArgument desc = Invalid transaction.
UUID-NOT-COLLECTED API-NOT-COLLECTED class.Upsert tx.Commit: rpc error: code = InvalidArgument desc = Invalid transaction.
2024/04/30 05:06:43 Error during Upsert: rpc error: code = InvalidArgument desc = Invalid transaction.

Additional context

if keys are unique then why transaction is failing?

[bhshkh]

On which version of Datastore client library are you facing this issue?

[mehulparmariitr]

@bhshkh I see its 1.0.0.

But we have replaced also this lib in go.mod

replace cloud.google.com/go/datastore => github.com/googleapis/google-cloud-go/datastore v1.0.0

[mehulparmariitr]

Even with the latest version the error is there. Seems like a logical error. But if keys are unique then it shouldnt throw concurrent transaction error right?

go 1.12

require (
	cloud.google.com/go/bigquery v1.60.0
	cloud.google.com/go/datastore v1.16.0
	cloud.google.com/go/logging v1.9.0
	cloud.google.com/go/profiler v0.1.0
	cloud.google.com/go/pubsub v1.37.0
	cloud.google.com/go/storage v1.39.1
	github.com/dimfeld/httptreemux v5.0.1+incompatible
	github.com/dimfeld/httptreemux/v5 v5.0.2
	github.com/evanphx/json-patch v4.5.0+incompatible
	github.com/getlantern/deepcopy v0.0.0-20160317154340-7f45deb8130a
	github.com/go-redis/redis v6.15.7+incompatible
	github.com/gomodule/redigo v2.0.0+incompatible
	github.com/kelseyhightower/envconfig v1.4.0
	github.com/mitchellh/mapstructure v1.1.2
	github.com/oleiade/reflections v1.0.1
	github.com/openzipkin/zipkin-go v0.2.5
	github.com/pborman/uuid v1.2.1
	github.com/pkg/errors v0.9.1
	github.com/schwarmco/go-cartesian-product v0.0.0-20180515110546-d5ee747a6dc9
	github.com/signalfx/golib/v3 v3.3.19
	github.com/spf13/cobra v1.5.0
	github.com/stretchr/testify v1.9.0
	github.com/thedevsaddam/gojsonq v2.3.0+incompatible
	go.opencensus.io v0.24.0
	google.golang.org/api v0.176.1
	google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda
	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
	gopkg.in/go-playground/validator.v8 v8.18.2
	gopkg.in/yaml.v2 v2.4.0
)

replace cloud.google.com/go/datastore => github.com/googleapis/google-cloud-go/datastore v1.16.0

The error -

 UUID-NOT-COLLECTED API-NOT-COLLECTED abc.Upsert tx.Commit: datastore: concurrent transaction
2024/05/01 02:46:59 Error during Upsert: datastore: concurrent transaction
 UUID-NOT-COLLECTED API-NOT-COLLECTED abc.Upsert tx.Commit: datastore: concurrent transaction
2024/05/01 02:46:59 Error during Upsert: datastore: concurrent transaction
 UUID-NOT-COLLECTED API-NOT-COLLECTED abc.Upsert tx.Commit: rpc error: code = InvalidArgument desc = The referenced transaction has expired or is no longer valid.
2024/05/01 02:47:00 Error during Upsert: rpc error: code = InvalidArgument desc = The referenced transaction has expired or is no longer valid.

[mehulparmariitr]

This is the method I am using to upsert into DB -

// Upsert - insert in gds aggregation and update using transaction if entries previously exist with same MdInstanceHashKey it gets updated
func Upsert(ctx context.Context, logger *slog.Logger, qSp *gds.QuerySpecs, tableName string, msgs []AggregationEntity) (error, []*datastore.Key, []*datastore.Key, []*AggregationEntity) {
	ctx, span := trace.StartSpan(ctx, "internal.dataanalytics.Update")
	defer span.End()
	logger.Infof(ctx, "Started putting entries in GDS(aggregation) in %v table", tableName)

	client, err := gds.GetDatastoreClient(ctx, logger, qSp)
	if err != nil {
		logger.Errorf(ctx, "Failed to create GDS client: %v", err)
		return err, nil, nil, nil
	}

	// make keys array use MdInstanceHashKey column
	var keys []*datastore.Key
	for id := 0; id < len(msgs); id++ {
		newKey := datastore.NameKey(tableName, msgs[id].MdInstanceHashKey, nil)
		newKey.Namespace = qSp.Namespace
		keys = append(keys, newKey)
	}
	logger.Infof(ctx, "Length of keys to insert in data store are : %v", len(keys))
	tx, err := client.NewTransaction(ctx)
	if err != nil {
		logger.Errorf(ctx, "client.NewTransaction: %v", err)
		return err, nil, nil, nil
	}
	gdsEntry := make([]AggregationEntity, len(msgs))
	if err := tx.GetMulti(keys, gdsEntry); err == datastore.ErrNoSuchEntity {
		logger.Errorf(ctx, "error in update in GDS %v", err)
		return err, nil, nil, nil
	}

	var createKeys []*datastore.Key
	var updateKeys []*datastore.Key
	var createValues []*AggregationEntity
	// checking if entry with same key exists or not if it exists then update it
	for i := 0; i < len(msgs); i++ {

		if gdsEntry[i].GitTestRepoURL == "" {
			//create
			gdsEntry[i] = msgs[i]
			createKeys = append(createKeys, keys[i])
			createValues = append(createValues, &msgs[i])
		} else {
			//update
			if gdsEntry[i].Type == Pass {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Passed)
				updateKeys = append(updateKeys, keys[i])
			}
			if gdsEntry[i].Type == Fail {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Failed)
				updateKeys = append(updateKeys, keys[i])
			}
			if gdsEntry[i].Type == Skip {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Skipped)
				updateKeys = append(updateKeys, keys[i])
			}
			if gdsEntry[i].Type == MostCommonFailure {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Failed)
				updateKeys = append(updateKeys, keys[i])
			}
			if gdsEntry[i].Type == MostFailed {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Failed)
				updateKeys = append(updateKeys, keys[i])
			}
			if gdsEntry[i].Type == MostSkipped {
				gdsEntry[i] = AggregateUpdateSum(logger, gdsEntry[i], msgs[i], Skipped)
			}
			if gdsEntry[i].Type == MostDuration {
				gdsEntry[i] = AggregateUpdateMax(logger, gdsEntry[i], msgs[i], Duration)
				updateKeys = append(updateKeys, keys[i])
			}
		}
	}

	// batch operation to put in GDS using transaction
	if _, e := tx.PutMulti(keys, gdsEntry); e != nil {
		tx.Rollback()
		logger.Errorf(ctx, "tx.PutMulti: %v", e)
		//logger.Errorf(ctx, "gdsEntry1 are: %v", gdsEntry)
		//logger.Errorf(ctx, "keys1 are: %v", gdsEntry)
		return e, nil, nil, nil
	}
	if _, er := tx.Commit(); er != nil {
		tx.Rollback()
		logger.Errorf(ctx, "tx.Commit: %v", er) // error: concurrent transaction
		//logger.Errorf(ctx, "gdsEntry2 are: %v", gdsEntry)
		//logger.Errorf(ctx, "keys2 are: %v", gdsEntry)
		return er, nil, nil, nil
	}

	logger.Infof(ctx, "Entries stored successfully")
	return nil, createKeys, updateKeys, createValues
}