[2/N] [List API] feat(context): add immutable request admission store

submitqueue/entity/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "push_result.go",
         "queue_config.go",
         "request.go",
+        "request_context.go",
         "request_log.go",
         "speculation_tree.go",
     ],

submitqueue/entity/request_context.go

@@ -0,0 +1,27 @@
+// Copyright (c) 2025 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
+
+// RequestContext is immutable gateway-owned admission data for a request.
+type RequestContext struct {
+	// RequestID identifies the request this context belongs to.
+	RequestID string
+	// Queue identifies the queue that admitted the request.
+	Queue string
+	// ChangeURIs contains the original change URIs submitted with the request.
+	ChangeURIs []string
+	// AdmittedAtMs is the request admission time in Unix epoch milliseconds.
+	AdmittedAtMs int64
+}

submitqueue/extension/storage/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store.go",
         "build_store.go",
         "change_store.go",
+        "request_context_store.go",
         "request_log_store.go",
         "request_store.go",
         "speculation_tree_store.go",

submitqueue/extension/storage/mock/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store_mock.go",
         "build_store_mock.go",
         "change_store_mock.go",
+        "request_context_store_mock.go",
         "request_log_store_mock.go",
         "request_store_mock.go",
         "speculation_tree_store_mock.go",

submitqueue/extension/storage/mysql/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "batch_store.go",
         "build_store.go",
         "change_store.go",
+        "request_context_store.go",
         "request_log_store.go",
         "request_store.go",
         "speculation_tree_store.go",

submitqueue/extension/storage/mysql/request_context_store.go

@@ -0,0 +1,93 @@
+// Copyright (c) 2025 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"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	mysqlDriver "github.com/go-sql-driver/mysql"
+	"github.com/uber-go/tally"
+
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+type requestContextStore struct {
+	db    *sql.DB
+	scope tally.Scope
+}
+
+// NewRequestContextStore creates a MySQL-backed RequestContextStore.
+func NewRequestContextStore(db *sql.DB, scope tally.Scope) storage.RequestContextStore {
+	return &requestContextStore{db: db, scope: scope}
+}
+
+// Create persists immutable request admission data.
+func (s *requestContextStore) Create(ctx context.Context, requestContext entity.RequestContext) (retErr error) {
+	op := metrics.Begin(s.scope, "create")
+	defer func() { op.Complete(retErr) }()
+
+	changeURIs := requestContext.ChangeURIs
+	if changeURIs == nil {
+		changeURIs = []string{}
+	}
+	changeURIsJSON, err := json.Marshal(changeURIs)
+	if err != nil {
+		return fmt.Errorf("failed to marshal request context change URIs for request_id=%s: %w", requestContext.RequestID, err)
+	}
+
+	_, err = s.db.ExecContext(ctx,
+		"INSERT INTO request_context (request_id, queue, change_uri, admitted_at_ms) VALUES (?, ?, ?, ?)",
+		requestContext.RequestID, requestContext.Queue, changeURIsJSON, requestContext.AdmittedAtMs,
+	)
+	if err != nil {
+		var mysqlErr *mysqlDriver.MySQLError
+		if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
+			return fmt.Errorf("request context request_id=%s: %w", requestContext.RequestID, storage.ErrAlreadyExists)
+		}
+		return fmt.Errorf("failed to create request context request_id=%s: %w", requestContext.RequestID, err)
+	}
+	return nil
+}
+
+// Get returns immutable request admission data.
+func (s *requestContextStore) Get(ctx context.Context, requestID string) (ret entity.RequestContext, retErr error) {
+	op := metrics.Begin(s.scope, "get")
+	defer func() { op.Complete(retErr) }()
+
+	var changeURIsJSON []byte
+	err := s.db.QueryRowContext(ctx,
+		"SELECT request_id, queue, change_uri, admitted_at_ms FROM request_context WHERE request_id = ?",
+		requestID,
+	).Scan(&ret.RequestID, &ret.Queue, &changeURIsJSON, &ret.AdmittedAtMs)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.RequestContext{}, fmt.Errorf("request context request_id=%s: %w", requestID, storage.ErrNotFound)
+	}
+	if err != nil {
+		return entity.RequestContext{}, fmt.Errorf("failed to get request context request_id=%s: %w", requestID, err)
+	}
+	if err := json.Unmarshal(changeURIsJSON, &ret.ChangeURIs); err != nil {
+		return entity.RequestContext{}, fmt.Errorf("failed to unmarshal request context change URIs for request_id=%s: %w", requestID, err)
+	}
+	if ret.ChangeURIs == nil {
+		ret.ChangeURIs = []string{}
+	}
+	return ret, nil
+}

