fix: Immediately release locks by pending workflows that are shutting down. Fixes #10733

workflow/controller/operator.go

@@ -246,13 +246,15 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 		}
 		woc.updated = wfUpdate
 		if !acquired {
-			woc.log.Warn("Workflow processing has been postponed due to concurrency limit")
-			phase := woc.wf.Status.Phase
-			if phase == wfv1.WorkflowUnknown {
-				phase = wfv1.WorkflowPending
+			if !woc.releaseLocksForPendingTerminatingWfs(ctx) {
+				woc.log.Warn("Workflow processing has been postponed due to concurrency limit")
+				phase := woc.wf.Status.Phase
+				if phase == wfv1.WorkflowUnknown {
+					phase = wfv1.WorkflowPending
+				}
+				woc.markWorkflowPhase(ctx, phase, msg)
+				return
 			}
-			woc.markWorkflowPhase(ctx, phase, msg)
-			return
 		}
 	}
 
@@ -490,6 +492,17 @@ func (woc *wfOperationCtx) operate(ctx context.Context) {
 	}
 }
 
+func (woc *wfOperationCtx) releaseLocksForPendingTerminatingWfs(ctx context.Context) bool {
+	if sdStrategy := woc.GetShutdownStrategy(); sdStrategy == wfv1.ShutdownStrategyTerminate && woc.execWf.Status.Phase == wfv1.WorkflowPending {
+		if woc.controller.syncManager.ReleaseAll(woc.execWf) {
+			woc.log.WithFields(log.Fields{"key": woc.execWf.Name}).Info("Released all locks since this pending workflow is terminating")
+			woc.markWorkflowSuccess(ctx)
+			return true
+		}
+	}
+	return false
+}
+
 // set Labels and Annotations for the Workflow
 // Also, since we're setting Labels and Annotations we need to find any
 // parameters formatted as "workflow.labels.<param>" or "workflow.annotations.<param>"
@@ -1868,13 +1881,15 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 			if node == nil {
 				node = woc.initializeExecutableNode(nodeName, wfutil.GetNodeType(processedTmpl), templateScope, processedTmpl, orgTmpl, opts.boundaryID, wfv1.NodePending, msg)
 			}
-			lockName, err := argosync.GetLockName(processedTmpl.Synchronization, woc.wf.Namespace)
-			if err != nil {
-				// If an error were to be returned here, it would have been caught by TryAcquire. If it didn't, then it is
-				// unexpected behavior and is a bug.
-				panic("bug: GetLockName should not return an error after a call to TryAcquire")
+			if !woc.releaseLocksForPendingTerminatingWfs(ctx) {
+				lockName, err := argosync.GetLockName(processedTmpl.Synchronization, woc.wf.Namespace)
+				if err != nil {
+					// If an error were to be returned here, it would have been caught by TryAcquire. If it didn't, then it is
+					// unexpected behavior and is a bug.
+					panic("bug: GetLockName should not return an error after a call to TryAcquire")
+				}
+				return woc.markNodeWaitingForLock(node.Name, lockName.EncodeName()), nil
 			}
-			return woc.markNodeWaitingForLock(node.Name, lockName.EncodeName()), nil
 		} else {
 			woc.log.Infof("Node %s acquired synchronization lock", nodeName)
 			if node != nil {