feat(orchestrator): keep per-state request membership records for listing a queue's requests by state

AGENTS.md

@@ -143,7 +143,7 @@ Vendor-agnostic, pluggable interfaces with implementations in subdirectories:
 3. **Service-scoped resolution** — when only one service of a multi-service domain resolves an extension, its `Factory`, its aggregate, its implementations, its mocks and its schema live at `{domain}/{service}/extension/{ext}/`. The behavioural contracts stay at `{domain}/extension/{ext}/`.
 4. Factory interface for dependency injection and lifecycle management (constructed in wiring, not inside `platform/extension` packages).
 
-**Split on reachability, not on declaration.** What a service may resolve is decided by the accessors on its aggregate, so that is the part worth scoping to a service — a gateway holding a four-accessor `Storage` cannot obtain an orchestrator store whatever else it can name. The contracts stay shared because moving them adds no enforcement and costs every domain-level caller a dependency on a service package. `submitqueue/extension/storage` is the worked example: thirteen store contracts, their mocks and the error vocabulary shared, while each service owns its own aggregate, MySQL implementation and schema.
+**Split on reachability, not on declaration.** What a service may resolve is decided by the accessors on its aggregate, so that is the part worth scoping to a service — a gateway holding a four-accessor `Storage` cannot obtain an orchestrator store whatever else it can name. The contracts stay shared because moving them adds no enforcement and costs every domain-level caller a dependency on a service package. `submitqueue/extension/storage` is the worked example: fourteen store contracts, their mocks and the error vocabulary shared, while each service owns its own aggregate, MySQL implementation and schema.
 
 **Extensions hold contracts and implementations only — not factories or routing.**
 

doc/rfc/service-scoped-extensions.md

@@ -26,7 +26,7 @@ Separating the services onto their own databases already worked — the orchestr
 2. An extension shared between a domain's services stays wholly at `{domain}/extension/{ext}/` — including one used by a single service today but expected in both. `submitqueue/extension/queueconfig/` is gateway-only now and stays where it is on that basis. A single-service domain is unaffected: its domain root is its service root, so nothing moves.
 3. Storage is the extension this split applied to. `changeprovider`, `validator`, `conflict`, `buildrunner` and `speculation` are orchestrator-only and would qualify under Decision 1, but none has a schema and none was the reason for this change; they stay at `{domain}/extension/` until a later RFC takes them. The trigger for moving storage was a schema that a separate-database deployment must not over-provision.
 4. SubmitQueue's storage aggregate splits in two: `submitqueue/gateway/extension/storage/` declares a `Factory` and a four-accessor `Storage`, `submitqueue/orchestrator/extension/storage/` a `Factory` and a nine-accessor `Storage`. Neither can resolve the other's stores.
-5. `submitqueue/extension/storage/` keeps the thirteen store contracts, their mocks, the error vocabulary, `Config`, and the package README, which documents the contracts. Its import path does not change, so callers that name only a contract or an error are untouched.
+5. `submitqueue/extension/storage/` keeps the fourteen store contracts, their mocks, the error vocabulary, `Config`, and the package README, which documents the contracts. Its import path does not change, so callers that name only a contract or an error are untouched.
 6. The contract package is not promoted to `platform/`. It is not cross-domain — `stovepipe/extension/storage/storage.go` already declares its own verbatim copy of the same symbols, so a per-domain error vocabulary is the existing pattern rather than something this RFC introduces.
 7. Each half owns its own MySQL schema directory, so the tables a service creates are the tables it reads. A union target over both stays available for deployments that colocate the services — the end-to-end tests and the local compose stack take it and name no service directory. Split schemas are the opt-in, not the default.
 
@@ -49,6 +49,7 @@ Every store, verified against actual usage rather than intent.
 | `BatchStore` | `batch` |
 | `BatchDependentStore` | `batch_dependent` |
 | `QueueBatchStateStore` | `queue_batch_state` |
+| `QueueRequestStateStore` | `queue_request_state` |
 | `BuildStore` | `build` |
 | `SpeculationPathSetStore` | `speculation_path_set` |
 | `PathBuildStore` | `path_build` |

platform/metrics/README.md

@@ -20,6 +20,7 @@ For any operation with a clear start and end, use `Begin` and `Complete`:
 |----------|-------|
 | `Begin(scope, name, buckets, ...tags)` | `{name}.start` counter +1 and returns an `Op` |
 | `op.Complete(err, ...tags)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` and any completion tags |
+| `op.CompleteAs(result, ...tags)` | `{name}.finish` histogram tagged with the given `result` and any completion tags, for outcomes the error alone does not classify (for example `result=not_found`) |
 
 `buckets` is required at `Begin` because operations differ widely in expected latency. The finish histogram records both the duration distribution and the number of completed operations, so `Complete` does not emit a separate counter.
 
