Skip to content

Commit a91ce72

Browse files
committed
feat(submitqueue): read List from receipt lookup
Summary: Builds on #799 to serve List from immutable request_receipt keys and authoritative request_summary rows, instead of the duplicated queue-summary projection. - Scan receipt keys, then point-read summaries for visible results; keep bounds, string-ID ordering, and page-token format unchanged. - Use the existing RequestSummary response type and wire mapper. - Fail the page on missing, mismatched, or hidden summaries using the existing InternalConsistencyError. - Add controller and gateway integration coverage; update the RFC. Keep legacy projection writes for rollback. Deploy only after #799 reaches all writers. No backfill or legacy-read fallback: requests without receipt keys are absent from List. Duplicate-write and old-table removal remain follow-up work. None — no tracking issue was supplied. Test Plan: After CD rollout, confirm new Land requests appear in List with the same summary fields as ID lookup, check continuation pages, and monitor page latency. Revert Plan: Revert the read cutover to restore the old List path. Legacy projection writes remain active; this PR makes no schema changes. API Changes: No RPC/proto changes. The Go ListResult now contains RequestSummary values.
1 parent 74e3cc3 commit a91ce72

9 files changed

Lines changed: 329 additions & 102 deletions

File tree

‎service/submitqueue/gateway/server/mapper/list.go‎

