Skip to content

Commit 8f30300

Browse files
authored
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. ### Changes - 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. ### Rollout 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. ## Issue Links 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 e241aaf commit 8f30300

10 files changed

Lines changed: 345 additions & 120 deletions

File tree

‎doc/rfc/submitqueue/status-list-api.md‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -87,11 +87,11 @@ The immutable context is queue, change URIs, and receipt time. The mutable respo
8787

8888
The key supports authoritative lookup and conditional status updates for one request without a secondary index, and leads with the queue so the table is shardable by queue.
8989

90-
### Request Summaries by Queue
90+
### Requests by Receipt Time
9191

92-
The queue projection is logically keyed by `(queue, received_at_ms, sqid)` and must support a bounded descending scan over `(received_at_ms, sqid)`. A backend may satisfy that contract with a reverse range scan or by descending-encoding the ordered key components; cursors always carry the original receipt time and sqid values.
92+
The immutable `request_receipt` lookup contains only `(queue, received_at_ms, request_id)` as its primary key and supports a bounded descending scan over `(received_at_ms, request_id)`. Request IDs break timestamp ties in string order, not numeric order. A backend may use a reverse range scan or descending-encode the ordered key components; cursors always carry the original receipt time and request ID.
9393

94-
The current `request_summary_by_queue` row duplicates the complete `List` response so a page is served by one bounded range scan. Status updates still propagate to this projection during the receipt-lookup transition below; it is not the target storage model.
94+
List resolves these keys through `request_summary` rather than storing response fields in the lookup. The old `request_summary_by_queue` projection still receives writes for rollback, but List no longer reads it.
9595

9696
The logical key covers the `List` queue predicate, receipt-time range, newest-first ordering, and complete keyset cursor in one bounded scan.
9797

@@ -111,23 +111,23 @@ The URI is stored in the canonical form received from the validated Land request
111111

112112
After synchronous validation, Land generates the sqid and one receipt timestamp. The gateway persists the authoritative summary in the internal `accepting` state, then publishes the request to the orchestrator. It does not create the URI or queue projections while the request remains `accepting`.
113113

114-
After publication succeeds, Land appends the initial `accepted` request log. Materializing `accepted`, `started`, or any later event promotes the authoritative summary out of `accepting` and creates the URI and queue projections. This handles the race where the orchestrator emits `started` before Land finishes persisting `accepted`.
114+
After publication succeeds, Land appends the initial `accepted` request log. Materializing `accepted`, `started`, or any later lifecycle state promotes the authoritative summary out of `accepting`, creates the URI and legacy queue projections, and ensures the receipt lookup key exists. This handles the race where the orchestrator emits `started` before Land finishes persisting `accepted`.
115115

116116
An `accepted` event is the lowest public lifecycle state. A late `accepted` event is retained in request history but must not replace `started` or any later materialized status.
117117

118118
Pipeline publication is the Land success boundary. If publication fails, Land returns an error and leaves the hidden `accepting` receipt for operational cleanup. If publication succeeds but appending or materializing `accepted` fails, Land still returns the sqid because retrying the RPC would submit a duplicate request that is already in the pipeline. A later pipeline event activates and repairs the public projections.
119119

120120
### Request-Log Materialization
121121

122-
Every gateway request-log persistence path uses the same materialization component. It appends the audit log, compares the incoming entry with the authoritative summary, conditionally advances the winner, and propagates the authoritative value to the queue projection.
122+
Every gateway request-log persistence path uses the same materialization component. It appends the audit log, compares the incoming entry with the authoritative summary, and conditionally advances the winner. For public summaries, it completes the URI and legacy queue projection writes before ensuring the immutable receipt key, including on unchanged or stale-log retries. Hidden `accepting` summaries remain excluded; a failed receipt write is returned through the existing projection-failure path.
123123

124124
The winner comparison preserves the existing current-status reconciliation behavior:
125125

126126
1. A terminal request-state entry with a positive request version beats every non-terminal or unversioned winner.
127127
2. Between versioned terminal entries, the greater request version wins.
128128
3. Equal terminal versions use the greater log timestamp as a tie-breaker.
129129
4. When no versioned terminal winner exists, the greater log timestamp wins.
130-
5. Any retained lifecycle event promotes an `accepting` summary into the public projections.
130+
5. A retained lifecycle status entry promotes an `accepting` summary into the public projections; audit-only events do not.
131131
6. `accepted` cannot replace `started` or any later status, even when the accepted log arrives later.
132132

133133
Materialization uses optimistic concurrency so stale or out-of-order consumers cannot replace a newer winner. Version arithmetic and reconciliation decisions belong to the materialization component; stores perform only mechanical creates, reads, conditional updates, and bounded page queries.
@@ -144,15 +144,17 @@ The gateway performs a bounded newest-first scan of the URI reverse mapping and
144144

145145
### List by Queue and Receipt Time
146146

