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.

Author: mnoah1

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. **Default 0** disables general throttling. |
 
 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.
 

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,
 	}
 }
 
@@ -230,6 +233,11 @@ func (c *Controller) coalesce(ctx context.Context, store storage.Storage, reques
 // re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate
 // defers by holding the delivery (redeliver after the gate wait delay) rather than failing.
 func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, delivery consumer.Delivery, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error {
+	if cfg.MinimumBuildAdmissionIntervalMs < 0 {
+		metrics.NamedCounter(c.metricsScope, _opName, "config_errors", 1, metrics.TagsFromContext(ctx)...)
+		return fmt.Errorf("minimum build admission interval must not be negative for queue %s, got %dms", request.Queue, cfg.MinimumBuildAdmissionIntervalMs)
+	}
+
 	var sc sourcecontrol.SourceControl
 	var strategy entity.BuildStrategy
 	var baseURI string
@@ -239,6 +247,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 +267,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 +361,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 +512,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")

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"
@@ -48,6 +49,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 +73,25 @@ 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 }
 
+type staticQueueConfigStore struct{ config entity.QueueConfig }
+
+func (s staticQueueConfigStore) Get(context.Context, string) (entity.QueueConfig, error) {
+	return s.config, nil
+}
+
+func (s staticQueueConfigStore) List(context.Context) ([]entity.QueueConfig, error) {
+	return []entity.QueueConfig{s.config}, 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 +132,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 +530,7 @@ func TestProcess(t *testing.T) {
 		wantHoldMs int64
 		wantErr    bool
 		wantRetry  bool
+		config     entity.QueueConfig
 	}{
 		{
 			name: "superseded redelivery repairs its state log",
@@ -730,6 +753,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 is rejected before claiming a slot",
+			wantErr: true,
+			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)
+			},
+		},
 		{
 			name:       "gate closed after slot claim race holds",
 			wantHoldMs: 5000,
@@ -754,6 +838,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 +1103,9 @@ 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 != "" {
+				c.queueConfigs = staticQueueConfigStore{config: tt.config}
+			}
 			if tt.setup != nil {
 				tt.setup(m)
 			}

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. Zero disables time-based throttling.
+	MinimumBuildAdmissionIntervalMs int64 `json:"minimum_build_admission_interval_ms" yaml:"minimum_build_admission_interval_ms"`
 }