Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 21 additions & 3 deletions internal/policy/source/source.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ type PolicyUrl struct {
Url string
Kind PolicyType
pinOnce sync.Once
urlMu sync.RWMutex
}

// downloadCache is a concurrent map used to cache downloaded files.
Expand Down Expand Up @@ -171,7 +172,7 @@ func (p *PolicyUrl) GetPolicy(ctx context.Context, workDir string, showMsg bool)
if trace.IsEnabled() {
region := trace.StartRegion(ctx, "ec:get-policy")
defer region.End()
trace.Logf(ctx, "", "policy=%q", p.Url)
trace.Logf(ctx, "", "policy=%q", p.url())
}

dl := func(source string, dest string) (metadata.Metadata, error) {
Expand All @@ -189,8 +190,19 @@ func (p *PolicyUrl) GetPolicy(ctx context.Context, workDir string, showMsg bool)

var pinErr error
p.pinOnce.Do(func() {
originalUrl := p.url()
p.urlMu.Lock()
p.Url, pinErr = metadata.GetPinnedURL(p.Url)
log.Debug("Pinned URL: ", p.Url)
p.urlMu.Unlock()
log.Debug("Pinned URL: ", p.url())
// Register the cached download under the pinned URL too, so
// subsequent lookups (which use the now-mutated p.Url) still hit.
pinnedUrl := p.url()
if pinErr == nil && pinnedUrl != originalUrl {
if cached, ok := downloadCache.Load(originalUrl); ok {
downloadCache.LoadOrStore(pinnedUrl, cached)
}
}
})
if pinErr != nil {
return "", pinErr
Expand All @@ -199,10 +211,16 @@ func (p *PolicyUrl) GetPolicy(ctx context.Context, workDir string, showMsg bool)
return dest, nil
}

func (p *PolicyUrl) PolicyUrl() string {
func (p *PolicyUrl) url() string {
p.urlMu.RLock()
defer p.urlMu.RUnlock()
return p.Url
}

func (p *PolicyUrl) PolicyUrl() string {
return p.url()
}

func (p *PolicyUrl) Subdir() string {
// Be lazy and assume the kind value is the same as the subdirectory we want
return string(p.Kind)
Expand Down
54 changes: 48 additions & 6 deletions internal/policy/source/source_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import (
"path"
"path/filepath"
"regexp"
"sync"
"testing"

ecc "github.com/conforma/crds/api/v1alpha1"
Expand Down Expand Up @@ -163,7 +162,7 @@ func TestInlineDataSource(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
// Clear download cache for each test
t.Cleanup(func() {
downloadCache = sync.Map{}
ClearDownloadCache()
})

s := InlineData(tt.inputData)
Expand Down Expand Up @@ -276,7 +275,7 @@ func TestInlineDataGetPolicy(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
// Clear download cache for each test
t.Cleanup(func() {
downloadCache = sync.Map{}
ClearDownloadCache()
})

s := InlineData(tt.inputData)
Expand Down Expand Up @@ -539,7 +538,7 @@ func (m mockPolicySource) Type() PolicyType {
func TestGetPolicyThroughCache(t *testing.T) {
test := func(t *testing.T, fs afero.Fs, expectedDownloads int) {
t.Cleanup(func() {
downloadCache = sync.Map{}
ClearDownloadCache()
})

ctx := utils.WithFS(context.Background(), fs)
Expand Down Expand Up @@ -604,7 +603,7 @@ func TestGetPolicyThroughCache(t *testing.T) {
// causing Rego compile issue
func TestDownloadCacheWorkdirMismatch(t *testing.T) {
t.Cleanup(func() {
downloadCache = sync.Map{}
ClearDownloadCache()
})
tmp := t.TempDir()

Expand Down Expand Up @@ -634,12 +633,55 @@ func TestDownloadCacheWorkdirMismatch(t *testing.T) {
assert.Equal(t, destination1, destination2)
}

// TestGetPolicyPinnedURLCacheConsistency verifies that URL pinning doesn't
// cause duplicate policy directories. After the first GetPolicy call, the URL
// is pinned (e.g. github.com/org/repo -> git::github.com/org/repo?ref=abc123),
// which changes the cache key. Without the fix, the second call would miss the
// cache and re-download into a new directory, causing OPA duplicate package errors.
Comment thread
simonbaird marked this conversation as resolved.
func TestGetPolicyPinnedURLCacheConsistency(t *testing.T) {
t.Cleanup(ClearDownloadCache)

workDir := t.TempDir()
policyDir := filepath.Join(workDir, "policy")
require.NoError(t, os.MkdirAll(policyDir, 0o755))

originalUrl := "github.com/org/repo//policy"
p := &PolicyUrl{Url: originalUrl, Kind: PolicyKind}

dl := &mockDownloader{}
dl.On("Download", mock.Anything, mock.Anything, originalUrl, false).
Run(func(args mock.Arguments) {
dest := args.String(1)
require.NoError(t, os.MkdirAll(dest, 0o755))
}).
Return(&gitMetadata.GitMetadata{LatestCommit: "abc123def456"}, nil)

ctx := usingDownloader(context.TODO(), dl)

// First call: downloads and pins URL
dest1, err := p.GetPolicy(ctx, workDir, false)
require.NoError(t, err)
assert.NotEqual(t, originalUrl, p.Url, "URL should be pinned after first call")

// Second call: should hit cache despite pinned URL
dest2, err := p.GetPolicy(ctx, workDir, false)
require.NoError(t, err)
assert.Equal(t, dest1, dest2, "second call should return same directory")

// Verify only one directory exists under policy/
entries, err := os.ReadDir(policyDir)
require.NoError(t, err)
assert.Len(t, entries, 1, "only one policy directory should exist, not a duplicate from pinned URL")

dl.AssertNumberOfCalls(t, "Download", 1)
}

// 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{}
ClearDownloadCache()
})

tmp := t.TempDir()
Expand Down
2 changes: 2 additions & 0 deletions internal/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/conforma/cli/internal/evaluation_target/input"
"github.com/conforma/cli/internal/evaluator"
"github.com/conforma/cli/internal/policy"
"github.com/conforma/cli/internal/version"
)

type Config struct {
Expand Down Expand Up @@ -69,6 +70,7 @@ func (s *Server) Start(ctx context.Context) error {
"address": s.cfg.Address,
"port": s.cfg.Port,
"sources": len(s.cfg.Policy.Spec().Sources),
"version": version.Version,
}).Info("Starting server")

log.Info("Loading policy sources...")
Expand Down
Loading