@@ -79,13 +80,16 @@ metrics.NamedCounter(c.scope, "publish", "attempts", 1, metrics.NewTag("topic",
 
 ## Duration Buckets
 
-There is no default bucket set. The package exports four common sets:
+There is no default bucket set. The package exports five common sets:
 
 | Set | Range | Use for |
 |-----|-------|---------|
 | `FastLatencyBuckets` | ~100µs – 5s | Fast in-process work such as scoring, cache lookups, and CPU-bound operations |
 | `StorageLatencyBuckets` | ~1ms – 1m | Storage and message-queue round trips such as database reads, writes, publishing, and consuming |
 | `LongLatencyBuckets` | ~5ms – 4h | Long-running pipeline work and external calls such as builds, lands, pushes, and provider calls |
 | `ChangeAgeBuckets` | ~1m – 30d | Elapsed time measured from a source-control change's commit timestamp rather than from work this system started |
+| `LandLatencyBuckets` | ~1m – 24h | End-to-end request lifecycles, from accepting a land request to its terminal outcome |
 
 Pass one of these sets or a custom `tally.DurationBuckets` to `Begin` or `NamedHistogram`.
+
+For counts recorded with `RecordValue`, such as how many items a unit of work carries or depends on, use `CountBuckets` (0 – 144, Fibonacci-spaced, with zero in its own bucket).

platform/metrics/metrics.go

@@ -153,8 +153,40 @@ var (
 		14 * 24 * time.Hour,
 		30 * 24 * time.Hour,
 	}
+
+	// LandLatencyBuckets suits end-to-end request lifecycles, from accepting a
+	// land request to its terminal outcome. Resolution is densest between 5 and
+	// 60 minutes, where typical lands fall, and the tail reaches a day so that
+	// requests stuck overnight stay distinguishable instead of overflowing.
+	LandLatencyBuckets = tally.DurationBuckets{
+		1 * time.Minute,
+		2 * time.Minute,
+		3 * time.Minute,
+		5 * time.Minute,
+		7*time.Minute + 30*time.Second,
+		10 * time.Minute,
+		15 * time.Minute,
+		20 * time.Minute,
+		30 * time.Minute,
+		45 * time.Minute,
+		1 * time.Hour,
+		90 * time.Minute,
+		2 * time.Hour,
+		3 * time.Hour,
+		4 * time.Hour,
+		6 * time.Hour,
+		8 * time.Hour,
+		12 * time.Hour,
+		24 * time.Hour,
+	}
 )
 
+// CountBuckets suits small cardinalities recorded with RecordValue, such as how
+// many items a unit of work carries or depends on. Zero has its own bucket
+// because "none" is a meaningful answer, and the Fibonacci spacing keeps
+// resolution where counts usually fall while still separating outliers.
+var CountBuckets = tally.ValueBuckets{0, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144}
+
 // Op tracks the lifecycle of a named operation. It captures the start time on
 // creation, emits a {name}.start counter, and records the duration and result
 // on a {name}.finish histogram when Complete is called.
@@ -200,6 +232,13 @@ func (o Op) Complete(err error, tags ...Tag) {
 		}
 	}
 