submitqueue/extension/storage/mysql/schema/README.md

@@ -22,3 +22,10 @@ As the `batch` table grows, the secondary index will grow with it, increasing st
 
 The `change` table records per-URI claims by in-flight requests. `request_id` is part of the primary key so that concurrent claims on the same URI by different requests coexist as distinct rows — a same-request retry collides on the PK and is a no-op (`INSERT IGNORE`), while a different-request claim is a new row that `GetByURI` surfaces for overlap detection. `queue` leads the key so queue-scoped lookups are primary-key-prefix scans and the table is shardable by queue.
 
+## request_context table
+
+`request_context` stores immutable gateway-owned admission data. It is created once by `Land` and supplies queue, original change URIs, and admission time to the List read model. The `(queue, admitted_at_ms, request_id)` index supports queue-scoped admission-order scans and future context backfills.
+
+## request_log table
+
+`request_log` stores immutable request status records.

submitqueue/extension/storage/mysql/schema/request_context.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS request_context (
+    request_id VARCHAR(255) NOT NULL,
+    queue VARCHAR(255) NOT NULL,
+    change_uri JSON NOT NULL,
+    admitted_at_ms BIGINT NOT NULL,
+    PRIMARY KEY (request_id),
+    KEY idx_queue_admitted (queue, admitted_at_ms, request_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

submitqueue/extension/storage/mysql/storage.go

@@ -32,6 +32,7 @@ type mysqlStorage struct {
 	buildStore           storage.BuildStore
 	speculationTreeStore storage.SpeculationTreeStore
 	requestLogStore      storage.RequestLogStore
+	requestContextStore  storage.RequestContextStore
 }
 
 // NewStorage creates a new MySQL storage.
@@ -45,6 +46,7 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) {
 		buildStore:           NewBuildStore(db, scope.SubScope("build_store")),
 		speculationTreeStore: NewSpeculationTreeStore(db, scope.SubScope("speculation_tree_store")),
 		requestLogStore:      NewRequestLogStore(db, scope.SubScope("request_log_store")),
+		requestContextStore:  NewRequestContextStore(db, scope.SubScope("request_context_store")),
 	}, nil
 }
 
@@ -83,6 +85,11 @@ func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore {
 	return f.requestLogStore
 }
 
+// GetRequestContextStore returns the MySQL-backed RequestContextStore.
+func (f *mysqlStorage) GetRequestContextStore() storage.RequestContextStore {
+	return f.requestContextStore
+}
+
 // Close closes the underlying database connection.
 func (f *mysqlStorage) Close() error {
 	return f.db.Close()

submitqueue/extension/storage/request_context_store.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2025 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=request_context_store.go -destination=mock/request_context_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+// RequestContextStore persists immutable gateway-owned request admission data.
+type RequestContextStore interface {
+	// Create persists requestContext. Returns ErrAlreadyExists when any context already exists for the request ID; callers own retry idempotency and conflict detection.
+	Create(ctx context.Context, requestContext entity.RequestContext) error
+
+	// Get returns the context for requestID. Returns ErrNotFound when it does not exist.
+	Get(ctx context.Context, requestID string) (entity.RequestContext, error)
+}

submitqueue/extension/storage/storage.go

@@ -65,6 +65,9 @@ type Storage interface {
 	// GetRequestLogStore returns the RequestLogStore instance.
 	GetRequestLogStore() RequestLogStore
 
+	// GetRequestContextStore returns the RequestContextStore instance.
+	GetRequestContextStore() RequestContextStore
+
 	// Close closes the storage and all underlying connections. Should only be called once at the end of the program.
 	Close() error
 }