perf: Use circuit breaker to fail open Malysis query requests (#196)

* perf: Use circuit breaker to fail open Malysis query requests

* fix: Linter fixes

* test: Add test to confirm grpc status error unwrapping

go.mod

@@ -8,13 +8,13 @@ require (
 	github.com/Masterminds/semver v1.5.0
 	github.com/elazarl/goproxy v1.8.1
 	github.com/fatih/color v1.18.0
-	github.com/goccy/go-yaml v1.19.2
 	github.com/google/osv-scalibr v0.2.1
 	github.com/google/uuid v1.6.0
 	github.com/jedib0t/go-pretty/v6 v6.6.7
 	github.com/posthog/posthog-go v1.5.12
 	github.com/safedep/dry v0.0.0-20260331131405-bd4c66ef7083
 	github.com/safedep/ptyx v0.2.1-0.20260119085117-f667570c2d12
+	github.com/sony/gobreaker/v2 v2.4.0
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.10
 	github.com/spf13/viper v1.21.0
@@ -42,6 +42,7 @@ require (
 	github.com/go-playground/validator/v10 v10.28.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/gobwas/glob v0.2.3 // indirect
+	github.com/goccy/go-yaml v1.19.2 // indirect
 	github.com/golang-jwt/jwt/v5 v5.3.0 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/go-github/v74 v74.0.0 // indirect

go.sum

@@ -157,6 +157,8 @@ github.com/safedep/ptyx v0.2.1-0.20260119085117-f667570c2d12/go.mod h1:fyt+PACz6
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
 github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/sony/gobreaker/v2 v2.4.0 h1:g2KJRW1Ubty3+ZOcSEUN7K+REQJdN6yo6XvaML+jptg=
+github.com/sony/gobreaker/v2 v2.4.0/go.mod h1:pTyFJgcZ3h2tdQVLZZruK2C0eoFL1fb/G83wK1ZQl+s=
 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw=
 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U=
 github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I=

proxy/interceptors/base_registry.go

@@ -12,6 +12,9 @@ import (
 	"github.com/safedep/pmg/config"
 	"github.com/safedep/pmg/internal/eventlog"
 	"github.com/safedep/pmg/proxy"
+	gobreaker "github.com/sony/gobreaker/v2"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // baseRegistryInterceptor provides common functionality for registry interceptors
@@ -21,6 +24,25 @@ type baseRegistryInterceptor struct {
 	cache            AnalysisCache
 	statsCollector   *AnalysisStatsCollector
 	confirmationChan chan *ConfirmationRequest
+	circuitBreaker   *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult]
+}
+
+func newAnalyzerCircuitBreaker(name string) *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult] {
+	return newAnalyzerCircuitBreakerWithTimeout(name, 30*time.Second)
+}
+
+func newAnalyzerCircuitBreakerWithTimeout(name string, cooldown time.Duration) *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult] {
+	return gobreaker.NewCircuitBreaker[*analyzer.PackageVersionAnalysisResult](gobreaker.Settings{
+		Name:        name,
+		MaxRequests: 1,
+		Timeout:     cooldown,
+		ReadyToTrip: func(counts gobreaker.Counts) bool {
+			return counts.ConsecutiveFailures >= 3
+		},
+		OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
+			log.Infof("Circuit breaker %s: %s -> %s", name, from, to)
+		},
+	})
 }
 
 var _ proxy.Interceptor = (*baseRegistryInterceptor)(nil)
