feat(submitqueue): terminate a request only on a validator.ErrRejected rejection and classify every other validator error

Author: roychying

submitqueue/extension/validator/composite/validator_test.go

@@ -17,6 +17,7 @@ package composite
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -104,3 +105,35 @@ func TestNew(t *testing.T) {
 		})
 	}
 }
+
+func TestNew_RejectsWhenAnyChildRejects(t *testing.T) {
+	rejection := fmt.Errorf("%w: missing approval", validator.ErrRejected)
+	failure := errors.New("could not reach the policy service")
+
+	tests := []struct {
+		name         string
+		children     []error
+		wantRejected bool
+	}{
+		{name: "one child rejects", children: []error{nil, rejection}, wantRejected: true},
+		{name: "a rejection beside a failure still rejects", children: []error{failure, rejection}, wantRejected: true},
+		{name: "failures alone do not reject", children: []error{failure, nil}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			var children []validator.Validator
+			for _, err := range tt.children {
+				child := validatormock.NewMockValidator(ctrl)
+				child.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(err)
+				children = append(children, child)
+			}
+
+			err := New(validator.Config{QueueName: "test-queue"}, children).Validate(context.Background(), entity.Request{})
+
+			require.Error(t, err)
+			assert.Equal(t, tt.wantRejected, errors.Is(err, validator.ErrRejected))
+		})
+	}
+}

submitqueue/extension/validator/github/README.md

@@ -33,7 +33,7 @@ URIs that are not GitHub change IDs are skipped, so the validator composes with
 
 ## Failures
 
-A rejection is an error naming the pull request and the rule it breaks. A failure to read GitHub — a transport error, an HTTP status, a GraphQL error such as a pull request that does not exist — is returned wrapped, and an HTTP status stays in the chain as `platform/http.StatusError` so `platform/errs/http` can classify it. Neither is classified here, as for any extension, and the validate stage currently rejects the request on either.
+A rejection is an error naming the pull request and the rule it breaks, wrapping `validator.ErrRejected`; the validate stage terminates the request with it as the reason. A failure to read GitHub — a transport error, an HTTP status, a GraphQL error such as a pull request that does not exist — does not wrap it, so the validate stage hands it to the consumer's classifiers like any other stage failure: an HTTP status stays in the chain as `platform/http.StatusError`, so `platform/errs/http` retries a 5xx or 429 and dead-letters a 403 or 404.
 
 ## Auth and hosts
 

submitqueue/extension/validator/github/evaluate.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 
 	entitygithub "github.com/uber/submitqueue/platform/base/change/github"
+	"github.com/uber/submitqueue/submitqueue/extension/validator"
 )
 
 // rejectionReason names why a pull request was rejected. The values are metric
@@ -53,6 +54,9 @@ func (r rejection) Error() string {
 	return fmt.Sprintf("%s: %s", r.uri, r.detail)
 }
 
+// Unwrap marks every rejection as validator.ErrRejected for the validate stage.
+func (r rejection) Unwrap() error { return validator.ErrRejected }
+
 // checkOutcome is ordered worst to best; requiredChecksNotPassed relies on it.
 type checkOutcome int
 

submitqueue/extension/validator/github/validator.go

@@ -81,10 +81,10 @@ func New(params Params) (validator.Validator, error) {
 }
 
 // Validate checks the request's GitHub pull requests bottom first and returns
