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

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

@@ -4,7 +4,7 @@ Deciding which service of a multi-service domain an extension belongs to, and mo
 
 ## Status
 
-Implemented for SubmitQueue storage and for the core packages that served one service. `submitqueue/gateway/extension/storage` and `submitqueue/orchestrator/extension/storage` each declare a `Factory`, a `Storage` aggregate, and a MySQL schema. The thirteen store contracts, their mocks, and the error vocabulary remain at `submitqueue/extension/storage`. There is no `submitqueue/extension/storage/mysql` package. `submitqueue/extension/storage/BUILD.bazel` filegroups the two service schemas for deployments that colocate them.
+Implemented for SubmitQueue storage and for the core packages that served one service. `submitqueue/gateway/extension/storage` and `submitqueue/orchestrator/extension/storage` each declare a `Factory`, a `Storage` aggregate, and a MySQL schema. The fourteen store contracts, their mocks, and the error vocabulary remain at `submitqueue/extension/storage`. There is no `submitqueue/extension/storage/mysql` package. `submitqueue/extension/storage/BUILD.bazel` filegroups the two service schemas for deployments that colocate them.
 
 `submitqueue/core/batch` and `submitqueue/core/request` are not on disk. Batch helpers live in `submitqueue/orchestrator/core/batch`. Request-log publish and terminate live in `submitqueue/orchestrator/core/request`. The gateway materializer lives in `submitqueue/gateway/core/request`. `submitqueue/core` holds `changeset`, `messagequeue`, and `topickey`. `changeset` still declares `Stores` and `Resolve`, because `buildrunner`, `conflict`, and `speculation` are still domain-level packages that depend on it. Stovepipe keeps `stovepipe/extension/storage/mysql` because that domain is the service; that package is not a leftover SubmitQueue backend.
 
@@ -25,8 +25,8 @@ Separating the services onto their own databases already worked — the orchestr
 1. An extension whose **aggregate** is used by exactly one service of a multi-service domain puts that aggregate, its implementations, its mocks and its schema at `{domain}/{service}/extension/{ext}/`. The store contracts themselves stay at `{domain}/extension/{ext}/`: the aggregate is what decides reachability, so moving the contracts adds no enforcement while forcing domain-level callers to import a service package.
 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.
+4. SubmitQueue's storage aggregate splits in two: `submitqueue/gateway/extension/storage/` declares a `Factory` and a five-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 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.
 
@@ -40,6 +40,7 @@ Every store, verified against actual usage rather than intent.
 | `RequestSummaryStore` | `request_summary` |
 | `RequestReceiptStore` | `request_receipt` |
 | `RequestURIStore` | `change_uri_request_mapping` |
+| `QueueRequestStatusStore` | `queue_request_status` |
 
 | Orchestrator | Table |
 |---|---|

service/submitqueue/gateway/server/main.go