@@ -87,10 +109,26 @@ func (b *baseRegistryInterceptor) analyzePackage(
 
 	log.Debugf("[%s] Analyzing package %s@%s", ctx.RequestID, packageName, packageVersion)
 
-	analysisCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
+	result, err := b.circuitBreaker.Execute(func() (*analyzer.PackageVersionAnalysisResult, error) {
+		analysisCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
 
-	result, err := b.analyzer.Analyze(analysisCtx, pkgVersion)
+		res, err := b.analyzer.Analyze(analysisCtx, pkgVersion)
+		if err != nil {
+			// NotFound means the package is not in the analysis DB — this is expected
+			// and should not count as a circuit breaker failure.
+			// Since gRPC v1.75.0, status.FromError unwraps error chains via errors.As.
+			if s, ok := status.FromError(err); ok && s.Code() == codes.NotFound {
+				log.Debugf("[%s] Package %s@%s not found in analysis DB, allowing", ctx.RequestID, packageName, packageVersion)
+				return &analyzer.PackageVersionAnalysisResult{
+					PackageVersion: pkgVersion,
+					Action:         analyzer.ActionAllow,
+				}, nil
+			}
+		}
+
+		return res, err
+	})
 	if err != nil {
 		return nil, fmt.Errorf("analyzer failed: %w", err)
 	}

proxy/interceptors/circuit_breaker_test.go

@@ -0,0 +1,214 @@
+package interceptors
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"testing"
+	"time"
+
+	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
+	"github.com/safedep/pmg/analyzer"
+	"github.com/safedep/pmg/proxy"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type mockAnalyzer struct {
+	callCount int
+	err       error
+	result    *analyzer.PackageVersionAnalysisResult
+}
+
+func (m *mockAnalyzer) Name() string { return "mock" }
+
+func (m *mockAnalyzer) Analyze(_ context.Context, pv *packagev1.PackageVersion) (*analyzer.PackageVersionAnalysisResult, error) {
+	m.callCount++
+	if m.err != nil {
+		return nil, m.err
+	}
+	return m.result, nil
+}
+
+func newTestBaseInterceptor(a analyzer.PackageVersionAnalyzer) *baseRegistryInterceptor {
+	return &baseRegistryInterceptor{
+		analyzer:         a,
+		cache:            NewInMemoryAnalysisCache(),
+		statsCollector:   NewAnalysisStatsCollector(),
+		confirmationChan: make(chan *ConfirmationRequest, 10),
+		circuitBreaker:   newAnalyzerCircuitBreaker("test"),
+	}
+}
+
+func newTestRequestContext() *proxy.RequestContext {
+	parsedURL, _ := url.Parse("https://registry.npmjs.org/test/-/test-1.0.0.tgz")
+	return &proxy.RequestContext{
+		URL:       parsedURL,
+		Method:    "GET",
+		RequestID: "test-req",
+		StartTime: time.Now(),
+		Data:      make(map[string]interface{}),
+	}
+}
+
+func TestCircuitBreaker_TripsAfterConsecutiveFailures(t *testing.T) {
+	mock := &mockAnalyzer{err: fmt.Errorf("rpc error: deadline exceeded")}
+	base := newTestBaseInterceptor(mock)
+	ctx := newTestRequestContext()
+
+	// First 3 calls should reach the analyzer (and fail)
+	for i := 0; i < 3; i++ {
+		_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg", "1.0.0")
+		require.Error(t, err)
+	}
+	assert.Equal(t, 3, mock.callCount)
+
+	// 4th call should be blocked by circuit breaker without calling analyzer
+	_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg", "1.0.0")
+	require.Error(t, err)
+	assert.Equal(t, 3, mock.callCount, "circuit breaker should prevent further analyzer calls")
+}
+
+func TestCircuitBreaker_SuccessResetsFailureCount(t *testing.T) {
+	mock := &mockAnalyzer{err: fmt.Errorf("transient error")}
+	base := newTestBaseInterceptor(mock)
+	ctx := newTestRequestContext()
+
+	// 2 failures (not enough to trip)
+	for i := 0; i < 2; i++ {
+		_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg", "1.0.0")
+		require.Error(t, err)
+	}
+
+	// Success resets the count
+	mock.err = nil
+	mock.result = &analyzer.PackageVersionAnalysisResult{Action: analyzer.ActionAllow}
+	result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg2", "1.0.0")
+	require.NoError(t, err)
+	assert.Equal(t, analyzer.ActionAllow, result.Action)
+
+	// 2 more failures should not trip (count was reset)
+	mock.err = fmt.Errorf("transient error")
+	for i := 0; i < 2; i++ {
+		_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg3", "1.0.0")
+		require.Error(t, err)
+	}
+	assert.Equal(t, 5, mock.callCount, "all calls should reach analyzer (breaker never tripped)")
+}
+
+func TestCircuitBreaker_RecoveryAfterCooldown(t *testing.T) {
+	mock := &mockAnalyzer{err: fmt.Errorf("rpc error: deadline exceeded")}
+
+	base := &baseRegistryInterceptor{
+		analyzer:         mock,
+		cache:            NewInMemoryAnalysisCache(),
+		statsCollector:   NewAnalysisStatsCollector(),
+		confirmationChan: make(chan *ConfirmationRequest, 10),
+		// Use a very short cooldown for testing
+		circuitBreaker: newAnalyzerCircuitBreakerWithTimeout("test-recovery", 1*time.Second),
+	}
+	ctx := newTestRequestContext()
+
+	// Trip the breaker
+	for i := 0; i < 3; i++ {
+		_, _ = base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg", "1.0.0")
+	}
+	assert.Equal(t, 3, mock.callCount)
+
+	// Breaker is open — calls don't reach analyzer
+	_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg", "1.0.0")
+	require.Error(t, err)
+	assert.Equal(t, 3, mock.callCount)
+
+	// Wait for cooldown, then the breaker enters half-open and allows a probe
+	time.Sleep(1500 * time.Millisecond)
+
+	mock.err = nil
+	mock.result = &analyzer.PackageVersionAnalysisResult{Action: analyzer.ActionAllow}
+	result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "pkg4", "1.0.0")
+	require.NoError(t, err)
+	assert.Equal(t, analyzer.ActionAllow, result.Action)
+	assert.Equal(t, 4, mock.callCount, "probe request should reach analyzer")
+}
+
+func TestCircuitBreaker_CacheBypassesBreaker(t *testing.T) {
+	mock := &mockAnalyzer{err: fmt.Errorf("rpc error: deadline exceeded")}
+	base := newTestBaseInterceptor(mock)
+	ctx := newTestRequestContext()
+
+	// Pre-populate cache
+	base.cache.Set(packagev1.Ecosystem_ECOSYSTEM_NPM.String(), "cached-pkg", "1.0.0", &analyzer.PackageVersionAnalysisResult{
+		Action: analyzer.ActionAllow,
+	})
+
+	// Trip the breaker with other packages
+	for i := 0; i < 3; i++ {
+		_, _ = base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, fmt.Sprintf("fail-%d", i), "1.0.0")
+	}
+
+	// Cached package should still be served even though breaker is open
+	result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "cached-pkg", "1.0.0")
+	require.NoError(t, err)
+	assert.Equal(t, analyzer.ActionAllow, result.Action)
+}
+
+func TestCircuitBreaker_NotFoundDoesNotCountAsFailure(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "single wrapped gRPC error (analyzer layer)",
+			err:  fmt.Errorf("failed to query package analysis: %w", status.Error(codes.NotFound, "package not found")),
+		},
+		{
+			name: "double wrapped gRPC error (interceptor + analyzer layers)",
+			err:  fmt.Errorf("analyzer failed: %w", fmt.Errorf("failed to query package analysis: %w", status.Error(codes.NotFound, "package not found"))),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mock := &mockAnalyzer{err: tt.err}
+			base := newTestBaseInterceptor(mock)
+			ctx := newTestRequestContext()
+
+			for i := 0; i < 5; i++ {
+				result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, fmt.Sprintf("unknown-%d", i), "1.0.0")
+				require.NoError(t, err)
+				assert.Equal(t, analyzer.ActionAllow, result.Action)
+			}
+
+			assert.Equal(t, 5, mock.callCount, "all calls should reach analyzer (breaker never tripped)")
+		})
+	}
+}
+
+func TestCircuitBreaker_NotFoundFollowedByRealFailures(t *testing.T) {
+	mock := &mockAnalyzer{err: fmt.Errorf("failed to query package analysis: %w", status.Error(codes.NotFound, "not found"))}
+	base := newTestBaseInterceptor(mock)
+	ctx := newTestRequestContext()
+
+	// 3 NotFound calls — should NOT trip the breaker
+	for i := 0; i < 3; i++ {
+		result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, fmt.Sprintf("notfound-%d", i), "1.0.0")
+		require.NoError(t, err)
+		assert.Equal(t, analyzer.ActionAllow, result.Action)
+	}
+
+	// Switch to real failures
+	mock.err = fmt.Errorf("rpc error: deadline exceeded")
+
+	// 3 real failures should trip the breaker
+	for i := 0; i < 3; i++ {
+		_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, fmt.Sprintf("fail-%d", i), "1.0.0")
+		require.Error(t, err)
+	}
+
+	// 7th call should be blocked by breaker
+	_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "blocked", "1.0.0")
+	require.Error(t, err)
+	assert.Equal(t, 6, mock.callCount, "breaker should prevent 7th call from reaching analyzer")
+}

proxy/interceptors/npm_registry.go

@@ -52,6 +52,7 @@ func NewNpmRegistryInterceptor(
 			cache:            cache,
 			statsCollector:   statsCollector,
 			confirmationChan: confirmationChan,
+			circuitBreaker:   newAnalyzerCircuitBreaker("malysis-analyzer-npm"),
 		},
 	}
 }
@@ -110,7 +111,7 @@ func (i *NpmRegistryInterceptor) HandleRequest(ctx *proxy.RequestContext) (*prox
 		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, nil
 	}
 
-	result, err := i.baseRegistryInterceptor.analyzePackage(
+	result, err := i.analyzePackage(
 		ctx,
 		packagev1.Ecosystem_ECOSYSTEM_NPM,
 		pkgInfo.GetName(),

proxy/interceptors/pypi_registry.go

@@ -53,6 +53,7 @@ func NewPypiRegistryInterceptor(
 			cache:            cache,
 			statsCollector:   statsCollector,
 			confirmationChan: confirmationChan,
+			circuitBreaker:   newAnalyzerCircuitBreaker("malysis-analyzer-pypi"),
 		},
 	}
 }