Fix/graphql client token refresh (#8791)

* fix(github-graphql): prevent panic in graphql rate limit polling goroutine

Replace panic in GraphqlAsyncClient rate-limit polling goroutine with graceful error handling.

Previously, any error while fetching rate limit (e.g., transient network issues or 401 responses) would trigger a panic inside a background goroutine, crashing the entire DevLake process.

Now, errors are logged and the client retries in the next cycle while retaining the last known rate limit.

Design decisions:
- Avoid panic in background goroutines: rate-limit polling is non-critical and should not bring down the entire pipeline.
- Use last known rateRemaining on runtime failures instead of resetting or blocking, ensuring continued progress with eventual consistency.
- Retry via existing polling mechanism instead of immediate retry to prevent tight retry loops and unnecessary API pressure.
- Introduce a default fallback (5000) only for initial rate-limit fetch failures, since no prior state exists at startup.
- Separate handling of initial vs runtime failures:
  - Initial failure → fallback to default (5000)
  - Runtime failure → retain previous value

Fixes #8788 (bug 1)

* fix(github-graphql): reuse ApiClient transport for GraphQL to enable token refresh

Replace oauth2.StaticTokenSource-based HTTP client with the underlying http.Client from ApiAsyncClient.

Previously, the GraphQL client constructed its own HTTP client using StaticTokenSource, which froze the access token at task start time. This caused GitHub App installation tokens (which expire after ~1 hour) to become invalid during long-running pipelines, leading to persistent 401 errors.

Now, the GraphQL client reuses apiClient.GetClient(), which is already configured with RefreshRoundTripper and TokenProvider. This enables automatic token refresh on 401 responses, aligning GraphQL behavior with the REST client.

Design decisions:
- Reuse transport layer instead of duplicating authentication logic to ensure consistency across REST and GraphQL clients.
- Avoid StaticTokenSource, as it prevents token refresh and breaks long-running pipelines.
- Leverage existing RefreshRoundTripper for transparent token rotation without modifying GraphQL query logic.
- Keep protocol-specific logic (GraphQL vs REST) separate while sharing the underlying HTTP transport.

This ensures GraphQL pipelines using GitHub App authentication can run beyond token expiry without failure.

Fixes #8788 (bug 2)

* refactor(github): extract shared authenticated http client from api client

- moved token provider and refresh round tripper setup into a reusable helper
- introduced CreateAuthenticatedHttpClient to centralize auth + transport logic
- updated CreateApiClient to use shared http client instead of inline setup

Rationale:
- decouples authentication (transport layer) from REST-specific client logic
- enables reuse for GraphQL client without duplicating token refresh logic
- aligns architecture with separation of concerns (http transport vs api clients)

* feat(github-graphql): introduce graphql client with shared auth and integrate into task flow

- added CreateGraphqlClient to encapsulate graphql client construction
- reused CreateAuthenticatedHttpClient from github/tasks to inject token refresh via RoundTripper
- replaced manual graphql client setup in PrepareTaskData with new factory function
- preserved existing rate limit handling via getRateRemaining callback
- preserved query cost calculation using SetGetRateCost

Technical details:
- graphql client now uses http transport with TokenProvider and RefreshRoundTripper
- removes dependency on oauth2 client and avoids token expiration issues
- decouples graphql client from REST ApiClient by avoiding reuse of apiClient.GetClient()
- maintains compatibility with github.com and enterprise graphql endpoints

Note:
- shared auth logic remains in github/tasks and is imported with alias to avoid package name collision
- introduces cross-plugin dependency (github_graphql → github/tasks) as a pragmatic tradeoff to avoid duplication

* feat(github): support static token transport for GraphQL and REST clients

add StaticRoundTripper for PAT authentication and use it in the shared http client.

since the same client is used by both REST and GraphQL, auth handling must distinguish
between refreshable tokens and static tokens. avoid applying refresh/retry logic to PAT.

ensures correct behavior across clients and prevents unnecessary retries for static auth.

* feat(github-graphql): introduce hierarchical fallback for GraphQL rate limit

Implement a layered fallback mechanism for GraphQL rate limiting:

1. Dynamic rate limit from provider (getRateRemaining)
2. Per-client override (WithFallbackRateLimit)
3. Config override (GRAPHQL_RATE_LIMIT)
4. Default fallback (1000)

Also moved GitHub-specific fallback (5000) via WithFallbackRateLimit
to the Graphql client.

* feat(github-graphql): Add graphql rate limit to .env example

* fix(github): Fix leaked debug statement

* fix(github-graphql): reuse http.Client proxy, auth configurations

Reused `http.Client` inside the apiClient returned by `CreateApiClient` method, so keeping the proxy and auth configurations the same.That also keep the centralized management of logic.

* fix(helpers): fix the priority order of fallback rate limit

Priority order fixed for fallback rate limit, priority order is:
1.Env variable
2.Value set with `WithFallbackRateLimit`
3.default value in the code
This all works only when the `getRateRemaining` fails: hence the fallback

* fix(github): StaticRoundTripper now owns token splitting and rotation for AccessToken connections

Previously, connection.Token (comma-separated PATs) was injected as-is
into the Authorization header, sending "Bearer tok1,tok2,tok3" instead
of a single rotated token.

StaticRoundTripper now splits the raw token string on comma and rotates
through tokens round-robin using an atomic counter.

For REST: StaticRoundTripper operates at transport level and always
overwrites the Authorization header set by SetupAuthentication.
SetupAuthentication is retained because conn.tokens is still required
by GetTokensCount() for rate limit calculation — but its header write
is superseded by StaticRoundTripper on every request.

For GraphQL: SetupAuthentication is never called by the graphql client,
so StaticRoundTripper is the only auth mechanism on this path — without
this fix, GraphQL requests were sent with the full unsplit token string.

* refactor(github-graphql): Downgrade fetch failure logs from Warn to Info

* fix(helper): use inline func type for GraphqlClientOption to avoid mock cycle

Replace exported GraphqlClientOption type with inline func(*GraphqlAsyncClient)
in CreateAsyncGraphqlClient signature. The named type caused mockery to generate
a mock file (GraphqlClientOption.go) that created an import cycle in tests.

* style(github): fix linting

* fix(github): token rotation start from index

* fix(helper): prevent graphql deadlock when rate limit fetch keeps failing

---------

Co-authored-by: Klesh Wong <klesh@qq.com>

Author: Ke-vin-S

backend/helpers/pluginhelper/api/graphql_async_client.go

@@ -20,12 +20,14 @@ package api
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"sync"
+	"time"
+
 	"github.com/apache/incubator-devlake/core/errors"
 	"github.com/apache/incubator-devlake/core/log"
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/core/utils"
-	"sync"
-	"time"
 
 	"github.com/merico-ai/graphql"
 )
