diff --git a/cache/credentials_cache.go b/cache/credentials_cache.go index ee672b5d9..063d80c91 100644 --- a/cache/credentials_cache.go +++ b/cache/credentials_cache.go @@ -14,6 +14,8 @@ package cache import ( + "time" + "github.com/cloudbase/garm/params" ) @@ -52,6 +54,11 @@ func (g *credentialCache) UpdateCredentialsUsingEndpoint(ep params.ForgeEndpoint func (g *credentialCache) SetCredentials(credentials params.ForgeCredentials) { g.Update(func(cache map[uint]params.ForgeCredentials) { + // Credentials sourced from the database carry no rate limit info, so + // an update clears the recorded values. That is fine as an update may + // be a token swap, and rate limits are per token. The clients record + // fresh values on every forge response and the cache worker's rate + // limit loop repolls within 30 seconds either way. cache[credentials.ID] = credentials UpdateCredentialsInAffectedEntities(credentials) }) @@ -110,3 +117,52 @@ func GetAllGiteaCredentialsAsMap() map[uint]params.ForgeCredentials { func UpdateCredentialsUsingEndpoint(ep params.ForgeEndpoint) { giteaCredentialsCache.UpdateCredentialsUsingEndpoint(ep) } + +// GetForgeCredentials returns the cached credentials with the given ID for +// the specified forge type. Unlike the copies workers hold on their entities, +// the cached credentials carry the most recently observed rate limit values. +func GetForgeCredentials(forgeType params.EndpointType, id uint) (params.ForgeCredentials, bool) { + switch forgeType { + case params.GithubEndpointType: + return GetGithubCredentials(id) + case params.GiteaEndpointType: + return GetGiteaCredentials(id) + } + return params.ForgeCredentials{}, false +} + +// entityCredentials resolves the freshest credentials for an entity: the +// entity cache tracks credential swaps, and the credentials cache carries +// the most recently observed rate limit values. Workers hold set-once +// copies of both, so rate limit checks must go through here. +func entityCredentials(entityID string) (params.ForgeCredentials, bool) { + entity, ok := GetEntity(entityID) + if !ok { + return params.ForgeCredentials{}, false + } + return GetForgeCredentials(entity.Credentials.ForgeType, entity.Credentials.ID) +} + +// EntityRateLimitReached reports whether the credentials currently assigned +// to the given entity should be considered rate limited for normal +// (non-critical) operations, and when the quota resets. Entities or +// credentials missing from the cache are never limited. +func EntityRateLimitReached(entityID string) (bool, time.Time) { + creds, ok := entityCredentials(entityID) + if !ok { + return false, time.Time{} + } + return creds.RateLimitReached() +} + +// EntityRateLimitExhausted reports whether the quota of the credentials +// currently assigned to the given entity is fully spent, meaning even +// critical operations cannot succeed, and when it resets. Entities or +// credentials missing from the cache are never limited. +func EntityRateLimitExhausted(entityID string) (bool, time.Time) { + creds, ok := entityCredentials(entityID) + if !ok { + return false, time.Time{} + } + return creds.CriticalRateLimitReached() +} diff --git a/cmd/garm-cli/cmd/github_credentials.go b/cmd/garm-cli/cmd/github_credentials.go index 6f9b64090..0929001dc 100644 --- a/cmd/garm-cli/cmd/github_credentials.go +++ b/cmd/garm-cli/cmd/github_credentials.go @@ -38,6 +38,8 @@ var ( credentialsPrivateKeyPath string credentialsType string credentialsEndpoint string + credentialsReserveEnabled bool + credentialsReservePercent int ) // credentialsCmd represents the credentials command @@ -117,7 +119,7 @@ var githubCredentialsUpdateCmd = &cobra.Command{ Short: "Update a github credential", Long: "Update a github credential", SilenceUsage: true, - RunE: func(_ *cobra.Command, args []string) error { + RunE: func(cmd *cobra.Command, args []string) error { if needsInit { return errNeedsInitError } @@ -135,7 +137,7 @@ var githubCredentialsUpdateCmd = &cobra.Command{ return fmt.Errorf("invalid credential ID: %s", args[0]) } - updateParams, err := parseCredentialsUpdateParams() + updateParams, err := parseCredentialsUpdateParams(cmd) if err != nil { return err } @@ -222,6 +224,8 @@ func init() { githubCredentialsUpdateCmd.Flags().Int64Var(&credentialsAppInstallationID, "app-installation-id", 0, "If the credential is an app, the installation ID") githubCredentialsUpdateCmd.Flags().Int64Var(&credentialsAppID, "app-id", 0, "If the credential is an app, the app ID") githubCredentialsUpdateCmd.Flags().StringVar(&credentialsPrivateKeyPath, "private-key-path", "", "If the credential is an app, the path to the private key file") + githubCredentialsUpdateCmd.Flags().BoolVar(&credentialsReserveEnabled, "reserve-usage-enabled", false, "Reserve a percentage of the rate limit for critical operations (such as runner deletion)") + githubCredentialsUpdateCmd.Flags().IntVar(&credentialsReservePercent, "reserve-usage-percentage", 0, "Percentage of the rate limit to reserve for critical operations (0-100). A value between 5 and 20 should be safe on most setups") githubCredentialsListCmd.Flags().BoolVarP(&long, "long", "l", false, "Include additional info.") @@ -238,6 +242,8 @@ func init() { githubCredentialsAddCmd.Flags().StringVar(&credentialsPrivateKeyPath, "private-key-path", "", "If the credential is an app, the path to the private key file") githubCredentialsAddCmd.Flags().StringVar(&credentialsType, "auth-type", "", "The type of the credential") githubCredentialsAddCmd.Flags().StringVar(&credentialsEndpoint, "endpoint", "", "The endpoint to associate the credential with") + githubCredentialsAddCmd.Flags().BoolVar(&credentialsReserveEnabled, "reserve-usage-enabled", false, "Reserve a percentage of the rate limit for critical operations (such as runner deletion)") + githubCredentialsAddCmd.Flags().IntVar(&credentialsReservePercent, "reserve-usage-percentage", 0, "Percentage of the rate limit to reserve for critical operations (0-100). A value between 5 and 20 should be safe on most setups") githubCredentialsAddCmd.MarkFlagsMutuallyExclusive("pat-oauth-token", "app-installation-id") githubCredentialsAddCmd.MarkFlagsMutuallyExclusive("pat-oauth-token", "app-id") @@ -285,6 +291,8 @@ func parseCredentialsAddParams() (ret params.CreateGithubCredentialsParams, err ret.Description = credentialsDescription ret.AuthType = params.ForgeAuthType(credentialsType) ret.Endpoint = credentialsEndpoint + ret.ReserveUsageEnabled = credentialsReserveEnabled + ret.ReserveUsagePercentage = credentialsReservePercent switch ret.AuthType { case params.ForgeAuthTypePAT: ret.PAT.OAuth2Token = credentialsOAuthToken @@ -303,7 +311,7 @@ func parseCredentialsAddParams() (ret params.CreateGithubCredentialsParams, err return ret, nil } -func parseCredentialsUpdateParams() (params.UpdateGithubCredentialsParams, error) { +func parseCredentialsUpdateParams(cmd *cobra.Command) (params.UpdateGithubCredentialsParams, error) { var updateParams params.UpdateGithubCredentialsParams if credentialsAppInstallationID != 0 || credentialsAppID != 0 || credentialsPrivateKeyPath != "" { @@ -318,6 +326,16 @@ func parseCredentialsUpdateParams() (params.UpdateGithubCredentialsParams, error updateParams.Description = &credentialsDescription } + // The zero values are meaningful for these two, so only send them if + // the flag was explicitly set on the command line. + if cmd.Flags().Changed("reserve-usage-enabled") { + updateParams.ReserveUsageEnabled = &credentialsReserveEnabled + } + + if cmd.Flags().Changed("reserve-usage-percentage") { + updateParams.ReserveUsagePercentage = &credentialsReservePercent + } + if credentialsOAuthToken != "" { if updateParams.PAT == nil { updateParams.PAT = ¶ms.GithubPAT{} @@ -390,6 +408,10 @@ func formatOneGithubCredential(cred params.ForgeCredentials) { t.AppendRow(table.Row{"Upload URL", cred.UploadBaseURL}) t.AppendRow(table.Row{"Type", cred.AuthType}) t.AppendRow(table.Row{"Endpoint", cred.Endpoint.Name}) + t.AppendRow(table.Row{"Reserve usage enabled", cred.ReserveUsageEnabled}) + if cred.ReserveUsageEnabled { + t.AppendRow(table.Row{"Reserve usage percentage", fmt.Sprintf("%d%%", cred.ReserveUsagePercentage)}) + } if resetMinutes > 0 { t.AppendRow(table.Row{"", ""}) t.AppendRow(table.Row{"Remaining API requests", cred.RateLimit.Remaining}) diff --git a/database/sql/file_store_test.go b/database/sql/file_store_test.go index 2c0577f71..8b26fed4b 100644 --- a/database/sql/file_store_test.go +++ b/database/sql/file_store_test.go @@ -207,7 +207,7 @@ func (s *FileStoreTestSuite) TestListFileObjects() { func (s *FileStoreTestSuite) TestListFileObjectsPagination() { // Create more files to test pagination for i := 0; i < 5; i++ { - content := []byte(fmt.Sprintf("File %d", i)) + content := fmt.Appendf(nil, "File %d", i) param := params.CreateFileObjectParams{ Name: fmt.Sprintf("page-test-%d.txt", i), Size: int64(len(content)), @@ -548,7 +548,7 @@ func (s *FileStoreTestSuite) TestSearchFileObjectByTagsEmptyTags() { func (s *FileStoreTestSuite) TestSearchFileObjectByTagsPagination() { // Create multiple files with the same tag for i := 0; i < 5; i++ { - content := []byte(fmt.Sprintf("Pagination test file %d", i)) + content := fmt.Appendf(nil, "Pagination test file %d", i) param := params.CreateFileObjectParams{ Name: fmt.Sprintf("page-search-%d.txt", i), Size: int64(len(content)), @@ -768,7 +768,7 @@ func (s *FileStoreTestSuite) TestSearchFileObjectByTagsOrderByCreatedAt() { func (s *FileStoreTestSuite) TestPaginationFieldsLastPage() { // Create exactly 5 files for i := 0; i < 5; i++ { - content := []byte(fmt.Sprintf("Last page test %d", i)) + content := fmt.Appendf(nil, "Last page test %d", i) param := params.CreateFileObjectParams{ Name: fmt.Sprintf("last-page-test-%d.txt", i), Size: int64(len(content)), diff --git a/database/sql/github.go b/database/sql/github.go index 50daa9047..760718548 100644 --- a/database/sql/github.go +++ b/database/sql/github.go @@ -274,12 +274,14 @@ func (s *sqlDatabase) CreateGithubCredentials(ctx context.Context, param params. } creds = GithubCredentials{ - Name: param.Name, - Description: param.Description, - EndpointName: &endpoint.Name, - AuthType: param.AuthType, - Payload: data, - UserID: &userID, + Name: param.Name, + Description: param.Description, + EndpointName: &endpoint.Name, + AuthType: param.AuthType, + Payload: data, + UserID: &userID, + ReserveUsageEnabled: param.ReserveUsageEnabled, + ReserveUsagePercentage: param.ReserveUsagePercentage, } if err := tx.Create(&creds).Error; err != nil { @@ -432,6 +434,12 @@ func (s *sqlDatabase) UpdateGithubCredentials(ctx context.Context, id uint, para if param.Description != nil && *param.Description != creds.Description { updates["description"] = *param.Description } + if param.ReserveUsageEnabled != nil && *param.ReserveUsageEnabled != creds.ReserveUsageEnabled { + updates["reserve_usage_enabled"] = *param.ReserveUsageEnabled + } + if param.ReserveUsagePercentage != nil && *param.ReserveUsagePercentage != creds.ReserveUsagePercentage { + updates["reserve_usage_percentage"] = *param.ReserveUsagePercentage + } var data []byte var err error diff --git a/database/sql/migrations/0007_credentials_reserve_usage.go b/database/sql/migrations/0007_credentials_reserve_usage.go new file mode 100644 index 000000000..4d6f8fd66 --- /dev/null +++ b/database/sql/migrations/0007_credentials_reserve_usage.go @@ -0,0 +1,41 @@ +// Copyright 2026 Cloudbase Solutions SRL +// +// Licensed 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 migrations + +import ( + "github.com/go-gormigrate/gormigrate/v2" + "gorm.io/gorm" +) + +// githubCredentials0007 is a minimal stub that only declares the new +// columns: whether a slice of the credential's rate limit budget is +// reserved for critical operations (such as runner deletion), and how +// large that slice is, as a percentage of the hourly limit. + +type githubCredentials0007 struct { + ReserveUsageEnabled bool + ReserveUsagePercentage int +} + +func (githubCredentials0007) TableName() string { return "github_credentials" } + +func init() { + Register(&gormigrate.Migration{ + ID: "0007_credentials_reserve_usage", + Migrate: func(tx *gorm.DB) error { + return tx.AutoMigrate(&githubCredentials0007{}) + }, + }) +} diff --git a/database/sql/models.go b/database/sql/models.go index 2dec823e5..01463581b 100644 --- a/database/sql/models.go +++ b/database/sql/models.go @@ -565,6 +565,12 @@ type GithubCredentials struct { Endpoint GithubEndpoint `gorm:"foreignKey:EndpointName"` EndpointName *string `gorm:"index"` + // ReserveUsageEnabled toggles whether or not to allocate a certain + // percentage of the available rate limit to critical operations such + // as delete operations for runners that have finished their jobs. + ReserveUsageEnabled bool + ReserveUsagePercentage int + Repositories []Repository `gorm:"foreignKey:CredentialsID"` Organizations []Organization `gorm:"foreignKey:CredentialsID"` Enterprises []Enterprise `gorm:"foreignKey:CredentialsID"` diff --git a/database/sql/util.go b/database/sql/util.go index 0432519c8..02c4b6c92 100644 --- a/database/sql/util.go +++ b/database/sql/util.go @@ -1139,19 +1139,21 @@ func (s *sqlDatabase) sqlToCommonForgeCredentials(creds GithubCredentials) (para } commonCreds := params.ForgeCredentials{ - ID: creds.ID, - Name: creds.Name, - Description: creds.Description, - APIBaseURL: creds.Endpoint.APIBaseURL, - BaseURL: creds.Endpoint.BaseURL, - UploadBaseURL: creds.Endpoint.UploadBaseURL, - CABundle: creds.Endpoint.CACertBundle, - AuthType: creds.AuthType, - CreatedAt: creds.CreatedAt, - UpdatedAt: creds.UpdatedAt, - ForgeType: creds.Endpoint.EndpointType, - Endpoint: ep, - CredentialsPayload: data, + ID: creds.ID, + Name: creds.Name, + Description: creds.Description, + APIBaseURL: creds.Endpoint.APIBaseURL, + BaseURL: creds.Endpoint.BaseURL, + UploadBaseURL: creds.Endpoint.UploadBaseURL, + CABundle: creds.Endpoint.CACertBundle, + AuthType: creds.AuthType, + CreatedAt: creds.CreatedAt, + UpdatedAt: creds.UpdatedAt, + ForgeType: creds.Endpoint.EndpointType, + Endpoint: ep, + CredentialsPayload: data, + ReserveUsageEnabled: creds.ReserveUsageEnabled, + ReserveUsagePercentage: creds.ReserveUsagePercentage, } for _, repo := range creds.Repositories { diff --git a/params/params.go b/params/params.go index b7a80aa3e..276d36cd3 100644 --- a/params/params.go +++ b/params/params.go @@ -1276,6 +1276,16 @@ type ForgeCredentials struct { ForgeType EndpointType `json:"forge_type,omitempty"` + // ReserveUsageEnabled toggles whether or not to allocate a certain + // percentage of the available rate limit to critical operations such + // as delete operations for runners that have finished their jobs. + ReserveUsageEnabled bool `json:"reserve_usage_enabled,omitempty"` + // ReserveUsagePercentage is the percentage of available rate limit reserved + // for critical operations. Setting this value too high will negatively impact + // normal operations, so it is capped at 50%. A value between 5% and 20% + // should be safe on most setups. Adjust this based on your usage patterns. + ReserveUsagePercentage int `json:"reserve_usage_percentage,omitempty"` + Repositories []Repository `json:"repositories,omitempty"` Organizations []Organization `json:"organizations,omitempty"` Enterprises []Enterprise `json:"enterprises,omitempty"` @@ -1292,6 +1302,52 @@ func (g ForgeCredentials) GetID() uint { return g.ID } +// reserveThreshold returns the number of API calls held in reserve for +// critical operations (such as removing runners that finished their jobs). +// Zero when usage reservation is disabled or no rate limit was observed. +func (g ForgeCredentials) reserveThreshold() int { + if !g.ReserveUsageEnabled || g.RateLimit == nil { + return 0 + } + return g.RateLimit.Limit * g.ReserveUsagePercentage / 100 +} + +// rateLimitReached reports whether the remaining quota has dropped to or +// below the given threshold, and when the quota resets. The reset time is +// returned whenever rate limit info was recorded, regardless of the verdict. +// It is the zero time only when no info is available (Gitea, GHES with rate +// limiting disabled, or no forge response observed yet). When no rate limits +// are available, the credentials are considered unlimited. A quota whose +// reset time has passed is treated as refreshed, even if we have not yet +// observed a fresh response confirming it, otherwise a controller that +// stopped making API calls due to the limit would never notice the reset. +func (g ForgeCredentials) rateLimitReached(threshold int) (bool, time.Time) { + if g.RateLimit == nil || g.RateLimit.Limit == 0 { + return false, time.Time{} + } + resetAt := g.RateLimit.ResetAt() + if !time.Now().Before(resetAt) { + return false, resetAt + } + return g.RateLimit.Remaining <= threshold, resetAt +} + +// RateLimitReached reports whether these credentials should be considered +// rate limited for normal (non-critical) operations, and when the quota +// resets. When usage reservation is enabled, normal operations are +// considered limited once the remaining quota dips into the reserved +// percentage, keeping the reserve available for critical operations. +func (g ForgeCredentials) RateLimitReached() (bool, time.Time) { + return g.rateLimitReached(g.reserveThreshold()) +} + +// CriticalRateLimitReached reports whether even critical operations (such +// as removing runners that finished their jobs) should back off, meaning +// the quota is fully exhausted, and when it resets. +func (g ForgeCredentials) CriticalRateLimitReached() (bool, time.Time) { + return g.rateLimitReached(0) +} + func (g ForgeCredentials) GetHTTPClient(ctx context.Context) (*http.Client, error) { var roots *x509.CertPool if g.CABundle != nil { diff --git a/params/requests.go b/params/requests.go index 202114e78..81794bdab 100644 --- a/params/requests.go +++ b/params/requests.go @@ -520,6 +520,12 @@ func (g GithubApp) Validate() error { return nil } +// MaxReserveUsagePercentage is the highest allowed value for +// ReserveUsagePercentage. Reserving more than half of the quota for +// critical operations leaves too little for normal operations and +// effectively breaks scaling. +const MaxReserveUsagePercentage = 50 + // swagger:model CreateGithubCredentialsParams type CreateGithubCredentialsParams struct { Name string `json:"name,omitempty"` @@ -528,6 +534,16 @@ type CreateGithubCredentialsParams struct { AuthType ForgeAuthType `json:"auth_type,omitempty"` PAT GithubPAT `json:"pat,omitempty"` App GithubApp `json:"app,omitempty"` + + // ReserveUsageEnabled toggles whether or not to allocate a certain + // percentage of the available rate limit to critical operations such + // as delete operations for runners that have finished their jobs. + ReserveUsageEnabled bool `json:"reserve_usage_enabled,omitempty"` + // ReserveUsagePercentage is the percentage of available rate limit reserved + // for critical operations. Setting this value too high will negatively impact + // normal operations, so it is capped at 50%. A value between 5% and 20% + // should be safe on most setups. Adjust this based on your usage patterns. + ReserveUsagePercentage int `json:"reserve_usage_percentage,omitempty"` } func (c CreateGithubCredentialsParams) Validate() error { @@ -539,6 +555,10 @@ func (c CreateGithubCredentialsParams) Validate() error { return runnerErrors.NewBadRequestError("missing endpoint") } + if c.ReserveUsagePercentage > MaxReserveUsagePercentage || c.ReserveUsagePercentage < 0 { + return runnerErrors.NewBadRequestError("value for reserve_usage_percentage must be an int between 0 and %d", MaxReserveUsagePercentage) + } + switch c.AuthType { case ForgeAuthTypePAT, ForgeAuthTypeApp: default: @@ -566,6 +586,15 @@ type UpdateGithubCredentialsParams struct { Description *string `json:"description,omitempty"` PAT *GithubPAT `json:"pat,omitempty"` App *GithubApp `json:"app,omitempty"` + // ReserveUsageEnabled toggles whether or not to allocate a certain + // percentage of the available rate limit to critical operations such + // as delete operations for runners that have finished their jobs. + ReserveUsageEnabled *bool `json:"reserve_usage_enabled,omitempty"` + // ReserveUsagePercentage is the percentage of available rate limit reserved + // for critical operations. Setting this value too high will negatively impact + // normal operations, so it is capped at 50%. A value between 5% and 20% + // should be safe on most setups. Adjust this based on your usage patterns. + ReserveUsagePercentage *int `json:"reserve_usage_percentage,omitempty"` } func (u UpdateGithubCredentialsParams) Validate() error { @@ -585,6 +614,10 @@ func (u UpdateGithubCredentialsParams) Validate() error { } } + if u.ReserveUsagePercentage != nil && (*u.ReserveUsagePercentage > MaxReserveUsagePercentage || *u.ReserveUsagePercentage < 0) { + return runnerErrors.NewBadRequestError("value for reserve_usage_percentage must be an int between 0 and %d", MaxReserveUsagePercentage) + } + return nil } diff --git a/runner/pool/pool.go b/runner/pool/pool.go index c5ac462aa..5c1d726e9 100644 --- a/runner/pool/pool.go +++ b/runner/pool/pool.go @@ -528,7 +528,40 @@ func jobIDFromLabels(labels []string) int64 { return 0 } -func (r *basePoolManager) startLoopForFunction(f func() error, interval time.Duration, name string, alwaysRun bool) { +// rateLimitTier classifies pool loops by how they consume forge API quota. +type rateLimitTier int + +const ( + // tierInternal marks loops that make no forge API calls. They are never + // paused by rate limits. + tierInternal rateLimitTier = iota + // tierNormal marks loops whose forge API usage is not critical (scaling + // up, reconciling state, reaping). They pause once the remaining quota + // dips into the configured reserve. + tierNormal + // tierCritical marks loops that execute already-decided work, such as + // removing runners that finished their jobs. They pause only when the + // quota is fully exhausted. + tierCritical +) + +// rateLimitReached reports whether a loop of the given tier should pause due +// to the entity credentials' forge rate limit, and when the quota resets. +// This is orthogonal to the manager running state: an unauthorized error +// disables the manager until valid credentials produce a successful tools +// update, regardless of the rate limit quota resetting in the meantime. +func (r *basePoolManager) rateLimitReached(tier rateLimitTier) (bool, time.Time) { + switch tier { + case tierCritical: + return cache.EntityRateLimitExhausted(r.entity.ID) + case tierNormal: + return cache.EntityRateLimitReached(r.entity.ID) + default: + return false, time.Time{} + } +} + +func (r *basePoolManager) startLoopForFunction(f func() error, interval time.Duration, name string, alwaysRun bool, tier rateLimitTier) { slog.InfoContext( r.ctx, "starting loop for entity", "loop_name", name) @@ -543,6 +576,7 @@ func (r *basePoolManager) startLoopForFunction(f func() error, interval time.Dur r.wg.Done() }() + rateLimited := false for { shouldRun := r.managerIsRunning if alwaysRun { @@ -552,6 +586,21 @@ func (r *basePoolManager) startLoopForFunction(f func() error, interval time.Dur case true: select { case <-ticker.C: + if limited, resetAt := r.rateLimitReached(tier); limited { + if !rateLimited { + slog.InfoContext( + r.ctx, "rate limit reached; pausing loop until the quota resets", + "loop_name", name, "reset_at", resetAt) + rateLimited = true + } + continue + } + if rateLimited { + slog.InfoContext( + r.ctx, "rate limit lifted; resuming loop", + "loop_name", name) + rateLimited = false + } if err := f(); err != nil { slog.With(slog.Any("error", err)).ErrorContext( r.ctx, "error in loop", @@ -1938,16 +1987,18 @@ func (r *basePoolManager) Start() error { case <-initializeEntity: } defer close(initializeEntity) - go r.startLoopForFunction(r.runnerCleanup, common.PoolReapTimeoutInterval, "timeout_reaper", false) - go r.startLoopForFunction(r.scaleDown, common.PoolScaleDownInterval, "scale_down", false) + go r.startLoopForFunction(r.runnerCleanup, common.PoolReapTimeoutInterval, "timeout_reaper", false, tierNormal) + go r.startLoopForFunction(r.scaleDown, common.PoolScaleDownInterval, "scale_down", false, tierNormal) // always run the delete pending instances routine. This way we can still remove existing runners, even if the pool is not running. - go r.startLoopForFunction(r.deletePendingInstances, common.PoolConsilitationInterval, "consolidate[delete_pending]", true) - go r.startLoopForFunction(r.addPendingInstances, common.PoolConsilitationInterval, "consolidate[add_pending]", false) - go r.startLoopForFunction(r.ensureMinIdleRunners, common.PoolConsilitationInterval, "consolidate[ensure_min_idle]", false) - go r.startLoopForFunction(r.retryFailedInstances, common.PoolConsilitationInterval, "consolidate[retry_failed]", false) - go r.startLoopForFunction(r.updateTools, common.PoolToolUpdateInterval, "update_tools", true) - go r.startLoopForFunction(r.consumeQueuedJobs, common.PoolConsilitationInterval, "job_queue_consumer", false) - go r.startLoopForFunction(r.reconcileStaleJobs, common.PoolStaleJobReconcileInterval, "stale_job_reconciler", false) + go r.startLoopForFunction(r.deletePendingInstances, common.PoolConsilitationInterval, "consolidate[delete_pending]", true, tierCritical) + go r.startLoopForFunction(r.addPendingInstances, common.PoolConsilitationInterval, "consolidate[add_pending]", false, tierNormal) + go r.startLoopForFunction(r.ensureMinIdleRunners, common.PoolConsilitationInterval, "consolidate[ensure_min_idle]", false, tierNormal) + go r.startLoopForFunction(r.retryFailedInstances, common.PoolConsilitationInterval, "consolidate[retry_failed]", false, tierNormal) + // updateTools reads the tools cache; it makes no forge API calls and is + // also the path that re-enables the manager after an unauthorized error. + go r.startLoopForFunction(r.updateTools, common.PoolToolUpdateInterval, "update_tools", true, tierInternal) + go r.startLoopForFunction(r.consumeQueuedJobs, common.PoolConsilitationInterval, "job_queue_consumer", false, tierNormal) + go r.startLoopForFunction(r.reconcileStaleJobs, common.PoolStaleJobReconcileInterval, "stale_job_reconciler", false, tierNormal) }() return nil } diff --git a/webapp/src/lib/api/generated/api.ts b/webapp/src/lib/api/generated/api.ts index 0d6bfede7..375b5fe2e 100644 --- a/webapp/src/lib/api/generated/api.ts +++ b/webapp/src/lib/api/generated/api.ts @@ -465,6 +465,18 @@ export interface CreateGithubCredentialsParams { * @memberof CreateGithubCredentialsParams */ 'pat'?: GithubPAT; + /** + * ReserveUsageEnabled toggles whether or not to allocate a certain percentage of the available rate limit to critical operations such as delete operations for runners that have finished their jobs. + * @type {boolean} + * @memberof CreateGithubCredentialsParams + */ + 'reserve_usage_enabled'?: boolean; + /** + * ReserveUsagePercentage is the percentage of available rate limit reserved for critical operations. Setting this value too high will negatively impact normal operations, so it is capped at 50%. A value between 5% and 20% should be safe on most setups. Adjust this based on your usage patterns. + * @type {number} + * @memberof CreateGithubCredentialsParams + */ + 'reserve_usage_percentage'?: number; } /** * @@ -1294,6 +1306,18 @@ export interface ForgeCredentials { * @memberof ForgeCredentials */ 'repositories'?: Array; + /** + * ReserveUsageEnabled toggles whether or not to allocate a certain percentage of the available rate limit to critical operations such as delete operations for runners that have finished their jobs. + * @type {boolean} + * @memberof ForgeCredentials + */ + 'reserve_usage_enabled'?: boolean; + /** + * ReserveUsagePercentage is the percentage of available rate limit reserved for critical operations. Setting this value too high will negatively impact normal operations, so it is capped at 50%. A value between 5% and 20% should be safe on most setups. Adjust this based on your usage patterns. + * @type {number} + * @memberof ForgeCredentials + */ + 'reserve_usage_percentage'?: number; /** * * @type {string} @@ -3565,6 +3589,18 @@ export interface UpdateGithubCredentialsParams { * @memberof UpdateGithubCredentialsParams */ 'pat'?: GithubPAT; + /** + * ReserveUsageEnabled toggles whether or not to allocate a certain percentage of the available rate limit to critical operations such as delete operations for runners that have finished their jobs. + * @type {boolean} + * @memberof UpdateGithubCredentialsParams + */ + 'reserve_usage_enabled'?: boolean; + /** + * ReserveUsagePercentage is the percentage of available rate limit reserved for critical operations. Setting this value too high will negatively impact normal operations, so it is capped at 50%. A value between 5% and 20% should be safe on most setups. Adjust this based on your usage patterns. + * @type {number} + * @memberof UpdateGithubCredentialsParams + */ + 'reserve_usage_percentage'?: number; } /** * diff --git a/webapp/src/lib/components/forms/CredentialsForm.svelte b/webapp/src/lib/components/forms/CredentialsForm.svelte index dedc2a536..8a9728550 100644 --- a/webapp/src/lib/components/forms/CredentialsForm.svelte +++ b/webapp/src/lib/components/forms/CredentialsForm.svelte @@ -18,7 +18,9 @@ oauth2_token: '', app_id: '', installation_id: '', - private_key_bytes: '' + private_key_bytes: '', + reserve_usage_enabled: false, + reserve_usage_percentage: 0 }; export let selectedAuthType: typeof AuthType[keyof typeof AuthType] = AuthType.PAT; export let forgeType: 'github' | 'gitea' | '' = ''; @@ -41,6 +43,8 @@ formData.app_id = ''; formData.installation_id = ''; formData.private_key_bytes = ''; + formData.reserve_usage_enabled = false; + formData.reserve_usage_percentage = 0; selectedAuthType = AuthType.PAT; dispatch('forgeTypeSelect', event.detail); } @@ -188,6 +192,48 @@ {/if} + +{#if forgeType === 'github'} +
+
+ + +
+

+ Sets aside a slice of this credential's API rate limit for critical operations, such as deleting runners that finished their jobs. Runner creation pauses when only the reserved budget is left. +

+ {#if formData.reserve_usage_enabled} +
+ +
+ + {formData.reserve_usage_percentage}% +
+

+ Percentage of the hourly rate limit to reserve (at most 50%). A value between 5% and 20% should be safe on most setups. +

+
+ {/if} +
+{/if} + {#if selectedAuthType === AuthType.APP}
diff --git a/webapp/src/lib/components/setup/CredentialsStep.svelte b/webapp/src/lib/components/setup/CredentialsStep.svelte index 475a12ea3..120b20000 100644 --- a/webapp/src/lib/components/setup/CredentialsStep.svelte +++ b/webapp/src/lib/components/setup/CredentialsStep.svelte @@ -7,7 +7,7 @@ import { eagerCacheManager } from '$lib/stores/eager-cache.js'; import { toastStore } from '$lib/stores/toast.js'; import { extractAPIError } from '$lib/utils/apiError'; - import { getForgeIcon } from '$lib/utils/common.js'; + import { getForgeIcon, validateReservePercentage } from '$lib/utils/common.js'; const AuthType = { PAT: 'pat', APP: 'app' } as const; @@ -40,7 +40,9 @@ oauth2_token: '', app_id: '', installation_id: '', - private_key_bytes: '' + private_key_bytes: '', + reserve_usage_enabled: false, + reserve_usage_percentage: 0 }; $: isFormValid = (() => { @@ -72,6 +74,11 @@ }); async function handleCreate() { + const reserveError = validateReservePercentage(formData.reserve_usage_percentage); + if (reserveError) { + error = reserveError; + return; + } creating = true; error = ''; try { @@ -80,7 +87,9 @@ name: formData.name.trim(), description: formData.description.trim(), endpoint: endpointName, - auth_type: selectedAuthType + auth_type: selectedAuthType, + reserve_usage_enabled: formData.reserve_usage_enabled, + reserve_usage_percentage: formData.reserve_usage_percentage }; if (selectedAuthType === AuthType.PAT) { githubParams.pat = { oauth2_token: formData.oauth2_token.trim() }; diff --git a/webapp/src/lib/utils/common.ts b/webapp/src/lib/utils/common.ts index 2dedf4e6c..b73ba4582 100644 --- a/webapp/src/lib/utils/common.ts +++ b/webapp/src/lib/utils/common.ts @@ -306,3 +306,21 @@ export function getPaginationInfo(currentPage: number, perPage: number, totalIte return `Showing ${start} to ${end} of ${totalItems} results`; } + +/** + * Highest allowed value for the rate limit reserve percentage. Mirrors the + * server-side cap (params.MaxReserveUsagePercentage): reserving more than + * half of the quota leaves too little for normal operations. + */ +export const MAX_RESERVE_USAGE_PERCENTAGE = 50; + +/** + * Validates the rate limit reserve percentage. Returns an error message, or + * null when the value is valid. + */ +export function validateReservePercentage(percentage: number): string | null { + if (!Number.isInteger(percentage) || percentage < 0 || percentage > MAX_RESERVE_USAGE_PERCENTAGE) { + return `Reserved percentage must be an integer between 0 and ${MAX_RESERVE_USAGE_PERCENTAGE}.`; + } + return null; +} diff --git a/webapp/src/routes/credentials/+page.svelte b/webapp/src/routes/credentials/+page.svelte index 7b50b94a1..a139ed759 100644 --- a/webapp/src/routes/credentials/+page.svelte +++ b/webapp/src/routes/credentials/+page.svelte @@ -13,7 +13,7 @@ import DataTable from '$lib/components/DataTable.svelte'; import { eagerCache, eagerCacheManager } from '$lib/stores/eager-cache.js'; import { toastStore } from '$lib/stores/toast.js'; - import { getForgeIcon, filterCredentials, changePerPage, paginateItems, getAuthTypeBadge } from '$lib/utils/common.js'; + import { getForgeIcon, filterCredentials, changePerPage, paginateItems, getAuthTypeBadge, validateReservePercentage } from '$lib/utils/common.js'; import { extractAPIError } from '$lib/utils/apiError'; import { handleFileInputAsBase64 } from '$lib/utils/file'; import Badge from '$lib/components/Badge.svelte'; @@ -72,6 +72,8 @@ app_id: string; installation_id: string; private_key_bytes: string; + reserve_usage_enabled: boolean; + reserve_usage_percentage: number; } = { name: '', description: '', @@ -80,7 +82,9 @@ oauth2_token: '', app_id: '', installation_id: '', - private_key_bytes: '' + private_key_bytes: '', + reserve_usage_enabled: false, + reserve_usage_percentage: 0 }; // Track original values for comparison during updates let originalFormData: typeof formData = { ...formData }; @@ -159,7 +163,9 @@ oauth2_token: '', app_id: '', installation_id: '', - private_key_bytes: '' + private_key_bytes: '', + reserve_usage_enabled: credential.reserve_usage_enabled || false, + reserve_usage_percentage: credential.reserve_usage_percentage || 0 }; selectedAuthType = (credential['auth-type'] as typeof AuthType[keyof typeof AuthType]) || AuthType.PAT; // Store original values for comparison @@ -184,7 +190,9 @@ oauth2_token: '', app_id: '', installation_id: '', - private_key_bytes: '' + private_key_bytes: '', + reserve_usage_enabled: false, + reserve_usage_percentage: 0 }; originalFormData = { ...formData }; selectedAuthType = AuthType.PAT; @@ -216,7 +224,18 @@ updateParams.description = formData.description.trim(); } } - + + // Rate limit reserve settings only exist on GitHub credentials. + if (editingCredential?.forge_type === 'github') { + if (formData.reserve_usage_enabled !== originalFormData.reserve_usage_enabled) { + updateParams.reserve_usage_enabled = formData.reserve_usage_enabled; + } + + if (formData.reserve_usage_percentage !== originalFormData.reserve_usage_percentage) { + updateParams.reserve_usage_percentage = formData.reserve_usage_percentage; + } + } + // Only include credential fields if the checkbox is checked and fields have values if (wantToChangeCredentials && editingCredential) { if (editingCredential['auth-type'] === AuthType.PAT) { @@ -263,6 +282,11 @@ async function handleCreateCredentials() { try { + const reserveError = validateReservePercentage(formData.reserve_usage_percentage); + if (reserveError) { + toastStore.error('Create Failed', reserveError); + return; + } // Use selected forge type to determine which API to call if (selectedForgeType === 'github') { // Build the correct nested structure for GitHub credentials @@ -270,7 +294,9 @@ name: formData.name.trim(), description: formData.description.trim(), endpoint: formData.endpoint.trim(), - auth_type: formData.auth_type + auth_type: formData.auth_type, + reserve_usage_enabled: formData.reserve_usage_enabled, + reserve_usage_percentage: formData.reserve_usage_percentage }; if (formData.auth_type === AuthType.PAT) { @@ -318,8 +344,13 @@ async function handleUpdateCredentials() { if (!editingCredential || !editingCredential.id) return; - + try { + const reserveError = validateReservePercentage(formData.reserve_usage_percentage); + if (reserveError) { + toastStore.error('Update Failed', reserveError); + return; + } const updateParams = buildUpdateParams(); // Only proceed if there are changes to apply @@ -716,6 +747,48 @@

Authentication type cannot be changed after creation

+ {#if editingCredential.forge_type === 'github'} + +
+
+ + +
+

+ Sets aside a slice of this credential's API rate limit for critical operations, such as deleting runners that finished their jobs. Runner creation pauses when only the reserved budget is left. +

+ {#if formData.reserve_usage_enabled} +
+ +
+ + {formData.reserve_usage_percentage}% +
+

+ Percentage of the hourly rate limit to reserve (at most 50%). A value between 5% and 20% should be safe on most setups. +

+
+ {/if} +
+ {/if} +
diff --git a/webapp/src/routes/credentials/[forge_type]/[id]/+page.svelte b/webapp/src/routes/credentials/[forge_type]/[id]/+page.svelte index 91699f899..53b83a54d 100644 --- a/webapp/src/routes/credentials/[forge_type]/[id]/+page.svelte +++ b/webapp/src/routes/credentials/[forge_type]/[id]/+page.svelte @@ -418,6 +418,12 @@ {new Date((credential.rate_limit.reset || 0) * 1000).toLocaleString()}
+
+
Reserved for critical operations
+
+ {credential.reserve_usage_enabled ? `${credential.reserve_usage_percentage || 0}%` : 'Disabled'} +
+
{/if} diff --git a/webapp/swagger.yaml b/webapp/swagger.yaml index 6a13ec8ce..5e1746ddb 100644 --- a/webapp/swagger.yaml +++ b/webapp/swagger.yaml @@ -291,6 +291,22 @@ definitions: x-go-name: Name pat: $ref: '#/definitions/GithubPAT' + reserve_usage_enabled: + description: |- + ReserveUsageEnabled toggles whether or not to allocate a certain + percentage of the available rate limit to critical operations such + as delete operations for runners that have finished their jobs. + type: boolean + x-go-name: ReserveUsageEnabled + reserve_usage_percentage: + description: |- + ReserveUsagePercentage is the percentage of available rate limit reserved + for critical operations. Setting this value too high will negatively impact + normal operations, so it is capped at 50%. A value between 5% and 20% + should be safe on most setups. Adjust this based on your usage patterns. + format: int64 + type: integer + x-go-name: ReserveUsagePercentage type: object x-go-package: github.com/cloudbase/garm/params CreateGithubEndpointParams: @@ -791,6 +807,22 @@ definitions: $ref: '#/definitions/Repository' type: array x-go-name: Repositories + reserve_usage_enabled: + description: |- + ReserveUsageEnabled toggles whether or not to allocate a certain + percentage of the available rate limit to critical operations such + as delete operations for runners that have finished their jobs. + type: boolean + x-go-name: ReserveUsageEnabled + reserve_usage_percentage: + description: |- + ReserveUsagePercentage is the percentage of available rate limit reserved + for critical operations. Setting this value too high will negatively impact + normal operations, so it is capped at 50%. A value between 5% and 20% + should be safe on most setups. Adjust this based on your usage patterns. + format: int64 + type: integer + x-go-name: ReserveUsagePercentage updated_at: format: date-time type: string @@ -2328,6 +2360,22 @@ definitions: x-go-name: Name pat: $ref: '#/definitions/GithubPAT' + reserve_usage_enabled: + description: |- + ReserveUsageEnabled toggles whether or not to allocate a certain + percentage of the available rate limit to critical operations such + as delete operations for runners that have finished their jobs. + type: boolean + x-go-name: ReserveUsageEnabled + reserve_usage_percentage: + description: |- + ReserveUsagePercentage is the percentage of available rate limit reserved + for critical operations. Setting this value too high will negatively impact + normal operations, so it is capped at 50%. A value between 5% and 20% + should be safe on most setups. Adjust this based on your usage patterns. + format: int64 + type: integer + x-go-name: ReserveUsagePercentage type: object x-go-package: github.com/cloudbase/garm/params UpdateGithubEndpointParams: diff --git a/workers/cache/tool_cache.go b/workers/cache/tool_cache.go index f442c8091..754018c83 100644 --- a/workers/cache/tool_cache.go +++ b/workers/cache/tool_cache.go @@ -278,7 +278,16 @@ reset: now := time.Now().UTC() if now.After(t.lastUpdate.Add(time.Duration(githubToolsUpdateDeadline) * time.Minute)) { slog.DebugContext(t.ctx, "last update after deadline", "last_update", t.lastUpdate, "deadline", t.lastUpdate.Add(time.Duration(githubToolsUpdateDeadline)*time.Minute)) - if err := t.updateTools(); err != nil { + // The tools updater is a critical consumer: pool managers are + // disabled when the tools cache expires (1 hour validity), so it + // keeps refreshing even when the remaining quota has dipped into the + // configured reserve, and only skips calls that are guaranteed to + // fail. Skip silently: the call cannot succeed and a status event + // per tick would just spam the entity event log. With the 1 minute + // tick, a refresh happens at most a minute after the quota resets. + if limited, resetAt := cache.EntityRateLimitExhausted(t.entity.ID); limited { + slog.DebugContext(t.ctx, "rate limit exhausted; deferring tools update", "reset_at", resetAt) + } else if err := t.updateTools(); err != nil { slog.ErrorContext(t.ctx, "updating tools", "error", err) t.addStatusEvent(fmt.Sprintf("failed to update tools: %q", err), params.EventError) } else { @@ -304,6 +313,11 @@ reset: if !now.After(t.lastUpdate.Add(time.Duration(githubToolsUpdateDeadline) * time.Minute)) { continue } + // See the comment on the check in the reset section above. + if limited, resetAt := cache.EntityRateLimitExhausted(t.entity.ID); limited { + slog.DebugContext(t.ctx, "rate limit exhausted; deferring tools update", "reset_at", resetAt) + continue + } slog.DebugContext(t.ctx, "updating tools") if err := t.updateTools(); err != nil { slog.ErrorContext(t.ctx, "updating tools", "error", err) diff --git a/workers/entity/worker.go b/workers/entity/worker.go index d2f795c4e..601b31cd2 100644 --- a/workers/entity/worker.go +++ b/workers/entity/worker.go @@ -239,6 +239,7 @@ func (w *Worker) consolidateRunnerLoop() { ticker := time.NewTicker(common.PoolReapTimeoutInterval) defer ticker.Stop() + rateLimited := false for { select { case _, ok := <-ticker.C: @@ -246,6 +247,20 @@ func (w *Worker) consolidateRunnerLoop() { slog.InfoContext(w.ctx, "consolidate ticker closed") return } + // Consolidation lists all runners from the forge and fans out + // into the scale set workers' reap and cleanup routines which are + // all non-critical forge API consumers. Skip while rate limited. + if limited, resetAt := cache.EntityRateLimitReached(w.Entity.ID); limited { + if !rateLimited { + slog.InfoContext(w.ctx, "rate limit reached; pausing runner consolidation until the quota resets", "reset_at", resetAt) + rateLimited = true + } + continue + } + if rateLimited { + slog.InfoContext(w.ctx, "rate limit lifted; resuming runner consolidation") + rateLimited = false + } if err := w.consolidateRunnerState(); err != nil { w.addStatusEvent(fmt.Sprintf("failed to consolidate runner state: %q", err.Error()), params.EventError) slog.With(slog.Any("error", err)).Error("failed to consolidate runner state") diff --git a/workers/entity/worker_watcher.go b/workers/entity/worker_watcher.go index ce8fd2444..06a752267 100644 --- a/workers/entity/worker_watcher.go +++ b/workers/entity/worker_watcher.go @@ -91,7 +91,7 @@ func (w *Worker) handleEntityCredentialsEventPayload(event dbCommon.ChangePayloa switch event.Operation { case dbCommon.UpdateOperation: - slog.DebugContext(w.ctx, "got delete operation") + slog.DebugContext(w.ctx, "got update operation") w.mux.Lock() defer w.mux.Unlock() if w.Entity.Credentials.GetID() != creds.GetID() { diff --git a/workers/scaleset/controller.go b/workers/scaleset/controller.go index 3cb5ede0b..af72eca2b 100644 --- a/workers/scaleset/controller.go +++ b/workers/scaleset/controller.go @@ -22,6 +22,7 @@ import ( "golang.org/x/sync/errgroup" + "github.com/cloudbase/garm/cache" dbCommon "github.com/cloudbase/garm/database/common" "github.com/cloudbase/garm/database/watcher" "github.com/cloudbase/garm/params" @@ -246,6 +247,14 @@ func (c *Controller) retryFailedScaleSets() { return } + // Starting a scale set worker involves several forge calls (runner + // group lookup, scale set reconciliation), which cannot succeed on an + // exhausted quota. Retrying would only churn logs and backoff state. + if limited, resetAt := cache.EntityRateLimitExhausted(c.Entity.ID); limited { + slog.DebugContext(c.ctx, "rate limit exhausted; deferring scale set retries", "reset_at", resetAt) + return + } + c.ScaleSets.Range(func(key, value any) bool { scaleSetID := key.(uint) set := value.(*scaleSet) diff --git a/workers/scaleset/scaleset.go b/workers/scaleset/scaleset.go index 5341f2059..2d5f9b3bb 100644 --- a/workers/scaleset/scaleset.go +++ b/workers/scaleset/scaleset.go @@ -102,6 +102,20 @@ type Worker struct { mux sync.Mutex running bool quit chan struct{} + + // authFailed marks that the forge rejected our credentials. While set, + // the listener is not restarted and scaling is paused, instead of + // hammering the forge with calls that cannot succeed. The latch clears + // when the entity worker installs a fresh client after a credentials + // update, and is retried periodically as a safety valve. This is kept + // separate from rate limiting: a quota reset must not resume a worker + // whose credentials are still rejected. + authFailed bool + authFailedCli common.GithubClient + authFailureAt time.Time + // scalingPauseReason tracks why scaling is paused, so pause/resume + // transitions are logged once instead of on every autoscale tick. + scalingPauseReason string } func (w *Worker) ensureScaleSetInGitHub() error { @@ -890,10 +904,109 @@ func (w *Worker) sleepWithCancel(sleepTime time.Duration) (canceled bool) { return true } +// forgeAuthRetryInterval is how often we probe the forge again while the +// auth failure latch is set, in case access was restored out of band (for +// example an app installation that was unsuspended, or a PAT authorized +// for an org enforcing SAML SSO) without a credentials update in GARM. +const forgeAuthRetryInterval = 5 * time.Minute + +// markAuthFailure latches the auth failure state after a forge call was +// rejected as unauthorized. Callers must not hold w.mux. +func (w *Worker) markAuthFailure() { + // The scale set API maps both 401 and 403 to ErrUnauthorized. A fully + // exhausted quota can also produce 403 responses; treat that as a rate + // limit condition, not an auth failure — the rate limit gate handles it + // and clears on its own when the quota resets. + if limited, _ := cache.EntityRateLimitExhausted(w.entity.ID); limited { + slog.WarnContext(w.ctx, "forge call rejected while rate limit is exhausted; treating as rate limited") + return + } + cli, ok := cache.GetGithubClient(w.entity.ID) + if !ok { + cli = nil + } + w.mux.Lock() + w.authFailed = true + w.authFailedCli = cli + w.authFailureAt = time.Now() + w.mux.Unlock() + slog.WarnContext(w.ctx, "forge rejected our credentials; pausing scale set forge operations until credentials are updated") +} + +func (w *Worker) clearAuthFailureLocked() { + if w.authFailed { + slog.InfoContext(w.ctx, "forge credentials accepted again; resuming scale set forge operations") + } + w.authFailed = false + w.authFailedCli = nil +} + +func (w *Worker) clearAuthFailure() { + w.mux.Lock() + defer w.mux.Unlock() + w.clearAuthFailureLocked() +} + +// forgeAuthBlockedLocked reports whether forge operations should stay paused +// due to a previously latched auth failure. Callers must hold w.mux. +func (w *Worker) forgeAuthBlockedLocked() bool { + if !w.authFailed { + return false + } + // A credentials update makes the entity worker install a fresh client in + // the cache. Give the new client a chance immediately. + if cli, ok := cache.GetGithubClient(w.entity.ID); ok && cli != w.authFailedCli { + w.clearAuthFailureLocked() + return false + } + // Safety valve: probe again periodically even without a credentials + // change. On failure the latch is simply re-armed with a fresh timestamp. + if time.Since(w.authFailureAt) >= forgeAuthRetryInterval { + return false + } + return true +} + +// scalingPausedLocked reports whether scaling in either direction should +// pause. Scaling talks to the forge (JIT config generation on the way up, +// runner removal on the way down), so it pauses while the credentials are +// rejected or while the remaining quota has dipped into the configured +// reserve. The two conditions are independent: a rate limit reset does not +// resume a worker whose credentials are still rejected, and vice versa. +// Pause and resume transitions are logged once. Callers must hold w.mux. +func (w *Worker) scalingPausedLocked() bool { + var reason string + if w.forgeAuthBlockedLocked() { + reason = "forge rejected our credentials" + } else if limited, resetAt := cache.EntityRateLimitReached(w.entity.ID); limited { + reason = fmt.Sprintf("rate limit reached; quota resets at %s", resetAt.UTC().Format(time.RFC3339)) + } + if reason != w.scalingPauseReason { + if reason == "" { + slog.InfoContext(w.ctx, "resuming scale set scaling") + } else { + slog.InfoContext(w.ctx, "pausing scale set scaling", "reason", reason) + } + w.scalingPauseReason = reason + } + return reason != "" +} + func (w *Worker) sessionLoopMayRun() bool { w.mux.Lock() defer w.mux.Unlock() - return w.scaleSet.Enabled + if !w.scaleSet.Enabled || w.forgeAuthBlockedLocked() { + return false + } + // While the quota is fully exhausted, session (re)creation cannot + // succeed either: it needs a runner registration token, which is the + // one REST call in the message pipeline. Everything downstream (the + // broker long poll, session refresh against the actions service) uses + // session or admin tokens and does not consume the REST quota, so an + // established listener keeps running and only session (re)creation is + // held back. It resumes on its own once the quota resets. + limited, _ := cache.EntityRateLimitExhausted(w.entity.ID) + return !limited } func (w *Worker) keepListenerAlive() { @@ -910,6 +1023,11 @@ Loop: // noop if already started. if err := w.listener.Start(); err != nil { slog.ErrorContext(w.ctx, "error starting listener", "error", err, "consumer_id", w.consumerID) + if errors.Is(err, runnerErrors.ErrUnauthorized) { + // Latch the auth failure; sessionLoopMayRun() keeps us parked + // until credentials are updated or the retry interval elapses. + w.markAuthFailure() + } if canceled := w.sleepWithCancel(2 * time.Second); canceled { slog.InfoContext(w.ctx, "worker is stopped; exiting keepListenerAlive") return @@ -917,6 +1035,7 @@ Loop: // we failed to start the listener. Try again. continue } + w.clearAuthFailure() select { case <-w.quit: @@ -939,15 +1058,20 @@ Loop: w.mux.Unlock() for { w.mux.Lock() - // In case the scaleset was disabled while we were in the - // backoff sleep. - if !w.scaleSet.Enabled { + // In case the scaleset was disabled or the credentials were + // rejected while we were in the backoff sleep. The outer loop + // parks until the condition clears. + if !w.scaleSet.Enabled || w.forgeAuthBlockedLocked() { w.mux.Unlock() continue Loop } slog.DebugContext(w.ctx, "attempting to restart") if err := w.listener.Start(); err != nil { w.mux.Unlock() + if errors.Is(err, runnerErrors.ErrUnauthorized) { + w.markAuthFailure() + continue Loop + } switch backoff { case 0: backoff = 5 * time.Second @@ -963,6 +1087,7 @@ Loop: continue } backoff = 0 + w.clearAuthFailureLocked() w.mux.Unlock() continue Loop } @@ -1222,6 +1347,14 @@ func (w *Worker) handleAutoScale() { } } + // Instance cleanup above is database-only and always runs; the + // scaling decisions below talk to the forge and pause while rate + // limited or while our credentials are rejected. + if w.scalingPausedLocked() { + w.mux.Unlock() + continue + } + if w.runnerCount() == w.targetRunners() { lastMsgDebugLog("desired runner count reached", w.targetRunners(), w.runnerCount()) w.mux.Unlock()