fix: addres buggy transactions

Signed-off-by: isubasinghe <[email protected]>

Author: isubasinghe

persist/sqldb/workflow_archive.go

@@ -112,16 +112,16 @@ func (r *workflowArchive) ArchiveWorkflow(ctx context.Context, wf *wfv1.Workflow
 	if r.dbType == sqldb.Postgres {
 		workflow = bytes.ReplaceAll(workflow, []byte("\\u0000"), []byte(postgresNullReplacement))
 	}
-	return r.sessionProxy.TxWith(ctx, func(s db.Session) error {
-		_, err := s.SQL().
+	return r.sessionProxy.TxWith(ctx, func(sp *sqldb.SessionProxy) error {
+		_, err := sp.Session().SQL().
 			DeleteFrom(archiveTableName).
 			Where(r.clusterManagedNamespaceAndInstanceID()).
 			And(db.Cond{"uid": wf.UID}).
 			Exec()
 		if err != nil {
 			return err
 		}
-		_, err = s.Collection(archiveTableName).
+		_, err = sp.Session().Collection(archiveTableName).
 			Insert(&archivedWorkflowRecord{
 				archivedWorkflowMetadata: archivedWorkflowMetadata{
 					ClusterName:       r.clusterName,
@@ -140,7 +140,7 @@ func (r *workflowArchive) ArchiveWorkflow(ctx context.Context, wf *wfv1.Workflow
 			return err
 		}
 
-		_, err = s.SQL().
+		_, err = sp.Session().SQL().
 			DeleteFrom(archiveLabelsTableName).
 			Where(db.Cond{"clustername": r.clusterName}).
 			And(db.Cond{"uid": wf.UID}).
@@ -150,7 +150,7 @@ func (r *workflowArchive) ArchiveWorkflow(ctx context.Context, wf *wfv1.Workflow
 		}
 		// insert the labels
 		for key, value := range wf.GetLabels() {
-			_, err := s.Collection(archiveLabelsTableName).
+			_, err := sp.Session().Collection(archiveLabelsTableName).
 				Insert(&archivedWorkflowLabelRecord{
 					ClusterName: r.clusterName,
 					UID:         string(wf.UID),

util/sqldb/session.go

@@ -105,9 +105,12 @@ func (sp *SessionProxy) Tx() *SessionProxy {
 }
 
 // TxWith runs a With transaction
-func (sp *SessionProxy) TxWith(ctx context.Context, fn func(db.Session) error, opts *sql.TxOptions) error {
-	return sp.Tx().With(ctx, func(s db.Session) error {
-		return s.TxContext(ctx, fn, opts)
+func (sp *SessionProxy) TxWith(ctx context.Context, fn func(*SessionProxy) error, opts *sql.TxOptions) error {
+	return sp.With(ctx, func(s db.Session) error {
+		return s.TxContext(ctx, func(sess db.Session) error {
+			newSp := SessionProxy{sp.kubectlConfig, sp.namespace, sp.dbConfig, sp.username, sp.password, sess, sync.RWMutex{}, sp.closed, sp.maxRetries, sp.baseDelay, sp.maxDelay, sp.retryMultiple, sp.insideTransaction}
+			return fn(newSp.Tx())
+		}, opts)
 	})
 }
 

workflow/sync/sync_manager.go

@@ -317,11 +317,10 @@ func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName s
 		var already bool
 		var msg string
 		var failedLockName string
-		err := sm.dbInfo.SessionProxy.TxWith(ctx, func(sess db.Session) error {
+		err := sm.dbInfo.SessionProxy.TxWith(ctx, func(sp *sqldb.SessionProxy) error {
 			sm.log.WithField("holderKey", holderKey).Info(ctx, "TryAcquire - starting transaction")
 			var err error
-			s := sqldb.NewSessionProxyFromSession(sess, nil, "", "")
-			tx := &transaction{s}
+			tx := &transaction{sp}
 			already, updated, msg, failedLockName, err = sm.tryAcquireImpl(ctx, wf, tx, holderKey, failedLockName, syncItems, lockKeys)
 			sm.log.WithField("holderKey", holderKey).Info(ctx, "TryAcquire - transaction completed")
 			return err