From 666c21c14b135f22c4c45507bc337fb6b356d18d Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 10:59:26 -0700
Subject: [PATCH 1/7] feat(entity): Add QueueRequestStatus membership records
 and the in-queue request status vocabulary

---
 submitqueue/entity/BUILD.bazel                |  2 +
 submitqueue/entity/queue_request_status.go    | 77 +++++++++++++++++++
 .../entity/queue_request_status_test.go       | 57 ++++++++++++++
 3 files changed, 136 insertions(+)
 create mode 100644 submitqueue/entity/queue_request_status.go
 create mode 100644 submitqueue/entity/queue_request_status_test.go

diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel
index 6e1561ab1..5dc29d5e1 100644
--- a/submitqueue/entity/BUILD.bazel
+++ b/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",
diff --git a/submitqueue/entity/queue_request_status.go b/submitqueue/entity/queue_request_status.go
new file mode 100644
index 000000000..73feba549
--- /dev/null
+++ b/submitqueue/entity/queue_request_status.go
@@ -0,0 +1,77 @@
+// 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
+
+// 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 transiently, or after an interrupted write, file a
+// request under a status it has already left. Readers load each listed request's summary
+// and classify it by the summary's own Status.
+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 {
+	switch status {
+	case RequestStatusAccepted, RequestStatusStarted, RequestStatusValidating, RequestStatusValidated,
+		RequestStatusBatching, RequestStatusBatched, RequestStatusSpeculating, RequestStatusSpeculated,
+		RequestStatusLanding, RequestStatusCancelling:
+		return true
+	default:
+		return false
+	}
+}
diff --git a/submitqueue/entity/queue_request_status_test.go b/submitqueue/entity/queue_request_status_test.go
new file mode 100644
index 000000000..24b8aab87
--- /dev/null
+++ b/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")
+			}
+		})
+	}
+}

From 9aded2cbbdafd2b432af444a759d894ff8a397ba Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 10:59:28 -0700
Subject: [PATCH 2/7] feat(gateway): Add an opt-in MySQL
 QueueRequestStatusStore backed by the queue_request_status table

---
 submitqueue/extension/storage/BUILD.bazel     |   1 +
 .../extension/storage/mock/BUILD.bazel        |   1 +
 .../mock/queue_request_status_store_mock.go   |  85 +++++++
 .../storage/queue_request_status_store.go     |  52 +++++
 .../extension/storage/mock/storage_mock.go    |  14 ++
 .../extension/storage/mysql/BUILD.bazel       |   2 +
 .../mysql/queue_request_status_store.go       | 124 +++++++++++
 .../mysql/queue_request_status_store_test.go  | 209 ++++++++++++++++++
 .../mysql/schema/queue_request_status.sql     |  12 +
 .../extension/storage/mysql/storage.go        |  42 +++-
 .../extension/storage/mysql/storage_test.go   |  57 +++++
 .../gateway/extension/storage/storage.go      |   3 +
 .../extension/storage/mysql/storage_test.go   |   2 +-
 .../submitqueue/extension/storage/suite.go    |  54 +++++
 14 files changed, 651 insertions(+), 7 deletions(-)
 create mode 100644 submitqueue/extension/storage/mock/queue_request_status_store_mock.go
 create mode 100644 submitqueue/extension/storage/queue_request_status_store.go
 create mode 100644 submitqueue/gateway/extension/storage/mysql/queue_request_status_store.go
 create mode 100644 submitqueue/gateway/extension/storage/mysql/queue_request_status_store_test.go
 create mode 100644 submitqueue/gateway/extension/storage/mysql/schema/queue_request_status.sql

diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel
index ec9a4d025..850229d5c 100644
--- a/submitqueue/extension/storage/BUILD.bazel
+++ b/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",
diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel
index 49f2907c6..b0af1b987 100644
--- a/submitqueue/extension/storage/mock/BUILD.bazel
+++ b/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",
diff --git a/submitqueue/extension/storage/mock/queue_request_status_store_mock.go b/submitqueue/extension/storage/mock/queue_request_status_store_mock.go
new file mode 100644
index 000000000..04e7e2388
--- /dev/null
+++ b/submitqueue/extension/storage/mock/queue_request_status_store_mock.go
@@ -0,0 +1,85 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: queue_request_status_store.go
+//
+// Generated by this command:
+//
+//	mockgen -source=queue_request_status_store.go -destination=mock/queue_request_status_store_mock.go -package=mock
+//
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	context "context"
+	reflect "reflect"
+
+	entity "github.com/uber/submitqueue/submitqueue/entity"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockQueueRequestStatusStore is a mock of QueueRequestStatusStore interface.
+type MockQueueRequestStatusStore struct {
+	ctrl     *gomock.Controller
+	recorder *MockQueueRequestStatusStoreMockRecorder
+	isgomock struct{}
+}
+
+// MockQueueRequestStatusStoreMockRecorder is the mock recorder for MockQueueRequestStatusStore.
+type MockQueueRequestStatusStoreMockRecorder struct {
+	mock *MockQueueRequestStatusStore
+}
+
+// NewMockQueueRequestStatusStore creates a new mock instance.
+func NewMockQueueRequestStatusStore(ctrl *gomock.Controller) *MockQueueRequestStatusStore {
+	mock := &MockQueueRequestStatusStore{ctrl: ctrl}
+	mock.recorder = &MockQueueRequestStatusStoreMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockQueueRequestStatusStore) EXPECT() *MockQueueRequestStatusStoreMockRecorder {
+	return m.recorder
+}
+
+// Delete mocks base method.
+func (m *MockQueueRequestStatusStore) Delete(ctx context.Context, status entity.RequestStatus, requestID string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Delete", ctx, status, requestID)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Delete indicates an expected call of Delete.
+func (mr *MockQueueRequestStatusStoreMockRecorder) Delete(ctx, status, requestID any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockQueueRequestStatusStore)(nil).Delete), ctx, status, requestID)
+}
+
+// List mocks base method.
+func (m *MockQueueRequestStatusStore) List(ctx context.Context, status entity.RequestStatus) ([]entity.QueueRequestStatus, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "List", ctx, status)
+	ret0, _ := ret[0].([]entity.QueueRequestStatus)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// List indicates an expected call of List.
+func (mr *MockQueueRequestStatusStoreMockRecorder) List(ctx, status any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockQueueRequestStatusStore)(nil).List), ctx, status)
+}
+
+// Put mocks base method.
+func (m *MockQueueRequestStatusStore) Put(ctx context.Context, record entity.QueueRequestStatus) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Put", ctx, record)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Put indicates an expected call of Put.
+func (mr *MockQueueRequestStatusStoreMockRecorder) Put(ctx, record any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Put", reflect.TypeOf((*MockQueueRequestStatusStore)(nil).Put), ctx, record)
+}
diff --git a/submitqueue/extension/storage/queue_request_status_store.go b/submitqueue/extension/storage/queue_request_status_store.go
new file mode 100644
index 000000000..682a309cf
--- /dev/null
+++ b/submitqueue/extension/storage/queue_request_status_store.go
@@ -0,0 +1,52 @@
+// 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 storage
+
+//go:generate mockgen -source=queue_request_status_store.go -destination=mock/queue_request_status_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+// QueueRequestStatusStore manages the per-queue membership records that file each in-queue
+// request under its public status, so "the requests of queue Q at status S" is a single read
+// keyed by (queue, status) — a primary-key prefix on any backend, with no secondary index on
+// the request summary.
+//
+// Records are advisory: the authoritative status is on the RequestSummary, and a record may
+// file a request under a status it has already left. Readers therefore treat a listing as a
+// set of candidate request IDs — they load each summary by key and classify it by its own
+// Status, never by the bucket the record was found in.
+//
+// The interface is per-status and per-record so that any backend can implement it without
+// multi-key queries or batch atomicity. A request moves buckets via Put of the new record
+// followed by Delete of the old one, which keeps at least one record visible throughout. All
+// writes are idempotent so log redeliveries can safely repeat them.
+type QueueRequestStatusStore interface {
+	// List returns every record filed under the bound queue's given status. An empty slice
+	// means the bucket is empty. Order is unspecified.
+	List(ctx context.Context, status entity.RequestStatus) ([]entity.QueueRequestStatus, error)
+
+	// Put persists a record. The record's Queue must match the instance's bound queue.
+	// Writing a record whose (queue, status, requestID) already exists is a no-op success
+	// that keeps the stored EnteredAtMs, so the call is idempotent under redeliveries.
+	Put(ctx context.Context, record entity.QueueRequestStatus) error
+
+	// Delete removes the bound queue's record identified by (status, requestID). Deleting an
+	// absent record is a no-op success, so the call is idempotent under redeliveries.
+	Delete(ctx context.Context, status entity.RequestStatus, requestID string) error
+}
diff --git a/submitqueue/gateway/extension/storage/mock/storage_mock.go b/submitqueue/gateway/extension/storage/mock/storage_mock.go
index 2b37d4710..1824a2c6b 100644
--- a/submitqueue/gateway/extension/storage/mock/storage_mock.go
+++ b/submitqueue/gateway/extension/storage/mock/storage_mock.go
@@ -80,6 +80,20 @@ func (m *MockStorage) EXPECT() *MockStorageMockRecorder {
 	return m.recorder
 }
 
