fix: requeue workflow on transient sync lock errors instead of failing (cherry-pick #16745 for 4.0) (#16792)

Signed-off-by: Alan Clucas <alan@clucas.org>

Author: Joibel

go.mod

@@ -300,7 +300,7 @@ require (
 	github.com/klauspost/compress v1.18.7 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
-	github.com/lib/pq v1.10.9 // indirect
+	github.com/lib/pq v1.10.9
 	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect
 	github.com/lithammer/dedent v1.1.0 // indirect
 	github.com/mailru/easyjson v0.9.0 // indirect

util/sqldb/errors.go

@@ -0,0 +1,36 @@
+package sqldb
+
+import (
+	"errors"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/lib/pq"
+)
+
+// PostgreSQL transaction-conflict error codes: serialization_failure and
+// deadlock_detected. lib/pq v1.10 exports no code constants (the pqerror
+// subpackage arrived in a later release), so they are defined here.
+const (
+	pqSerializationFailure pq.ErrorCode = "40001"
+	pqDeadlockDetected     pq.ErrorCode = "40P01"
+)
+
+// mysqlErLockDeadlock is ER_LOCK_DEADLOCK: the transaction was chosen as the
+// deadlock victim and rolled back. go-sql-driver/mysql exports no error-number
+// constants. ER_LOCK_WAIT_TIMEOUT (1205) is deliberately not treated as a
+// conflict: it only rolls back the statement, not the transaction, and it
+// signals a held-too-long lock rather than a retryable ordering conflict.
+const mysqlErLockDeadlock = 1213
+
+// IsSerializationConflict reports whether err is a transaction conflict
+// reported by the database driver itself: PostgreSQL serialization_failure
+// (40001) or deadlock_detected (40P01), or MySQL ER_LOCK_DEADLOCK (1213).
+// These are the errors where the database aborted the transaction because of
+// concurrent activity and the whole transaction can safely be retried.
+func IsSerializationConflict(err error) bool {
+	var pqErr *pq.Error
+	if errors.As(err, &pqErr) && (pqErr.Code == pqSerializationFailure || pqErr.Code == pqDeadlockDetected) {
+		return true
+	}
+	return errors.Is(err, &mysql.MySQLError{Number: mysqlErLockDeadlock})
+}

util/sqldb/errors_test.go

@@ -0,0 +1,35 @@
+package sqldb
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/lib/pq"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsSerializationConflict(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"postgres serialization failure", &pq.Error{Code: pqSerializationFailure, Message: "could not serialize access"}, true},
+		{"postgres deadlock detected", &pq.Error{Code: pqDeadlockDetected, Message: "deadlock detected"}, true},
+		{"postgres wrapped", fmt.Errorf("commit: %w", &pq.Error{Code: pqSerializationFailure}), true},
+		{"postgres unrelated code", &pq.Error{Code: "23505", Message: "duplicate key value"}, false},
+		{"mysql deadlock", &mysql.MySQLError{Number: 1213, Message: "Deadlock found when trying to get lock; try restarting transaction"}, true},
+		{"mysql wrapped", fmt.Errorf("commit: %w", &mysql.MySQLError{Number: 1213}), true},
+		{"mysql lock wait timeout", &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded; try restarting transaction"}, false},
+		{"plain error mentioning deadlock", errors.New("deadlock detected"), false},
+		{"plain error mentioning serialization", errors.New("could not serialize access"), false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, IsSerializationConflict(tt.err))
+		})
+	}
+}

workflow/controller/operator.go