@@ -47,30 +49,52 @@ type GraphqlAsyncClient struct {
 	getRateCost      func(q interface{}) int
 }
 
+// defaultRateLimitConst is the generic fallback rate limit for GraphQL requests.
+// It is used as the initial remaining quota when dynamic rate limit
+// information is unavailable from the provider.
+const defaultRateLimitConst = 1000
+
 // CreateAsyncGraphqlClient creates a new GraphqlAsyncClient
 func CreateAsyncGraphqlClient(
 	taskCtx plugin.TaskContext,
 	graphqlClient *graphql.Client,
 	logger log.Logger,
 	getRateRemaining func(context.Context, *graphql.Client, log.Logger) (rateRemaining int, resetAt *time.Time, err errors.Error),
+	opts ...func(*GraphqlAsyncClient),
 ) (*GraphqlAsyncClient, errors.Error) {
 	ctxWithCancel, cancel := context.WithCancel(taskCtx.GetContext())
+
 	graphqlAsyncClient := &GraphqlAsyncClient{
 		ctx:              ctxWithCancel,
 		cancel:           cancel,
 		client:           graphqlClient,
 		logger:           logger,
 		rateExhaustCond:  sync.NewCond(&sync.Mutex{}),
-		rateRemaining:    0,
+		rateRemaining:    defaultRateLimitConst,
 		getRateRemaining: getRateRemaining,
 	}
 
+	// apply options
+	for _, opt := range opts {
+		opt(graphqlAsyncClient)
+	}
+
+	// Env config wins over everything, only if explicitly set
+	if rateLimit := resolveRateLimit(taskCtx, logger); rateLimit != -1 {
+		logger.Info("GRAPHQL_RATE_LIMIT env override applied: %d (was %d)", rateLimit, graphqlAsyncClient.rateRemaining)
+		graphqlAsyncClient.rateRemaining = rateLimit
+	}
+
 	if getRateRemaining != nil {
 		rateRemaining, resetAt, err := getRateRemaining(taskCtx.GetContext(), graphqlClient, logger)
 		if err != nil {
-			panic(err)
+			graphqlAsyncClient.logger.Info("failed to fetch initial graphql rate limit, fallback to default: %v", err)
+			graphqlAsyncClient.updateRateRemaining(graphqlAsyncClient.rateRemaining, nil)
+		} else {
+			graphqlAsyncClient.updateRateRemaining(rateRemaining, resetAt)
 		}
-		graphqlAsyncClient.updateRateRemaining(rateRemaining, resetAt)
+	} else {
+		graphqlAsyncClient.updateRateRemaining(graphqlAsyncClient.rateRemaining, nil)
 	}
 
 	// load retry/timeout from configuration
@@ -115,6 +139,10 @@ func (apiClient *GraphqlAsyncClient) updateRateRemaining(rateRemaining int, rese
 		apiClient.rateExhaustCond.Signal()
 	}
 	go func() {
+		if apiClient.getRateRemaining == nil {
+			return
+		}
+
 		nextDuring := 3 * time.Minute
 		if resetAt != nil && resetAt.After(time.Now()) {
 			nextDuring = time.Until(*resetAt)
@@ -126,7 +154,15 @@ func (apiClient *GraphqlAsyncClient) updateRateRemaining(rateRemaining int, rese
 		case <-time.After(nextDuring):
 			newRateRemaining, newResetAt, err := apiClient.getRateRemaining(apiClient.ctx, apiClient.client, apiClient.logger)
 			if err != nil {
-				panic(err)
+				apiClient.logger.Info("failed to update graphql rate limit, will retry next cycle: %v", err)
+				// Floor the reused value so Signal() always fires; prevents deadlock when
+				// rateRemaining is 0 and the rate-limit endpoint keeps erroring (e.g. GHE).
+				fallback := apiClient.rateRemaining
+				if fallback < defaultRateLimitConst {
+					fallback = defaultRateLimitConst
+				}
+				apiClient.updateRateRemaining(fallback, nil)
+				return
 			}
 			apiClient.updateRateRemaining(newRateRemaining, newResetAt)
 		}
@@ -218,3 +254,25 @@ func (apiClient *GraphqlAsyncClient) Wait() {
 func (apiClient *GraphqlAsyncClient) Release() {
 	apiClient.cancel()
 }
+
+// WithFallbackRateLimit sets the initial/fallback rate limit used when
+// rate limit information cannot be fetched dynamically.
+// This value may be overridden later by getRateRemaining.
+func WithFallbackRateLimit(limit int) func(*GraphqlAsyncClient) {
+	return func(c *GraphqlAsyncClient) {
+		if limit > 0 {
+			c.rateRemaining = limit
+		}
+	}
+}
+
+// resolveRateLimit returns -1 if GRAPHQL_RATE_LIMIT is not set or invalid
+func resolveRateLimit(taskCtx plugin.TaskContext, logger log.Logger) int {
+	if v := taskCtx.GetConfig("GRAPHQL_RATE_LIMIT"); v != "" {
+		if parsed, err := strconv.Atoi(v); err == nil {
+			return parsed
+		}
+		logger.Warn(nil, "invalid GRAPHQL_RATE_LIMIT, using default")
+	}
+	return -1
+}

backend/plugins/github/tasks/api_client.go

@@ -26,7 +26,6 @@ import (
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
 	"github.com/apache/incubator-devlake/plugins/github/models"
-	"github.com/apache/incubator-devlake/plugins/github/token"
 )
 
 func CreateApiClient(taskCtx plugin.TaskContext, connection *models.GithubConnection) (*api.ApiAsyncClient, errors.Error) {
@@ -35,40 +34,10 @@ func CreateApiClient(taskCtx plugin.TaskContext, connection *models.GithubConnec
 		return nil, err
 	}
 
-	logger := taskCtx.GetLogger()
-	db := taskCtx.GetDal()
-	encryptionSecret := taskCtx.GetConfig(plugin.EncodeKeyEnvStr)
-
-	// Inject TokenProvider for OAuth refresh or GitHub App installation tokens.
-	var tp *token.TokenProvider
-	if connection.RefreshToken != "" {
-		tp = token.NewTokenProvider(connection, db, apiClient.GetClient(), logger, encryptionSecret)
-	} else if connection.AuthMethod == models.AppKey && connection.InstallationID != 0 {
-		tp = token.NewAppInstallationTokenProvider(connection, db, apiClient.GetClient(), logger, encryptionSecret)
-	}
-	if tp != nil {
-		// Wrap the transport
-		baseTransport := apiClient.GetClient().Transport
-		if baseTransport == nil {
-			baseTransport = http.DefaultTransport
-		}
-
-		rt := token.NewRefreshRoundTripper(baseTransport, tp)
-		apiClient.GetClient().Transport = rt
-		logger.Info("Installed token refresh round tripper for connection %d (authMethod=%s)",
-			connection.ID, connection.AuthMethod)
-	}
-
-	// Persist the freshly minted token so the DB has a correctly encrypted value.
-	// PrepareApiClient (called by NewApiClientFromConnection) mints the token
-	// in-memory but does not persist it; without this, the DB may contain a stale
-	// or corrupted token that breaks GET /connections.
-	if connection.AuthMethod == models.AppKey && connection.Token != "" {
-		if err := token.PersistEncryptedTokenColumns(db, connection, encryptionSecret, logger, false); err != nil {
-			logger.Warn(err, "Failed to persist initial token for connection %d", connection.ID)
-		} else {
-			logger.Info("Persisted initial token for connection %d", connection.ID)
-		}
+	// inject the shared auth layer
+	_, err = CreateAuthenticatedHttpClient(taskCtx, connection, apiClient.GetClient())
+	if err != nil {
+		return nil, err
 	}
 
 	// create rate limit calculator

backend/plugins/github/tasks/http_client.go

@@ -0,0 +1,88 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"net/http"
+
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+	"github.com/apache/incubator-devlake/plugins/github/models"
+	"github.com/apache/incubator-devlake/plugins/github/token"
+)
+
+func CreateAuthenticatedHttpClient(
+	taskCtx plugin.TaskContext,
+	connection *models.GithubConnection,
+	baseClient *http.Client,
+) (*http.Client, errors.Error) {
+
+	logger := taskCtx.GetLogger()
+	db := taskCtx.GetDal()
+	encryptionSecret := taskCtx.GetConfig(plugin.EncodeKeyEnvStr)
+
+	if baseClient == nil {
+		baseClient = &http.Client{}
+	}
+
+	// Inject TokenProvider for OAuth refresh or GitHub App installation tokens.
+	var tp *token.TokenProvider
+	if connection.RefreshToken != "" {
+		tp = token.NewTokenProvider(connection, db, baseClient, logger, encryptionSecret)
+	} else if connection.AuthMethod == models.AppKey && connection.InstallationID != 0 {
+		tp = token.NewAppInstallationTokenProvider(connection, db, baseClient, logger, encryptionSecret)
+	}
+
+	baseTransport := baseClient.Transport
+	if baseTransport == nil {
+		baseTransport = http.DefaultTransport
+	}
+
+	if tp != nil {
+		baseClient.Transport = token.NewRefreshRoundTripper(baseTransport, tp)
+		logger.Info(
+			"Installed token refresh round tripper for connection %d (authMethod=%s)",
+			connection.ID,
+			connection.AuthMethod,
+		)
+
+	} else if connection.Token != "" {
+		baseClient.Transport = token.NewStaticRoundTripper(
+			baseTransport,
+			connection.Token,
+		)
+		logger.Info(
+			"Installed static token round tripper for connection %d",
+			connection.ID,
+		)
+	}
+
+	// Persist the freshly minted token so the DB has a correctly encrypted value.
+	// PrepareApiClient (called by NewApiClientFromConnection) mints the token
+	// in-memory but does not persist it; without this, the DB may contain a stale
+	// or corrupted token that breaks GET /connections.
+	if connection.AuthMethod == models.AppKey && connection.Token != "" {
+		if err := token.PersistEncryptedTokenColumns(db, connection, encryptionSecret, logger, false); err != nil {
+			logger.Warn(err, "Failed to persist initial token for connection %d", connection.ID)
+		} else {
+			logger.Info("Persisted initial token for connection %d", connection.ID)
+		}
+	}
+
+	return baseClient, nil
+}

backend/plugins/github/token/round_tripper.go

@@ -19,6 +19,8 @@ package token
 
 import (
 	"net/http"
+	"strings"
+	"sync/atomic"
 )
 
 // RefreshRoundTripper is an HTTP transport middleware that automatically manages OAuth token refreshes.
@@ -93,3 +95,37 @@ func (rt *RefreshRoundTripper) roundTripWithRetry(req *http.Request, refreshAtte
 
 	return resp, nil
 }
+
+// StaticRoundTripper is an HTTP transport that injects a fixed bearer token.
+// Unlike RefreshRoundTripper, it does NOT attempt refresh or retries.
+type StaticRoundTripper struct {
+	base   http.RoundTripper
+	tokens []string
+	idx    atomic.Uint64
+}
+
+func NewStaticRoundTripper(base http.RoundTripper, rawToken string) *StaticRoundTripper {
+	if base == nil {
+		base = http.DefaultTransport
+	}
+	parts := strings.Split(rawToken, ",")
+	tokens := make([]string, 0, len(parts))
+	for _, t := range parts {
+		if t = strings.TrimSpace(t); t != "" {
+			tokens = append(tokens, t)
+		}
+	}
+	if len(tokens) == 0 {
+		tokens = []string{rawToken}
+	}
+	return &StaticRoundTripper{base: base, tokens: tokens}
+}
+
+func (rt *StaticRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	// always overrides headers put by SetupAuthentication, to make sure the token is always injected
+	// Add(1)-1 yields a 0-based sequence (0, 1, 2, ...) so rotation starts at tokens[0].
+	tok := rt.tokens[(rt.idx.Add(1)-1)%uint64(len(rt.tokens))]
+	reqClone := req.Clone(req.Context())
+	reqClone.Header.Set("Authorization", "Bearer "+tok)
+	return rt.base.RoundTrip(reqClone)
+}

backend/plugins/github_graphql/impl/impl.go

@@ -20,10 +20,7 @@ package impl
 import (
 	"context"
 	"fmt"
-	"net/http"
-	"net/url"
 	"reflect"
-	"strings"
 	"time"
 
 	"github.com/apache/incubator-devlake/core/models/domainlayer/devops"
@@ -39,7 +36,6 @@ import (
 	"github.com/apache/incubator-devlake/plugins/github_graphql/model/migrationscripts"
 	"github.com/apache/incubator-devlake/plugins/github_graphql/tasks"
 	"github.com/merico-ai/graphql"
-	"golang.org/x/oauth2"
 )
 
 // make sure interface is implemented
@@ -180,46 +176,10 @@ func (p GithubGraphql) PrepareTaskData(taskCtx plugin.TaskContext, options map[s
 		return nil, err
 	}
 
-	tokens := strings.Split(connection.Token, ",")
-	src := oauth2.StaticTokenSource(
-		&oauth2.Token{AccessToken: tokens[0]},
-	)
-	oauthContext := taskCtx.GetContext()
-	proxy := connection.GetProxy()
-	if proxy != "" {
-		pu, err := url.Parse(proxy)
-		if err != nil {
-			return nil, errors.Convert(err)
-		}
-		if pu.Scheme == "http" || pu.Scheme == "socks5" {
-			proxyClient := &http.Client{
-				Transport: &http.Transport{Proxy: http.ProxyURL(pu)},
-			}
-			oauthContext = context.WithValue(
-				taskCtx.GetContext(),
-				oauth2.HTTPClient,
-				proxyClient,
-			)
-			logger.Debug("Proxy set in oauthContext to %s", proxy)
-		} else {
-			return nil, errors.BadInput.New("Unsupported scheme set in proxy")
-		}
-	}
-
-	httpClient := oauth2.NewClient(oauthContext, src)
-	endpoint, err := errors.Convert01(url.Parse(connection.Endpoint))
-	if err != nil {
-		return nil, errors.BadInput.Wrap(err, fmt.Sprintf("malformed connection endpoint supplied: %s", connection.Endpoint))
-	}
-
-	// github.com and github enterprise have different graphql endpoints
-	endpoint.Path = "/graphql" // see https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
-	if endpoint.Hostname() != "api.github.com" {
-		// see https://docs.github.com/en/enterprise-server@3.11/graphql/guides/forming-calls-with-graphql
-		endpoint.Path = "/api/graphql"
-	}
-	client := graphql.NewClient(endpoint.String(), httpClient)
-	graphqlClient, err := helper.CreateAsyncGraphqlClient(taskCtx, client, taskCtx.GetLogger(),
+	graphqlClient, err := tasks.CreateGraphqlClient(
+		taskCtx,
+		connection,
+		apiClient.ApiClient.GetClient(),
 		func(ctx context.Context, client *graphql.Client, logger log.Logger) (rateRemaining int, resetAt *time.Time, err errors.Error) {
 			var query GraphQueryRateLimit
 			dataErrors, err := errors.Convert01(client.Query(taskCtx.GetContext(), &query, nil))
@@ -230,8 +190,7 @@ func (p GithubGraphql) PrepareTaskData(taskCtx plugin.TaskContext, options map[s
 				return 0, nil, errors.Default.Wrap(dataErrors[0], `query rate limit fail`)
 			}
 			if query.RateLimit == nil {
-				logger.Info(`github graphql rate limit are disabled, fallback to 5000req/hour`)
-				return 5000, nil, nil
+				return 0, nil, errors.Default.New("rate limit unavailable")
 			}
 			logger.Info(`github graphql init success with remaining %d/%d and will reset at %s`,
 				query.RateLimit.Remaining, query.RateLimit.Limit, query.RateLimit.ResetAt)