+// GetQueueRequestStatusStore mocks base method.
+func (m *MockStorage) GetQueueRequestStatusStore() storage.QueueRequestStatusStore {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetQueueRequestStatusStore")
+	ret0, _ := ret[0].(storage.QueueRequestStatusStore)
+	return ret0
+}
+
+// GetQueueRequestStatusStore indicates an expected call of GetQueueRequestStatusStore.
+func (mr *MockStorageMockRecorder) GetQueueRequestStatusStore() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetQueueRequestStatusStore", reflect.TypeOf((*MockStorage)(nil).GetQueueRequestStatusStore))
+}
+
 // GetRequestLogStore mocks base method.
 func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore {
 	m.ctrl.T.Helper()
diff --git a/submitqueue/gateway/extension/storage/mysql/BUILD.bazel b/submitqueue/gateway/extension/storage/mysql/BUILD.bazel
index 02473333a..38bef1268 100644
--- a/submitqueue/gateway/extension/storage/mysql/BUILD.bazel
+++ b/submitqueue/gateway/extension/storage/mysql/BUILD.bazel
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "queue_request_status_store.go",
         "request_log_store.go",
         "request_receipt_store.go",
         "request_summary_store.go",
@@ -24,6 +25,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "queue_request_status_store_test.go",
         "request_log_store_test.go",
         "request_receipt_store_test.go",
         "request_summary_store_test.go",
diff --git a/submitqueue/gateway/extension/storage/mysql/queue_request_status_store.go b/submitqueue/gateway/extension/storage/mysql/queue_request_status_store.go
new file mode 100644
index 000000000..9e2befc5b
--- /dev/null
+++ b/submitqueue/gateway/extension/storage/mysql/queue_request_status_store.go
@@ -0,0 +1,124 @@
+// 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 mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"github.com/uber-go/tally"
+
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+// errQueueRequestStatusRecordsDisabled is returned by List on a storage built without
+// WithQueueRequestStatusRecords: the records are not kept, so an empty listing would be
+// indistinguishable from an empty queue.
+var errQueueRequestStatusRecordsDisabled = errors.New("queue request status records are not enabled for this storage")
+
+type queueRequestStatusStore struct {
+	db    *sql.DB
+	scope tally.Scope
+	// queue is the queue name this store instance is bound to; every read and
+	// write is scoped to it.
+	queue string
+}
+
+// NewQueueRequestStatusStore creates a new MySQL-backed QueueRequestStatusStore.
+func NewQueueRequestStatusStore(db *sql.DB, scope tally.Scope, queue string) storage.QueueRequestStatusStore {
+	return &queueRequestStatusStore{db: db, scope: scope, queue: queue}
+}
+
+// List returns every record filed under the bound queue's status bucket. The WHERE clause
+// is a prefix of the (queue, status, request_id) PK, so this is a PK-prefix scan.
+func (s *queueRequestStatusStore) List(ctx context.Context, status entity.RequestStatus) (ret []entity.QueueRequestStatus, retErr error) {
+	op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets)
+	defer func() { storage.CompleteReadMetricsOp(op, retErr) }()
+
+	queue := s.queue
+	const query = "SELECT queue, status, request_id, received_at_ms, entered_at_ms FROM queue_request_status WHERE queue = ? AND status = ?"
+	rows, err := s.db.QueryContext(ctx, query, queue, string(status))
+	if err != nil {
+		return nil, fmt.Errorf("failed to query queue request status records for queue=%s status=%s: %w", queue, status, err)
+	}
+	defer rows.Close()
+
+	var results []entity.QueueRequestStatus
+	for rows.Next() {
+		var rec entity.QueueRequestStatus
+		if err := rows.Scan(&rec.Queue, &rec.Status, &rec.RequestID, &rec.ReceivedAtMs, &rec.EnteredAtMs); err != nil {
+			return nil, fmt.Errorf("failed to scan queue request status record for queue=%s status=%s: %w", queue, status, err)
+		}
+		results = append(results, rec)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("failed to iterate queue request status records for queue=%s status=%s: %w", queue, status, err)
+	}
+	return results, nil
+}
+
+// Put inserts a record. A primary-key conflict on (queue, status, request_id) is a no-op
+// update that keeps the stored entered_at_ms; unlike INSERT IGNORE, it does not downgrade
+// other errors to warnings.
+func (s *queueRequestStatusStore) Put(ctx context.Context, record entity.QueueRequestStatus) (retErr error) {
+	op := metrics.Begin(s.scope, "put", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if record.Queue != s.queue {
+		return fmt.Errorf("queue request status record request_id=%s queue %q does not match the store's bound queue %q", record.RequestID, record.Queue, s.queue)
+	}
+
+	const query = "INSERT INTO queue_request_status (queue, status, request_id, received_at_ms, entered_at_ms) VALUES (?, ?, ?, ?, ?) " +
+		"ON DUPLICATE KEY UPDATE entered_at_ms = entered_at_ms"
+	if _, err := s.db.ExecContext(ctx, query, record.Queue, string(record.Status), record.RequestID, record.ReceivedAtMs, record.EnteredAtMs); err != nil {
+		return fmt.Errorf("failed to put queue request status record queue=%s status=%s request_id=%s: %w", record.Queue, record.Status, record.RequestID, err)
+	}
+	return nil
+}
+
+// Delete removes the bound queue's record identified by (status, requestID). Deleting an
+// absent record is a no-op success — rows-affected is intentionally not checked.
+func (s *queueRequestStatusStore) Delete(ctx context.Context, status entity.RequestStatus, requestID string) (retErr error) {
+	op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	queue := s.queue
+	const query = "DELETE FROM queue_request_status WHERE queue = ? AND status = ? AND request_id = ?"
+	if _, err := s.db.ExecContext(ctx, query, queue, string(status), requestID); err != nil {
+		return fmt.Errorf("failed to delete queue request status record queue=%s status=%s request_id=%s: %w", queue, status, requestID, err)
+	}
+	return nil
+}
+
+// disabledQueueRequestStatusStore stands in when the deployment keeps no records, for
+// example before the table exists in its datastore. Writes are no-ops so the materializer
+// runs unchanged; List fails rather than report every queue as empty.
+type disabledQueueRequestStatusStore struct{}
+
+func (disabledQueueRequestStatusStore) List(context.Context, entity.RequestStatus) ([]entity.QueueRequestStatus, error) {
+	return nil, errQueueRequestStatusRecordsDisabled
+}
+
+func (disabledQueueRequestStatusStore) Put(context.Context, entity.QueueRequestStatus) error {
+	return nil
+}
+
+func (disabledQueueRequestStatusStore) Delete(context.Context, entity.RequestStatus, string) error {
+	return nil
+}
diff --git a/submitqueue/gateway/extension/storage/mysql/queue_request_status_store_test.go b/submitqueue/gateway/extension/storage/mysql/queue_request_status_store_test.go
new file mode 100644
index 000000000..03523db8d
--- /dev/null
+++ b/submitqueue/gateway/extension/storage/mysql/queue_request_status_store_test.go
@@ -0,0 +1,209 @@
+// 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 mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+var queueRequestStatusColumns = []string{"queue", "status", "request_id", "received_at_ms", "entered_at_ms"}
+
+func setupQueueRequestStatusStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.QueueRequestStatusStore) {
+	t.Helper()
+
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	return db, mock, NewQueueRequestStatusStore(db, testMetrics(), "monorepo")
+}
+
+func TestQueueRequestStatusStore_List(t *testing.T) {
+	record1 := entity.QueueRequestStatus{Queue: "monorepo", Status: entity.RequestStatusSpeculating, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+	record2 := entity.QueueRequestStatus{Queue: "monorepo", Status: entity.RequestStatusSpeculating, RequestID: "2", ReceivedAtMs: 11, EnteredAtMs: 21}
+	storeErr := errors.New("storage failed")
+	addRow := func(rows *sqlmock.Rows, rec entity.QueueRequestStatus) *sqlmock.Rows {
+		return rows.AddRow(rec.Queue, string(rec.Status), rec.RequestID, rec.ReceivedAtMs, rec.EnteredAtMs)
+	}
+	tests := map[string]struct {
+		setup  func(sqlmock.Sqlmock)
+		want   []entity.QueueRequestStatus
+		errMsg string
+	}{
+		"query fails": {
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectQuery("SELECT queue, status, request_id, received_at_ms, entered_at_ms FROM queue_request_status").
+					WithArgs(record1.Queue, string(record1.Status)).
+					WillReturnError(fmt.Errorf("connection reset"))
+			},
+			errMsg: "connection reset",
+		},
+		"empty bucket": {
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectQuery("SELECT queue, status, request_id, received_at_ms, entered_at_ms FROM queue_request_status").
+					WithArgs(record1.Queue, string(record1.Status)).
+					WillReturnRows(sqlmock.NewRows(queueRequestStatusColumns))
+			},
+		},
+		"row iteration fails": {
+			setup: func(mock sqlmock.Sqlmock) {
+				rows := addRow(sqlmock.NewRows(queueRequestStatusColumns), record1).RowError(0, storeErr)
+				mock.ExpectQuery("SELECT queue, status, request_id, received_at_ms, entered_at_ms FROM queue_request_status").
+					WithArgs(record1.Queue, string(record1.Status)).
+					WillReturnRows(rows)
+			},
+			errMsg: storeErr.Error(),
+		},
+		"success": {
+			setup: func(mock sqlmock.Sqlmock) {
+				rows := addRow(addRow(sqlmock.NewRows(queueRequestStatusColumns), record1), record2)
+				mock.ExpectQuery("SELECT queue, status, request_id, received_at_ms, entered_at_ms FROM queue_request_status").
+					WithArgs(record1.Queue, string(record1.Status)).
+					WillReturnRows(rows)
+			},
+			want: []entity.QueueRequestStatus{record1, record2},
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			db, mock, store := setupQueueRequestStatusStoreTest(t)
+			defer db.Close()
+			tt.setup(mock)
+
+			got, err := store.List(context.Background(), record1.Status)
+			if tt.errMsg != "" {
+				assert.ErrorContains(t, err, tt.errMsg)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.want, got)
+			}
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestQueueRequestStatusStore_Put(t *testing.T) {
+	record := entity.QueueRequestStatus{Queue: "monorepo", Status: entity.RequestStatusValidating, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+	const insert = "INSERT INTO queue_request_status \\(queue, status, request_id, received_at_ms, entered_at_ms\\) VALUES \\(\\?, \\?, \\?, \\?, \\?\\) ON DUPLICATE KEY UPDATE entered_at_ms = entered_at_ms"
+	tests := map[string]struct {
+		record entity.QueueRequestStatus
+		setup  func(sqlmock.Sqlmock)
+		errMsg string
+	}{
+		"record for another queue is rejected": {
+			record: entity.QueueRequestStatus{Queue: "other", Status: record.Status, RequestID: record.RequestID},
+			setup:  func(sqlmock.Sqlmock) {},
+			errMsg: "does not match the store's bound queue",
+		},
+		"insert fails": {
+			record: record,
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec(insert).
+					WithArgs(record.Queue, string(record.Status), record.RequestID, record.ReceivedAtMs, record.EnteredAtMs).
+					WillReturnError(fmt.Errorf("connection reset"))
+			},
+			errMsg: "connection reset",
+		},
+		"existing record is a no-op success": {
+			record: record,
+			setup: func(mock sqlmock.Sqlmock) {
+				// The no-op ON DUPLICATE KEY UPDATE reports zero affected rows on a PK conflict.
+				mock.ExpectExec(insert).
+					WithArgs(record.Queue, string(record.Status), record.RequestID, record.ReceivedAtMs, record.EnteredAtMs).
+					WillReturnResult(sqlmock.NewResult(0, 0))
+			},
+		},
+		"success": {
+			record: record,
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec(insert).
+					WithArgs(record.Queue, string(record.Status), record.RequestID, record.ReceivedAtMs, record.EnteredAtMs).
+					WillReturnResult(sqlmock.NewResult(0, 1))
+			},
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			db, mock, store := setupQueueRequestStatusStoreTest(t)
+			defer db.Close()
+			tt.setup(mock)
+
+			err := store.Put(context.Background(), tt.record)
+			if tt.errMsg != "" {
+				assert.ErrorContains(t, err, tt.errMsg)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestQueueRequestStatusStore_Delete(t *testing.T) {
+	tests := map[string]struct {
+		setup  func(sqlmock.Sqlmock)
+		errMsg string
+	}{
+		"delete fails": {
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec("DELETE FROM queue_request_status").
+					WithArgs("monorepo", string(entity.RequestStatusStarted), "1").
+					WillReturnError(fmt.Errorf("connection reset"))
+			},
+			errMsg: "connection reset",
+		},
+		"absent record is a no-op success": {
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec("DELETE FROM queue_request_status").
+					WithArgs("monorepo", string(entity.RequestStatusStarted), "1").
+					WillReturnResult(sqlmock.NewResult(0, 0))
+			},
+		},
+		"success": {
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec("DELETE FROM queue_request_status").
+					WithArgs("monorepo", string(entity.RequestStatusStarted), "1").
+					WillReturnResult(sqlmock.NewResult(0, 1))
+			},
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			db, mock, store := setupQueueRequestStatusStoreTest(t)
+			defer db.Close()
+			tt.setup(mock)
+
+			err := store.Delete(context.Background(), entity.RequestStatusStarted, "1")
+			if tt.errMsg != "" {
+				assert.ErrorContains(t, err, tt.errMsg)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
diff --git a/submitqueue/gateway/extension/storage/mysql/schema/queue_request_status.sql b/submitqueue/gateway/extension/storage/mysql/schema/queue_request_status.sql
new file mode 100644
index 000000000..62cf102ed
--- /dev/null
+++ b/submitqueue/gateway/extension/storage/mysql/schema/queue_request_status.sql
@@ -0,0 +1,12 @@
+-- Membership records filing each in-queue request under its public status, mirroring
+-- request_summary.status. The (queue, status, request_id) PK *is* the lookup: "requests of
+-- queue Q at status S" is a PK-prefix scan, so request_summary needs no secondary index.
+-- Records are advisory: readers confirm each listed request against its summary.
+CREATE TABLE IF NOT EXISTS queue_request_status (
+    queue          VARCHAR(255) NOT NULL,
+    status         VARCHAR(64)  NOT NULL,
+    request_id     VARCHAR(255) NOT NULL,
+    received_at_ms BIGINT       NOT NULL,
+    entered_at_ms  BIGINT       NOT NULL,
+    PRIMARY KEY (queue, status, request_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/submitqueue/gateway/extension/storage/mysql/storage.go b/submitqueue/gateway/extension/storage/mysql/storage.go
index ed6793038..840ba15a9 100644
--- a/submitqueue/gateway/extension/storage/mysql/storage.go
+++ b/submitqueue/gateway/extension/storage/mysql/storage.go
@@ -36,11 +36,28 @@ const mysqlErrDuplicateEntry = 1062
 type Storage struct {
 	db    *sql.DB
 	scope tally.Scope
+	// queueRequestStatusRecords reports whether the queue_request_status table is
+	// kept; see WithQueueRequestStatusRecords.
+	queueRequestStatusRecords bool
+}
+
+// StorageOption configures a Storage.
+type StorageOption func(*Storage)
+
+// WithQueueRequestStatusRecords keeps queue_request_status membership records. A
+// deployment enables it once the table exists in its datastore; without it, record
+// writes are no-ops and QueueRequestStatusStore.List fails.
+func WithQueueRequestStatusRecords() StorageOption {
+	return func(s *Storage) { s.queueRequestStatusRecords = true }
 }
 
 // NewStorage creates a new MySQL storage backend over the given connection pool.
-func NewStorage(db *sql.DB, scope tally.Scope) (*Storage, error) {
-	return &Storage{db: db, scope: scope}, nil
+func NewStorage(db *sql.DB, scope tally.Scope, opts ...StorageOption) (*Storage, error) {
+	s := &Storage{db: db, scope: scope}
+	for _, opt := range opts {
+		opt(s)
+	}
+	return s, nil
 }
 
 // For returns the queue-scoped store aggregate bound to queueName over the
@@ -50,11 +67,16 @@ func (s *Storage) For(queueName string) (storage.Storage, error) {
 	if queueName == "" {
 		return nil, fmt.Errorf("queue name must not be empty")
 	}
+	var queueRequestStatusStore basestorage.QueueRequestStatusStore = disabledQueueRequestStatusStore{}
+	if s.queueRequestStatusRecords {
+		queueRequestStatusStore = NewQueueRequestStatusStore(s.db, s.scope.SubScope("queue_request_status_store"), queueName)
+	}
 	return &boundStorage{
-		requestReceiptStore: NewRequestReceiptStore(s.db, s.scope.SubScope("request_receipt_store"), queueName),
-		requestSummaryStore: NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName),
-		requestLogStore:     NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName),
-		requestURIStore:     NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName),
+		requestReceiptStore:     NewRequestReceiptStore(s.db, s.scope.SubScope("request_receipt_store"), queueName),
+		requestSummaryStore:     NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName),
+		requestLogStore:         NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName),
+		requestURIStore:         NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName),
+		queueRequestStatusStore: queueRequestStatusStore,
 	}, nil
 }
 
@@ -69,6 +91,8 @@ type boundStorage struct {
 	requestSummaryStore basestorage.RequestSummaryStore
 	requestLogStore     basestorage.RequestLogStore
 	requestURIStore     basestorage.RequestURIStore
+
+	queueRequestStatusStore basestorage.QueueRequestStatusStore
 }
 
 // Verify boundStorage implements the queue-scoped aggregate at compile time.
@@ -93,3 +117,9 @@ func (f *boundStorage) GetRequestLogStore() basestorage.RequestLogStore {
 func (f *boundStorage) GetRequestURIStore() basestorage.RequestURIStore {
 	return f.requestURIStore
 }
+
+// GetQueueRequestStatusStore returns the bound QueueRequestStatusStore: MySQL-backed when
+// the storage keeps records, otherwise one whose writes are no-ops and whose List fails.
+func (f *boundStorage) GetQueueRequestStatusStore() basestorage.QueueRequestStatusStore {
+	return f.queueRequestStatusStore
+}
diff --git a/submitqueue/gateway/extension/storage/mysql/storage_test.go b/submitqueue/gateway/extension/storage/mysql/storage_test.go
index 7d4244706..74bb7419a 100644
--- a/submitqueue/gateway/extension/storage/mysql/storage_test.go
+++ b/submitqueue/gateway/extension/storage/mysql/storage_test.go
@@ -15,12 +15,15 @@
 package mysql
 
 import (
+	"context"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
 )
 
 // testMetrics returns a test metrics scope for use in tests.
@@ -41,6 +44,7 @@ func TestNewStorage(t *testing.T) {
 	assert.NotNil(t, bound.GetRequestLogStore())
 	assert.NotNil(t, bound.GetRequestSummaryStore())
 	assert.NotNil(t, bound.GetRequestURIStore())
+	assert.NotNil(t, bound.GetQueueRequestStatusStore())
 
 	_, err = s.For("")
 	assert.Error(t, err, "resolving an empty queue name must fail")
@@ -57,3 +61,56 @@ func TestMysqlStorage_Close(t *testing.T) {
 	require.NoError(t, s.Close())
 	require.NoError(t, mock.ExpectationsWereMet())
 }
+
+// TestNewStorage_QueueRequestStatusRecords covers the opt-in: without it the deployment may
+// not have the table, so record writes touch nothing and List refuses to report an empty
+// queue; with it the records are read from and written to the table.
+func TestNewStorage_QueueRequestStatusRecords(t *testing.T) {
+	record := entity.QueueRequestStatus{Queue: "monorepo", Status: entity.RequestStatusStarted, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+	tests := map[string]struct {
+		opts        []StorageOption
+		setup       func(sqlmock.Sqlmock)
+		wantListErr bool
+	}{
+		"records off": {
+			setup:       func(sqlmock.Sqlmock) {},
+			wantListErr: true,
+		},
+		"records on": {
+			opts: []StorageOption{WithQueueRequestStatusRecords()},
+			setup: func(mock sqlmock.Sqlmock) {
+				mock.ExpectExec("INSERT INTO queue_request_status").WillReturnResult(sqlmock.NewResult(0, 1))
+				mock.ExpectExec("DELETE FROM queue_request_status").WillReturnResult(sqlmock.NewResult(0, 1))
+				mock.ExpectQuery("FROM queue_request_status").
+					WithArgs("monorepo", string(entity.RequestStatusStarted)).
+					WillReturnRows(sqlmock.NewRows(queueRequestStatusColumns))
+			},
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			tt.setup(mock)
+
+			s, err := NewStorage(db, testMetrics(), tt.opts...)
+			require.NoError(t, err)
+			bound, err := s.For("monorepo")
+			require.NoError(t, err)
+			records := bound.GetQueueRequestStatusStore()
+
+			require.NoError(t, records.Put(context.Background(), record))
+			require.NoError(t, records.Delete(context.Background(), record.Status, record.RequestID))
+			got, err := records.List(context.Background(), entity.RequestStatusStarted)
+			if tt.wantListErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Empty(t, got)
+			}
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
diff --git a/submitqueue/gateway/extension/storage/storage.go b/submitqueue/gateway/extension/storage/storage.go
index 37883e4e2..d2898ac7c 100644
--- a/submitqueue/gateway/extension/storage/storage.go
+++ b/submitqueue/gateway/extension/storage/storage.go
@@ -55,4 +55,7 @@ type Storage interface {
 
 	// GetRequestURIStore returns the RequestURIStore instance.
 	GetRequestURIStore() basestorage.RequestURIStore
+
+	// GetQueueRequestStatusStore returns the QueueRequestStatusStore instance.
+	GetQueueRequestStatusStore() basestorage.QueueRequestStatusStore
 }
diff --git a/test/integration/submitqueue/extension/storage/mysql/storage_test.go b/test/integration/submitqueue/extension/storage/mysql/storage_test.go
index bbc0f07a2..197eeca58 100644
--- a/test/integration/submitqueue/extension/storage/mysql/storage_test.go
+++ b/test/integration/submitqueue/extension/storage/mysql/storage_test.go
@@ -81,7 +81,7 @@ func (s *MySQLStorageIntegrationSuite) SetupSuite() {
 	store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope)
 	require.NoError(t, err, "failed to create storage")
 
-	gwStore, err := gwmysqlstorage.NewStorage(s.db, tally.NoopScope)
+	gwStore, err := gwmysqlstorage.NewStorage(s.db, tally.NoopScope, gwmysqlstorage.WithQueueRequestStatusRecords())
 	require.NoError(t, err, "failed to create gateway storage")
 
 	// Provide the storage backend to the contract suite through the
diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go
index a85ad023e..3ea006282 100644
--- a/test/integration/submitqueue/extension/storage/suite.go
+++ b/test/integration/submitqueue/extension/storage/suite.go
@@ -419,6 +419,60 @@ func (s *StorageContractSuite) TestStorage_QueueBatchStateRecordLifecycle() {
 	assert.ElementsMatch(t, []entity.QueueBatchState{otherQueue}, got)
 }
 
+// TestStorage_QueueRequestStatusRecordLifecycle exercises the QueueRequestStatusStore
+// contract: records list by exact (queue, status) bucket, re-putting keeps the first
+// entered_at_ms, a move is a put followed by a delete, and deletes of absent records succeed.
+func (s *StorageContractSuite) TestStorage_QueueRequestStatusRecordLifecycle() {
+	t := s.T()
+	ctx := s.ctx
+	storeA := s.forGatewayQueue("qrs-gw-queue-a").GetQueueRequestStatusStore()
+	storeB := s.forGatewayQueue("qrs-gw-queue-b").GetQueueRequestStatusStore()
+
+	started1 := entity.QueueRequestStatus{Queue: "qrs-gw-queue-a", Status: entity.RequestStatusStarted, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+	started2 := entity.QueueRequestStatus{Queue: "qrs-gw-queue-a", Status: entity.RequestStatusStarted, RequestID: "2", ReceivedAtMs: 11, EnteredAtMs: 21}
+	landing := entity.QueueRequestStatus{Queue: "qrs-gw-queue-a", Status: entity.RequestStatusLanding, RequestID: "3", ReceivedAtMs: 12, EnteredAtMs: 22}
+	otherQueue := entity.QueueRequestStatus{Queue: "qrs-gw-queue-b", Status: entity.RequestStatusStarted, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+
+	require.NoError(t, storeA.Put(ctx, started1))
+	require.NoError(t, storeA.Put(ctx, started2))
+	require.NoError(t, storeA.Put(ctx, landing))
+	require.NoError(t, storeB.Put(ctx, otherQueue))
+
+	// Re-putting an existing record is a no-op success that keeps the first entered_at_ms.
+	refiled := started1
+	refiled.EnteredAtMs = 99
+	require.NoError(t, storeA.Put(ctx, refiled))
+
+	got, err := storeA.List(ctx, entity.RequestStatusStarted)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []entity.QueueRequestStatus{started1, started2}, got)
+
+	// An empty bucket lists empty, not an error.
+	got, err = storeA.List(ctx, entity.RequestStatusSpeculating)
+	require.NoError(t, err)
+	assert.Empty(t, got)
+
+	// A record move: file under the new status, then remove the old bucket's record.
+	moved := entity.QueueRequestStatus{Queue: started1.Queue, Status: entity.RequestStatusValidating, RequestID: started1.RequestID, ReceivedAtMs: 10, EnteredAtMs: 30}
+	require.NoError(t, storeA.Put(ctx, moved))
+	require.NoError(t, storeA.Delete(ctx, started1.Status, started1.RequestID))
+
+	got, err = storeA.List(ctx, entity.RequestStatusStarted)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []entity.QueueRequestStatus{started2}, got)
+	got, err = storeA.List(ctx, entity.RequestStatusValidating)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []entity.QueueRequestStatus{moved}, got)
+
+	// Deleting an absent record is a no-op success.
+	require.NoError(t, storeA.Delete(ctx, started1.Status, started1.RequestID))
+
+	// The other queue is untouched by all of the above.
+	got, err = storeB.List(ctx, entity.RequestStatusStarted)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []entity.QueueRequestStatus{otherQueue}, got)
+}
+
 // TestStorage_QueueIsolation verifies a store aggregate bound to one queue can
 // neither read nor write another queue's records: cross-queue reads miss, and
 // writes whose entity queue disagrees with the binding are rejected.

From fd1bd1e4f7a5d407d3b13d778f20d46dc8c18f82 Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 10:59:30 -0700
Subject: [PATCH 3/7] feat(gateway): File each request's status record from the
 materializer on every log delivery and enable records in the reference server

---
 service/submitqueue/gateway/server/main.go    |   2 +-
 submitqueue/gateway/controller/land_test.go   |   1 +
 .../gateway/controller/log/log_test.go        |  10 ++
 .../controller/storage_fixture_test.go        |  11 ++
 .../gateway/core/request/materializer.go      |  36 ++++-
 .../gateway/core/request/materializer_test.go | 126 ++++++++++++++++++
 .../core/request/request_receipt_test.go      |  13 ++
 7 files changed, 197 insertions(+), 2 deletions(-)

diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go
index 27a56497a..2326cd4c9 100644
--- a/service/submitqueue/gateway/server/main.go
+++ b/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)
 	}
diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go
index 6f31bf913..3cdc7ff51 100644
--- a/submitqueue/gateway/controller/land_test.go
+++ b/submitqueue/gateway/controller/land_test.go
@@ -442,6 +442,7 @@ func TestLand_PublishesToQueue(t *testing.T) {
 	store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
 	store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
 	store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
+	store.EXPECT().GetQueueRequestStatusStore().Return(permissiveStatusRecords(ctrl)).AnyTimes()
 	factory := gwstoragemock.NewMockFactory(ctrl)
 	factory.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes()
 	materializer := requestcore.NewMaterializer(factory)
diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go
index b014f6e4e..12b3e9ac1 100644
--- a/submitqueue/gateway/controller/log/log_test.go
+++ b/submitqueue/gateway/controller/log/log_test.go
@@ -146,6 +146,7 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr
 	store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
 	store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
 	store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
+	store.EXPECT().GetQueueRequestStatusStore().Return(permissiveStatusRecords(ctrl)).AnyTimes()
 	factory := gwstoragemock.NewMockFactory(ctrl)
 	factory.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes()
 	materializer := requestcore.NewMaterializer(factory)
@@ -172,3 +173,12 @@ func newRequestLog(requestID string, status entity.RequestStatus, requestVersion
 	log := entity.NewRequestStatusLog("test-queue", requestID, status, requestVersion, lastError, metadata)
 	return &log
 }
+
+// permissiveStatusRecords returns a QueueRequestStatusStore that accepts every record write;
+// these tests are not about status records.
+func permissiveStatusRecords(ctrl *gomock.Controller) *storagemock.MockQueueRequestStatusStore {
+	records := storagemock.NewMockQueueRequestStatusStore(ctrl)
+	records.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	records.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	return records
+}
diff --git a/submitqueue/gateway/controller/storage_fixture_test.go b/submitqueue/gateway/controller/storage_fixture_test.go
index 36a89da7b..96d83cb02 100644
--- a/submitqueue/gateway/controller/storage_fixture_test.go
+++ b/submitqueue/gateway/controller/storage_fixture_test.go
@@ -53,6 +53,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt
 	fixture.storage.EXPECT().GetRequestSummaryStore().Return(fixture.summaryStore).AnyTimes()
 	fixture.storage.EXPECT().GetRequestLogStore().Return(fixture.logStore).AnyTimes()
 	fixture.storage.EXPECT().GetRequestURIStore().Return(fixture.uriStore).AnyTimes()
+	fixture.storage.EXPECT().GetQueueRequestStatusStore().Return(permissiveStatusRecords(ctrl)).AnyTimes()
 
 	fixture.summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestSummary) error {
 		fixture.mu.Lock()
@@ -152,6 +153,7 @@ func readModelFactory(
 	}
 	if uris != nil {
 		store.EXPECT().GetRequestURIStore().Return(uris).AnyTimes()
+		store.EXPECT().GetQueueRequestStatusStore().Return(permissiveStatusRecords(ctrl)).AnyTimes()
 	}
 	return factoryForStorage(ctrl, store)
 }
@@ -161,3 +163,12 @@ func readModelFactory(
 func (f *controllerStorageFixture) newMaterializer(ctrl *gomock.Controller) *requestcore.Materializer {
 	return requestcore.NewMaterializer(f.newFactory(ctrl))
 }
+
+// permissiveStatusRecords returns a QueueRequestStatusStore that accepts every record write;
+// these tests are not about status records.
+func permissiveStatusRecords(ctrl *gomock.Controller) *storagemock.MockQueueRequestStatusStore {
+	records := storagemock.NewMockQueueRequestStatusStore(ctrl)
+	records.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	records.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	return records
+}
diff --git a/submitqueue/gateway/core/request/materializer.go b/submitqueue/gateway/core/request/materializer.go
index 273bb68b4..c6b149e9b 100644
--- a/submitqueue/gateway/core/request/materializer.go
+++ b/submitqueue/gateway/core/request/materializer.go
@@ -70,6 +70,9 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 		return fmt.Errorf("failed to insert request log request_id=%s: %w", log.RequestID, err)
 	}
 
+	// previous is the status this delivery moved the summary away from, when it moved it.
+	var previous entity.RequestStatus
+	moved := false
 	for {
 		summary, err := summaries.Get(ctx, log.RequestID)
 		if err != nil {
@@ -94,6 +97,7 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 			}
 			updated.Version = newVersion
 			m.recordLifecycleTransition(summary, updated)
+			previous, moved = summary.Status, true
 			summary = updated
 		}
 
@@ -106,10 +110,40 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 		if err := ensureRequestReceiptMapping(ctx, stores.GetRequestReceiptStore(), summary); err != nil {
 			return err
 		}
-		return nil
+		return fileStatusRecord(ctx, stores.GetQueueRequestStatusStore(), summary, previous, moved)
 	}
 }
 
+// fileStatusRecord keeps the request's status membership record in step with its summary.
+//
+// It runs on every delivery, like the immutable mappings: putting the record for the
+// summary's current status repairs a delivery that failed after the summary update, and is
+// a no-op when the record already exists. When this delivery moved the summary, the record
+// under the previous status is deleted only after the new one is filed, so a request in the
+// queue is never without a record. A record left under an old status by a delivery that
+// failed before that delete is not found again here — the redelivery no longer knows the old
+// status — which is why records are advisory and readers confirm them against the summary.
+func fileStatusRecord(ctx context.Context, records basestorage.QueueRequestStatusStore, summary entity.RequestSummary, previous entity.RequestStatus, moved bool) error {
+	if entity.IsRequestStatusInQueue(summary.Status) {
+		record := entity.QueueRequestStatus{
+			Queue:        summary.Queue,
+			Status:       summary.Status,
+			RequestID:    summary.RequestID,
+			ReceivedAtMs: summary.ReceivedAtMs,
+			EnteredAtMs:  summary.StatusTimestampMs,
+		}
+		if err := records.Put(ctx, record); err != nil {
+			return fmt.Errorf("failed to file request status record request_id=%s status=%s: %w", summary.RequestID, summary.Status, err)
+		}
+	}
+	if moved && previous != summary.Status && entity.IsRequestStatusInQueue(previous) {
+		if err := records.Delete(ctx, previous, summary.RequestID); err != nil {
+			return fmt.Errorf("failed to remove request status record request_id=%s status=%s: %w", summary.RequestID, previous, err)
+		}
+	}
+	return nil
+}
+
 // recordLifecycleTransition counts a request's public lifecycle milestones from
 // one successful summary update.
 //
diff --git a/submitqueue/gateway/core/request/materializer_test.go b/submitqueue/gateway/core/request/materializer_test.go
index e35ccab36..4117cbebe 100644
--- a/submitqueue/gateway/core/request/materializer_test.go
+++ b/submitqueue/gateway/core/request/materializer_test.go
@@ -162,6 +162,130 @@ func TestMaterializer_RecordsLifecycleMetrics(t *testing.T) {
 	}
 }
 