Lines changed: 1 addition & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -32,25 +32,8 @@ func ProtoToListRequest(req *pb.ListRequest) entity.ListRequest {
3232

3333
// ListResultToProto maps the domain result to the wire response.
3434
func ListResultToProto(result entity.ListResult) *pb.ListResponse {
35-
requests := make([]*pb.RequestSummary, 0, len(result.Requests))
36-
for _, summary := range result.Requests {
37-
requests = append(requests, RequestQueueSummaryToProto(summary))
38-
}
3935
return &pb.ListResponse{
40-
Requests: requests,
36+
Requests: RequestSummariesToProto(result.Requests),
4137
NextPageToken: result.NextPageToken,
4238
}
4339
}
44-
45-
// RequestQueueSummaryToProto maps a queue projection to the wire request summary.
46-
func RequestQueueSummaryToProto(summary entity.RequestQueueSummary) *pb.RequestSummary {
47-
return &pb.RequestSummary{
48-
Sqid: summary.RequestID,
49-
Queue: summary.Queue,
50-
ChangeUris: append([]string{}, summary.ChangeURIs...),
51-
ReceivedAtMs: summary.ReceivedAtMs,
52-
Status: string(summary.Status),
53-
LastError: summary.LastError,
54-
Metadata: cloneStringMap(summary.Metadata),
55-
}
56-
}

‎service/submitqueue/gateway/server/mapper/list_test.go‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,19 +42,26 @@ func TestProtoToListRequest(t *testing.T) {
4242

4343
func TestListResultToProto(t *testing.T) {
4444
result := entity.ListResult{
45-
Requests: []entity.RequestQueueSummary{{
45+
Requests: []entity.RequestSummary{{
4646
RequestID: "1",
4747
Queue: "q",
4848
ChangeURIs: []string{"github://uber/repo/pull/1/abc"},
4949
ReceivedAtMs: 100,
50-
Status: entity.RequestStatusAccepted,
51-
Metadata: map[string]string{},
50+
Status: entity.RequestStatusError,
51+
LastError: "build failed",
52+
Metadata: map[string]string{"build": "url"},
5253
}},
5354
NextPageToken: "next",
5455
}
5556

5657
response := ListResultToProto(result)
5758

58-
assert.Equal(t, "1", response.Requests[0].Sqid)
59-
assert.Equal(t, "next", response.NextPageToken)
59+
assert.Equal(t, &pb.ListResponse{
60+
Requests: []*pb.RequestSummary{{
61+
Sqid: "1", Queue: "q", ChangeUris: []string{"github://uber/repo/pull/1/abc"},
62+
ReceivedAtMs: 100, Status: string(entity.RequestStatusError), LastError: "build failed",
63+
Metadata: map[string]string{"build": "url"},
64+
}},
65+
NextPageToken: "next",
66+
}, response)
6067
}

‎submitqueue/entity/list.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ type ListRequest struct {
3131
// ListResult contains one page of queue receipt history.
3232
type ListResult struct {
3333
// Requests are ordered by receipt time descending, then request ID descending.
34-
Requests []RequestQueueSummary
34+
Requests []RequestSummary
3535
// NextPageToken is an opaque continuation token. Empty means this is the last page.
3636
NextPageToken string
3737
}

‎submitqueue/entity/request_summary.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ type RequestSummary struct {
5757
Metadata map[string]string
5858
}
5959

60-
// RequestQueueSummary is the queue-ordered projection returned by List.
60+
// RequestQueueSummary is a queue-ordered copy of a public request summary.
6161
type RequestQueueSummary struct {
6262
// RequestID is the canonical decimal request identifier, unique within Queue.
6363
RequestID string

‎submitqueue/gateway/controller/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ go_test(
3939
srcs = [
4040
"cancel_test.go",
4141
"land_test.go",
42+
"list_consistency_test.go",
4243
"list_queues_test.go",
4344
"list_test.go",
4445
"ping_test.go",

‎submitqueue/gateway/controller/list.go‎

Lines changed: 36 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ func (c *listController) List(ctx context.Context, req entity.ListRequest) (resu
9999
return entity.ListResult{}, fmt.Errorf("failed to resolve storage for queue %q: %w", req.Queue, err)
100100
}
101101

102-
query := basestorage.RequestQueueSummaryQuery{
102+
query := basestorage.RequestReceiptRange{
103103
ReceivedAtOrAfterMs: req.ReceivedAtOrAfterMs,
104104
ReceivedBeforeMs: req.ReceivedBeforeMs,
105105
Limit: pageSize + 1,
@@ -112,19 +112,27 @@ func (c *listController) List(ctx context.Context, req entity.ListRequest) (resu
112112
if token.Queue != req.Queue || token.ReceivedAtOrAfterMs != req.ReceivedAtOrAfterMs || token.ReceivedBeforeMs != req.ReceivedBeforeMs {
113113
return entity.ListResult{}, fmt.Errorf("page token does not match query: %w", ErrInvalidRequest)
114114
}
115-
query.HasCursor = true
116-
query.Cursor = basestorage.RequestQueueSummaryCursor{ReceivedAtMs: token.LastReceivedAtMs, RequestID: token.LastRequestID}
115+
query.Before = basestorage.RequestReceiptCursor{ReceivedAtMs: token.LastReceivedAtMs, RequestID: token.LastRequestID}
117116
}
118117

119-
summaries, err := store.GetRequestQueueSummaryStore().List(ctx, query)
118+
receipts, err := store.GetRequestReceiptStore().List(ctx, query)
120119
if err != nil {
121-
return entity.ListResult{}, fmt.Errorf("failed to list queue=%s: %w", req.Queue, err)
120+
return entity.ListResult{}, fmt.Errorf("failed to list request receipts queue=%q: %w", req.Queue, err)
121+
}
122+
123+
visible := receipts[:min(len(receipts), pageSize)]
124+
result.Requests = make([]entity.RequestSummary, 0, len(visible))
125+
if len(visible) > 0 {
126+
summaries := store.GetRequestSummaryStore()
127+
for _, receipt := range visible {
128+
summary, err := readListSummary(ctx, summaries, req.Queue, receipt)
129+
if err != nil {
130+
return entity.ListResult{}, err
131+
}
132+
result.Requests = append(result.Requests, summary)
133+
}
122134
}
123-
124-
visible := summaries
125-
result = entity.ListResult{Requests: make([]entity.RequestQueueSummary, 0, min(len(visible), pageSize))}
126-
if len(visible) > pageSize {
127-
visible = visible[:pageSize]
135+
if len(receipts) > pageSize {
128136
last := visible[len(visible)-1]
129137
result.NextPageToken = encodeListPageToken(listPageToken{
130138
Queue: req.Queue,
@@ -134,11 +142,27 @@ func (c *listController) List(ctx context.Context, req entity.ListRequest) (resu
134142
LastRequestID: last.RequestID,
135143
})
136144
}
137-
result.Requests = append(result.Requests, visible...)
138145
c.logger.Debugw("queue requests listed", "queue", req.Queue, "request_count", len(result.Requests), "has_next_page", result.NextPageToken != "")
139146
return result, nil
140147
}
141148

149+
func readListSummary(ctx context.Context, summaries basestorage.RequestSummaryStore, queue string, receipt entity.RequestReceipt) (entity.RequestSummary, error) {
150+
if receipt.Queue != queue || receipt.ReceivedAtMs <= 0 || receipt.RequestID == "" {
151+
return entity.RequestSummary{}, &InternalConsistencyError{Message: fmt.Sprintf("invalid request receipt queue=%q request_id=%q", queue, receipt.RequestID)}
152+
}
153+
summary, err := summaries.Get(ctx, receipt.RequestID)
154+
if err != nil {
155+
if basestorage.IsNotFound(err) {
156+
return entity.RequestSummary{}, &InternalConsistencyError{Message: fmt.Sprintf("request summary missing for receipt queue=%q request_id=%q", queue, receipt.RequestID)}
157+
}
158+
return entity.RequestSummary{}, fmt.Errorf("failed to read request summary queue=%q request_id=%q: %w", queue, receipt.RequestID, err)
159+
}
160+
if summary.Queue != queue || summary.RequestID != receipt.RequestID || summary.ReceivedAtMs != receipt.ReceivedAtMs || summary.Status == entity.RequestStatusAccepting {
161+
return entity.RequestSummary{}, &InternalConsistencyError{Message: fmt.Sprintf("request receipt disagrees with public summary queue=%q request_id=%q", queue, receipt.RequestID)}
162+
}
163+
return summary, nil
164+
}
165+
142166
func encodeListPageToken(token listPageToken) string {
143167
values := url.Values{
144168
"queue": {token.Queue},
@@ -178,7 +202,7 @@ func decodeListPageToken(encoded string) (listPageToken, error) {
178202
LastReceivedAtMs: lastReceivedAtMs,
179203
LastRequestID: values.Get("last_request_id"),
180204
}
181-
if token.Queue == "" || token.LastRequestID == "" || token.ReceivedAtOrAfterMs >= token.ReceivedBeforeMs {
205+
if token.Queue == "" || token.LastRequestID == "" || token.LastReceivedAtMs <= 0 || token.ReceivedAtOrAfterMs >= token.ReceivedBeforeMs {
182206
return listPageToken{}, fmt.Errorf("invalid token fields")
183207
}
184208
return token, nil
Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package controller
16+
17+
import (
18+
"context"
19+
"fmt"
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/stretchr/testify/require"
24+
"github.com/uber/submitqueue/platform/errs"
25+
"github.com/uber/submitqueue/submitqueue/entity"
26+
basestorage "github.com/uber/submitqueue/submitqueue/extension/storage"
27+
"go.uber.org/mock/gomock"
28+
)
29+
30+
func TestList_ConsistencyErrors(t *testing.T) {
31+
receipt := entity.RequestReceipt{Queue: "q", RequestID: "1", ReceivedAtMs: 100}
32+
summary := entity.RequestSummary{Queue: "q", RequestID: "1", ReceivedAtMs: 100, Status: entity.RequestStatusAccepted}
33+
tests := []struct {
34+
name string
35+
changeReceipt func(*entity.RequestReceipt)
36+
changeSummary func(*entity.RequestSummary)
37+
readErr error
38+
}{
39+
{name: "receipt from another queue", changeReceipt: func(r *entity.RequestReceipt) { r.Queue = "other" }},
40+
{name: "receipt has no request ID", changeReceipt: func(r *entity.RequestReceipt) { r.RequestID = "" }},
41+
{name: "receipt has no time", changeReceipt: func(r *entity.RequestReceipt) { r.ReceivedAtMs = 0 }},
42+
{name: "receipt has negative time", changeReceipt: func(r *entity.RequestReceipt) { r.ReceivedAtMs = -1 }},
43+
{name: "missing summary", readErr: fmt.Errorf("get summary: %w", basestorage.ErrNotFound)},
44+
{name: "summary from another queue", changeSummary: func(s *entity.RequestSummary) { s.Queue = "other" }},
45+
{name: "summary for another request", changeSummary: func(s *entity.RequestSummary) { s.RequestID = "2" }},
46+
{name: "summary has another receipt time", changeSummary: func(s *entity.RequestSummary) { s.ReceivedAtMs = 101 }},
47+
{name: "hidden accepting summary", changeSummary: func(s *entity.RequestSummary) { s.Status = entity.RequestStatusAccepting }},
48+
}
49+
for _, tt := range tests {
50+
t.Run(tt.name, func(t *testing.T) {
51+
f := newListTestFixture(t)
52+
badReceipt, badSummary := receipt, summary
53+
if tt.changeReceipt != nil {
54+
tt.changeReceipt(&badReceipt)
55+
} else {
56+
if tt.changeSummary != nil {
57+
tt.changeSummary(&badSummary)
58+
}
59+
f.summaries.EXPECT().Get(gomock.Any(), "1").Return(badSummary, tt.readErr)
60+
}
61+
f.queueConfigs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil)
62+
f.receipts.EXPECT().List(gomock.Any(), gomock.Any()).Return([]entity.RequestReceipt{badReceipt}, nil)
63+
64+
result, err := f.controller.List(context.Background(), entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 200})
65+
66+
require.Error(t, err)
67+
assert.True(t, IsInternalConsistency(err))
68+
assert.False(t, errs.IsUserError(err))
69+
assert.Equal(t, entity.ListResult{}, result)
70+
})
71+
}
72+
}
73+
74+
func TestList_DoesNotReturnPartialPage(t *testing.T) {
75+
f := newListTestFixture(t)
76+
f.queueConfigs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil)
77+
f.receipts.EXPECT().List(gomock.Any(), gomock.Any()).Return([]entity.RequestReceipt{
78+
{Queue: "q", RequestID: "2", ReceivedAtMs: 100},
79+
{Queue: "q", RequestID: "1", ReceivedAtMs: 90},
80+
}, nil)
81+
f.summaries.EXPECT().Get(gomock.Any(), "2").Return(entity.RequestSummary{
82+
Queue: "q", RequestID: "2", ReceivedAtMs: 100, Status: entity.RequestStatusAccepted,
83+
}, nil)
84+
f.summaries.EXPECT().Get(gomock.Any(), "1").Return(entity.RequestSummary{}, basestorage.ErrNotFound)
85+
86+
result, err := f.controller.List(context.Background(), entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 200})
87+
88+
assert.True(t, IsInternalConsistency(err))
89+
assert.Equal(t, entity.ListResult{}, result)
90+
}

0 commit comments

Comments
 (0)