fix: address semaphore/mutex unsoundness for Initalize (cherry-pick #16160 for 4.0) (#16252)

Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: Alan Clucas <alan@clucas.org>
Co-authored-by: Isitha Subasinghe <isitha@pipekit.io>

Author: Joibel

workflow/controller/controller.go

@@ -463,7 +463,24 @@ func (wfc *WorkflowController) initManagers(ctx context.Context) error {
 		return err
 	}
 
-	wfc.syncManager.Initialize(ctx, wfList.Items)
+	// A non-nil error means a recorded lock holder could not be re-established
+	// (undecodable lock name, or an unavailable database session). This is fatal
+	// by design: we fail closed rather than risk a silent double-acquire.
+	staleHolds, err := wfc.syncManager.Initialize(ctx, wfList.Items)
+	if err != nil {
+		return err
+	}
+	// Stale holds are workflows whose recorded hold on a database-backed lock
+	// the database no longer has (e.g. expired while the controller was down,
+	// possibly acquired by someone else since). The database is the source of
+	// truth, so these workflows must not keep running on a hold it does not
+	// back: fail them; persistUpdates releases any locks they still hold.
+	for _, stale := range staleHolds {
+		woc := newWorkflowOperationCtx(ctx, stale.WF, wfc)
+		wocCtx := logging.WithLogger(ctx, woc.log)
+		woc.markWorkflowFailed(wocCtx, fmt.Sprintf("Failed to re-establish synchronization lock at controller startup: %s", stale.Reason))
+		woc.persistUpdates(wocCtx)
+	}
 
 	if err := wfc.throttler.Init(wfList.Items); err != nil {
 		return err

workflow/sync/common.go

@@ -7,6 +7,20 @@ import (
 
 type semaphore interface {
 	acquire(ctx context.Context, holderKey string, tx *transaction) (bool, error)
+	// reacquire re-establishes a recorded holder at controller startup.
+	//
+	// For an in-memory lock it force-registers the holder, ignoring the current
+	// limit, so the in-memory count reflects persisted reality even when recorded
+	// holders exceed a (since lowered) limit - new acquisitions then correctly
+	// wait until the count drains below the limit, rather than dropping a holder
+	// (a double-acquire) or poisoning the lock over a routine limit change.
+	//
+	// For a database-backed lock the database is the single source of truth:
+	// reacquire mutates nothing and only asserts the recorded hold still exists
+	// there. An error means the hold could not be verified - either the held row
+	// is gone (e.g. expired while the controller was down) or the database could
+	// not be queried - and the caller fails the holding workflow.
+	reacquire(ctx context.Context, holderKey string, tx *transaction) error
 	checkAcquire(ctx context.Context, holderKey string, tx *transaction) (bool, bool, string)
 	tryAcquire(ctx context.Context, holderKey string, tx *transaction) (bool, string, error)
 	release(ctx context.Context, key string) bool

workflow/sync/database_semaphore.go

@@ -377,6 +377,25 @@ func (s *databaseSemaphore) acquire(ctx context.Context, holderKey string, tx *t
 	return false, nil
 }
 
+// reacquire asserts at startup that the recorded holder still holds this lock
+// in the database. The database is the single source of truth for a
+// database-backed lock: the held row is durable and survives the controller
+// restart, so nothing is inserted or mutated here. A missing row means the
+// hold no longer exists - e.g. it was expired by ExpireInactiveLocks while the
+// controller was down and may since have been acquired by another holder - so
+// the workflow's recorded hold is stale and the caller fails the workflow
+// rather than resurrect a hold the database does not back.
+func (s *databaseSemaphore) reacquire(ctx context.Context, holderKey string, tx *transaction) error {
+	holders, err := s.currentHoldersSession(ctx, *tx.db)
+	if err != nil {
+		return fmt.Errorf("could not verify hold on %s for %s: %w", s.longDBKey(), holderKey, err)
+	}
+	if !slices.Contains(holders, holderKey) {
+		return fmt.Errorf("hold on %s for %s is not present in the database", s.longDBKey(), holderKey)
+	}
+	return nil
+}
+
 func (s *databaseSemaphore) tryAcquire(ctx context.Context, holderKey string, tx *transaction) (bool, string, error) {
 	logger := s.logger(ctx)
 	acq, already, msg := s.checkAcquire(ctx, holderKey, tx)

workflow/sync/mutex_test.go

@@ -122,7 +122,9 @@ func TestMutexLock(t *testing.T) {
 
 		wfList, err := wfclientset.ArgoprojV1alpha1().Workflows("default").List(ctx, metav1.ListOptions{})
 		require.NoError(t, err)
-		syncManager.Initialize(ctx, wfList.Items)
+		staleHolds, err := syncManager.Initialize(ctx, wfList.Items)
+		require.NoError(t, err)
+		require.Empty(t, staleHolds)
 		assert.Len(t, syncManager.syncLockMap, 1)
 	})
 	t.Run("WfLevelMutexAcquireAndRelease", func(t *testing.T) {

workflow/sync/poison.go

@@ -0,0 +1,86 @@
+package sync
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// poisonedLock is a sentinel lock installed into the Manager's syncLockMap when,
+// during Initialize, the controller cannot re-establish a holder that a Running
+// workflow's status claims to hold.
+//
+// The soundness invariant is: if a Workflow's status records that it is holding
+// a lock, the in-memory lock map must reflect that hold after Initialize.
+// Otherwise a racing workflow's TryAcquire would find the lock absent, create a
+// fresh one, and acquire a lock that is - per persisted state - already held.
+// For a mutex that means two workflows running concurrently under the same
+// mutex.
+//
+// Rather than silently dropping the holder (the previous behaviour), we install
+// this lock, which refuses every acquire and reports a poisoned-state message.
+// That message surfaces on the waiting node's synchronization status, marking
+// the node/workflow as blocked by a poisoned lock so an operator can intervene.
+//
+// The poison is in-memory only and is cleared on the next controller restart,
+// at which point Initialize re-evaluates: if the offending workflow is no longer
+// Running the lock is recreated clean; if it is still Running and still
+// unresolvable, it is poisoned again.
+type poisonedLock struct {
+	name   string
+	reason string
+}
+
+var _ semaphore = &poisonedLock{}
+
+func newPoisonedLock(name, reason string) *poisonedLock {
+	return &poisonedLock{name: name, reason: reason}
+}
+
+func (p *poisonedLock) message() string {
+	return fmt.Sprintf("lock %s is in a poisoned state: %s; manual intervention required", p.name, p.reason)
+}
+
+func (p *poisonedLock) acquire(_ context.Context, _ string, _ *transaction) (bool, error) {
+	return false, nil
+}
+
+// reacquire is a no-op: a poisoned lock refuses all holds until restart. It
+// returns nil because the poison already protects the recorded hold; failing
+// the holding workflow on top of that would punish it for an unrelated
+// holder's poisoning.
+func (p *poisonedLock) reacquire(_ context.Context, _ string, _ *transaction) error {
+	return nil
+}
+
+func (p *poisonedLock) checkAcquire(_ context.Context, _ string, _ *transaction) (bool, bool, string) {
+	return false, false, p.message()
+}
+
+func (p *poisonedLock) tryAcquire(_ context.Context, _ string, _ *transaction) (bool, string, error) {
+	return false, p.message(), nil
+}
+
+func (p *poisonedLock) release(_ context.Context, _ string) bool { return false }
+
+func (p *poisonedLock) getName() string { return p.name }
+
+func (p *poisonedLock) addToQueue(_ context.Context, _ string, _ int32, _ time.Time) error {
+	return nil
+}
+
+func (p *poisonedLock) removeFromQueue(_ context.Context, _ string) error { return nil }
+
+func (p *poisonedLock) getCurrentHolders(_ context.Context) ([]string, error) { return nil, nil }
+
+func (p *poisonedLock) getCurrentPending(_ context.Context) ([]string, error) { return nil, nil }
+
+func (p *poisonedLock) getLimit(_ context.Context) int { return 0 }
+
+func (p *poisonedLock) probeWaiting(_ context.Context) {}
+
+// lock returns true so that tryAcquireImpl proceeds to checkAcquire, which
+// returns the poisoned-state message rather than a generic "failed to lock()".
+func (p *poisonedLock) lock(_ context.Context) bool { return true }
+
+func (p *poisonedLock) unlock(_ context.Context) {}

workflow/sync/semaphore.go

@@ -175,6 +175,24 @@ func (s *prioritySemaphore) acquire(_ context.Context, holderKey string, _ *tran
 	return false, nil
 }
 
+// reacquire re-establishes a recorded holder at startup, ignoring the limit. It
+// always registers the holder, even when the recorded holders already exceed the
+// current limit (e.g. the limit was lowered while held). The weighted semaphore
+// is capped at the limit, so a slot is only taken when one is free; the excess is
+// tracked solely in lockHolder, exactly as a downward resize leaves it. release()
+// already tolerates len(lockHolder) > limit and only frees a weighted slot once
+// the count drops below the limit, so new acquisitions wait until every recorded
+// holder has drained. It never fails: the in-memory map is the source of truth
+// here, so registering the holder is always possible.
+func (s *prioritySemaphore) reacquire(_ context.Context, holderKey string, _ *transaction) error {
+	if _, ok := s.lockHolder[holderKey]; ok {
+		return nil
+	}
+	s.semaphore.TryAcquire(1) // best effort: take a slot if one is free
+	s.lockHolder[holderKey] = true
+	return nil
+}
+
 func isSameWorkflowNodeKeys(firstKey, secondKey string) bool {
 	firstItems := strings.Split(firstKey, "/")
 	secondItems := strings.Split(secondKey, "/")