feat(policies): parallelize policy evaluations with bounded concurrency

Use errgroup to evaluate policy attachments concurrently within
VerifyStatement and VerifyMaterial, bounded by max(NumCPU*2, 10).
This reduces evaluation time from O(n*t) to O(t) where n is the
number of policies and t is the slowest single evaluation.

Concurrency safety fixes:
- Move remotePolicyCache/remoteGroupCache mutexes from instance-level
  to package-level to prevent data races under concurrent access
- Use singleflight to coalesce concurrent gRPC fetches for the same
  policy/group ref, ensuring exactly one fetch per key
- Move extism.SetLogLevel() to sync.Once to guarantee it is called
  exactly once across all concurrent WASM engine instances

Adds race detector tests covering:
- Concurrent VerifyStatement from 10 goroutines
- Concurrent VerifyMaterial from 10 goroutines
- WithMaxConcurrency option validation
- Sequential (maxConcurrency=1) vs parallel result equivalence
- errgroup cancellation on policy load failure

Fixes #2899

Signed-off-by: Vibhav Bobade <vibhav.bobde@gmail.com>

Author: waveywaves

pkg/policies/concurrency_test.go

@@ -0,0 +1,236 @@
+//
+// Copyright 2024-2026 The Chainloop Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package policies
+
+import (
+	"context"
+	"os"
+	"sync"
+	"testing"
+
+	v12 "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1"
+	v1 "github.com/chainloop-dev/chainloop/pkg/attestation/crafter/api/attestation/v1"
+	intoto "github.com/in-toto/attestation/go/v1"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+// TestConcurrentVerifyStatement runs VerifyStatement from multiple goroutines
+// to exercise errgroup parallelism and catch race conditions.
+// Run with: go test -race -count=1 -run TestConcurrentVerifyStatement ./pkg/policies/...
+func TestConcurrentVerifyStatement(t *testing.T) {
+	logger := zerolog.Nop()
+
+	schema := &v12.CraftingSchema{
+		Policies: &v12.Policies{
+			Attestation: []*v12.PolicyAttachment{
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}},
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/materials.yaml"}},
+			},
+		},
+	}
+
+	statement := loadStatementForTest(t, "testdata/statement.json")
+
+	pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger)
+
+	// Run multiple VerifyStatement calls concurrently
+	const goroutines = 10
+	var wg sync.WaitGroup
+	errs := make([]error, goroutines)
+	results := make([][]*v1.PolicyEvaluation, goroutines)
+
+	for i := range goroutines {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			res, err := pv.VerifyStatement(context.Background(), statement)
+			errs[i] = err
+			results[i] = res
+		}()
+	}
+
+	wg.Wait()
+
+	// All calls should succeed
+	for i := range goroutines {
+		assert.NoError(t, errs[i], "goroutine %d failed", i)
+	}
+
+	// All calls should return the same number of evaluations
+	for i := 1; i < goroutines; i++ {
+		assert.Equal(t, len(results[0]), len(results[i]),
+			"goroutine %d returned different number of evaluations", i)
+	}
+}
+
+// TestConcurrentVerifyMaterial runs VerifyMaterial from multiple goroutines.
+func TestConcurrentVerifyMaterial(t *testing.T) {
+	logger := zerolog.Nop()
+
+	schema := &v12.CraftingSchema{
+		Policies: &v12.Policies{
+			Materials: []*v12.PolicyAttachment{
+				{
+					Policy:   &v12.PolicyAttachment_Ref{Ref: "file://testdata/materials.yaml"},
+					Selector: &v12.PolicyAttachment_MaterialSelector{Name: "sbom"},
+				},
+			},
+		},
+	}
+
+	material := &v1.Attestation_Material{
+		MaterialType: v12.CraftingSchema_Material_SBOM_CYCLONEDX_JSON,
+		M: &v1.Attestation_Material_String_{
+			String_: &v1.Attestation_Material_KeyVal{
+				Id:    "sbom",
+				Value: "testdata/sbom.cyclonedx.json",
+			},
+		},
+	}
+
+	pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger)
+
+	const goroutines = 10
+	var wg sync.WaitGroup
+	errs := make([]error, goroutines)
+	results := make([][]*v1.PolicyEvaluation, goroutines)
+
+	for i := range goroutines {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			res, err := pv.VerifyMaterial(context.Background(), material, "")
+			errs[i] = err
+			results[i] = res
+		}()
+	}
+
+	wg.Wait()
+
+	for i := range goroutines {
+		assert.NoError(t, errs[i], "goroutine %d failed", i)
+	}
+
+	for i := 1; i < goroutines; i++ {
+		assert.Equal(t, len(results[0]), len(results[i]),
+			"goroutine %d returned different number of evaluations", i)
+	}
+}
+
+// TestWithMaxConcurrency verifies the WithMaxConcurrency option is applied.
+func TestWithMaxConcurrency(t *testing.T) {
+	logger := zerolog.Nop()
+
+	// Test default
+	pv := NewPolicyVerifier(nil, nil, &logger)
+	assert.Greater(t, pv.maxConcurrency, 0, "default maxConcurrency should be positive")
+
+	// Test custom value
+	pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(5))
+	assert.Equal(t, 5, pv.maxConcurrency)
+
+	// Test zero defaults to computed default
+	pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(0))
+	assert.Equal(t, defaultMaxConcurrency, pv.maxConcurrency)
+
+	// Test negative defaults to computed default
+	pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(-1))
+	assert.Equal(t, defaultMaxConcurrency, pv.maxConcurrency)
+}
+
+// TestVerifyStatementWithConcurrencyOne ensures sequential mode (maxConcurrency=1)
+// produces identical results to default parallel mode.
+func TestVerifyStatementWithConcurrencyOne(t *testing.T) {
+	logger := zerolog.Nop()
+
+	schema := &v12.CraftingSchema{
+		Policies: &v12.Policies{
+			Attestation: []*v12.PolicyAttachment{
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}},
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/materials.yaml"}},
+			},
+		},
+	}
+
+	statement := loadStatementForTest(t, "testdata/statement.json")
+
+	// Sequential
+	pvSeq := NewPolicyVerifier(schema.GetPolicies(), nil, &logger, WithMaxConcurrency(1))
+	seqResults, seqErr := pvSeq.VerifyStatement(context.Background(), statement)
+
+	// Parallel (default)
+	pvPar := NewPolicyVerifier(schema.GetPolicies(), nil, &logger)
+	parResults, parErr := pvPar.VerifyStatement(context.Background(), statement)
+
+	require.NoError(t, seqErr)
+	require.NoError(t, parErr)
+	assert.Equal(t, len(seqResults), len(parResults),
+		"sequential and parallel should return same number of evaluations")
+
+	// Build name→result maps for comparison (order may differ)
+	seqByName := make(map[string]*v1.PolicyEvaluation)
+	for _, ev := range seqResults {
+		seqByName[ev.Name] = ev
+	}
+
+	for _, ev := range parResults {
+		seqEv, ok := seqByName[ev.Name]
+		assert.True(t, ok, "parallel result has policy %q not found in sequential", ev.Name)
+		if ok {
+			assert.Equal(t, len(seqEv.Violations), len(ev.Violations),
+				"policy %q: violation count mismatch", ev.Name)
+			assert.Equal(t, seqEv.Skipped, ev.Skipped,
+				"policy %q: skipped mismatch", ev.Name)
+		}
+	}
+}
+
+// TestErrGroupCancellation verifies that when one policy evaluation fails,
+// the errgroup context is cancelled and propagated.
+func TestErrGroupCancellation(t *testing.T) {
+	logger := zerolog.Nop()
+
+	schema := &v12.CraftingSchema{
+		Policies: &v12.Policies{
+			Attestation: []*v12.PolicyAttachment{
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}},
+				// This policy ref does not exist — will cause an error
+				{Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/nonexistent_policy.yaml"}},
+			},
+		},
+	}
+
+	statement := loadStatementForTest(t, "testdata/statement.json")
+
+	pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger)
+	_, err := pv.VerifyStatement(context.Background(), statement)
+
+	assert.Error(t, err, "should fail when a policy file does not exist")
+	assert.IsType(t, &PolicyError{}, err, "error should be a PolicyError")
+}
+
+func loadStatementForTest(t *testing.T, file string) *intoto.Statement {
+	t.Helper()
+	stContent, err := os.ReadFile(file)
+	require.NoError(t, err)
+	var statement intoto.Statement
+	err = protojson.Unmarshal(stContent, &statement)
+	require.NoError(t, err)
+	return &statement
+}