-// the first rule one breaks. A failure to read GitHub is returned wrapped,
-// keeping an HTTP status code in the chain for platform/errs/http. URIs that
-// are not GitHub change IDs are skipped, so this validator composes with
-// validators for other change providers.
+// the first rule one breaks, wrapping validator.ErrRejected. A failure to read
+// GitHub is returned without it, keeping an HTTP status code in the chain for
+// platform/errs/http. URIs that are not GitHub change IDs are skipped, so this
+// validator composes with validators for other change providers.
 func (v *githubValidator) Validate(ctx context.Context, request entity.Request) error {
 	op := coremetrics.Begin(v.metricsScope, "validate", coremetrics.LongLatencyBuckets)
 	rej, rejected, err := v.findRejection(ctx, request)

submitqueue/extension/validator/github/validator_test.go

@@ -208,6 +208,7 @@ func TestValidate(t *testing.T) {
 				assert.NoError(t, err)
 				return
 			}
+			require.ErrorIs(t, err, validator.ErrRejected)
 			var rej rejection
 			require.True(t, errors.As(err, &rej), "want a rejection, got %v", err)
 			assert.Equal(t, tt.wantReason, rej.reason)
@@ -250,7 +251,7 @@ func TestValidateReadFailures(t *testing.T) {
 			err := v.Validate(context.Background(), request(testURI(1)))
 
 			require.Error(t, err)
-			assert.False(t, errors.As(err, &rejection{}), "a failure to read GitHub is not a rejection")
+			assert.NotErrorIs(t, err, validator.ErrRejected, "a failure to read GitHub is not a rejection")
 			var statusErr *phttp.StatusError
 			if tt.wantStatus != 0 {
 				require.True(t, errors.As(err, &statusErr))

submitqueue/extension/validator/validator.go

@@ -18,18 +18,27 @@ package validator
 
 import (
 	"context"
+	"errors"
 
 	"github.com/uber/submitqueue/submitqueue/entity"
 )
 
+// ErrRejected signals that the request must not land: a policy verdict, not a
+// failure. Validators wrap it into the error they return for a rejection, and
+// the validate stage terminates the request with that error's message as the
+// reason. Any other error is a failure to validate, classified and retried or
+// dead-lettered like any other stage failure.
+var ErrRejected = errors.New("rejected")
+
 // Validator runs custom validation checks against a request.
 // Implementations are provided by integrators for company-specific checks
 // (e.g. org-policy enforcement, repo-specific rules). The built-in validate
 // controller invokes this after fetching change metadata and before claiming
 // URIs in the change store.
 type Validator interface {
-	// Validate runs custom validation on the request.
-	// Returns nil if validation passes, error if the request should be rejected.
+	// Validate runs custom validation on the request. It returns nil if the
+	// request passes, an error wrapping ErrRejected if it must not land, and any
+	// other error if it could not be validated.
 	Validate(ctx context.Context, request entity.Request) error
 }
 

submitqueue/orchestrator/controller/validate/BUILD.bazel

@@ -40,6 +40,7 @@ go_test(
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/extension/messagequeue/mock:go_default_library",
+        "//platform/http:go_default_library",
         "//submitqueue/core/messagequeue:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",

submitqueue/orchestrator/controller/validate/validate.go

@@ -202,6 +202,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		}
 		if v != nil {
 			if err := v.Validate(ctx, request); err != nil {
+				// Only a rejection is the request's outcome. A failure to validate
+				// is left to the consumer's classifiers, so a transient one retries
+				// instead of failing a request nothing was wrong with.
+				if !errors.Is(err, validator.ErrRejected) {
+					coremetrics.NamedCounter(c.metricsScope, "process", "custom_validation_errors", 1)
+					return fmt.Errorf("failed to validate request %s: %w", request.ID, err)
+				}
 				coremetrics.NamedCounter(c.metricsScope, "process", "custom_validation_failures", 1)
 				c.logger.Infow("custom validation rejected request",
 					"request_id", request.ID,

submitqueue/orchestrator/controller/validate/validate_test.go

@@ -16,6 +16,7 @@ package validate
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 
@@ -31,6 +32,7 @@ import (
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	"github.com/uber/submitqueue/platform/errs"
 	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	phttp "github.com/uber/submitqueue/platform/http"
 	sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
@@ -735,8 +737,7 @@ func TestController_Process_CustomValidatorFails(t *testing.T) {
 	cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes()
 
 	mockValidator := validatormock.NewMockValidator(ctrl)
-	// mockValidator returns an error - validation failed
-	mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("some validation error"))
+	mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("%w: some validation error", validator.ErrRejected))
 	mockValidatorFactory := validatormock.NewMockFactory(ctrl)
 	mockValidatorFactory.EXPECT().For(validator.Config{
 		QueueName: request.Queue,
@@ -758,6 +759,79 @@ func TestController_Process_CustomValidatorFails(t *testing.T) {
 	assert.Contains(t, gotLog.LastError, "some validation error")
 }
 
+// A validator that could not validate has said nothing about the request, so
+// the failure goes to the consumer's classifiers instead of terminating it.
+func TestController_Process_CustomValidatorCannotValidate(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{name: "plain failure", err: errors.New("policy service unreachable")},
+		{name: "HTTP status", err: fmt.Errorf("read pull request: %w", phttp.NewStatusError(502, nil))},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+
+			request := entity.Request{
+				ID:           "123",
+				Queue:        "test-queue",
+				Change:       change.Change{URIs: []string{"github://uber/service/pull/456/abcdef0123456789abcdef0123456789abcdef01"}},
+				LandStrategy: mergestrategy.MergeStrategyRebase,
+				State:        entity.RequestStateStarted,
+				Version:      1,
+			}
+			// No Update and no claim Create are expected: the request must stay
+			// live and unclaimed for the redelivery to validate it again.
+			store, _ := newMockStorage(ctrl, request)
+			cs := storagemock.NewMockChangeStore(ctrl)
+			cs.EXPECT().GetByURI(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+			store.EXPECT().GetChangeStore().Return(cs).AnyTimes()
+
+			var statuses []entity.RequestStatus
+			mockPub := queuemock.NewMockPublisher(ctrl)
+			mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+				func(_ context.Context, _ string, msg entityqueue.Message) error {
+					if log, err := sqmq.UnmarshalRequestLog(msg.Payload); err == nil {
+						statuses = append(statuses, log.Status)
+					}
+					return nil
+				},
+			).AnyTimes()
+			mockQ := queuemock.NewMockQueue(ctrl)
+			mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
+			registry, err := consumer.NewTopicRegistry(
+				[]consumer.TopicConfig{
+					{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check", Queue: mockQ},
+					{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+				},
+			)
+			require.NoError(t, err)
+
+			cpFactory := changeprovidermock.NewMockFactory(ctrl)
+			cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes()
+			mockValidator := validatormock.NewMockValidator(ctrl)
+			mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(tt.err)
+			mockValidatorFactory := validatormock.NewMockFactory(ctrl)
+			mockValidatorFactory.EXPECT().For(validator.Config{QueueName: request.Queue}).Return(mockValidator, nil)
+
+			controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: store}, registry, cpFactory, mockValidatorFactory, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate")
+			msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil)
+			msg.Tenant = request.Queue
+			delivery := consumermock.NewMockDelivery(ctrl)
+			delivery.EXPECT().Message().Return(msg).AnyTimes()
+			delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+			err = controller.Process(context.Background(), delivery)
+
+			require.ErrorIs(t, err, tt.err)
+			assert.NotErrorIs(t, err, validator.ErrRejected)
+			assert.NotContains(t, statuses, entity.RequestStatusError, "the request must not be terminated")
+		})
+	}
+}
+
 func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
@@ -800,7 +874,7 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te
 	cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes()
 
 	mockValidator := validatormock.NewMockValidator(ctrl)
-	mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("some validation error"))
+	mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("%w: some validation error", validator.ErrRejected))
 	mockValidatorFactory := validatormock.NewMockFactory(ctrl)
 	mockValidatorFactory.EXPECT().For(validator.Config{
 		QueueName: request.Queue,