From 705229ae0120af8b2457c28d178e2441202e795d Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Date: Tue, 4 Aug 2026 17:56:39 +0530 Subject: [PATCH] fix(router): abort router-transform on proactive invalid_grant instead of retrying (INT-6779) When an OAuth destination's refresh token is revoked/expired, rudder-auth returns ref_token_invalid_grant. During router transformation, a proactive FetchToken that fails this way was returned by the OAuth transport's preRoundTrip as a raw-text HTTP 400 body. The router-transform response handler collapses any non-200 (non-404) response to a hardcoded 500 (retryable), so the real 400 was discarded - the job retried for the full retry window (~3h) and was ultimately drained at 410 instead of aborting. This makes ref_token_invalid_grant abort at 400 in the router-transform flow: - services/oauth/v2/http/transport.go: preRoundTrip now reports a failed FetchToken through the TransportResponse/InterceptorResponse envelope, carrying the real status code and a clean message. OriginalResponse keeps the raw error text, so every caller that falls back to it is unaffected. - services/oauth/v2/types.go: OAuthInterceptorResponse gains an ErrorType field, so callers act on the specific failure instead of inferring terminality from the status code. - router/transformer/transformer.go: the non-200 else-branch aborts with a 400 only when the interceptor reports ref_token_invalid_grant. The error type alone decides; the status the interceptor attached is not consulted, since a revoked refresh token is terminal however it arrives. Every other interceptor outcome keeps the existing retryable path and its original message. invalid_grant still counts toward the per-account OAuth circuit breaker (kept as-is to preserve control-plane protection); a multi-account breaker-trip alert to distinguish a control-plane-wide issue from isolated token revocations is tracked as a follow-up. Co-Authored-By: Claude Opus 5 (1M context) --- router/transformer/transformer.go | 15 +++- router/transformer/transformer_test.go | 108 ++++++++++++++++++++++- services/oauth/v2/http/transport.go | 26 +++++- services/oauth/v2/http/transport_test.go | 96 ++++++++++++++++++++ services/oauth/v2/types.go | 5 +- 5 files changed, 244 insertions(+), 6 deletions(-) diff --git a/router/transformer/transformer.go b/router/transformer/transformer.go index 0cd8fc1512..cf40ea51bb 100644 --- a/router/transformer/transformer.go +++ b/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) } diff --git a/router/transformer/transformer_test.go b/router/transformer/transformer_test.go index faa05cacf6..036707c953 100644 --- a/router/transformer/transformer_test.go +++ b/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) diff --git a/services/oauth/v2/http/transport.go b/services/oauth/v2/http/transport.go index 91dc2f3029..097de7c350 100644 --- a/services/oauth/v2/http/transport.go +++ b/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)) diff --git a/services/oauth/v2/http/transport_test.go b/services/oauth/v2/http/transport_test.go index b11906c9a1..71410f13d9 100644 --- a/services/oauth/v2/http/transport_test.go +++ b/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 diff --git a/services/oauth/v2/types.go b/services/oauth/v2/types.go index 0d8a229bb0..83e3986206 100644 --- a/services/oauth/v2/types.go +++ b/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 {