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,
@@ -146,5 +164,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,21 @@ 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{}
+	}
+
+	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 +72,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 +90,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,13 +101,18 @@ 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 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)
 		}

submitqueue/extension/storage/mysql/request_log_store_test.go

@@ -0,0 +1,37 @@
+// 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)
+	require.Contains(t, err.Error(), "requires queue")
+}

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,