147-
The gateway performs one bounded range scan of the queue projection using queue, receipt-time bounds, and an optional keyset cursor. Queue projections are created only when the request reaches `accepted` or a later state, so `accepting` receipts are absent by construction. The ordering key is immutable, so later status updates cannot move an item across an issued cursor.
147+
The gateway scans `request_receipt` using the queue, receipt-time bounds, and optional keyset cursor, fetching one extra key to determine whether another page exists. It point-reads `request_summary` for each visible result, preserves receipt-key order, and forms the next token from the last returned receipt. The extra key does not require a summary read. Later status updates cannot move an item across the immutable cursor.
148+
149+
Receipt keys are created only for public summaries. An invalid key, missing summary, identity or receipt-time mismatch, or `accepting` summary fails the entire page as an internal consistency error rather than returning a user-facing not-found or silently omitting the item. Other backend failures retain their error chains.
148150

149151
## Consistency
150152

151-
The request log remains the append-only audit record. The authoritative summary and queue projection are eventually consistent views of its winning current state.
153+
The request log remains the append-only audit record. The authoritative summary is an eventually consistent view of its winning current state.
152154

153-
Because the authoritative summary and queue projection are separate writes, a short interval can exist where request-summary retrieval and `List` show different statuses. Retried materialization repairs the queue projection from the authoritative sqid summary until both converge. Neither API reconciles logs during reads to hide this interval.
155+
List and request-summary retrieval read the same authoritative projection, so List no longer waits for mutable state to propagate into a second table. Membership may lag until the receipt key is written and retries repair partial writes. Separate calls and per-item reads are not a shared snapshot and may observe different lifecycle versions. Neither API reconciles logs during reads.
154156

155-
Request context has a stronger guarantee than status convergence: the authoritative receipt is persisted before the request is published to the orchestrator. Public URI and queue projections are activated only by `accepted` or a later event. Once a queue projection is visible, its queue, sqid, change URIs, and receipt time are complete.
157+
Request context has a stronger guarantee than status convergence: the authoritative receipt is persisted before the request is published to the orchestrator. Public lookup keys are activated only by `accepted` or a later lifecycle state. Once a receipt key is visible, its authoritative summary has complete queue, sqid, change URIs, and receipt time.
156158

157159
## Compatibility and Rollout
158160

@@ -164,17 +166,13 @@ The request-summary RPCs replace the former polymorphic Status shape with explic
164166

165167
### Receipt-Lookup Transition
166168

167-
The target model keeps the full projection only in `request_summary`. An immutable `request_receipt` lookup contains only `(queue, received_at_ms, request_id)` as its primary key. List will scan these keys and point-read the corresponding authoritative summaries, preserving the existing response, required time window, string-ID tie ordering, and page tokens. Receipt time and all response fields are already captured; no additional data collection is needed.
168-
169-
The first phase adds the lookup and its writes, without changing List reads or removing the old projection. After materializing a public summary and completing the existing URI/queue projection writes, the materializer ensures the receipt key exists, including on unchanged or stale-log retries. Hidden `accepting` receipts are excluded. A failed lookup write is returned through the existing projection-failure path.
169+
The receipt lookup and writer path are implemented, and List now scans receipt keys and point-reads authoritative summaries. The response, required time window, string-ID tie ordering, and page-token format are unchanged. Receipt time and all response fields are already captured; no additional data collection is needed.
170170

171-
Deploy the additive table before enabling these writes. To finish the transition:
171+
CD must roll out the additive table and receipt writers to all instances before deploying the read cutover. The cutover performs one scan plus one point read per visible result; check page latency after rollout.
172172

173-
1. Upgrade all writers, backfill existing List membership from the old table's keys, validate each key against its public authoritative summary, and verify lookup coverage.
174-
2. Switch List to the receipt lookup plus authoritative summary reads, keeping old projection writes during the rollback window. This trades one page scan for one scan plus a point read per visible result; verify page latency before cutover.
175-
3. Stop duplicate projection writes, remove `RequestQueueSummary` and its store/mapper, and retire `request_summary_by_queue` after the rollback window. Only the small immutable lookup key remains duplicated.
173+
There is no historical backfill or fallback to the legacy queue projection. Requests without receipt keys are absent from List, even if their old queue-summary rows remain. Later log activity can create missing receipt keys.
176174

177-
This backfill preserves existing List history; it does not reconstruct requests predating the original read-model rollout. Backfill tooling, read cutover, and removal are follow-up work.
175+
Keep legacy projection writes during the rollback window; reverting the read cutover restores the old List path. Once stable, a follow-up stops duplicate projection writes and removes `RequestQueueSummary` and its store. Retire `request_summary_by_queue` after the rollback window. The final model keeps one full projection plus immutable lookup keys.
178176

179177
## Rejected Alternatives
180178

‎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

0 commit comments

Comments
 (0)