From 115531f6b3764c09607bb4a1c0d17cf637af4a1e Mon Sep 17 00:00:00 2001 From: Abhisek Datta Date: Fri, 3 Apr 2026 21:39:00 +0530 Subject: [PATCH] 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 | 3 +- go.sum | 2 + proxy/interceptors/base_registry.go | 44 ++++- proxy/interceptors/circuit_breaker_test.go | 214 +++++++++++++++++++++ proxy/interceptors/npm_registry.go | 3 +- proxy/interceptors/pypi_registry.go | 1 + 6 files changed, 262 insertions(+), 5 deletions(-) create mode 100644 proxy/interceptors/circuit_breaker_test.go diff --git a/go.mod b/go.mod index 176506f..6383f7c 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index aabdc01..a35f8a4 100644 --- a/go.sum +++ b/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= diff --git a/proxy/interceptors/base_registry.go b/proxy/interceptors/base_registry.go index 9b0c6fc..5be0c0e 100644 --- a/proxy/interceptors/base_registry.go +++ b/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) } diff --git a/proxy/interceptors/circuit_breaker_test.go b/proxy/interceptors/circuit_breaker_test.go new file mode 100644 index 0000000..dd3a0fb --- /dev/null +++ b/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") +} diff --git a/proxy/interceptors/npm_registry.go b/proxy/interceptors/npm_registry.go index 6e9161f..6cc683a 100644 --- a/proxy/interceptors/npm_registry.go +++ b/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(), diff --git a/proxy/interceptors/pypi_registry.go b/proxy/interceptors/pypi_registry.go index 3ebad2a..e1544e0 100644 --- a/proxy/interceptors/pypi_registry.go +++ b/proxy/interceptors/pypi_registry.go @@ -53,6 +53,7 @@ func NewPypiRegistryInterceptor( cache: cache, statsCollector: statsCollector, confirmationChan: confirmationChan, + circuitBreaker: newAnalyzerCircuitBreaker("malysis-analyzer-pypi"), }, } }