Merge pull request #2964 from Acepresso/file-exists-EC-1406

Fix concurrent symlink creation race condition

Author: dheerajodha

internal/policy/source/source.go

@@ -78,9 +78,13 @@ type PolicyUrl struct {
 // downloadCache is a concurrent map used to cache downloaded files.
 var downloadCache sync.Map
 
+// symlinkMutexes provides per-destination synchronization for symlink creation
+var symlinkMutexes sync.Map
+
 // ClearDownloadCache clears the download cache. This is primarily used for testing.
 func ClearDownloadCache() {
 	downloadCache = sync.Map{}
+	symlinkMutexes = sync.Map{}
 }
 
 type cacheContent struct {
@@ -111,13 +115,24 @@ func getPolicyThroughCache(ctx context.Context, s PolicySource, workDir string,
 	}
 
 	fs := utils.FS(ctx)
-	if _, err := fs.Stat(dest); err == nil {
-		return dest, c.metadata, nil
-	}
 
 	// If the destination directory is different from the source directory, we
 	// need to symlink the source directory to the destination directory.
+	// Use synchronization to prevent race conditions when multiple goroutines
+	// try to create the same symlink simultaneously.
 	if filepath.Dir(dest) != filepath.Dir(d) {
+		// Get or create a mutex for this specific destination
+		mutexValue, _ := symlinkMutexes.LoadOrStore(dest, &sync.Mutex{})
+		mutex := mutexValue.(*sync.Mutex)
+
+		mutex.Lock()
+		defer mutex.Unlock()
+
+		// Check again if the destination exists after acquiring the lock
+		if _, err := fs.Stat(dest); err == nil {
+			return dest, c.metadata, nil
+		}
+
 		base := filepath.Dir(dest)
 		if err := fs.MkdirAll(base, 0755); err != nil {
 			return "", nil, err
@@ -136,6 +151,12 @@ func getPolicyThroughCache(ctx context.Context, s PolicySource, workDir string,
 			logMetadata(m)
 			return dest, m, err
 		}
+	} else {
+		// If dest and d are in the same directory, no symlink needed,
+		// but still check if dest exists
+		if _, err := fs.Stat(dest); err == nil {
+			return dest, c.metadata, nil
+		}
 	}
 
 	if c.metadata != nil {

internal/policy/source/source_test.go

@@ -21,6 +21,7 @@ package source
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 	"path"
 	"path/filepath"
@@ -633,6 +634,84 @@ func TestDownloadCacheWorkdirMismatch(t *testing.T) {
 	assert.Equal(t, destination1, destination2)
 }
 
+// TestConcurrentPolicyCachingRaceCondition reproduces the "file exists" error
+// that occurs when multiple workers simultaneously try to create symlinks from
+// cached policy downloads to their individual work directories
+func TestConcurrentPolicyCachingRaceCondition(t *testing.T) {
+	t.Cleanup(func() {
+		downloadCache = sync.Map{}
+	})
+
+	tmp := t.TempDir()
+
+	source := &mockPolicySource{&mock.Mock{}}
+	source.On("PolicyUrl").Return("policy-url")
+	source.On("Subdir").Return("subdir")
+
+	// Simulate a pre-cached policy download in a shared location
+	// This represents the first worker that successfully downloaded the policy
+	sharedCacheDir := filepath.Join(tmp, "shared-cache")
+	cachedPolicyPath := uniqueDestination(sharedCacheDir, "subdir", source.PolicyUrl())
+	require.NoError(t, os.MkdirAll(cachedPolicyPath, 0755))
+
+	// Create test policy files
+	policyFile := filepath.Join(cachedPolicyPath, "policy.rego")
+	require.NoError(t, os.WriteFile(policyFile, []byte("package test"), 0600))
+
+	// Pre-populate the cache with the shared policy location
+	downloadCache.Store("policy-url", func() (string, cacheContent) {
+		return cachedPolicyPath, cacheContent{}
+	})
+
+	// ALL workers use the same work directory - this forces them to compete
+	// for the exact same symlink destination path, triggering the race condition
+	sharedWorkDir := filepath.Join(tmp, "shared-worker-dir")
+
+	// Setup concurrent workers that will try to create the SAME symlink
+	numWorkers := 50
+	results := make(chan error, numWorkers)
+
+	// Synchronization barrier to maximize race condition probability
+	startSignal := make(chan struct{})
+
+	for i := 0; i < numWorkers; i++ {
+		go func(workerID int) {
+			// Wait for all workers to be ready
+			<-startSignal
+
+			// All workers use the SAME work directory - this creates the race condition
+			_, _, err := getPolicyThroughCache(
+				context.Background(),
+				source,
+				sharedWorkDir, // Same workDir for all workers = same destination path
+				func(sourceURL, destDir string) (metadata.Metadata, error) {
+					// Should not be called since policy is already cached
+					return nil, fmt.Errorf("unexpected download call for worker %d", workerID)
+				},
+			)
+
+			results <- err
+		}(i)
+	}
+
+	// Release all workers simultaneously to trigger race condition
+	close(startSignal)
+
+	// Collect errors
+	var errors []error
+	for i := 0; i < numWorkers; i++ {
+		if err := <-results; err != nil {
+			errors = append(errors, err)
+		}
+	}
+
+	// The test should succeed (no errors) when the race condition is fixed
+	// Any errors indicate a concurrency bug
+	if len(errors) > 0 {
+		t.Errorf("Concurrent policy caching failed with %d errors: %v", len(errors), errors)
+	}
+}
+
 func TestLogMetadata(t *testing.T) {
 	tests := []struct {
 		name     string