fix: cleaner transactions

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

Author: isubasinghe

persist/sqldb/workflow_archive.go

@@ -112,59 +112,57 @@ 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.With(ctx, func(s db.Session) error {
-		return s.Tx(func(sess db.Session) error {
-			_, err := sess.SQL().
-				DeleteFrom(archiveTableName).
-				Where(r.clusterManagedNamespaceAndInstanceID()).
-				And(db.Cond{"uid": wf.UID}).
-				Exec()
-			if err != nil {
-				return err
-			}
-			_, err = sess.Collection(archiveTableName).
-				Insert(&archivedWorkflowRecord{
-					archivedWorkflowMetadata: archivedWorkflowMetadata{
-						ClusterName:       r.clusterName,
-						InstanceID:        r.instanceIDService.InstanceID(),
-						UID:               string(wf.UID),
-						Name:              wf.Name,
-						Namespace:         wf.Namespace,
-						Phase:             wf.Status.Phase,
-						StartedAt:         wf.Status.StartedAt.Time,
-						FinishedAt:        wf.Status.FinishedAt.Time,
-						CreationTimestamp: wf.CreationTimestamp.Time,
-					},
-					Workflow: string(workflow),
-				})
-			if err != nil {
-				return err
-			}
+	return r.sessionProxy.TxWith(ctx, func(s db.Session) error {
+		_, err := s.SQL().
+			DeleteFrom(archiveTableName).
+			Where(r.clusterManagedNamespaceAndInstanceID()).
+			And(db.Cond{"uid": wf.UID}).
+			Exec()
+		if err != nil {
+			return err
+		}
+		_, err = s.Collection(archiveTableName).
+			Insert(&archivedWorkflowRecord{
+				archivedWorkflowMetadata: archivedWorkflowMetadata{
+					ClusterName:       r.clusterName,
+					InstanceID:        r.instanceIDService.InstanceID(),
+					UID:               string(wf.UID),
+					Name:              wf.Name,
+					Namespace:         wf.Namespace,
+					Phase:             wf.Status.Phase,
+					StartedAt:         wf.Status.StartedAt.Time,
+					FinishedAt:        wf.Status.FinishedAt.Time,
+					CreationTimestamp: wf.CreationTimestamp.Time,
+				},
+				Workflow: string(workflow),
+			})
+		if err != nil {
+			return err
+		}
 
-			_, err = sess.SQL().
-				DeleteFrom(archiveLabelsTableName).
-				Where(db.Cond{"clustername": r.clusterName}).
-				And(db.Cond{"uid": wf.UID}).
-				Exec()
+		_, err = s.SQL().
+			DeleteFrom(archiveLabelsTableName).
+			Where(db.Cond{"clustername": r.clusterName}).
+			And(db.Cond{"uid": wf.UID}).
+			Exec()
+		if err != nil {
+			return err
+		}
+		// insert the labels
+		for key, value := range wf.GetLabels() {
+			_, err := s.Collection(archiveLabelsTableName).
+				Insert(&archivedWorkflowLabelRecord{
+					ClusterName: r.clusterName,
+					UID:         string(wf.UID),
+					Key:         key,
+					Value:       value,
+				})
 			if err != nil {
 				return err
 			}
-			// insert the labels
-			for key, value := range wf.GetLabels() {
-				_, err := sess.Collection(archiveLabelsTableName).
-					Insert(&archivedWorkflowLabelRecord{
-						ClusterName: r.clusterName,
-						UID:         string(wf.UID),
-						Key:         key,
-						Value:       value,
-					})
-				if err != nil {
-					return err
-				}
-			}
-			return nil
-		})
-	})
+		}
+		return nil
+	}, nil)
 }
 
 func (r *workflowArchive) ListWorkflows(ctx context.Context, options sutils.ListOptions) (wfv1.Workflows, error) {

util/sqldb/session.go

@@ -105,8 +105,10 @@ func (sp *SessionProxy) Tx() *SessionProxy {
 }
 
 // TxWith runs a With transaction
-func (sp *SessionProxy) TxWith(ctx context.Context, fn func(db.Session) error) error {
-	return sp.Tx().With(ctx, fn)
+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) connect(ctx context.Context) error {

workflow/sync/sync_manager.go

@@ -318,19 +318,14 @@ func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName s
 		var msg string
 		var failedLockName string
 		err := sm.dbInfo.SessionProxy.TxWith(ctx, func(sess db.Session) error {
-			return sess.TxContext(ctx, func(txSess db.Session) error {
-				sm.log.WithField("holderKey", holderKey).Info(ctx, "TryAcquire - starting transaction")
-				var err error
-				s := sqldb.NewSessionProxyFromSession(txSess, nil, "", "")
-				tx := &transaction{s}
-				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
-			}, &sql.TxOptions{
-				Isolation: sql.LevelSerializable,
-				ReadOnly:  false,
-			})
-		})
+			sm.log.WithField("holderKey", holderKey).Info(ctx, "TryAcquire - starting transaction")
+			var err error
+			s := sqldb.NewSessionProxyFromSession(sess, nil, "", "")
+			tx := &transaction{s}
+			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
+		}, &sql.TxOptions{Isolation: sql.LevelSerializable, ReadOnly: false})
 		if err != nil {
 			return false, false, "", failedLockName, err
 		}