@@ -62,6 +62,7 @@ import (
 	"github.com/argoproj/argo-workflows/v4/workflow/controller/indexes"
 	"github.com/argoproj/argo-workflows/v4/workflow/metrics"
 	"github.com/argoproj/argo-workflows/v4/workflow/progress"
+	wfsync "github.com/argoproj/argo-workflows/v4/workflow/sync"
 	"github.com/argoproj/argo-workflows/v4/workflow/templateresolution"
 	wfutil "github.com/argoproj/argo-workflows/v4/workflow/util"
 	"github.com/argoproj/argo-workflows/v4/workflow/validate"
@@ -264,6 +265,20 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 	if woc.execWf.Spec.Synchronization != nil {
 		acquired, wfUpdate, msg, failedLockName, err := woc.controller.syncManager.TryAcquire(ctx, woc.wf, "", woc.execWf.Spec.Synchronization)
 		if err != nil {
+			if wfsync.IsRetryableSyncError(err) || errorsutil.IsTransientErr(ctx, err) {
+				// Transient failure in the lock backend (e.g. a database
+				// serialization conflict that exhausted its in-place retries):
+				// leave the workflow pending and try again on a later
+				// reconcile instead of failing it.
+				woc.log.WithError(err).WithField("lockName", failedLockName).Warn(ctx, "Transient failure acquiring the synchronization lock, requeueing")
+				phase := woc.wf.Status.Phase
+				if phase == wfv1.WorkflowUnknown {
+					phase = wfv1.WorkflowPending
+				}
+				woc.markWorkflowPhase(ctx, phase, fmt.Sprintf("Waiting to acquire the synchronization lock. %v", err))
+				woc.requeue()
+				return
+			}
 			woc.log.WithField("lockName", failedLockName).Warn(ctx, "Failed to acquire the lock")
 			woc.markWorkflowFailed(ctx, fmt.Sprintf("Failed to acquire the synchronization lock. %s", err.Error()))
 			return
@@ -2239,6 +2254,15 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 	if processedTmpl.Synchronization != nil {
 		lockAcquired, wfUpdated, msg, failedLockName, err := woc.controller.syncManager.TryAcquire(ctx, woc.wf, woc.wf.NodeID(nodeName), processedTmpl.Synchronization)
 		if err != nil {
+			if wfsync.IsRetryableSyncError(err) || errorsutil.IsTransientErr(ctx, err) {
+				// Transient failure in the lock backend: leave the node pending
+				// and try again on a later reconcile instead of erroring it.
+				woc.requeue()
+				if node == nil {
+					node = woc.initializeExecutableNode(ctx, nodeName, wfutil.GetNodeType(processedTmpl), templateScope, processedTmpl, orgTmpl, opts.boundaryID, wfv1.NodePending, opts.nodeFlag, false, err.Error())
+				}
+				return node, nil
+			}
 			return woc.initializeNodeOrMarkError(ctx, node, nodeName, templateScope, orgTmpl, opts.boundaryID, opts.nodeFlag, err), err
 		}
 		if !lockAcquired {

workflow/controller/operator_sync_transient_test.go

@@ -0,0 +1,92 @@
+package controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo-workflows/v4/util/logging"
+	"github.com/argoproj/argo-workflows/v4/workflow/sync"
+)
+
+const wfWithWorkflowLevelSemaphore = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: hello-world-wf-level-sem
+  namespace: default
+spec:
+  entrypoint: whalesay
+  synchronization:
+    semaphores:
+      - configMapKeyRef:
+          key: workflow
+          name: my-config
+  templates:
+    - name: whalesay
+      container:
+        args:
+          - "hello world"
+        command:
+          - cowsay
+        image: "docker/whalesay:latest"
+`
+
+// The semaphore's configmap does not exist in these tests, so TryAcquire
+// returns an error. A transient error must leave the workflow/node pending
+// and requeue; only a non-transient error may fail it.
+
+func TestSyncErrorWorkflowLevel(t *testing.T) {
+	operateWorkflow := func(t *testing.T) *wfOperationCtx {
+		t.Helper()
+		cancel, controller := newController(logging.TestContext(t.Context()))
+		defer cancel()
+		ctx := logging.TestContext(t.Context())
+		controller.syncManager = sync.NewLockManager(ctx, controller.kubeclientset, controller.namespace, nil, getSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+		}, workflowExistenceFunc)
+
+		wf := wfv1.MustUnmarshalWorkflow(wfWithWorkflowLevelSemaphore)
+		wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Create(ctx, wf, metav1.CreateOptions{})
+		require.NoError(t, err)
+		woc := newWorkflowOperationCtx(ctx, wf, controller)
+		woc.operate(ctx)
+		return woc
+	}
+
+	t.Run("NonTransientErrorFailsWorkflow", func(t *testing.T) {
+		woc := operateWorkflow(t)
+		assert.Equal(t, wfv1.WorkflowFailed, woc.wf.Status.Phase)
+		assert.Contains(t, woc.wf.Status.Message, "Failed to acquire the synchronization lock")
+	})
+
+	t.Run("TransientErrorLeavesWorkflowPending", func(t *testing.T) {
+		t.Setenv("TRANSIENT_ERROR_PATTERN", "failed to initialize semaphore")
+		woc := operateWorkflow(t)
+		assert.Equal(t, wfv1.WorkflowPending, woc.wf.Status.Phase)
+		assert.Contains(t, woc.wf.Status.Message, "Waiting to acquire the synchronization lock")
+	})
+}
+
+func TestSyncTransientErrorNodeLevel(t *testing.T) {
+	t.Setenv("TRANSIENT_ERROR_PATTERN", "failed to initialize semaphore")
+	cancel, controller := newController(logging.TestContext(t.Context()))
+	defer cancel()
+	ctx := logging.TestContext(t.Context())
+	controller.syncManager = sync.NewLockManager(ctx, controller.kubeclientset, controller.namespace, nil, getSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	}, workflowExistenceFunc)
+
+	wf := wfv1.MustUnmarshalWorkflow(wfWithSemaphore)
+	wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Create(ctx, wf, metav1.CreateOptions{})
+	require.NoError(t, err)
+	woc := newWorkflowOperationCtx(ctx, wf, controller)
+	woc.operate(ctx)
+
+	assert.NotEqual(t, wfv1.WorkflowFailed, woc.wf.Status.Phase)
+	require.NotEmpty(t, woc.wf.Status.Nodes)
+	for _, node := range woc.wf.Status.Nodes {
+		assert.Equal(t, wfv1.NodePending, node.Phase, "a transient sync error must leave the node pending, not errored")
+	}
+}

workflow/sync/sync_manager.go

@@ -18,6 +18,7 @@ import (
 	"github.com/argoproj/argo-workflows/v4/config"
 	wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v4/util/logging"
+	"github.com/argoproj/argo-workflows/v4/util/sqldb"
 	syncdb "github.com/argoproj/argo-workflows/v4/util/sync/db"
 )
 
@@ -465,18 +466,16 @@ func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName s
 		var updated bool
 		var already bool
 		var msg string
-		// Backoff bounds: sm.lock is held for the whole loop, so cap each sleep
-		// modestly. Jitter prevents a fleet of replicas from retrying in lockstep
-		// after a shared conflict burst.
-		backoff := wait.Backoff{
-			Steps:    5,
-			Duration: 10 * time.Millisecond,
-			Factor:   2.0,
-			Jitter:   0.5,
-			Cap:      600 * time.Millisecond,
-		}
+		backoff := dbRetryBackoff
+		// tryAcquireImpl mutates wf.Status.Synchronization in memory before the
+		// transaction commits. Snapshot it and roll each failed attempt back, so
+		// that attempts are independent and an error return leaves the caller's
+		// status exactly as it was - otherwise an abort leaves a Holding entry
+		// for a row the database rolled back, which would be persisted and then
+		// failed as a stale hold on the next controller restart.
+		syncStatus := wf.Status.Synchronization.DeepCopy()
 		attempt := 0
-		err = retry.OnError(backoff, isRetryableSyncError, func() error {
+		err = retry.OnError(backoff, IsRetryableSyncError, func() error {
 			attempt++
 			sm.log.WithFields(logging.Fields{
 				"holderKey": holderKey,
@@ -489,11 +488,12 @@ func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName s
 				return implErr
 			}, &sql.TxOptions{Isolation: sql.LevelSerializable, ReadOnly: false})
 			if txErr != nil {
+				wf.Status.Synchronization = syncStatus.DeepCopy()
 				sm.log.WithFields(logging.Fields{
 					"holderKey": holderKey,
 					"attempt":   attempt,
 					"error":     txErr,
-					"retryable": isRetryableSyncError(txErr),
+					"retryable": IsRetryableSyncError(txErr),
 				}).Info(ctx, "TryAcquire - transaction failed")
 			}
 			return txErr
@@ -506,14 +506,32 @@ func (sm *Manager) TryAcquire(ctx context.Context, wf *wfv1.Workflow, nodeName s
 	return sm.tryAcquireImpl(ctx, wf, nil, holderKey, failedLockName, syncItems, lockKeys)
 }
 
-// isRetryableSyncError reports whether a TryAcquire transaction failure should
-// be retried. Matches PostgreSQL SERIALIZABLE conflict (40001), deadlock
-// (40P01), and explicit rollback messages by substring against the driver's
-// text.
-func isRetryableSyncError(err error) bool {
+// dbRetryBackoff bounds the in-place retries of the TryAcquire transaction.
+// sm.lock is held for the whole loop, so each sleep is capped modestly. Jitter
+// prevents a fleet of replicas from retrying in lockstep after a shared
+// conflict burst. An error that survives these retries is surfaced to the
+// caller, which classifies it (sqldb.IsSerializationConflict, IsTransientErr)
+// and requeues the workflow rather than failing it.
+var dbRetryBackoff = wait.Backoff{
+	Steps:    5,
+	Duration: 10 * time.Millisecond,
+	Factor:   2.0,
+	Jitter:   0.5,
+	Cap:      600 * time.Millisecond,
+}
+
+// IsRetryableSyncError reports whether a TryAcquire transaction failure should
+// be retried. A conflict reported by the database driver itself is always
+// retryable; the substring match against the driver's text (serialization
+// conflicts, deadlocks, explicit rollbacks) remains as a fallback for errors
+// that arrive with the driver type stringified away by an intermediate layer.
+func IsRetryableSyncError(err error) bool {
 	if err == nil {
 		return false
 	}
+	if sqldb.IsSerializationConflict(err) {
+		return true
+	}
 	s := strings.ToLower(err.Error())
 	return strings.Contains(s, "serialization") ||
 		strings.Contains(s, "dependencies") ||