feat(orchestrator): Keep exact per-state request membership records so a queue's requests can be listed by state

Author: roychying

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` |

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,16 @@ func IsRequestStateTerminal(s RequestState) bool {
 	return s == RequestStateLanded || s == RequestStateError || s == RequestStateCancelled
 }
 
+// InQueueRequestStates lists 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.
+var InQueueRequestStates = []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",

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",