+	o.CompleteAs(result, tags...)
+}
+
+// CompleteAs records elapsed time on the {name}.finish histogram like Complete,
+// but with a caller-chosen result tag. It serves outcomes that the error alone
+// does not classify, such as a lookup that found no record.
+func (o Op) CompleteAs(result string, tags ...Tag) {
 	tagged(o.scope, tags).
 		Tagged(map[string]string{"result": result}).
 		Histogram("finish", o.buckets).

platform/metrics/metrics_test.go

@@ -125,6 +125,46 @@ func TestOp_CompleteWithTags(t *testing.T) {
 	assert.True(t, ok, "expected finish-only tags, got keys: %v", histogramKeys(histograms))
 }
 
+func TestOp_CompleteAs(t *testing.T) {
+	tests := []struct {
+		name         string
+		result       string
+		tags         []Tag
+		histogramKey string
+	}{
+		{
+			name:         "records the given result",
+			result:       "not_found",
+			histogramKey: "process.finish+result=not_found",
+		},
+		{
+			name:         "records completion tags alongside the result",
+			result:       "not_found",
+			tags:         []Tag{NewTag("store", "build")},
+			histogramKey: "process.finish+result=not_found,store=build",
+		},
+		{
+			name:         "result tag wins over a completion tag of the same key",
+			result:       "not_found",
+			tags:         []Tag{NewTag("result", "success")},
+			histogramKey: "process.finish+result=not_found",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scope := tally.NewTestScope("", nil)
+			op := Begin(scope, "process", StorageLatencyBuckets)
+			op.CompleteAs(tt.result, tt.tags...)
+
+			histograms := scope.Snapshot().Histograms()
+			_, ok := histograms[tt.histogramKey]
+			assert.True(t, ok, "expected histogram key %s, got keys: %v", tt.histogramKey, histogramKeys(histograms))
+			assert.Len(t, histograms, 1)
+		})
+	}
+}
+
 func TestNamedCounter(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
 	NamedCounter(scope, "publish", "attempts", 5)
@@ -180,6 +220,7 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 		"StorageLatencyBuckets": StorageLatencyBuckets,
 		"LongLatencyBuckets":    LongLatencyBuckets,
 		"ChangeAgeBuckets":      ChangeAgeBuckets,
+		"LandLatencyBuckets":    LandLatencyBuckets,
 	}
 	for name, buckets := range sets {
 		t.Run(name, func(t *testing.T) {
@@ -192,6 +233,13 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 	}
 }
 
+func TestCountBuckets_Sorted(t *testing.T) {
+	for i := 1; i < len(CountBuckets); i++ {
+		assert.Greater(t, CountBuckets[i], CountBuckets[i-1],
+			"CountBuckets[%d] (%v) must be greater than CountBuckets[%d] (%v)", i, CountBuckets[i], i-1, CountBuckets[i-1])
+	}
+}
+
 // counterKeys extracts map keys for error messages.
 func counterKeys(m map[string]tally.CounterSnapshot) []string {
 	keys := make([]string, 0, len(m))

service/submitqueue/gateway/server/main.go

@@ -366,7 +366,7 @@ func run() error {
 	// Create controllers and wrap them for gRPC. Every store is queue-scoped and
 	// resolves through the factory adapter; land/cancel/log share one materializer.
 	storageFty := storageFactory{backend: store}
-	materializer := requestcore.NewMaterializer(storageFty)
+	materializer := requestcore.NewMaterializer(storageFty, requestcore.WithMetricsScope(scope))
 	pingController := controller.NewPingController(logger, scope)
 	landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry)
 	cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry)

service/submitqueue/orchestrator/server/main.go

@@ -143,7 +143,7 @@ func run() error {
 
 	cnt := counterFactory{db: appDB, scope: scope.SubScope("counter")}
 
-	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"))
+	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"), mysqlstorage.WithQueueRequestStateRecords())
 	if err != nil {
 		return fmt.Errorf("failed to create storage: %w", err)
 	}

submitqueue/entity/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "queue_batch_state.go",
         "queue_config.go",
         "queue_pause.go",
+        "queue_request_state.go",
         "request.go",
         "request_batch.go",
         "request_history.go",

submitqueue/entity/queue_request_state.go

@@ -0,0 +1,34 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// 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 entity
+
+// QueueRequestState is a membership record filing one in-queue request under its current
+// state. Exactly one record exists for a request from its creation until it reaches a
+// terminal state, and none afterwards.
+//
+// The (Queue, State, RequestID) triple is the record's identity. The record mirrors the
+// State of the Request identified by RequestID and changes together with it.
+type QueueRequestState struct {
+	// Queue is the name of the queue the request belongs to. Queue name is defined in the
+	// configuration and should be unique within the system.
+	Queue string
+	// State is the request's current state. Always one of InQueueRequestStates().
+	State RequestState
+	// RequestID is the request's identifier, unique within Queue.
+	RequestID string
+	// EnteredAtMs is when the request entered State, in Unix milliseconds. A version-only
+	// update that keeps the state does not change it.
+	EnteredAtMs int64
+}

submitqueue/entity/request.go

@@ -59,6 +59,18 @@ func IsRequestStateTerminal(s RequestState) bool {
 	return s == RequestStateLanded || s == RequestStateError || s == RequestStateCancelled
 }
 
+// InQueueRequestStates returns every non-terminal request state: the states a request occupies while it is in its
+// queue. Every valid RequestState is either listed here or terminal. Each call returns a fresh slice.
+func InQueueRequestStates() []RequestState {
+	return []RequestState{
+		RequestStateStarted,
+		RequestStateValidated,
+		RequestStateBatched,
+		RequestStateProcessing,
+		RequestStateCancelling,
+	}
+}
+
 // IsRequestStateHalted returns true if the request is either terminal or in the process of being cancelled.
 // Forward-progress controllers (validate, batch, ...) use this to short-circuit work for requests that the
 // user has asked to cancel — even though Cancelling is non-terminal, no further pipeline work should start.

submitqueue/entity/request_test.go

@@ -15,6 +15,7 @@
 package entity
 
 import (
+	"slices"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -41,6 +42,28 @@ func TestIsRequestStateTerminal(t *testing.T) {
 	}
 }
 
+func TestInQueueRequestStates(t *testing.T) {
+	tests := []struct {
+		state   RequestState
+		inQueue bool
+	}{
+		{RequestStateStarted, true},
+		{RequestStateValidated, true},
+		{RequestStateBatched, true},
+		{RequestStateProcessing, true},
+		{RequestStateCancelling, true},
+		{RequestStateLanded, false},
+		{RequestStateError, false},
+		{RequestStateCancelled, false},
+	}
+	for _, tt := range tests {
+		t.Run(string(tt.state), func(t *testing.T) {
+			assert.Equal(t, tt.inQueue, slices.Contains(InQueueRequestStates(), tt.state))
+			assert.NotEqual(t, tt.inQueue, IsRequestStateTerminal(tt.state), "a valid state is either in-queue or terminal")
+		})
+	}
+}
+
 func TestIsRequestStateHalted(t *testing.T) {
 	tests := []struct {
 		state  RequestState

submitqueue/extension/storage/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "change_store.go",
         "path_build_store.go",
         "queue_batch_state_store.go",
+        "queue_request_state_store.go",
         "request_batch_store.go",
         "request_log_store.go",
         "request_receipt_store.go",
@@ -22,6 +23,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//platform/errs:go_default_library",
+        "//platform/metrics:go_default_library",
         "//submitqueue/entity:go_default_library",
     ],
 )
@@ -32,7 +34,9 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//platform/errs:go_default_library",
+        "//platform/metrics:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
     ],
 )
 