@@ -358,7 +358,7 @@ func run() error {
 	// read materialized views, and request-history controllers read retained
 	// logs. Normal log-topic persistence and terminal DLQ repair both use the
 	// materializer.
-	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"))
+	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"), mysqlstorage.WithQueueRequestStatusRecords())
 	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_status.go",
         "request.go",
         "request_batch.go",
         "request_history.go",
@@ -40,6 +41,7 @@ go_test(
     srcs = [
         "batch_test.go",
         "build_test.go",
+        "queue_request_status_test.go",
         "request_log_test.go",
         "request_test.go",
         "speculation_test.go",

submitqueue/entity/queue_request_status.go

@@ -0,0 +1,73 @@
+// 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
+
+import "slices"
+
+// QueueRequestStatus is a membership record filing one in-queue request under its public
+// status in its queue, so a queue's requests can be listed by status without a secondary
+// index on the request summary.
+//
+// The (Queue, Status, RequestID) triple is the record's identity. A request changes status
+// by a record appearing under the new status before the record under the old status
+// disappears; a request that reaches a terminal status keeps no record.
+//
+// Records are advisory. The authoritative status is the Status of the RequestSummary
+// identified by RequestID; a record may file a request under a status it has already left,
+// briefly while writers race and until the request finishes after an interrupted write.
+// Readers must load each listed request's summary, count the request only under the
+// summary's own Status, and delete records that disagree.
+type QueueRequestStatus 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
+
+	// Status is the public status this record files the request under. Always one of
+	// InQueueRequestStatuses(). Advisory: the summary's Status may differ transiently.
+	Status RequestStatus
+
+	// RequestID is the request's identifier, unique within Queue.
+	RequestID string
+
+	// ReceivedAtMs is when the gateway received the request, in Unix milliseconds.
+	ReceivedAtMs int64
+
+	// EnteredAtMs is the timestamp of the status log that moved the request into Status, in
+	// Unix milliseconds. Re-filing the request under the same status keeps it.
+	EnteredAtMs int64
+}
+
+// InQueueRequestStatuses returns every public status a request holds while it is in its
+// queue: every status except the internal accepting status and the terminal statuses. Each
+// call returns a fresh slice.
+func InQueueRequestStatuses() []RequestStatus {
+	return []RequestStatus{
+		RequestStatusAccepted,
+		RequestStatusStarted,
+		RequestStatusValidating,
+		RequestStatusValidated,
+		RequestStatusBatching,
+		RequestStatusBatched,
+		RequestStatusSpeculating,
+		RequestStatusSpeculated,
+		RequestStatusLanding,
+		RequestStatusCancelling,
+	}
+}
+
+// IsRequestStatusInQueue reports whether status is one of InQueueRequestStatuses().
+func IsRequestStatusInQueue(status RequestStatus) bool {
+	return slices.Contains(InQueueRequestStatuses(), status)
+}

submitqueue/entity/queue_request_status_test.go

@@ -0,0 +1,57 @@
+// 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
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestInQueueRequestStatuses pins the vocabulary split: every status is in the queue,
+// terminal, the internal accepting status, or the unknown sentinel — never more than one.
+func TestInQueueRequestStatuses(t *testing.T) {
+	tests := []struct {
+		status  RequestStatus
+		inQueue bool
+	}{
+		{status: RequestStatusUnknown},
+		{status: RequestStatusAccepting},
+		{status: RequestStatusAccepted, inQueue: true},
+		{status: RequestStatusStarted, inQueue: true},
+		{status: RequestStatusValidating, inQueue: true},
+		{status: RequestStatusValidated, inQueue: true},
+		{status: RequestStatusBatching, inQueue: true},
+		{status: RequestStatusBatched, inQueue: true},
+		{status: RequestStatusSpeculating, inQueue: true},
+		{status: RequestStatusSpeculated, inQueue: true},
+		{status: RequestStatusLanding, inQueue: true},
+		{status: RequestStatusCancelling, inQueue: true},
+		{status: RequestStatusLanded},
+		{status: RequestStatusError},
+		{status: RequestStatusCancelled},
+	}
+
+	for _, tt := range tests {
+		t.Run(string(tt.status), func(t *testing.T) {
+			assert.Equal(t, tt.inQueue, slices.Contains(InQueueRequestStatuses(), tt.status))
+			assert.Equal(t, tt.inQueue, IsRequestStatusInQueue(tt.status))
+			if tt.inQueue {
+				assert.False(t, IsRequestStateTerminal(RequestState(tt.status)), "an in-queue status is never terminal")
+			}
+		})
+	}
+}

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_status_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. `QueueRequestStatus` applies the same advisory pattern to requests in the gateway: "which requests of this queue are at this status" is a by-key read on (queue, status), with records the gateway's request materializer moves alongside each summary update, so `request_summary` keeps no index. 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_status_store_mock.go",
         "request_batch_store_mock.go",
         "request_log_store_mock.go",
         "request_receipt_store_mock.go",