+func TestMaterializer_FilesStatusRecords(t *testing.T) {
+	summaryAt := func(status entity.RequestStatus, timestampMs int64) entity.RequestSummary {
+		summary := testRequestSummary()
+		summary.Status, summary.StatusTimestampMs = status, timestampMs
+		return summary
+	}
+	statusLog := func(status entity.RequestStatus, version int32, timestampMs int64) entity.RequestLog {
+		return entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: status, RequestVersion: version, TimestampMs: timestampMs}
+	}
+	record := func(status entity.RequestStatus, enteredAtMs int64) entity.QueueRequestStatus {
+		return entity.QueueRequestStatus{Queue: "q", Status: status, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: enteredAtMs}
+	}
+	landed := summaryAt(entity.RequestStatusLanded, 20)
+	landed.RequestVersion = 2
+
+	tests := []struct {
+		name       string
+		before     entity.RequestSummary
+		log        entity.RequestLog
+		wantUpdate bool
+		wantPut    *entity.QueueRequestStatus
+		wantDelete entity.RequestStatus
+	}{
+		{
+			name:       "accepting the request files its first record",
+			before:     testRequestSummary(),
+			log:        statusLog(entity.RequestStatusAccepted, 0, 20),
+			wantUpdate: true,
+			wantPut:    ptr(record(entity.RequestStatusAccepted, 20)),
+		},
+		{
+			name:       "a status change files the new record before removing the old one",
+			before:     summaryAt(entity.RequestStatusStarted, 15),
+			log:        statusLog(entity.RequestStatusValidating, 0, 20),
+			wantUpdate: true,
+			wantPut:    ptr(record(entity.RequestStatusValidating, 20)),
+			wantDelete: entity.RequestStatusStarted,
+		},
+		{
+			name:       "a terminal status removes the record and files none",
+			before:     summaryAt(entity.RequestStatusLanding, 15),
+			log:        statusLog(entity.RequestStatusLanded, 2, 20),
+			wantUpdate: true,
+			wantDelete: entity.RequestStatusLanding,
+		},
+		{
+			name:       "a newer log at the same status re-files without removing anything",
+			before:     summaryAt(entity.RequestStatusSpeculating, 15),
+			log:        statusLog(entity.RequestStatusSpeculating, 0, 20),
+			wantUpdate: true,
+			wantPut:    ptr(record(entity.RequestStatusSpeculating, 20)),
+		},
+		{
+			name:    "a redelivery that loses re-files the current record to repair an interrupted write",
+			before:  summaryAt(entity.RequestStatusBatched, 20),
+			log:     statusLog(entity.RequestStatusBatched, 0, 20),
+			wantPut: ptr(record(entity.RequestStatusBatched, 20)),
+		},
+		{
+			name:   "a finished request keeps no record on redelivery",
+			before: landed,
+			log:    statusLog(entity.RequestStatusLanded, 2, 20),
+		},
+		{
+			name:   "a request still accepting has no record",
+			before: testRequestSummary(),
+			log:    statusLog(entity.RequestStatusAccepting, 0, 20),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := newStrictMaterializerFixture(gomock.NewController(t))
+			f.logs.EXPECT().Insert(gomock.Any(), tt.log).Return(nil)
+			f.summaries.EXPECT().Get(gomock.Any(), tt.before.RequestID).Return(tt.before, nil)
+			if tt.wantUpdate {
+				f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), tt.before.Version, tt.before.Version+1).Return(nil)
+			}
+			f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			var calls []any
+			if tt.wantPut != nil {
+				calls = append(calls, f.statuses.EXPECT().Put(gomock.Any(), *tt.wantPut).Return(nil))
+			}
+			if tt.wantDelete != "" {
+				calls = append(calls, f.statuses.EXPECT().Delete(gomock.Any(), tt.wantDelete, "1").Return(nil))
+			}
+			gomock.InOrder(calls...)
+
+			require.NoError(t, f.materializer.PersistLog(context.Background(), tt.log))
+		})
+	}
+}
+
+func TestMaterializer_StatusRecordFailures(t *testing.T) {
+	writeErr := errors.New("storage unavailable")
+	tests := []struct {
+		name              string
+		putErr, deleteErr error
+		wantDeleteAttempt bool
+	}{
+		{name: "filing the new record", putErr: writeErr},
+		{name: "removing the old record", deleteErr: writeErr, wantDeleteAttempt: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := newStrictMaterializerFixture(gomock.NewController(t))
+			before := testRequestSummary()
+			before.Status = entity.RequestStatusStarted
+			log := entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusValidating, TimestampMs: 20}
+			f.logs.EXPECT().Insert(gomock.Any(), log).Return(nil)
+			f.summaries.EXPECT().Get(gomock.Any(), "1").Return(before, nil)
+			f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
+			f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			f.statuses.EXPECT().Put(gomock.Any(), gomock.Any()).Return(tt.putErr)
+			if tt.wantDeleteAttempt {
+				f.statuses.EXPECT().Delete(gomock.Any(), entity.RequestStatusStarted, "1").Return(tt.deleteErr)
+			}
+
+			require.ErrorIs(t, f.materializer.PersistLog(context.Background(), log), writeErr)
+		})
+	}
+}
+
 func TestMaterializer_PersistenceFailures(t *testing.T) {
 	writeErr := errors.New("storage unavailable")
 	tests := []struct {
@@ -335,3 +459,5 @@ func testRequestSummary() entity.RequestSummary {
 		Status: entity.RequestStatusAccepting, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{},
 	}
 }
+
+func ptr[T any](v T) *T { return &v }
diff --git a/submitqueue/gateway/core/request/request_receipt_test.go b/submitqueue/gateway/core/request/request_receipt_test.go
index a35dd05ec..54138cafb 100644
--- a/submitqueue/gateway/core/request/request_receipt_test.go
+++ b/submitqueue/gateway/core/request/request_receipt_test.go
@@ -34,14 +34,26 @@ type materializerReceiptFixture struct {
 	uris         *storagemock.MockRequestURIStore
 	logs         *storagemock.MockRequestLogStore
 	receipts     *storagemock.MockRequestReceiptStore
+	statuses     *storagemock.MockQueueRequestStatusStore
 	metrics      tally.TestScope
 }
 
+// newMaterializerReceiptFixture accepts every status record write, for tests that are not
+// about status records.
 func newMaterializerReceiptFixture(ctrl *gomock.Controller) materializerReceiptFixture {
+	f := newStrictMaterializerFixture(ctrl)
+	f.statuses.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	f.statuses.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	return f
+}
+
+// newStrictMaterializerFixture expects every status record write to be declared.
+func newStrictMaterializerFixture(ctrl *gomock.Controller) materializerReceiptFixture {
 	f := materializerReceiptFixture{
 		summaries: storagemock.NewMockRequestSummaryStore(ctrl),
 		uris:      storagemock.NewMockRequestURIStore(ctrl), logs: storagemock.NewMockRequestLogStore(ctrl),
 		receipts: storagemock.NewMockRequestReceiptStore(ctrl),
+		statuses: storagemock.NewMockQueueRequestStatusStore(ctrl),
 		metrics:  tally.NewTestScope("", nil),
 	}
 	stores := gwstoragemock.NewMockStorage(ctrl)
@@ -49,6 +61,7 @@ func newMaterializerReceiptFixture(ctrl *gomock.Controller) materializerReceiptF
 	stores.EXPECT().GetRequestURIStore().Return(f.uris).AnyTimes()
 	stores.EXPECT().GetRequestLogStore().Return(f.logs).AnyTimes()
 	stores.EXPECT().GetRequestReceiptStore().Return(f.receipts).AnyTimes()
+	stores.EXPECT().GetQueueRequestStatusStore().Return(f.statuses).AnyTimes()
 	factory := gwstoragemock.NewMockFactory(ctrl)
 	factory.EXPECT().For(gomock.Any()).Return(stores, nil).AnyTimes()
 	f.materializer = NewMaterializer(factory, WithMetricsScope(f.metrics))

From cc337082636b1636c65bbc02fe0a17636fb57825 Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 10:59:31 -0700
Subject: [PATCH 4/7] docs(gateway): Document queue_request_status and the
 advisory status records in the storage contracts

---
 AGENTS.md                                          |  2 +-
 doc/rfc/service-scoped-extensions.md               |  7 ++++---
 submitqueue/extension/storage/README.md            |  2 +-
 .../extension/storage/mysql/schema/README.md       | 14 +++++++++++---
 4 files changed, 17 insertions(+), 8 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index 54d1c70ec..dfed61469 100644
--- a/AGENTS.md
+++ b/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.**
 
diff --git a/doc/rfc/service-scoped-extensions.md b/doc/rfc/service-scoped-extensions.md
index 555447091..f8752b143 100644
--- a/doc/rfc/service-scoped-extensions.md
+++ b/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 |
 |---|---|
diff --git a/submitqueue/extension/storage/README.md b/submitqueue/extension/storage/README.md
index 026429d84..4ed70b9a5 100644
--- a/submitqueue/extension/storage/README.md
+++ b/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
 
diff --git a/submitqueue/gateway/extension/storage/mysql/schema/README.md b/submitqueue/gateway/extension/storage/mysql/schema/README.md
index 53463a5ee..60988b0bc 100644
--- a/submitqueue/gateway/extension/storage/mysql/schema/README.md
+++ b/submitqueue/gateway/extension/storage/mysql/schema/README.md
@@ -1,14 +1,14 @@
 # MySQL Schema
 
-The gateway's read model consists of an append-only request log, one authoritative summary, and two immutable lookup tables behind request-summary retrieval and `List`. The orchestrator's pipeline working state is a separate schema — see [../../../../../orchestrator/extension/storage/mysql/schema/README.md](../../../../../orchestrator/extension/storage/mysql/schema/README.md).
+The gateway's read model consists of an append-only request log, one authoritative summary, two immutable lookup tables behind request-summary retrieval and `List`, and advisory status membership records behind listing a queue's requests by status. The orchestrator's pipeline working state is a separate schema — see [../../../../../orchestrator/extension/storage/mysql/schema/README.md](../../../../../orchestrator/extension/storage/mysql/schema/README.md).
 
 ## Queue-leading primary keys
 
-Every table leads its primary key with `queue`: `request_summary` on `(queue, request_id)`, `request_log` on `(queue, request_id, timestamp_ms, salt)`, `change_uri_request_mapping` on `(queue, change_uri, received_at_ms, request_id)`, and `request_receipt` on `(queue, received_at_ms, request_id)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path.
+Every table leads its primary key with `queue`: `request_summary` on `(queue, request_id)`, `request_log` on `(queue, request_id, timestamp_ms, salt)`, `change_uri_request_mapping` on `(queue, change_uri, received_at_ms, request_id)`, `request_receipt` on `(queue, received_at_ms, request_id)`, and `queue_request_status` on `(queue, status, request_id)`. A queue-bound store instance prefixes every read and stamps every write with its bound queue, so one queue's rows are unreachable through another queue's binding and every table is shardable by queue. `//tool/linter/queueshard` enforces this, and also rejects any secondary index that does not itself lead with `queue`, since such an index would reintroduce a cross-queue access path.
 
 ## Read model
 
-`request_summary` holds the full projection. `request_receipt` and `change_uri_request_mapping` contain only immutable lookup keys.
+`request_summary` holds the full projection. `request_receipt` and `change_uri_request_mapping` contain only immutable lookup keys. `queue_request_status` is the one mutable index, and it is advisory: the summary stays the only source of truth.
 
 ### `request_summary`
 
@@ -22,6 +22,14 @@ Every table leads its primary key with `queue`: `request_summary` on `(queue, re
 
 `change_uri_request_mapping` is keyed by `(queue, change_uri, received_at_ms, request_id)` and serves bounded newest-first request-summary lookup by change URI within one queue. The gateway reads at most 101 mappings to enforce the API maximum of 100 results without silently truncating. A change URI landed into several queues has independent mappings in each, so looking it up across queues is one call per queue.
 
+### `queue_request_status`
+
+`queue_request_status` is keyed by `(queue, status, request_id)` and files each in-queue request under its public status, so "the requests of queue Q at status S" is a primary-key-prefix scan and `request_summary` keeps no secondary index. Each row carries only the request's `received_at_ms` and the `entered_at_ms` of the log that moved it into the status, enough to count and age a queue's requests by status.
+
+The materializer maintains it on every log delivery, after the summary update: it files the record for the summary's current status (a no-op when present, which also repairs a delivery that failed after the update) and, when the delivery moved the summary, then deletes the record under the previous status. Accepting and terminal statuses have no record. Every write is a single-key idempotent put or delete, so the records need no multi-row transaction. A delivery that fails between the summary update and that delete leaves a stale row under the old status that no later delivery revisits, so readers treat a listing as candidate request IDs, confirm each against its summary, and may delete rows that disagree.
+
+The records are kept only by a storage built with `WithQueueRequestStatusRecords`, which a deployment enables once the table exists; without it, record writes are no-ops and listing fails. Requests already in flight at that point gain a record on their next log delivery.
+
 ### `request_log`
 
 `request_log` is keyed by `(queue, request_id, timestamp_ms, salt)` and holds the append-only audit trail behind History. `salt` disambiguates entries sharing a request, queue and millisecond; it is part of the key but never exposed through the storage interface.

From 736088c791e94234e1ae90cfbd4a306bbc917b19 Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 11:34:01 -0700
Subject: [PATCH 5/7] fix(gateway): Re-check the summary after filing a status
 record, sweep records on terminal statuses, and skip event logs

---
 submitqueue/gateway/controller/land_test.go   |   6 +
 .../gateway/controller/log/log_test.go        |   5 +
 .../gateway/core/request/materializer.go      |  60 +++++-
 .../gateway/core/request/materializer_test.go | 172 ++++++++++++++----
 .../core/request/request_receipt_test.go      |  20 +-
 5 files changed, 213 insertions(+), 50 deletions(-)

diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go
index 3cdc7ff51..41debaa5f 100644
--- a/submitqueue/gateway/controller/land_test.go
+++ b/submitqueue/gateway/controller/land_test.go
@@ -492,6 +492,12 @@ func TestLand_PublishesToQueue(t *testing.T) {
 				return nil
 			},
 		),
+		// The status record is filed, then the summary re-read to confirm it still holds.
+		summaryStore.EXPECT().Get(gomock.Any(), "123").DoAndReturn(
+			func(context.Context, string) (entity.RequestSummary, error) {
+				return materializedSummary, nil
+			},
+		),
 	)
 
 	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), queuepausenoop.New(), registry)
diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go
index 12b3e9ac1..ac9dee5cd 100644
--- a/submitqueue/gateway/controller/log/log_test.go
+++ b/submitqueue/gateway/controller/log/log_test.go
@@ -166,6 +166,11 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr
 		return materializer
 	}
 	receiptStore.EXPECT().Create(gomock.Any(), entity.RequestReceipt{Queue: "test-queue", ReceivedAtMs: 1, RequestID: "2"}).Return(receiptErr)
+	if receiptErr != nil {
+		return materializer
+	}
+	// Filing the status record re-reads the summary to confirm the status still holds.
+	summaryStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestSummary{RequestID: "2", Queue: "test-queue", Status: entity.RequestStatusAccepted}, nil).AnyTimes()
 	return materializer
 }
 
diff --git a/submitqueue/gateway/core/request/materializer.go b/submitqueue/gateway/core/request/materializer.go
index c6b149e9b..9037f148f 100644
--- a/submitqueue/gateway/core/request/materializer.go
+++ b/submitqueue/gateway/core/request/materializer.go
@@ -110,20 +110,54 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 		if err := ensureRequestReceiptMapping(ctx, stores.GetRequestReceiptStore(), summary); err != nil {
 			return err
 		}
-		return fileStatusRecord(ctx, stores.GetQueueRequestStatusStore(), summary, previous, moved)
+		// An event log never moves the summary, so it leaves the status record to the
+		// status logs around it rather than spend a write re-filing an unchanged status.
+		if log.Type != entity.RequestLogTypeStatus {
+			return nil
+		}
+		return fileStatusRecord(ctx, stores.GetQueueRequestStatusStore(), summaries, summary, previous, moved)
 	}
 }
 
 // fileStatusRecord keeps the request's status membership record in step with its summary.
 //
-// It runs on every delivery, like the immutable mappings: putting the record for the
-// summary's current status repairs a delivery that failed after the summary update, and is
-// a no-op when the record already exists. When this delivery moved the summary, the record
+// It runs on every status-log delivery, like the immutable mappings: putting the record for
+// the summary's current status repairs a delivery that failed after the summary update, and
+// is a no-op when the record already exists. When this delivery moved the summary, the record
 // under the previous status is deleted only after the new one is filed, so a request in the
-// queue is never without a record. A record left under an old status by a delivery that
-// failed before that delete is not found again here — the redelivery no longer knows the old
-// status — which is why records are advisory and readers confirm them against the summary.
-func fileStatusRecord(ctx context.Context, records basestorage.QueueRequestStatusStore, summary entity.RequestSummary, previous entity.RequestStatus, moved bool) error {
+// queue is never without a record.
+//
+// Two rules keep records from outliving the status they file:
+//
+//   - Land and Cancel persist logs outside the log consumer's per-request ordering, so another
+//     writer may move the summary on, and delete this status's record, between this
+//     delivery's summary update and its Put. Every writer updates the summary before deleting
+//     the old record, so re-reading the summary after the Put sees any such move, and the
+//     record just filed is removed.
+//   - A redelivery no longer knows the status a failed delivery moved away from, so a
+//     terminal summary deletes the request's record under every in-queue status. Each
+//     terminal redelivery retries that sweep, and it also clears records stranded by earlier
+//     failed moves.
+//
+// A record stranded by a failed move while the request is still in the queue remains until
+// the request finishes, which is why readers must confirm records against the summary.
+func fileStatusRecord(
+	ctx context.Context,
+	records basestorage.QueueRequestStatusStore,
+	summaries basestorage.RequestSummaryStore,
+	summary entity.RequestSummary,
+	previous entity.RequestStatus,
+	moved bool,
+) error {
+	if isTerminalStatus(summary.Status) {
+		for _, status := range entity.InQueueRequestStatuses() {
+			if err := records.Delete(ctx, status, summary.RequestID); err != nil {
+				return fmt.Errorf("failed to remove request status record request_id=%s status=%s: %w", summary.RequestID, status, err)
+			}
+		}
+		return nil
+	}
+
 	if entity.IsRequestStatusInQueue(summary.Status) {
 		record := entity.QueueRequestStatus{
 			Queue:        summary.Queue,
@@ -135,7 +169,17 @@ func fileStatusRecord(ctx context.Context, records basestorage.QueueRequestStatu
 		if err := records.Put(ctx, record); err != nil {
 			return fmt.Errorf("failed to file request status record request_id=%s status=%s: %w", summary.RequestID, summary.Status, err)
 		}
+		current, err := summaries.Get(ctx, summary.RequestID)
+		if err != nil {
+			return fmt.Errorf("failed to re-read request summary request_id=%s: %w", summary.RequestID, err)
+		}
+		if current.Status != summary.Status {
+			if err := records.Delete(ctx, summary.Status, summary.RequestID); err != nil {
+				return fmt.Errorf("failed to remove superseded request status record request_id=%s status=%s: %w", summary.RequestID, summary.Status, err)
+			}
+		}
 	}
+
 	if moved && previous != summary.Status && entity.IsRequestStatusInQueue(previous) {
 		if err := records.Delete(ctx, previous, summary.RequestID); err != nil {
 			return fmt.Errorf("failed to remove request status record request_id=%s status=%s: %w", summary.RequestID, previous, err)
diff --git a/submitqueue/gateway/core/request/materializer_test.go b/submitqueue/gateway/core/request/materializer_test.go
index 4117cbebe..8c4cef7ef 100644
--- a/submitqueue/gateway/core/request/materializer_test.go
+++ b/submitqueue/gateway/core/request/materializer_test.go
@@ -138,6 +138,9 @@ func TestMaterializer_RecordsLifecycleMetrics(t *testing.T) {
 			f.summaries.EXPECT().Get(gomock.Any(), tt.before.RequestID).Return(tt.before, nil)
 			if tt.wantUpdate {
 				f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), tt.before.Version, tt.before.Version+1).Return(nil)
+				if entity.IsRequestStatusInQueue(tt.log.Status) {
+					f.expectRecheck(tt.log.Status)
+				}
 			}
 			f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 			f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
@@ -162,6 +165,59 @@ func TestMaterializer_RecordsLifecycleMetrics(t *testing.T) {
 	}
 }
 
+// recordCall is one expected status-record write, in order: a put (followed by the summary
+// re-read when recheck is set) or a delete.
+type recordCall struct {
+	put     bool
+	status  entity.RequestStatus
+	entered int64
+	// recheck is the status the summary re-read after a put reports.
+	recheck entity.RequestStatus
+}
+
+func putRecord(status entity.RequestStatus, entered int64, recheck entity.RequestStatus) recordCall {
+	return recordCall{put: true, status: status, entered: entered, recheck: recheck}
+}
+
+func deleteRecord(status entity.RequestStatus) recordCall {
+	return recordCall{status: status}
+}
+
+// sweepRecords is the terminal sweep: a delete under every in-queue status.
+func sweepRecords() []recordCall {
+	var calls []recordCall
+	for _, status := range entity.InQueueRequestStatuses() {
+		calls = append(calls, deleteRecord(status))
+	}
+	return calls
+}
+
+// expectRecordCalls declares calls in order, each put followed by its summary re-read. The
+// call at failAt returns err, and nothing after it is expected.
+func (f materializerReceiptFixture) expectRecordCalls(calls []recordCall, failAt int, err error) {
+	var ordered []any
+	for i, call := range calls {
+		failed := i == failAt
+		var callErr error
+		if failed {
+			callErr = err
+		}
+		if call.put {
+			record := entity.QueueRequestStatus{Queue: "q", Status: call.status, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: call.entered}
+			ordered = append(ordered, f.statuses.EXPECT().Put(gomock.Any(), record).Return(callErr))
+			if !failed {
+				ordered = append(ordered, f.expectRecheck(call.recheck))
+			}
+		} else {
+			ordered = append(ordered, f.statuses.EXPECT().Delete(gomock.Any(), call.status, "1").Return(callErr))
+		}
+		if failed {
+			break
+		}
+	}
+	gomock.InOrder(ordered...)
+}
+
 func TestMaterializer_FilesStatusRecords(t *testing.T) {
 	summaryAt := func(status entity.RequestStatus, timestampMs int64) entity.RequestSummary {
 		summary := testRequestSummary()
@@ -171,9 +227,6 @@ func TestMaterializer_FilesStatusRecords(t *testing.T) {
 	statusLog := func(status entity.RequestStatus, version int32, timestampMs int64) entity.RequestLog {
 		return entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: status, RequestVersion: version, TimestampMs: timestampMs}
 	}
-	record := func(status entity.RequestStatus, enteredAtMs int64) entity.QueueRequestStatus {
-		return entity.QueueRequestStatus{Queue: "q", Status: status, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: enteredAtMs}
-	}
 	landed := summaryAt(entity.RequestStatusLanded, 20)
 	landed.RequestVersion = 2
 
@@ -182,48 +235,76 @@ func TestMaterializer_FilesStatusRecords(t *testing.T) {
 		before     entity.RequestSummary
 		log        entity.RequestLog
 		wantUpdate bool
-		wantPut    *entity.QueueRequestStatus
-		wantDelete entity.RequestStatus
+		wantCalls  []recordCall
 	}{
 		{
 			name:       "accepting the request files its first record",
 			before:     testRequestSummary(),
 			log:        statusLog(entity.RequestStatusAccepted, 0, 20),
 			wantUpdate: true,
-			wantPut:    ptr(record(entity.RequestStatusAccepted, 20)),
+			wantCalls:  []recordCall{putRecord(entity.RequestStatusAccepted, 20, entity.RequestStatusAccepted)},
 		},
 		{
 			name:       "a status change files the new record before removing the old one",
 			before:     summaryAt(entity.RequestStatusStarted, 15),
 			log:        statusLog(entity.RequestStatusValidating, 0, 20),
 			wantUpdate: true,
-			wantPut:    ptr(record(entity.RequestStatusValidating, 20)),
-			wantDelete: entity.RequestStatusStarted,
+			wantCalls: []recordCall{
+				putRecord(entity.RequestStatusValidating, 20, entity.RequestStatusValidating),
+				deleteRecord(entity.RequestStatusStarted),
+			},
 		},
 		{
-			name:       "a terminal status removes the record and files none",
-			before:     summaryAt(entity.RequestStatusLanding, 15),
-			log:        statusLog(entity.RequestStatusLanded, 2, 20),
+			name:       "a newer log at the same status re-files without removing anything",
+			before:     summaryAt(entity.RequestStatusSpeculating, 15),
+			log:        statusLog(entity.RequestStatusSpeculating, 0, 20),
 			wantUpdate: true,
-			wantDelete: entity.RequestStatusLanding,
+			wantCalls:  []recordCall{putRecord(entity.RequestStatusSpeculating, 20, entity.RequestStatusSpeculating)},
 		},
 		{
-			name:       "a newer log at the same status re-files without removing anything",
+			name:      "a redelivery that loses re-files the current record to repair an interrupted write",
+			before:    summaryAt(entity.RequestStatusBatched, 20),
+			log:       statusLog(entity.RequestStatusBatched, 0, 20),
+			wantCalls: []recordCall{putRecord(entity.RequestStatusBatched, 20, entity.RequestStatusBatched)},
+		},
+		{
+			name:       "a record filed after another writer moved the summary on is removed again",
+			before:     testRequestSummary(),
+			log:        statusLog(entity.RequestStatusAccepted, 0, 20),
+			wantUpdate: true,
+			wantCalls: []recordCall{
+				putRecord(entity.RequestStatusAccepted, 20, entity.RequestStatusStarted),
+				deleteRecord(entity.RequestStatusAccepted),
+			},
+		},
+		{
+			name:       "a cancel overtaken by its terminal log removes both its own and the previous record",
 			before:     summaryAt(entity.RequestStatusSpeculating, 15),
-			log:        statusLog(entity.RequestStatusSpeculating, 0, 20),
+			log:        statusLog(entity.RequestStatusCancelling, 0, 20),
+			wantUpdate: true,
+			wantCalls: []recordCall{
+				putRecord(entity.RequestStatusCancelling, 20, entity.RequestStatusCancelled),
+				deleteRecord(entity.RequestStatusCancelling),
+				deleteRecord(entity.RequestStatusSpeculating),
+			},
+		},
+		{
+			name:       "a terminal status sweeps every in-queue record and files none",
+			before:     summaryAt(entity.RequestStatusLanding, 15),
+			log:        statusLog(entity.RequestStatusLanded, 2, 20),
 			wantUpdate: true,
-			wantPut:    ptr(record(entity.RequestStatusSpeculating, 20)),
+			wantCalls:  sweepRecords(),
 		},
 		{
-			name:    "a redelivery that loses re-files the current record to repair an interrupted write",
-			before:  summaryAt(entity.RequestStatusBatched, 20),
-			log:     statusLog(entity.RequestStatusBatched, 0, 20),
-			wantPut: ptr(record(entity.RequestStatusBatched, 20)),
+			name:      "a terminal redelivery sweeps again to retry a failed removal",
+			before:    landed,
+			log:       statusLog(entity.RequestStatusLanded, 2, 20),
+			wantCalls: sweepRecords(),
 		},
 		{
-			name:   "a finished request keeps no record on redelivery",
-			before: landed,
-			log:    statusLog(entity.RequestStatusLanded, 2, 20),
+			name:   "an event log leaves the record alone",
+			before: summaryAt(entity.RequestStatusSpeculating, 15),
+			log:    entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeEvent, Event: entity.RequestEventBuilding, TimestampMs: 20},
 		},
 		{
 			name:   "a request still accepting has no record",
@@ -241,44 +322,57 @@ func TestMaterializer_FilesStatusRecords(t *testing.T) {
 			}
 			f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 			f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-			var calls []any
-			if tt.wantPut != nil {
-				calls = append(calls, f.statuses.EXPECT().Put(gomock.Any(), *tt.wantPut).Return(nil))
-			}
-			if tt.wantDelete != "" {
-				calls = append(calls, f.statuses.EXPECT().Delete(gomock.Any(), tt.wantDelete, "1").Return(nil))
-			}
-			gomock.InOrder(calls...)
+			f.expectRecordCalls(tt.wantCalls, -1, nil)
 
 			require.NoError(t, f.materializer.PersistLog(context.Background(), tt.log))
 		})
 	}
 }
 
+// TestMaterializer_StatusRecordFailures covers every record write and the re-read failing:
+// each is returned so the delivery is retried, and nothing after it runs.
 func TestMaterializer_StatusRecordFailures(t *testing.T) {
 	writeErr := errors.New("storage unavailable")
+	moving := []recordCall{
+		putRecord(entity.RequestStatusValidating, 20, entity.RequestStatusValidating),
+		deleteRecord(entity.RequestStatusStarted),
+	}
+	superseded := []recordCall{
+		putRecord(entity.RequestStatusValidating, 20, entity.RequestStatusValidated),
+		deleteRecord(entity.RequestStatusValidating),
+	}
 	tests := []struct {
-		name              string
-		putErr, deleteErr error
-		wantDeleteAttempt bool
+		name       string
+		log        entity.RequestStatus
+		calls      []recordCall
+		failAt     int
+		recheckErr bool
 	}{
-		{name: "filing the new record", putErr: writeErr},
-		{name: "removing the old record", deleteErr: writeErr, wantDeleteAttempt: true},
+		{name: "filing the new record", log: entity.RequestStatusValidating, calls: moving, failAt: 0},
+		{name: "removing the previous record", log: entity.RequestStatusValidating, calls: moving, failAt: 1},
+		{name: "removing a superseded record", log: entity.RequestStatusValidating, calls: superseded, failAt: 1},
+		{name: "sweeping a finished request", log: entity.RequestStatusError, calls: sweepRecords()[:3], failAt: 2},
+		{name: "re-reading the summary", log: entity.RequestStatusValidating, recheckErr: true, failAt: -1},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			f := newStrictMaterializerFixture(gomock.NewController(t))
 			before := testRequestSummary()
 			before.Status = entity.RequestStatusStarted
-			log := entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: entity.RequestStatusValidating, TimestampMs: 20}
+			log := entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: tt.log, TimestampMs: 20}
 			f.logs.EXPECT().Insert(gomock.Any(), log).Return(nil)
 			f.summaries.EXPECT().Get(gomock.Any(), "1").Return(before, nil)
 			f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
 			f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 			f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-			f.statuses.EXPECT().Put(gomock.Any(), gomock.Any()).Return(tt.putErr)
-			if tt.wantDeleteAttempt {
-				f.statuses.EXPECT().Delete(gomock.Any(), entity.RequestStatusStarted, "1").Return(tt.deleteErr)
+			if tt.recheckErr {
+				record := entity.QueueRequestStatus{Queue: "q", Status: tt.log, RequestID: "1", ReceivedAtMs: 10, EnteredAtMs: 20}
+				gomock.InOrder(
+					f.statuses.EXPECT().Put(gomock.Any(), record).Return(nil),
+					f.summaries.EXPECT().Get(gomock.Any(), "1").Return(entity.RequestSummary{}, writeErr),
+				)
+			} else {
+				f.expectRecordCalls(tt.calls, tt.failAt, writeErr)
 			}
 
 			require.ErrorIs(t, f.materializer.PersistLog(context.Background(), log), writeErr)
diff --git a/submitqueue/gateway/core/request/request_receipt_test.go b/submitqueue/gateway/core/request/request_receipt_test.go
index 54138cafb..75d3f3b2a 100644
--- a/submitqueue/gateway/core/request/request_receipt_test.go
+++ b/submitqueue/gateway/core/request/request_receipt_test.go
@@ -74,14 +74,18 @@ func TestMaterializer_ActivatesReceiptMapping(t *testing.T) {
 			f := newMaterializerReceiptFixture(gomock.NewController(t))
 			current := testRequestSummary()
 			log := entity.RequestLog{Queue: current.Queue, RequestID: current.RequestID, Type: entity.RequestLogTypeStatus, Status: status, TimestampMs: 20}
-			gomock.InOrder(
+			calls := []any{
 				f.logs.EXPECT().Insert(gomock.Any(), log).Return(nil),
 				f.summaries.EXPECT().Get(gomock.Any(), current.RequestID).Return(current, nil),
 				f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil),
 				f.uris.EXPECT().Create(gomock.Any(), entity.RequestURI{Queue: current.Queue, ChangeURI: current.ChangeURIs[0], ReceivedAtMs: current.ReceivedAtMs, RequestID: current.RequestID}).Return(nil),
 				f.uris.EXPECT().Create(gomock.Any(), entity.RequestURI{Queue: current.Queue, ChangeURI: current.ChangeURIs[1], ReceivedAtMs: current.ReceivedAtMs, RequestID: current.RequestID}).Return(nil),
 				f.receipts.EXPECT().Create(gomock.Any(), receiptFromTestSummary(current)).Return(nil),
-			)
+			}
+			if entity.IsRequestStatusInQueue(status) {
+				calls = append(calls, f.expectRecheck(status))
+			}
+			gomock.InOrder(calls...)
 			require.NoError(t, f.materializer.PersistLog(context.Background(), log))
 		})
 	}
@@ -128,6 +132,7 @@ func TestMaterializer_RetriesReceiptMappingFailure(t *testing.T) {
 		f.receipts.EXPECT().Create(gomock.Any(), receiptFromTestSummary(current)).Return(writeErr),
 		f.summaries.EXPECT().Get(gomock.Any(), current.RequestID).Return(updated, nil),
 		f.receipts.EXPECT().Create(gomock.Any(), receiptFromTestSummary(current)).Return(nil),
+		f.expectRecheck(updated.Status),
 	)
 	require.ErrorIs(t, f.materializer.PersistLog(context.Background(), log), writeErr)
 	require.NoError(t, f.materializer.PersistLog(context.Background(), log))
@@ -161,6 +166,14 @@ func TestMaterializer_DoesNotCreateReceiptAfterURIMappingFailure(t *testing.T) {
 	require.ErrorIs(t, f.materializer.PersistLog(context.Background(), log), writeErr)
 }
 
+// expectRecheck expects the summary re-read that follows filing a status record, reporting
+// the summary still at status.
+func (f materializerReceiptFixture) expectRecheck(status entity.RequestStatus) *gomock.Call {
+	current := testRequestSummary()
+	current.Status = status
+	return f.summaries.EXPECT().Get(gomock.Any(), current.RequestID).Return(current, nil)
+}
+
 func receiptFromTestSummary(summary entity.RequestSummary) entity.RequestReceipt {
 	return entity.RequestReceipt{Queue: summary.Queue, ReceivedAtMs: summary.ReceivedAtMs, RequestID: summary.RequestID}
 }
@@ -175,7 +188,8 @@ func TestMaterializer_RetriesPartialURIActivation(t *testing.T) {
 	current.Status = entity.RequestStatusAccepted
 	log := entity.RequestLog{Queue: current.Queue, RequestID: current.RequestID, Type: entity.RequestLogTypeStatus, Status: current.Status, TimestampMs: current.StatusTimestampMs}
 	f.logs.EXPECT().Insert(gomock.Any(), log).Return(nil).Times(2)
-	f.summaries.EXPECT().Get(gomock.Any(), current.RequestID).Return(current, nil).Times(2)
+	// Two deliveries read the summary; the second, which files the status record, re-reads it.
+	f.summaries.EXPECT().Get(gomock.Any(), current.RequestID).Return(current, nil).Times(3)
 	mapping := func(uri string) entity.RequestURI {
 		return entity.RequestURI{Queue: current.Queue, ChangeURI: uri, ReceivedAtMs: current.ReceivedAtMs, RequestID: current.RequestID}
 	}

From 2132e394fdb805f5d3de734071a8dabf4930a0f8 Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 11:34:04 -0700
Subject: [PATCH 6/7] refactor(entity): Derive IsRequestStatusInQueue from
 InQueueRequestStatuses so the in-queue list lives in one place

---
 submitqueue/entity/queue_request_status.go | 18 +++++++-----------
 1 file changed, 7 insertions(+), 11 deletions(-)

diff --git a/submitqueue/entity/queue_request_status.go b/submitqueue/entity/queue_request_status.go
index 73feba549..2ea06e6fa 100644
--- a/submitqueue/entity/queue_request_status.go
+++ b/submitqueue/entity/queue_request_status.go
@@ -14,6 +14,8 @@
 
 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.
@@ -23,9 +25,10 @@ package entity
 // 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 transiently, or after an interrupted write, file a
-// request under a status it has already left. Readers load each listed request's summary
-// and classify it by the summary's own Status.
+// 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.
@@ -66,12 +69,5 @@ func InQueueRequestStatuses() []RequestStatus {
 
 // IsRequestStatusInQueue reports whether status is one of InQueueRequestStatuses().
 func IsRequestStatusInQueue(status RequestStatus) bool {
-	switch status {
-	case RequestStatusAccepted, RequestStatusStarted, RequestStatusValidating, RequestStatusValidated,
-		RequestStatusBatching, RequestStatusBatched, RequestStatusSpeculating, RequestStatusSpeculated,
-		RequestStatusLanding, RequestStatusCancelling:
-		return true
-	default:
-		return false
-	}
+	return slices.Contains(InQueueRequestStatuses(), status)
 }

From 08996adc5300e04ce49c676e4200156fd61ec450 Mon Sep 17 00:00:00 2001
From: Chenghan Ying <125171961+roychying@users.noreply.github.com>
Date: Tue, 20 Oct 2026 11:34:06 -0700
Subject: [PATCH 7/7] docs(gateway): State that status records go stale in
 normal operation and that readers must confirm and clean them up

---
 .../extension/storage/queue_request_status_store.go      | 8 +++++---
 .../gateway/extension/storage/mysql/schema/README.md     | 9 ++++++++-
 2 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/submitqueue/extension/storage/queue_request_status_store.go b/submitqueue/extension/storage/queue_request_status_store.go
index 682a309cf..0b9fca64e 100644
--- a/submitqueue/extension/storage/queue_request_status_store.go
+++ b/submitqueue/extension/storage/queue_request_status_store.go
@@ -28,9 +28,11 @@ import (
 // the request summary.
 //
 // Records are advisory: the authoritative status is on the RequestSummary, and a record may
-// file a request under a status it has already left. Readers therefore treat a listing as a
-// set of candidate request IDs — they load each summary by key and classify it by its own
-// Status, never by the bucket the record was found in.
+// file a request under a status it has already left. Readers must therefore treat a listing
+// as a set of candidate request IDs — load each summary by key, classify the request by the
+// summary's own Status rather than the bucket the record was found in, and delete records
+// whose summary is finished or at a different status. Writers do not clean up after every
+// interruption, so reader cleanup is part of the contract, not an optimization.
 //
 // The interface is per-status and per-record so that any backend can implement it without
 // multi-key queries or batch atomicity. A request moves buckets via Put of the new record
diff --git a/submitqueue/gateway/extension/storage/mysql/schema/README.md b/submitqueue/gateway/extension/storage/mysql/schema/README.md
index 60988b0bc..30f036593 100644
--- a/submitqueue/gateway/extension/storage/mysql/schema/README.md
+++ b/submitqueue/gateway/extension/storage/mysql/schema/README.md
@@ -26,7 +26,14 @@ Every table leads its primary key with `queue`: `request_summary` on `(queue, re
 
 `queue_request_status` is keyed by `(queue, status, request_id)` and files each in-queue request under its public status, so "the requests of queue Q at status S" is a primary-key-prefix scan and `request_summary` keeps no secondary index. Each row carries only the request's `received_at_ms` and the `entered_at_ms` of the log that moved it into the status, enough to count and age a queue's requests by status.
 
-The materializer maintains it on every log delivery, after the summary update: it files the record for the summary's current status (a no-op when present, which also repairs a delivery that failed after the update) and, when the delivery moved the summary, then deletes the record under the previous status. Accepting and terminal statuses have no record. Every write is a single-key idempotent put or delete, so the records need no multi-row transaction. A delivery that fails between the summary update and that delete leaves a stale row under the old status that no later delivery revisits, so readers treat a listing as candidate request IDs, confirm each against its summary, and may delete rows that disagree.
+The materializer maintains it on every status-log delivery, after the summary update: it files the record for the summary's current status (a no-op when present, which also repairs a delivery that failed after the update) and, when the delivery moved the summary, then deletes the record under the previous status. Accepting and terminal statuses have no record, and event logs, which never move the summary, write nothing. Every write is a single-key idempotent put or delete, so the records need no multi-row transaction.
+
+Stale rows are a normal occurrence, not only a failure mode, and two rules bound them:
+
+- Land and Cancel persist logs outside the log consumer's per-request ordering, so another writer can move the summary on and delete a status's record before a late put re-creates it. After each put the materializer re-reads the summary and deletes the record it just filed if the status has moved on; every writer updates the summary before deleting, so the re-read sees any such move.
+- A terminal summary deletes the request's record under every in-queue status, on every terminal delivery. That retries a failed removal and clears records stranded by earlier failed moves.
+
+A record stranded by a failed move while the request is still in the queue remains until the request finishes. Readers must therefore confirm each listed request against its summary, count it only under the summary's status, and delete rows whose summary is finished or at a different status. That costs one summary read per listed row per reader per sample, which is small while queues are short but grows with queue depth and the number of reading instances.
 
 The records are kept only by a storage built with `WithQueueRequestStatusRecords`, which a deployment enables once the table exists; without it, record writes are no-ops and listing fails. Requests already in flight at that point gain a record on their next log delivery.