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)
+		// When the OAuth interceptor propagates a terminal (non-retryable) status
+		// — e.g. a proactive token fetch that failed with invalid_grant (400) —
+		// honor it so the job aborts instead of being retried as a generic 500.
+		// Interceptor-propagated 500s are left on the default path: they are
+		// already retryable and honoring them would only alter the error message
+		// of unrelated (non-invalid_grant) failures.
+		if transResp.InterceptorResponse.StatusCode > 0 && transResp.InterceptorResponse.StatusCode < http.StatusInternalServerError {
+			statusCode = transResp.InterceptorResponse.StatusCode
+			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

@@ -597,6 +597,42 @@ 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"},
+		},
+	},
 }
 
 type mockIdentifier struct {

services/oauth/v2/http/transport.go

@@ -108,6 +108,26 @@ func (t *OAuthTransport) preRoundTrip(rts *roundTripState) *http.Response {
 	}
 	secret, scErr := t.oauthHandler.FetchToken(rts.tokenParams)
 	if scErr != nil {
+		// For permanent auth errors (invalid_grant), propagate the real status
+		// code via the interceptor envelope so the caller aborts instead of
+		// retrying. Other errors keep the raw body (mapped to a retryable 500).
+		if errors.Is(scErr, common.ErrInvalidGrant) {
+			message := scErr.Error()
+			var typeMessageError *v2.TypeMessageError
+			if errors.As(scErr, &typeMessageError) {
+				message = typeMessageError.Message
+			}
+			body, marshalErr := jsonrs.Marshal(v2.TransportResponse{
+				InterceptorResponse: v2.OAuthInterceptorResponse{
+					StatusCode: scErr.StatusCode(),
+					Response:   message,
+				},
+			})
+			if marshalErr != nil {
+				return httpResponseCreator(scErr.StatusCode(), []byte(scErr.Error()))
+			}
+			return httpResponseCreator(scErr.StatusCode(), body)
+		}
 		return httpResponseCreator(scErr.StatusCode(), []byte(scErr.Error()))
 	}
 	rts.req = rts.req.WithContext(cntx.CtxWithSecret(rts.req.Context(), secret))