From 567f176b4d29ca97e8dfe6bc1c95df7382212bc8 Mon Sep 17 00:00:00 2001 From: Anbraten Date: Tue, 13 Jan 2026 14:40:46 +0100 Subject: [PATCH] Improve forge token refreshing --- server/api/helper.go | 17 --------- server/api/hook.go | 4 +- server/api/pipeline.go | 15 +++++--- .../bitbucketdatacenter.go | 8 ++-- server/forge/refresh.go | 37 +++++++++++++++++-- server/grpc/rpc.go | 6 +-- 6 files changed, 53 insertions(+), 34 deletions(-) diff --git a/server/api/helper.go b/server/api/helper.go index 591e8f621..47d27cd0d 100644 --- a/server/api/helper.go +++ b/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 { diff --git a/server/api/hook.go b/server/api/hook.go index d6fc4148f..dd1a2900b 100644 --- a/server/api/hook.go +++ b/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 diff --git a/server/api/pipeline.go b/server/api/pipeline.go index ce3d1396f..d8d989078 100644 --- a/server/api/pipeline.go +++ b/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 { diff --git a/server/forge/bitbucketdatacenter/bitbucketdatacenter.go b/server/forge/bitbucketdatacenter/bitbucketdatacenter.go index f182b3788..3fb2200b9 100644 --- a/server/forge/bitbucketdatacenter/bitbucketdatacenter.go +++ b/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) { diff --git a/server/forge/refresh.go b/server/forge/refresh.go index 3a35bad25..eb9e8d136 100644 --- a/server/forge/refresh.go +++ b/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 } } } diff --git a/server/grpc/rpc.go b/server/grpc/rpc.go index bef1b7fdd..ab59ed2da 100644 --- a/server/grpc/rpc.go +++ b/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) }