submitqueue/extension/storage/README.md

@@ -61,7 +61,7 @@ Store interfaces are designed for the storage technology *space*, not for SQL (s
 
 **Domain state is often already the index.** Before adding any lookup, check whether an entity the caller already loads enumerates the children — an aggregate that references its parts by ID (e.g. a tree whose paths record their build identities) is the batch→children index, persisted and versioned as domain state. Duplicating that relationship as a database index adds a second source of truth for something the domain already owns.
 
-**When neither applies, the reverse lookup is real — give it its own mapping store.** In the KV space there is no third mechanism: the only way to look up by an attribute is to make that attribute a primary key somewhere. So promote the relationship to a first-class mapping entity — keyed by the lookup attribute, written by the same flow that creates the source entity with idempotent puts, and rebuildable as a projection if it drifts. `ChangeRecord` is the in-repo example: it exists so "which requests claimed this change URI" is a by-key read on (queue, URI). `QueueBatchState` is the same pattern for a mutable attribute: "which batches of this queue are in this state" is a by-key read on (queue, state), maintained as advisory records that move buckets alongside the batch's own state CAS (the shared primitives in `submitqueue/orchestrator/core/batch` own that protocol) — it replaced `BatchStore.GetByQueueAndStates`, which was the contract's one query-by-attribute. Unlike a `KEY idx_*`, the relationship is visible in the contract and portable to any backend.
+**When neither applies, the reverse lookup is real — give it its own mapping store.** In the KV space there is no third mechanism: the only way to look up by an attribute is to make that attribute a primary key somewhere. So promote the relationship to a first-class mapping entity — keyed by the lookup attribute, written by the same flow that creates the source entity with idempotent puts, and rebuildable as a projection if it drifts. `ChangeRecord` is the in-repo example: it exists so "which requests claimed this change URI" is a by-key read on (queue, URI). `QueueBatchState` is the same pattern for a mutable attribute: "which batches of this queue are in this state" is a by-key read on (queue, state), maintained as advisory records that move buckets alongside the batch's own state CAS (the shared primitives in `submitqueue/orchestrator/core/batch` own that protocol) — it replaced `BatchStore.GetByQueueAndStates`, which was the contract's one query-by-attribute. `QueueRequestState` applies the pattern to requests and is exact rather than advisory: a backend writes each record as part of the request write it mirrors, so `QueueRequestStateStore` is read-only and needs no transition protocol. Unlike a `KEY idx_*`, the relationship is visible in the contract and portable to any backend.
 
 ### Decision path
 

submitqueue/extension/storage/mock/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "change_store_mock.go",
         "path_build_store_mock.go",
         "queue_batch_state_store_mock.go",
+        "queue_request_state_store_mock.go",
         "request_batch_store_mock.go",
         "request_log_store_mock.go",
         "request_receipt_store_mock.go",