From c571f7f745a3aa86fd27e4d782fc1d906e559750 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 15 Sep 2026 17:08:36 +0000 Subject: [PATCH] feat(stovepipe): throttle logical build admissions Summary: Intent: - Allow queues to limit expensive logical build admissions independently of concurrency. Changes: - Gate process admission on the durable deadline and hold ineligible deliveries. - Advance the deadline atomically while claiming a build slot. - Add the general minimum admission interval policy, metrics, tests, and design documentation. This change builds on the durable admission state introduced by the parent PR. --- doc/rfc/stovepipe/steps/process.md | 29 ++--- stovepipe/controller/process/BUILD.bazel | 1 + stovepipe/controller/process/process.go | 43 +++++++- stovepipe/controller/process/process_test.go | 104 ++++++++++++++++++ stovepipe/entity/queue_config.go | 3 + stovepipe/extension/queueconfig/README.md | 2 +- .../extension/queueconfig/default/default.go | 12 +- .../queueconfig/default/default_test.go | 1 + 8 files changed, 172 insertions(+), 23 deletions(-) diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index 8f6962733..8f751d9af 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -25,11 +25,11 @@ For a delivery carrying request id `R`: 4. R.State is accepted. Load the Queue row Q. 5. Coalesce: if CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0: - a newer head exists -> mark R superseded, ack, return. (No slot consumed.) -6. R is the latest head. Gate: if Q.in_flight_count >= max_concurrent (from queue config; see below): - - defer (hold the delivery) -> re-check on redelivery until the slot frees (admit) or a newer head supersedes it. See [Waiting for a slot](#waiting-for-a-slot). +6. R is the latest head. Gate: if Q.in_flight_count >= max_concurrent or Q.build_admission_not_before_ms is in the future: + - defer (hold the delivery) -> re-check on redelivery until both gates open (admit) or a newer head supersedes it. See [Waiting for admission](#waiting-for-admission). 7. Admit R: a. Derive build strategy + baseline (see "Build-strategy decision"). - b. CAS the Queue row: in_flight_count += 1. + b. CAS the Queue row: in_flight_count += 1 and advance build_admission_not_before_ms by the configured minimum admission interval. c. CAS the Request: accepted -> processing, persist build_strategy + base_uri. d. Announce validation start on the hook topic (see "Hooks"). e. Publish R to build. @@ -60,8 +60,10 @@ Validation is expensive and shares a baseline, so heads arriving while an earlie | Source | Field | Meaning | |---|---|---| | Queue row | `last_green_uri` | Bookmark `record` advances on whole-repo green; empty until first green. | -| Queue row | `in_flight_count` | Requests past `process` and not yet terminal. `process` increments on admit; `record` (or DLQ reconciliation) decrements on terminal. | +| Queue row | `in_flight_count` | Requests past `process` and not yet terminal. `process` increments on admit; `buildsignal` (or DLQ reconciliation) decrements on terminal. | +| Queue row | `build_admission_not_before_ms` | Durable earliest time for the next logical admission; zero until a time policy advances it. | | Queue config | `max_concurrent` | Cap on concurrent in-flight validations. **Default 1** (global wiring default for MVP; per-queue override when a Stovepipe `queueconfig` extension lands). | +| Queue config | `minimum_build_admission_interval_ms` | Minimum start-to-start spacing between logical admissions. Positive values enable general throttling; non-positive values disable it. **Default 0**. | A slot is held from admit until the build goes terminal (`process → build → buildsignal`), not just while `process` runs. It is released when the Request reaches **any** terminal state and `in_flight_count` is decremented — `buildsignal` recording the build's outcome, success *or* failure, or the DLQ reconciler forcing a terminal `failed` (see [integrity](#in_flight_count-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot. @@ -115,7 +117,7 @@ The gate is **not** tied to `process` returning; a slot taken at admit is held u **Rules** -1. **One slot per in-flight validation** (MVP: one per Queue). `process` increments `in_flight_count` on admit; `record` decrements on terminal. +1. **One slot per in-flight validation** (MVP: one per Queue). `process` increments `in_flight_count` on admit; `buildsignal` decrements when the build becomes terminal. 2. **No skip-ahead while in-flight.** The latest head waits for a slot until the running validation completes; it never preempts. 3. **Intermediates are superseded on sight**, gate open or closed — no slot consumed (step 5). 4. **Coalesce-to-latest on gate open.** When a slot frees, the waiting latest head is admitted. @@ -135,9 +137,9 @@ A, D, F each get a full cycle; B, C, E end `superseded`. No intermediate is vali **What does not happen** -- `process` returning does **not** free a slot — only `record` (or DLQ reconciliation) does. +- `process` returning does **not** free a slot — only `buildsignal` (or DLQ reconciliation) does. - A newer head does **not** preempt an in-flight validation. -- Deferred messages are **not** failed or dead-lettered — they wait for the gate (see [Waiting for a slot](#waiting-for-a-slot)). +- Deferred messages are **not** failed or dead-lettered — they wait for the gate (see [Waiting for admission](#waiting-for-admission)). ## Hooks @@ -170,10 +172,10 @@ The window to handle is "count incremented, state not yet `processing`". Admit d `in_flight_count` is a cache; the source of truth is **the set of non-terminal Request rows for the Queue**. Two rules keep it from drifting: -1. **Decrement is bound to the terminal transition.** The single CAS that moves a Request non-terminal → terminal (in `record` or the DLQ reconciler) also decrements. Being CAS-guarded, it fires exactly once per Request even under redelivery. +1. **Decrement precedes the terminal transition.** `buildsignal` or the DLQ reconciler decrements before moving a Request non-terminal → terminal, so a terminal request never strands a slot. Redelivery may transiently over-release after a crash between the two entity writes, which is preferred to a permanent capacity leak. 2. **Increment is bound to the admit transition.** `process` increments only on the `accepted → processing` CAS; a redelivery of an already-`processing` Request takes step 3 and does not increment again. -On a crash between admit and `record`, the Request stays non-terminal; visibility-timeout redelivery drives it forward, and the fail-closed DLQ path eventually forces it terminal, decrementing as it does. The count can drift high only transiently and self-heals as stuck Requests terminate. A reconciler that recomputes the count from non-terminal rows can be added later if drift proves real, but isn't required for MVP. +On a crash between admit and the terminal outcome, the Request stays non-terminal; visibility-timeout redelivery drives it forward, and the fail-closed DLQ path eventually forces it terminal, decrementing as it does. The count can drift high only transiently and self-heals as stuck Requests terminate. A reconciler that recomputes the count from non-terminal rows can be added later if drift proves real, but isn't required for MVP. ## Edge cases @@ -192,7 +194,8 @@ Runtime coordination only — fields the pipeline writes under CAS: |---|---|---| | `name` | Stable logical id (`monorepo/main`); the string ingest accepts | ingest (create) | | `last_green_uri` | Bookmark; empty until first green | record | -| `in_flight_count` | Active Phase 1 validations | process (+1), record/DLQ (−1) | +| `in_flight_count` | Active Phase 1 validations | process (+1), buildsignal/DLQ (−1) | +| `build_admission_not_before_ms` | Earliest Unix-millisecond time for another logical admission | process and terminal-result policies | | `latest_request_id` | Request id of the newest head ingest accepted | ingest | | `version` | Optimistic-locking version | all writers | @@ -227,14 +230,14 @@ New key/value-shaped operations (single-key reads/writes, no server-side filteri No "list requests by queue/state" query is introduced; coalescing uses the single-row `latest_request_id` pointer instead, keeping the contract satisfiable by a plain KV backend. -## Waiting for a slot +## Waiting for admission -When the gate is closed, `process` must defer the latest head without admitting it (no `in_flight_count` increment, no publish to `build`). The mechanism is the consumer hold primitive ([consumer-hold.md](../../consumer-hold.md)): the controller records a hold for `gate_wait_delay_ms` and returns success, and the framework postpones the delivery — the same message redelivers after the delay, and the redelivery does not count toward `MaxAttempts`. +When either gate is closed, `process` must defer the latest head without admitting it (no `in_flight_count` increment, no publish to `build`). The mechanism is the consumer hold primitive ([consumer-hold.md](../../consumer-hold.md)): the controller records a hold for at most `gate_wait_delay_ms` and returns success, and the framework postpones the delivery — the same message redelivers after the delay, and the redelivery does not count toward `MaxAttempts`. Every wake-up re-runs the same **coalesce-then-gate** checks (steps 5 → 6): 1. **Stale? (checked first.)** If `CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0`, `R` is no longer latest → supersede it (ack). A newer head is admitted by its own delivery when its slot attempt runs. -2. **Slot free?** If `in_flight_count < max_concurrent` (from config) and `R` is still latest → admit (step 7). +2. **Capacity and time eligible?** If `in_flight_count < max_concurrent`, `build_admission_not_before_ms <= now`, and `R` is still latest → admit (step 7). Nothing is admitted to `build` until the gate opens. diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index 27fbe8d7d..fd3874cfa 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -44,6 +44,7 @@ go_test( "//stovepipe/core/requestlog/mock:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", + "//stovepipe/extension/queueconfig/mock:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/sourcecontrol/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 991899ea0..99bb0c04b 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -22,6 +22,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" @@ -55,6 +56,7 @@ type Controller struct { registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string + now func() time.Time } // Verify Controller implements consumer.Controller interface at compile time. @@ -85,6 +87,7 @@ func NewController( registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, + now: time.Now, } } @@ -239,6 +242,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, if queueRow.InFlightCount >= cfg.MaxConcurrent { return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightCount, cfg.GateWaitDelayMs) } + nowMs := c.now().UnixMilli() + if queueRow.BuildAdmissionNotBeforeMs > nowMs { + return c.holdForBuildThrottle(ctx, delivery, request, queueRow.BuildAdmissionNotBeforeMs, nowMs, cfg.GateWaitDelayMs) + } if queueRow.LastGreenURI != "" && sc == nil { sc, err = c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) @@ -255,7 +262,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, return err } - err = c.claimBuildSlot(ctx, store, &queueRow) + err = c.claimBuildSlot(ctx, store, &queueRow, nowMs, cfg.MinimumBuildAdmissionIntervalMs) if err == nil { break } @@ -349,13 +356,19 @@ func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.S return entity.BuildStrategyFull, "", nil } -// claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it -// reloads queueRow and returns ErrVersionMismatch so the caller can retry. -func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, queueRow *entity.Queue) error { +// claimBuildSlot atomically claims capacity and reserves the next logical admission time. +// On version mismatch it reloads queueRow and returns ErrVersionMismatch so the caller can retry. +func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, queueRow *entity.Queue, admittedAtMs, minimumIntervalMs int64) error { queueStore := store.GetQueueStore() updated := *queueRow updated.InFlightCount = queueRow.InFlightCount + 1 + if minimumIntervalMs > 0 { + notBeforeMs := admittedAtMs + minimumIntervalMs + if notBeforeMs > updated.BuildAdmissionNotBeforeMs { + updated.BuildAdmissionNotBeforeMs = notBeforeMs + } + } newVersion := queueRow.Version + 1 if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { @@ -494,6 +507,28 @@ func (c *Controller) holdForBuildSlot(ctx context.Context, delivery consumer.Del return nil } +func (c *Controller) holdForBuildThrottle(ctx context.Context, delivery consumer.Delivery, request entity.Request, notBeforeMs, nowMs, gateWaitDelayMs int64) error { + if gateWaitDelayMs <= 0 { + metrics.NamedCounter(c.metricsScope, _opName, "config_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("requires a positive gate wait delay for queue %s, got %dms", request.Queue, gateWaitDelayMs) + } + + delayMs := notBeforeMs - nowMs + if delayMs > gateWaitDelayMs { + delayMs = gateWaitDelayMs + } + delivery.Hold(delayMs) + metrics.NamedCounter(c.metricsScope, _opName, "admission_throttled", 1, metrics.TagsFromContext(ctx)...) + c.logger.Infow("holding latest head until build admission is eligible", + "request_id", request.ID, + "queue", request.Queue, + "uri", request.URI, + "not_before_ms", notBeforeMs, + "delay_ms", delayMs, + ) + return nil +} + // loadRequest returns the request for id. func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { return loader.ByID(ctx, id, store.GetRequestStore().Get, "request") diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 7677e1d86..5cfe28f38 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -35,6 +36,7 @@ import ( requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" "github.com/uber/submitqueue/stovepipe/entity" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" + queueconfigmock "github.com/uber/submitqueue/stovepipe/extension/queueconfig/mock" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -48,6 +50,7 @@ const ( testID = "request/monorepo/main/7" testOlderID = "request/monorepo/main/3" testURI = "git://repo/monorepo/main/abc123" + testNowMs = int64(2_000_000) ) func queueContext(queueName string) context.Context { @@ -71,6 +74,15 @@ type staticStorageFactory struct{ store storage.Storage } // For returns the fixed store aggregate for any queue. func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } +func queueConfig(minimumBuildAdmissionIntervalMs int64) entity.QueueConfig { + return entity.QueueConfig{ + Name: testQueue, + MaxConcurrent: 1, + GateWaitDelayMs: 5000, + MinimumBuildAdmissionIntervalMs: minimumBuildAdmissionIntervalMs, + } +} + func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() return newControllerWithScope(t, ctrl, tally.NewTestScope("test", nil)) @@ -111,6 +123,7 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S stovepipemq.TopicKeyProcess, "stovepipe-process", ) + c.now = func() time.Time { return time.UnixMilli(testNowMs) } return c, m } @@ -508,6 +521,7 @@ func TestProcess(t *testing.T) { wantHoldMs int64 wantErr bool wantRetry bool + config entity.QueueConfig }{ { name: "superseded redelivery repairs its state log", @@ -730,6 +744,67 @@ func TestProcess(t *testing.T) { }, nil) }, }, + { + name: "latest accepted head holds while admission deadline is active", + wantHoldMs: 5000, + config: queueConfig(3_600_000), + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + BuildAdmissionNotBeforeMs: testNowMs + 7500, + Version: 1, + }, nil) + }, + }, + { + name: "admission deadline uses remaining duration below gate wait", + wantHoldMs: 2500, + config: queueConfig(3_600_000), + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + BuildAdmissionNotBeforeMs: testNowMs + 2500, + Version: 1, + }, nil) + }, + }, + { + name: "admission claim reserves the next configured interval", + config: queueConfig(3_600_000), + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, Version: 1, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + InFlightCount: 1, + BuildAdmissionNotBeforeMs: testNowMs + 3_600_000, + Version: 1, + }, int32(1), int32(2)).Return(nil) + updatedReq := acceptedRequest(testID) + updatedReq.State = entity.RequestStateProcessing + updatedReq.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) + expectStartAnnounceAndBuildPublish(t, m, testID) + }, + }, + { + name: "negative admission interval leaves throttling disabled", + config: queueConfig(-1), + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, Version: 1, + }, nil) + expectAdmit(t, m, testID) + }, + }, { name: "gate closed after slot claim race holds", wantHoldMs: 5000, @@ -754,6 +829,30 @@ func TestProcess(t *testing.T) { }, nil) }, }, + { + name: "claim conflict reload observes a concurrent admission deadline", + wantHoldMs: 5000, + config: queueConfig(3_600_000), + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, Version: 1, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + InFlightCount: 1, + BuildAdmissionNotBeforeMs: testNowMs + 3_600_000, + Version: 1, + }, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + BuildAdmissionNotBeforeMs: testNowMs + 10_000, + Version: 2, + }, nil) + }, + }, { name: "claim slot retries on queue version mismatch then admits", setup: func(m processMocks) { @@ -995,6 +1094,11 @@ func TestProcess(t *testing.T) { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) + if tt.config.Name != "" { + queueConfigs := queueconfigmock.NewMockStore(ctrl) + queueConfigs.EXPECT().Get(gomock.Any(), testQueue).Return(tt.config, nil).AnyTimes() + c.queueConfigs = queueConfigs + } if tt.setup != nil { tt.setup(m) } diff --git a/stovepipe/entity/queue_config.go b/stovepipe/entity/queue_config.go index b6681cd83..14ba5c159 100644 --- a/stovepipe/entity/queue_config.go +++ b/stovepipe/entity/queue_config.go @@ -26,4 +26,7 @@ type QueueConfig struct { MaxConcurrent int32 `json:"max_concurrent" yaml:"max_concurrent"` // GateWaitDelayMs is the redelivery delay while the latest head waits for a slot. GateWaitDelayMs int64 `json:"gate_wait_delay_ms" yaml:"gate_wait_delay_ms"` + // MinimumBuildAdmissionIntervalMs is the minimum start-to-start spacing between logical + // build admissions for this queue. Non-positive values disable time-based throttling. + MinimumBuildAdmissionIntervalMs int64 `json:"minimum_build_admission_interval_ms" yaml:"minimum_build_admission_interval_ms"` } diff --git a/stovepipe/extension/queueconfig/README.md b/stovepipe/extension/queueconfig/README.md index 40f88558d..74708b661 100644 --- a/stovepipe/extension/queueconfig/README.md +++ b/stovepipe/extension/queueconfig/README.md @@ -10,7 +10,7 @@ Pipeline stages read mutable runtime state from storage and read knobs such as ` ## Entities -Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`) separate from the mutable `Queue` row. +Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`, `minimum_build_admission_interval_ms`) separate from the mutable `Queue` row. The minimum interval is start-to-start spacing between logical admissions; zero disables time-based throttling. ## Implementations diff --git a/stovepipe/extension/queueconfig/default/default.go b/stovepipe/extension/queueconfig/default/default.go index 5fb696710..29aa7d636 100644 --- a/stovepipe/extension/queueconfig/default/default.go +++ b/stovepipe/extension/queueconfig/default/default.go @@ -25,8 +25,9 @@ import ( ) const ( - _defaultMaxConcurrent = 1 - _defaultGateWaitDelayMs = 5000 + _defaultMaxConcurrent = 1 + _defaultGateWaitDelayMs = 5000 + _defaultMinimumBuildAdmissionIntervalMs = 0 ) // Store is a queueconfig.Store that returns the same defaults for every queue. @@ -43,9 +44,10 @@ func (Store) Get(_ context.Context, name string) (entity.QueueConfig, error) { return entity.QueueConfig{}, queueconfig.ErrNotFound } return entity.QueueConfig{ - Name: name, - MaxConcurrent: _defaultMaxConcurrent, - GateWaitDelayMs: _defaultGateWaitDelayMs, + Name: name, + MaxConcurrent: _defaultMaxConcurrent, + GateWaitDelayMs: _defaultGateWaitDelayMs, + MinimumBuildAdmissionIntervalMs: _defaultMinimumBuildAdmissionIntervalMs, }, nil } diff --git a/stovepipe/extension/queueconfig/default/default_test.go b/stovepipe/extension/queueconfig/default/default_test.go index 490ff35a7..5df4e50fd 100644 --- a/stovepipe/extension/queueconfig/default/default_test.go +++ b/stovepipe/extension/queueconfig/default/default_test.go @@ -32,6 +32,7 @@ func TestStore_Get(t *testing.T) { assert.Equal(t, "monorepo/main", cfg.Name) assert.Equal(t, int32(1), cfg.MaxConcurrent) assert.Equal(t, int64(5000), cfg.GateWaitDelayMs) + assert.Zero(t, cfg.MinimumBuildAdmissionIntervalMs) }) t.Run("empty name is not found", func(t *testing.T) {