Improve forge token refreshing

server/api/helper.go

@@ -19,13 +19,9 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
-	"github.com/rs/zerolog/log"
 
-	"go.woodpecker-ci.org/woodpecker/v3/server"
-	"go.woodpecker-ci.org/woodpecker/v3/server/forge"
 	"go.woodpecker-ci.org/woodpecker/v3/server/model"
 	"go.woodpecker-ci.org/woodpecker/v3/server/pipeline"
-	"go.woodpecker-ci.org/woodpecker/v3/server/store"
 	"go.woodpecker-ci.org/woodpecker/v3/server/store/types"
 )
 
@@ -52,19 +48,6 @@ func handleDBError(c *gin.Context, err error) {
 	_ = c.AbortWithError(http.StatusInternalServerError, err)
 }
 
-// If the forge has a refresh token, the current access token may be stale.
-// Therefore, we should refresh prior to dispatching the job.
-func refreshUserToken(c *gin.Context, user *model.User) {
-	_store := store.FromContext(c)
-	_forge, err := server.Config.Services.Manager.ForgeFromUser(user)
-	if err != nil {
-		log.Error().Err(err).Msg("Cannot get forge from user")
-		c.AbortWithStatus(http.StatusInternalServerError)
-		return
-	}
-	forge.Refresh(c, _forge, _store, user)
-}
-
 // pipelineDeleteAllowed checks if the given pipeline can be deleted based on its status.
 // It returns a bool indicating if delete is allowed, and the pipeline's status.
 func pipelineDeleteAllowed(pl *model.Pipeline) bool {

server/api/hook.go

@@ -296,12 +296,12 @@ func PostHook(c *gin.Context) {
 		return
 	}
 
-	user, err := _store.GetUser(repo.UserID)
+	repoUser, err := _store.GetUser(repo.UserID)
 	if err != nil {
 		handleDBError(c, err)
 		return
 	}
-	forge.Refresh(c, _forge, _store, user)
+	forge.Refresh(c, _forge, _store, repoUser)
 
 	//
 	// 4. Update the repo

server/api/pipeline.go

@@ -29,6 +29,7 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"go.woodpecker-ci.org/woodpecker/v3/server"
+	"go.woodpecker-ci.org/woodpecker/v3/server/forge"
 	"go.woodpecker-ci.org/woodpecker/v3/server/model"
 	"go.woodpecker-ci.org/woodpecker/v3/server/pipeline"
 	"go.woodpecker-ci.org/woodpecker/v3/server/pipeline/stepbuilder"
@@ -604,7 +605,7 @@ func PostPipeline(c *gin.Context) {
 		return
 	}
 
-	user, err := _store.GetUser(repo.UserID)
+	repoUser, err := _store.GetUser(repo.UserID)
 	if err != nil {
 		handleDBError(c, err)
 		return
@@ -617,9 +618,13 @@ func PostPipeline(c *gin.Context) {
 	}
 
 	// refresh the token to make sure, pipeline.Restart can still obtain the pipeline config if necessary again
-	refreshUserToken(c, user)
-
-	// make Deploy overridable
+	_forge, err := server.Config.Services.Manager.ForgeFromUser(repoUser)
+	if err != nil {
+		log.Error().Err(err).Msg("Cannot get forge from user")
+		_ = c.AbortWithError(http.StatusBadRequest, err)
+		return
+	}
+	forge.Refresh(c, _forge, _store, repoUser)
 
 	// make Deploy task overridable
 	pl.DeployTask = c.DefaultQuery("deploy_task", pl.DeployTask)
@@ -656,7 +661,7 @@ func PostPipeline(c *gin.Context) {
 		}
 	}
 
-	newPipeline, err := pipeline.Restart(c, _store, pl, user, repo, envs)
+	newPipeline, err := pipeline.Restart(c, _store, pl, repoUser, repo, envs)
 	if err != nil {
 		handlePipelineErr(c, err)
 	} else {

server/forge/bitbucketdatacenter/bitbucketdatacenter.go

@@ -554,15 +554,15 @@ func (c *client) getUserAndRepo(ctx context.Context, r *model.Repo) (*model.User
 	}
 	log.Trace().Any("repo", repo).Msg("got repo")
 
-	user, err := _store.GetUser(repo.UserID)
+	repoUser, err := _store.GetUser(repo.UserID)
 	if err != nil {
 		return nil, nil, fmt.Errorf("unable to get user: %w", err)
 	}
-	log.Trace().Any("user", user).Msg("got user")
+	log.Trace().Any("user", repoUser).Msg("got user")
 
-	forge.Refresh(ctx, c, _store, user)
+	forge.Refresh(ctx, c, _store, repoUser)
 
-	return user, repo, nil
+	return repoUser, repo, nil
 }
 
 func (c *client) updatePipelineFromCommits(ctx context.Context, u *model.User, r *model.Repo, p *model.Pipeline, currCommit, prevCommit string) (*model.Pipeline, error) {

server/forge/refresh.go

@@ -16,6 +16,7 @@ package forge
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog/log"
@@ -24,6 +25,9 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
 )
 
+// userRefreshLocks stores per-user mutexes to prevent concurrent token refresh.
+var userRefreshLocks sync.Map
+
 // Refresher is an optional interface for OAuth token refresh support.
 //
 // Tokens are checked before each operation. If expiring within 30 minutes,
@@ -50,12 +54,39 @@ func Refresh(c context.Context, forge Forge, _store store.Store, user *model.Use
 			return
 		}
 
-		userUpdated, err := refresher.Refresh(c, user)
+		lockValue, _ := userRefreshLocks.LoadOrStore(user.ID, &sync.Mutex{})
+		userLock := lockValue.(*sync.Mutex)
+
+		userLock.Lock()
+		defer userLock.Unlock()
+
+		// Re-fetch the user from the database after acquiring the lock to check if
+		// another goroutine already refreshed the token while we were waiting.
+		freshUser, err := _store.GetUser(user.ID)
 		if err != nil {
-			log.Error().Err(err).Msgf("refresh oauth token of user '%s' failed", user.Login)
+			log.Error().Err(err).Msgf("failed to fetch user '%s' from store during token refresh", user.Login)
+			return
+		}
+
+		if time.Now().UTC().Unix() < (freshUser.Expiry - tokenMinTTL) {
+			// Update the passed-in user object.
+			user.AccessToken = freshUser.AccessToken
+			user.RefreshToken = freshUser.RefreshToken
+			user.Expiry = freshUser.Expiry
+			return
+		}
+
+		userUpdated, err := refresher.Refresh(c, freshUser)
+		if err != nil {
+			log.Error().Err(err).Msgf("refresh oauth token of user '%s' failed", freshUser.Login)
 		} else if userUpdated {
-			if err := _store.UpdateUser(user); err != nil {
+			if err := _store.UpdateUser(freshUser); err != nil {
 				log.Error().Err(err).Msg("fail to save user to store after refresh oauth token")
+			} else {
+				// Update the passed-in user object.
+				user.AccessToken = freshUser.AccessToken
+				user.RefreshToken = freshUser.RefreshToken
+				user.Expiry = freshUser.Expiry
 			}
 		}
 	}

server/grpc/rpc.go

@@ -552,7 +552,7 @@ func (s *RPC) completeChildrenIfParentCompleted(completedWorkflow *model.Workflo
 }
 
 func (s *RPC) updateForgeStatus(ctx context.Context, repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow) {
-	user, err := s.store.GetUser(repo.UserID)
+	repoUser, err := s.store.GetUser(repo.UserID)
 	if err != nil {
 		log.Error().Err(err).Msgf("cannot get user with id '%d'", repo.UserID)
 		return
@@ -564,11 +564,11 @@ func (s *RPC) updateForgeStatus(ctx context.Context, repo *model.Repo, pipeline
 		return
 	}
 
-	forge.Refresh(ctx, _forge, s.store, user)
+	forge.Refresh(ctx, _forge, s.store, repoUser)
 
 	// only do status updates for parent steps
 	if workflow != nil {
-		err = _forge.Status(ctx, user, repo, pipeline, workflow)
+		err = _forge.Status(ctx, repoUser, repo, pipeline, workflow)
 		if err != nil {
 			log.Error().Err(err).Msgf("error setting commit status for %s/%d", repo.FullName, pipeline.Number)
 		}