pkg/policies/engine/wasm/engine.go

@@ -31,6 +31,8 @@ import (
 
 // setLogLevelOnce ensures extism.SetLogLevel is called exactly once,
 // since it modifies global state and is not safe for concurrent calls.
+// Note: the log level is determined by the first NewEngine() call in the
+// process. Subsequent engines with different log levels will not override it.
 var setLogLevelOnce sync.Once
 
 // Ensure Engine implements PolicyEngine interface

pkg/policies/group_loader.go

@@ -28,6 +28,7 @@ import (
 	pb "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1"
 	v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1"
 	crv1 "github.com/google/go-containerregistry/pkg/v1"
+	"golang.org/x/sync/singleflight"
 )
 
 // GroupLoader defines the interface for policy loaders from contract attachments
@@ -109,6 +110,7 @@ type groupWithReference struct {
 var (
 	remoteGroupCache      = make(map[string]*groupWithReference)
 	remoteGroupCacheMutex sync.Mutex
+	remoteGroupFlight     singleflight.Group
 )
 
 func NewChainloopGroupLoader(client pb.AttestationServiceClient) *ChainloopGroupLoader {
@@ -118,49 +120,64 @@ func NewChainloopGroupLoader(client pb.AttestationServiceClient) *ChainloopGroup
 func (c *ChainloopGroupLoader) Load(ctx context.Context, attachment *v1.PolicyGroupAttachment) (*v1.PolicyGroup, *PolicyDescriptor, error) {
 	ref := attachment.GetRef()
 
-	// Check cache (read under lock, release before any I/O)
+	// Fast path: check cache under lock
 	remoteGroupCacheMutex.Lock()
 	if v, ok := remoteGroupCache[ref]; ok {
 		remoteGroupCacheMutex.Unlock()
 		return v.group, v.reference, nil
 	}
 	remoteGroupCacheMutex.Unlock()
 
-	if !IsProviderScheme(ref) {
-		return nil, nil, fmt.Errorf("invalid group reference %q", ref)
-	}
+	// Use singleflight to coalesce concurrent fetches for the same ref.
+	result, err, _ := remoteGroupFlight.Do(ref, func() (interface{}, error) {
+		// Re-check cache (another goroutine may have populated it)
+		remoteGroupCacheMutex.Lock()
+		if v, ok := remoteGroupCache[ref]; ok {
+			remoteGroupCacheMutex.Unlock()
+			return v, nil
+		}
+		remoteGroupCacheMutex.Unlock()
 
-	providerRef := ProviderParts(ref)
+		if !IsProviderScheme(ref) {
+			return nil, fmt.Errorf("invalid group reference %q", ref)
+		}
 
-	// gRPC call happens outside the lock
-	resp, err := c.Client.GetPolicyGroup(ctx, &pb.AttestationServiceGetPolicyGroupRequest{
-		Provider:  providerRef.Provider,
-		GroupName: providerRef.Name,
-		OrgName:   providerRef.OrgName,
-	})
-	if err != nil {
-		return nil, nil, fmt.Errorf("requesting remote group (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err)
-	}
+		providerRef := ProviderParts(ref)
 
-	h, err := crv1.NewHash(resp.Reference.GetDigest())
-	if err != nil {
-		return nil, nil, fmt.Errorf("parsing digest: %w", err)
-	}
+		resp, err := c.Client.GetPolicyGroup(ctx, &pb.AttestationServiceGetPolicyGroupRequest{
+			Provider:  providerRef.Provider,
+			GroupName: providerRef.Name,
+			OrgName:   providerRef.OrgName,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("requesting remote group (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err)
+		}
 
-	orgName := providerRef.OrgName
-	// Extract organization name from URL if present
-	if u, err := url.Parse(resp.Reference.GetUrl()); err == nil {
-		if orgParam := u.Query().Get("org"); orgParam != "" {
-			orgName = orgParam
+		h, err := crv1.NewHash(resp.Reference.GetDigest())
+		if err != nil {
+			return nil, fmt.Errorf("parsing digest: %w", err)
 		}
-	}
 
-	reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h)
+		orgName := providerRef.OrgName
+		if u, err := url.Parse(resp.Reference.GetUrl()); err == nil {
+			if orgParam := u.Query().Get("org"); orgParam != "" {
+				orgName = orgParam
+			}
+		}
 
-	// Write to cache under lock
-	remoteGroupCacheMutex.Lock()
-	remoteGroupCache[ref] = &groupWithReference{group: resp.GetGroup(), reference: reference}
-	remoteGroupCacheMutex.Unlock()
+		reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h)
+		cached := &groupWithReference{group: resp.GetGroup(), reference: reference}
+
+		remoteGroupCacheMutex.Lock()
+		remoteGroupCache[ref] = cached
+		remoteGroupCacheMutex.Unlock()
+
+		return cached, nil
+	})
+	if err != nil {
+		return nil, nil, err
+	}
 
-	return resp.GetGroup(), reference, nil
+	cached := result.(*groupWithReference)
+	return cached.group, cached.reference, nil
 }