fix(router): abort router-transform on proactive invalid_grant instead of retrying (INT-6779)

router/transformer/transformer.go

@@ -377,6 +377,19 @@ func (trans *handle) Transform(transformType string, transformMessage *types.Tra
 		if resp.StatusCode == http.StatusNotFound {
 			statusCode = 404
 		}
+		errBody := string(respData)
+		// invalid_grant means the refresh token has been revoked: retrying cannot help, so the
+		// job is aborted instead of being retried as a generic 500. The error type alone
+		// decides this — not the status the interceptor attached, which is 400 for a token
+		// fetch but 500 when the control plane reports the same error type at the top level.
+		// Every other interceptor outcome, terminal-looking or not, keeps the default
+		// retryable path so that no unrelated failure changes behaviour or message.
+		if transResp.InterceptorResponse.ErrorType == common.RefTokenInvalidGrant {
+			statusCode = http.StatusBadRequest
+			if transResp.InterceptorResponse.Response != "" {
+				errBody = transResp.InterceptorResponse.Response
+			}
+		}
 		for i := range transformMessage.Data {
 			routerJob := &transformMessage.Data[i]
 			resp := types.DestinationJobT{
@@ -385,7 +398,7 @@ func (trans *handle) Transform(transformType string, transformMessage *types.Tra
 				Destination:      routerJob.Destination,
 				Connection:       routerJob.Connection,
 				StatusCode:       statusCode,
-				Error:            string(respData),
+				Error:            errBody,
 			}
 			destinationJobs = append(destinationJobs, resp)
 		}

router/transformer/transformer_test.go

@@ -450,10 +450,40 @@ type oauthV2TestCase struct {
 	description              string
 	cpResponses              []testutils.CpResponseParams
 	routerTransformResponses []types.DestinationJobT
-	inputEvents              []types.RouterJobT
-	expected                 []types.DestinationJobT
+	// routerTransformStatusCode overrides the status code the mock transformer replies with;
+	// defaults to 200 when routerTransformResponses is set.
+	routerTransformStatusCode int
+	inputEvents               []types.RouterJobT
+	expected                  []types.DestinationJobT
 }
 
+// authStatusInactiveResponses is a transformer response carrying a non-invalid_grant auth error
+// category. The interceptor turns AUTH_STATUS_INACTIVE into a terminal 400, which must NOT be
+// mistaken for invalid_grant by the router transform response handler.
+var authStatusInactiveResponses = []types.DestinationJobT{
+	{
+		JobMetadataArray:  []types.JobMetadataT{{JobID: 1, WorkspaceID: "wsp"}},
+		StatusCode:        http.StatusUnauthorized,
+		AuthErrorCategory: common.CategoryAuthStatusInactive,
+		Destination:       oauthDests[0],
+		Message:           []byte("{}"),
+	},
+}
+
+// authStatusInactiveBody is the exact payload the mock transformer serves for the case above,
+// which is also what the handler records as the job error on the retryable path.
+var authStatusInactiveBody = func() string {
+	b, err := jsonrs.Marshal(authStatusInactiveResponses)
+	if err != nil {
+		panic(err)
+	}
+	out, err := sjson.SetRawBytes([]byte(`{}`), "output", b)
+	if err != nil {
+		panic(err)
+	}
+	return string(out)
+}()
+
 var oauthDests = []backendconfig.DestinationT{
 	{
 		ID:          "d1",
@@ -597,6 +627,77 @@ var oauthV2RtTcs = []oauthV2TestCase{
 			{JobMetadataArray: []types.JobMetadataT{{JobID: 2, WorkspaceID: "wsp"}}, StatusCode: http.StatusInternalServerError, Error: "Reset Content", Destination: oauthDests[0]},
 		},
 	},
+	{
+		description: "when proactive fetch token fails with invalid_grant, all jobs abort with 400 instead of retrying with 500",
+		cpResponses: []testutils.CpResponseParams{
+			// fetch token http request -> invalid_grant
+			{
+				Code:     403,
+				Response: `{"status":403,"body":{"message":"[google_analytics] \"invalid_grant\" error, refresh token has been revoked","status":403,"code":"ref_token_invalid_grant"},"code":"ref_token_invalid_grant","access_token":"invalid_grant_access_token","refresh_token":"invalid_grant_refresh_token"}`,
+			},
+		},
+		inputEvents: []types.RouterJobT{
+			{JobMetadata: types.JobMetadataT{JobID: 1, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+			{JobMetadata: types.JobMetadataT{JobID: 2, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+		},
+		expected: []types.DestinationJobT{
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 1, WorkspaceID: "wsp"}}, StatusCode: http.StatusBadRequest, Error: `[google_analytics] "invalid_grant" error, refresh token has been revoked`},
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 2, WorkspaceID: "wsp"}}, StatusCode: http.StatusBadRequest, Error: `[google_analytics] "invalid_grant" error, refresh token has been revoked`},
+		},
+	},
+	{
+		description: "when proactive fetch token fails with a non-invalid_grant error, jobs stay 500 (retryable), not aborted",
+		cpResponses: []testutils.CpResponseParams{
+			// fetch token http request -> empty/invalid secret (non-invalid_grant failure)
+			{
+				Code:     200,
+				Response: `{}`,
+			},
+		},
+		inputEvents: []types.RouterJobT{
+			{JobMetadata: types.JobMetadataT{JobID: 1, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+			{JobMetadata: types.JobMetadataT{JobID: 2, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+		},
+		expected: []types.DestinationJobT{
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 1, WorkspaceID: "wsp"}}, StatusCode: http.StatusInternalServerError, Error: "status 500: empty secret received from CP"},
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 2, WorkspaceID: "wsp"}}, StatusCode: http.StatusInternalServerError, Error: "status 500: empty secret received from CP"},
+		},
+	},
+	{
+		description: "when fetch token fails with invalid_grant reported at the top level (status 500), jobs still abort with 400",
+		cpResponses: []testutils.CpResponseParams{
+			// fetch token http request -> invalid_grant surfaced via the top-level errorType key,
+			// which the oauth handler maps to a 500. The error type, not the status, decides.
+			{
+				Code:     500,
+				Response: `{"errorType":"ref_token_invalid_grant","message":"refresh token has been revoked"}`,
+			},
+		},
+		inputEvents: []types.RouterJobT{
+			{JobMetadata: types.JobMetadataT{JobID: 1, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+		},
+		expected: []types.DestinationJobT{
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 1, WorkspaceID: "wsp"}}, StatusCode: http.StatusBadRequest, Error: "refresh token has been revoked"},
+		},
+	},
+	{
+		description: "when a non-200 transform response carries a non-invalid_grant terminal status (authStatus inactive), jobs stay 500 (retryable), not aborted",
+		cpResponses: []testutils.CpResponseParams{
+			// fetch token http request -> succeeds, so the failure comes from the transform response
+			{
+				Code:     200,
+				Response: `{"secret": {"access_token": "valid_token","refresh_token":"refresh_token"}}`,
+			},
+		},
+		routerTransformResponses:  authStatusInactiveResponses,
+		routerTransformStatusCode: http.StatusBadRequest,
+		inputEvents: []types.RouterJobT{
+			{JobMetadata: types.JobMetadataT{JobID: 1, WorkspaceID: "wsp"}, Destination: oauthDests[0]},
+		},
+		expected: []types.DestinationJobT{
+			{Destination: oauthDests[0], JobMetadataArray: []types.JobMetadataT{{JobID: 1, WorkspaceID: "wsp"}}, StatusCode: http.StatusInternalServerError, Error: authStatusInactiveBody},
+		},
+	},
 }
 
 type mockIdentifier struct {
@@ -632,6 +733,9 @@ func TestRouterTransformationWithOAuthV2(t *testing.T) {
 					outputJson, _ = sjson.SetRawBytes([]byte(`{}`), "output", b)
 					statusCode = http.StatusOK
 				}
+				if tc.routerTransformStatusCode != 0 {
+					statusCode = tc.routerTransformStatusCode
+				}
 				require.NoError(t, err)
 				w.WriteHeader(statusCode)
 				_, err = w.Write(outputJson)

services/oauth/v2/http/transport.go

@@ -108,7 +108,31 @@ func (t *OAuthTransport) preRoundTrip(rts *roundTripState) *http.Response {
 	}
 	secret, scErr := t.oauthHandler.FetchToken(rts.tokenParams)
 	if scErr != nil {
-		return httpResponseCreator(scErr.StatusCode(), []byte(scErr.Error()))
+		// Propagate the token fetch failure through the interceptor envelope, so that
+		// callers which ignore the raw HTTP status code (e.g. router transformation, which
+		// otherwise collapses every non-200 to a retryable 500) can act on it. ErrorType
+		// carries the specific failure (e.g. common.RefTokenInvalidGrant) so callers decide
+		// on the error itself rather than inferring terminality from the status code.
+		// OriginalResponse preserves the raw error text for the callers that fall back to it.
+		message := scErr.Error()
+		var errorType string
+		var typeMessageError *v2.TypeMessageError
+		if errors.As(scErr, &typeMessageError) { // use the message from the underlying TypeMessageError if possible
+			message = typeMessageError.Message
+			errorType = typeMessageError.Type
+		}
+		respBody, marshalErr := jsonrs.Marshal(v2.TransportResponse{
+			OriginalResponse: scErr.Error(),
+			InterceptorResponse: v2.OAuthInterceptorResponse{
+				StatusCode: scErr.StatusCode(),
+				Response:   message,
+				ErrorType:  errorType,
+			},
+		})
+		if marshalErr != nil { // should never happen, the payload is a plain struct of scalars
+			return httpResponseCreator(scErr.StatusCode(), []byte(scErr.Error()))
+		}
+		return httpResponseCreator(scErr.StatusCode(), respBody)
 	}
 	rts.req = rts.req.WithContext(cntx.CtxWithSecret(rts.req.Context(), secret))
 

services/oauth/v2/http/transport_test.go

@@ -10,6 +10,7 @@ import (
 	. "github.com/onsi/gomega"
 	"go.uber.org/mock/gomock"
 
+	"github.com/rudderlabs/rudder-go-kit/jsonrs"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 	kitsync "github.com/rudderlabs/rudder-go-kit/sync"
@@ -346,6 +347,101 @@ var _ = Describe("OAuthTransport Error Handling", func() {
 			Expect(string(respBody)).To(Equal(`{"valid": "json"}`))
 		})
 
+		// Tests for a proactive token fetch (preRoundTrip) that fails: the failure is reported
+		// through the interceptor envelope so that callers which ignore the raw HTTP status
+		// code can still tell a terminal failure apart from a retryable one.
+		DescribeTable("should report a failing proactive token fetch through the interceptor envelope",
+			func(cpResponseCode int, cpResponse string, expectedStatusCode int, expectedResponse, expectedOriginalResponse, expectedErrorType string) {
+				cache := v2.NewOauthTokenCache()
+				ctrl := gomock.NewController(GinkgoT())
+
+				// the original transport must never be reached: preRoundTrip short circuits
+				mockRoundTrip := mockoauthv2.NewMockRoundTripper(ctrl)
+
+				mockAuthIdentityProvider := mockoauthv2.NewMockAuthIdentityProvider(ctrl)
+				mockAuthIdentityProvider.EXPECT().Identity().Return(nil).AnyTimes()
+
+				mockCpConnector := mockoauthv2.NewMockConnector(ctrl)
+				mockCpConnector.EXPECT().CpApiCall(gomock.Any()).Return(cpResponseCode, cpResponse)
+
+				oauthHandler := v2.NewOAuthHandler(mockAuthIdentityProvider,
+					v2.WithCache(v2.NewOauthTokenCache()),
+					v2.WithLocker(kitsync.NewPartitionRWLocker()),
+					v2.WithStats(stats.Default),
+					v2.WithLogger(logger.NewLogger().Child("MockOAuthHandler")),
+					v2.WithCpClient(mockCpConnector),
+				)
+
+				transport := httpClient.NewOAuthTransport(&httpClient.TransportArgs{
+					FlowType:             common.RudderFlowDelivery,
+					TokenCache:           &cache,
+					Locker:               kitsync.NewPartitionRWLocker(),
+					GetAuthErrorCategory: func([]byte) (string, error) { return "", nil },
+					Augmenter:            extensions.RouterBodyAugmenter,
+					OAuthHandler:         oauthHandler,
+					OriginalTransport:    mockRoundTrip,
+				})
+
+				req, _ := http.NewRequest("POST", "http://example.com", bytes.NewReader([]byte(`{}`)))
+				req = req.WithContext(cntx.CtxWithDestination(req.Context(), &backendconfig.DestinationT{
+					ID:              "test-destination-id",
+					WorkspaceID:     "test-workspace-id",
+					Config:          map[string]any{"rudderAccountId": "test-account-id"},
+					DeliveryAccount: &backendconfig.Account{ID: "test-account-id"},
+					DestinationDefinition: backendconfig.DestinationDefinitionT{
+						Name: "test-definition-name",
+						Config: map[string]any{
+							"auth": map[string]any{
+								"type":         "OAuth",
+								"rudderScopes": []any{"delivery"},
+							},
+						},
+					},
+				}))
+
+				res, err := transport.RoundTrip(req)
+				Expect(err).To(BeNil())
+				Expect(res).NotTo(BeNil())
+				Expect(res.StatusCode).To(Equal(expectedStatusCode))
+
+				respBody, err := io.ReadAll(res.Body)
+				Expect(err).To(BeNil())
+
+				var transportResponse v2.TransportResponse
+				Expect(jsonrs.Unmarshal(respBody, &transportResponse)).To(Succeed())
+				Expect(transportResponse.InterceptorResponse.StatusCode).To(Equal(expectedStatusCode))
+				Expect(transportResponse.InterceptorResponse.Response).To(Equal(expectedResponse))
+				// ErrorType is what callers key the abort decision on, not the status code
+				Expect(transportResponse.InterceptorResponse.ErrorType).To(Equal(expectedErrorType))
+				// OriginalResponse keeps the raw error text for the callers that fall back to it
+				Expect(transportResponse.OriginalResponse).To(Equal(expectedOriginalResponse))
+			},
+			Entry("terminal: invalid_grant is tagged as such and propagated as a 400 so the caller can abort",
+				403,
+				`{"status":403,"body":{"message":"[google_analytics] \"invalid_grant\" error, refresh token has been revoked","status":403,"code":"ref_token_invalid_grant"},"code":"ref_token_invalid_grant"}`,
+				http.StatusBadRequest,
+				`[google_analytics] "invalid_grant" error, refresh token has been revoked`,
+				`status 400: type: ref_token_invalid_grant, message: [google_analytics] "invalid_grant" error, refresh token has been revoked`,
+				common.RefTokenInvalidGrant,
+			),
+			Entry("retryable: a different error type is tagged with its own type and stays a 500",
+				403,
+				`{"status":403,"body":{"message":"invalid auth token refresh response","status":403,"code":"INVALID_REFRESH_RESPONSE"},"code":"INVALID_REFRESH_RESPONSE"}`,
+				http.StatusInternalServerError,
+				"invalid auth token refresh response",
+				"status 500: type: INVALID_REFRESH_RESPONSE, message: invalid auth token refresh response",
+				common.RefTokenInvalidResponse,
+			),
+			Entry("retryable: a failure with no error type carries an empty ErrorType and stays a 500",
+				200,
+				`{}`,
+				http.StatusInternalServerError,
+				"status 500: empty secret received from CP",
+				"status 500: empty secret received from CP",
+				"",
+			),
+		)
+
 		// Test for Bad Request errors
 		It("should convert Bad Request errors to 500 errors", func() {
 			// Setup

services/oauth/v2/types.go

@@ -63,8 +63,9 @@ type StatusRequestParams struct {
 }
 
 type OAuthInterceptorResponse struct {
-	StatusCode int    `json:"statusCode"`         // This is non-zero when the OAuth interceptor, upon completing its functions, intends to pass on the status code to the caller.
-	Response   string `json:"response,omitempty"` // This is non-empty when the OAuth interceptor, upon completing its functions, intends to pass on the response body to the caller.
+	StatusCode int    `json:"statusCode"`          // This is non-zero when the OAuth interceptor, upon completing its functions, intends to pass on the status code to the caller.
+	Response   string `json:"response,omitempty"`  // This is non-empty when the OAuth interceptor, upon completing its functions, intends to pass on the response body to the caller.
+	ErrorType  string `json:"errorType,omitempty"` // This is non-empty when the failure carries an error type (e.g. common.RefTokenInvalidGrant), letting callers act on the specific error rather than inferring it from the status code.
 }
 
 type TransportResponse struct {