feat(requestlog): persist queue and change metadata

Summary:

Request logs now carry queue and change metadata so downstream read models can build queue-scoped views without reloading the original request. Existing SQID parsing remains as a fallback for older or minimal log producers.

Test Plan:

✅ `make fmt && make build && make test && make check-mocks && make e2e-test`

Author: albertywu

submitqueue/entity/request_log.go

@@ -16,6 +16,7 @@ package entity
 
 import (
 	"encoding/json"
+	"strings"
 	"time"
 )
 
@@ -96,6 +97,10 @@ const (
 type RequestLog struct {
 	// RequestID is the ID of the request this log entry belongs to. References entity.Request.ID.
 	RequestID string `json:"request_id"`
+	// Queue is the queue this request belongs to. New log producers should set it explicitly.
+	Queue string `json:"queue"`
+	// ChangeURIs are the original change URIs submitted with the request. They are populated by gateway-originated accepted logs.
+	ChangeURIs []string `json:"change_uris"`
 	// TimestampMs is the time this log entry was created, in milliseconds since Unix epoch.
 	TimestampMs int64 `json:"timestamp_ms"`
 	// Status is the request status at the time this log entry was created. It may contain requests states from the state machine and also display-friendly intermediate statuses.
@@ -117,11 +122,24 @@ type RequestLog struct {
 // lastError is the last error message associated with the status at the time of this log entry, empty string if no error.
 // metadata is a set of key-value pairs providing additional context for this log entry. Not constrained to any specific format or schema, used for display or debugging purposes.
 func NewRequestLog(requestID string, status RequestStatus, requestVersion int32, lastError string, metadata map[string]string) RequestLog {
+	return NewRequestLogWithDetails(requestID, QueueFromRequestID(requestID), nil, status, requestVersion, lastError, metadata)
+}
+
+// NewRequestLogWithDetails creates a new RequestLog with queue and change information for list summaries.
+func NewRequestLogWithDetails(requestID, queue string, changeURIs []string, status RequestStatus, requestVersion int32, lastError string, metadata map[string]string) RequestLog {
 	if metadata == nil {
 		metadata = make(map[string]string)
 	}
+	if queue == "" {
+		queue = QueueFromRequestID(requestID)
+	}
+	if changeURIs == nil {
+		changeURIs = []string{}
+	}
 	return RequestLog{
 		RequestID:      requestID,
+		Queue:          queue,
+		ChangeURIs:     changeURIs,
 		TimestampMs:    time.Now().UnixMilli(),
 		Status:         status,
 		RequestVersion: requestVersion,
@@ -135,8 +153,9 @@ func (r RequestLog) ToBytes() ([]byte, error) {
 	return json.Marshal(r)
 }
 
-// RequestLogFromBytes deserializes a RequestLog from JSON bytes.
-// If metadata is absent from the JSON, it will be initialized as an empty map.
+// RequestLogFromBytes deserializes a RequestLog from JSON bytes. If metadata or change URIs are
+// absent from the JSON, they will be initialized as empty collections. If queue is absent, it will
+// be derived from the request ID when possible.
 func RequestLogFromBytes(data []byte) (RequestLog, error) {
 	var log RequestLog
 	err := json.Unmarshal(data, &log)
@@ -146,5 +165,53 @@ func RequestLogFromBytes(data []byte) (RequestLog, error) {
 	if log.Metadata == nil {
 		log.Metadata = make(map[string]string)
 	}
+	if log.ChangeURIs == nil {
+		log.ChangeURIs = []string{}
+	}
+	if log.Queue == "" {
+		log.Queue = QueueFromRequestID(log.RequestID)
+	}
 	return log, nil
 }
+
+// QueueFromRequestID extracts the queue from the current "<queue>/<number>" request ID format.
+// It strips only a trailing numeric path segment so queue names may contain slashes.
+func QueueFromRequestID(requestID string) string {
+	idx := strings.LastIndex(requestID, "/")
+	if idx <= 0 || idx == len(requestID)-1 {
+		return ""
+	}
+	for _, r := range requestID[idx+1:] {
+		if r < '0' || r > '9' {
+			return ""
+		}
+	}
+	return requestID[:idx]
+}
+
+// IsKnownRequestStatus returns true if status is a public request status emitted by SubmitQueue.
+func IsKnownRequestStatus(status RequestStatus) bool {
+	switch status {
+	case RequestStatusAccepted,
+		RequestStatusStarted,
+		RequestStatusValidating,
+		RequestStatusValidated,
+		RequestStatusBatching,
+		RequestStatusBatched,
+		RequestStatusScored,
+		RequestStatusSpeculating,
+		RequestStatusSpeculated,
+		RequestStatusBuilding,
+		RequestStatusBuilt,
+		RequestStatusWaitingPath,
+		RequestStatusLanding,
+		RequestStatusProcessing,
+		RequestStatusLanded,
+		RequestStatusError,
+		RequestStatusCancelling,
+		RequestStatusCancelled:
+		return true
+	default:
+		return false
+	}
+}

submitqueue/entity/request_log_test.go

@@ -51,6 +51,8 @@ func TestRequestLog_ToBytes(t *testing.T) {
 func TestRequestLogFromBytes(t *testing.T) {
 	original := RequestLog{
 		RequestID:      "my-queue/999",
+		Queue:          "my-queue",
+		ChangeURIs:     []string{"github://uber/repo/pull/1/abcdef"},
 		TimestampMs:    1709568000000,
 		Status:         RequestStatusProcessing,
 		RequestVersion: 3,
@@ -65,6 +67,8 @@ func TestRequestLogFromBytes(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.Equal(t, original.RequestID, deserialized.RequestID)
+	assert.Equal(t, original.Queue, deserialized.Queue)
+	assert.Equal(t, original.ChangeURIs, deserialized.ChangeURIs)
 	assert.Equal(t, original.TimestampMs, deserialized.TimestampMs)
 	assert.Equal(t, original.Status, deserialized.Status)
 	assert.Equal(t, original.RequestVersion, deserialized.RequestVersion)
@@ -103,6 +107,8 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) {
 			name: "with all fields populated",
 			log: RequestLog{
 				RequestID:      "queue1/100",
+				Queue:          "queue1",
+				ChangeURIs:     []string{},
 				TimestampMs:    1709568000000,
 				Status:         RequestStatusLanded,
 				RequestVersion: 5,
@@ -114,6 +120,8 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) {
 			name: "with error",
 			log: RequestLog{
 				RequestID:      "queue2/200",
+				Queue:          "queue2",
+				ChangeURIs:     []string{},
 				TimestampMs:    1709568001000,
 				Status:         RequestStatusError,
 				RequestVersion: 2,
@@ -125,6 +133,8 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) {
 			name: "with zero version",
 			log: RequestLog{
 				RequestID:      "queue3/300",
+				Queue:          "queue3",
+				ChangeURIs:     []string{},
 				TimestampMs:    1709568002000,
 				Status:         RequestStatusStarted,
 				RequestVersion: 0,
@@ -146,3 +156,10 @@ func TestRequestLog_SerializationRoundTrip(t *testing.T) {
 		})
 	}
 }
+
+func TestQueueFromRequestID(t *testing.T) {
+	assert.Equal(t, "queue", QueueFromRequestID("queue/100"))
+	assert.Equal(t, "org/queue", QueueFromRequestID("org/queue/100"))
+	assert.Empty(t, QueueFromRequestID("queue/not-a-number"))
+	assert.Empty(t, QueueFromRequestID("queue"))
+}

submitqueue/extension/storage/mysql/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "mysql",
@@ -22,3 +22,14 @@ go_library(
         "@com_github_uber_go_tally//:tally",
     ],
 )
+
+go_test(
+    name = "mysql_test",
+    srcs = ["request_log_store_test.go"],
+    embed = [":mysql"],
+    deps = [
+        "//submitqueue/entity",
+        "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally//:tally",
+    ],
+)

submitqueue/extension/storage/mysql/request_log_store.go

@@ -46,6 +46,24 @@ func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (re
 	op := metrics.Begin(r.scope, "insert")
 	defer func() { op.Complete(retErr) }()
 
+	if log.Queue == "" {
+		log.Queue = entity.QueueFromRequestID(log.RequestID)
+	}
+	if log.Queue == "" {
+		return fmt.Errorf("request log insert requires queue for request_id=%s", log.RequestID)
+	}
+	if log.ChangeURIs == nil {
+		log.ChangeURIs = []string{}
+	}
+	if log.Metadata == nil {
+		log.Metadata = map[string]string{}
+	}
+
+	changeURIsJSON, err := json.Marshal(log.ChangeURIs)
+	if err != nil {
+		return fmt.Errorf("failed to marshal change URIs for request log request_id=%s: %w", log.RequestID, err)
+	}
+
 	metadataJSON, err := json.Marshal(log.Metadata)
 	if err != nil {
 		return fmt.Errorf("failed to marshal metadata for request log request_id=%s: %w", log.RequestID, err)
@@ -57,8 +75,8 @@ func (r *requestLogStore) Insert(ctx context.Context, log entity.RequestLog) (re
 	salt := rand.Int64()
 
 	_, err = r.db.ExecContext(ctx,
-		"INSERT INTO request_log (request_id, timestamp_ms, salt, status, request_version, last_error, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
-		log.RequestID, log.TimestampMs, salt, log.Status, log.RequestVersion, log.LastError, metadataJSON,
+		"INSERT INTO request_log (request_id, queue, change_uri, timestamp_ms, salt, status, request_version, last_error, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+		log.RequestID, log.Queue, changeURIsJSON, log.TimestampMs, salt, log.Status, log.RequestVersion, log.LastError, metadataJSON,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to insert request log for request_id=%s timestamp_ms=%d: %w", log.RequestID, log.TimestampMs, err)
@@ -75,7 +93,7 @@ func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []ent
 	defer func() { op.Complete(retErr) }()
 
 	rows, err := r.db.QueryContext(ctx,
-		"SELECT request_id, timestamp_ms, status, request_version, last_error, metadata FROM request_log WHERE request_id = ? ORDER BY timestamp_ms ASC, salt ASC",
+		"SELECT request_id, queue, change_uri, timestamp_ms, status, request_version, last_error, metadata FROM request_log WHERE request_id = ? ORDER BY timestamp_ms ASC, salt ASC",
 		requestID,
 	)
 	if err != nil {
@@ -86,16 +104,27 @@ func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []ent
 	var logs []entity.RequestLog
 	for rows.Next() {
 		var log entity.RequestLog
+		var changeURIsJSON []byte
 		var metadataJSON []byte
 
-		err := rows.Scan(&log.RequestID, &log.TimestampMs, &log.Status, &log.RequestVersion, &log.LastError, &metadataJSON)
+		err := rows.Scan(&log.RequestID, &log.Queue, &changeURIsJSON, &log.TimestampMs, &log.Status, &log.RequestVersion, &log.LastError, &metadataJSON)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan request log row for request_id=%s: %w", requestID, err)
 		}
 
+		if err := json.Unmarshal(changeURIsJSON, &log.ChangeURIs); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal change URIs for request log request_id=%s: %w", requestID, err)
+		}
+		if log.ChangeURIs == nil {
+			log.ChangeURIs = []string{}
+		}
+
 		if err := json.Unmarshal(metadataJSON, &log.Metadata); err != nil {
 			return nil, fmt.Errorf("failed to unmarshal metadata for request log request_id=%s: %w", requestID, err)
 		}
+		if log.Metadata == nil {
+			log.Metadata = map[string]string{}
+		}
 
 		logs = append(logs, log)
 	}

submitqueue/extension/storage/mysql/request_log_store_test.go

@@ -0,0 +1,36 @@
+// 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"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+func TestRequestLogStoreInsert_RejectsMissingQueue(t *testing.T) {
+	store := NewRequestLogStore(nil, tally.NoopScope)
+
+	err := store.Insert(context.Background(), entity.RequestLog{
+		RequestID:   "not-an-sqid",
+		TimestampMs: 100,
+		Status:      entity.RequestStatusAccepted,
+	})
+
+	require.Error(t, err)
+}

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

@@ -22,3 +22,6 @@ 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_log table
+
+`request_log` stores immutable request status records. Schema application uses `CREATE TABLE IF NOT EXISTS` files, so existing local databases created before the `queue` and `change_uri` columns were added must be recreated or manually altered before running newer binaries.

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

@@ -1,5 +1,7 @@
 CREATE TABLE IF NOT EXISTS request_log (
     request_id VARCHAR(255) NOT NULL,
+    queue VARCHAR(255) NOT NULL,
+    change_uri JSON NOT NULL,
     timestamp_ms BIGINT NOT NULL,
     salt BIGINT NOT NULL,
     status VARCHAR(64) NOT NULL,