Merge pull request #1799 from gruntwork-io/james/oss-3398-concurrency-races

fix(opa): deduplicate concurrent policy downloads via singleflight

Author: james00012

go.mod

@@ -99,6 +99,7 @@ require (
 	github.com/lib/pq v1.10.9
 	github.com/microsoft/go-mssqldb v1.9.8
 	github.com/slack-go/slack v0.15.0
+	golang.org/x/sync v0.20.0
 	google.golang.org/grpc v1.80.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/v3 v3.5.2
@@ -223,7 +224,6 @@ require (
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/mod v0.33.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/term v0.41.0 // indirect
 	golang.org/x/text v0.35.0 // indirect

modules/opa/download_policy.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	getter "github.com/hashicorp/go-getter/v2"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/gruntwork-io/terratest/modules/logger"
 	"github.com/gruntwork-io/terratest/modules/testing"
@@ -16,6 +17,10 @@ import (
 var (
 	// A map that maps the go-getter base URL to the temporary directory where it is downloaded.
 	policyDirCache sync.Map
+
+	// downloadGroup deduplicates concurrent downloads for the same baseDir so that N parallel callers requesting the
+	// same rulePath result in a single underlying download rather than N separate downloads racing into N temp dirs.
+	downloadGroup singleflight.Group
 )
 
 // DownloadPolicyE takes in a rule path written in go-getter syntax and downloads it to a temporary directory so that it
@@ -52,13 +57,38 @@ func DownloadPolicyE(t testing.TestingT, rulePath string) (string, error) {
 	// First, check if we had already downloaded the source and it is in our cache.
 	baseDir, subDir := getter.SourceDirSubdir(rulePath)
 
-	downloadPath, hasDownloaded := policyDirCache.Load(baseDir)
-	if hasDownloaded {
+	if downloadPath, hasDownloaded := policyDirCache.Load(baseDir); hasDownloaded {
 		logger.Default.Logf(t, "Previously downloaded %s: returning cached path", baseDir)
 		return filepath.Join(downloadPath.(string), subDir), nil
 	}
 
-	// Not downloaded, so use go-getter to download the remote source to a temp dir.
+	// Cache miss. Use singleflight to ensure that only one goroutine actually performs the download for a given
+	// baseDir; any concurrent callers block on the same call and reuse its result.
+	v, err, _ := downloadGroup.Do(baseDir, func() (any, error) {
+		// Re-check the cache in case another goroutine populated it while we were waiting to enter the singleflight.
+		if downloadPath, hasDownloaded := policyDirCache.Load(baseDir); hasDownloaded {
+			return downloadPath.(string), nil
+		}
+
+		tempDir, err := downloadPolicyToTempDir(t, rulePath, baseDir)
+		if err != nil {
+			return "", err
+		}
+
+		policyDirCache.Store(baseDir, tempDir)
+
+		return tempDir, nil
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return filepath.Join(v.(string), subDir), nil
+}
+
+// downloadPolicyToTempDir downloads the given baseDir using go-getter into a fresh temp directory and returns the path
+// to the directory containing the downloaded source.
+func downloadPolicyToTempDir(t testing.TestingT, rulePath, baseDir string) (string, error) {
 	tempDir, err := os.MkdirTemp("", "terratest-opa-policy-*")
 	if err != nil {
 		return "", fmt.Errorf("creating temp directory for policy download: %w", err)
@@ -74,7 +104,5 @@ func DownloadPolicyE(t testing.TestingT, rulePath string) (string, error) {
 		return "", fmt.Errorf("downloading policy from %s: %w", baseDir, err)
 	}
 
-	policyDirCache.Store(baseDir, tempDir)
-
-	return filepath.Join(tempDir, subDir), nil
+	return tempDir, nil
 }

modules/opa/download_policy_test.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -64,6 +65,62 @@ func TestDownloadPolicyDownloadsRemote(t *testing.T) {
 	assert.Equal(t, localContents, remoteContents)
 }
 
+// TestDownloadPolicyDeduplicatesConcurrentDownloads makes sure concurrent calls for the same rulePath collapse to a
+// single cache entry rather than racing into separate temp directories.
+//
+//nolint:paralleltest // go-getter's Client.configure has an internal race on its global Getters map, so we cannot run alongside other go-getter tests.
+func TestDownloadPolicyDeduplicatesConcurrentDownloads(t *testing.T) {
+	baseDir := "git::https://github.com/gruntwork-io/terratest.git?ref=v0.50.0"
+	remotePath := "git::https://github.com/gruntwork-io/terratest.git//examples/terraform-opa-example/policy/enforce_source.rego?ref=v0.50.0"
+
+	defer func() {
+		if cached, ok := opa.PolicyDirCache.Load(baseDir); ok {
+			downloadPath := cached.(string)
+			if strings.HasSuffix(downloadPath, "/getter") {
+				downloadPath = filepath.Dir(downloadPath)
+			}
+
+			os.RemoveAll(downloadPath)
+		}
+	}()
+
+	tempDirGlob := filepath.Join(os.TempDir(), "terratest-opa-policy-*")
+	before, _ := filepath.Glob(tempDirGlob)
+
+	const numGoroutines = 5
+
+	var wg sync.WaitGroup
+
+	results := make([]string, numGoroutines)
+	errs := make([]error, numGoroutines)
+
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+
+		go func(idx int) {
+			defer wg.Done()
+
+			path, err := opa.DownloadPolicyE(t, remotePath)
+
+			errs[idx] = err
+			results[idx] = path
+		}(i)
+	}
+
+	wg.Wait()
+
+	for i := 0; i < numGoroutines; i++ {
+		require.NoError(t, errs[i])
+	}
+
+	for i := 1; i < numGoroutines; i++ {
+		assert.Equal(t, results[0], results[i])
+	}
+
+	after, _ := filepath.Glob(tempDirGlob)
+	assert.Len(t, after, len(before)+1, "expected exactly one new temp dir; dedup may have failed")
+}
+
 // TestDownloadPolicyReusesCachedDir makes sure the DownloadPolicyE function uses the cache if it has already downloaded
 // an existing base path.
 func TestDownloadPolicyReusesCachedDir(t *testing.T) {