Skip to content

Commit 2331282

Browse files
committed
feat(submitqueue): add immutable receipt lookup
Summary: First step toward keeping one full request projection in request_summary, with a small time-based lookup instead of duplicating lifecycle data for List. All required response data and receipt time are already captured. ### Changes - Add request_receipt with only (queue, received_at_ms, request_id), its immutable storage contract, and gateway wiring. - Ensure receipt keys after public projection writes, including unchanged/stale-log retries. Keep accepting receipts hidden and surface partial-write failures for existing retry handling. - Keep List reads and the existing queue-summary projection writes in place for a safe cutover. - Add storage, materializer, and integration coverage; document the transition in the RFC. ### Migration plan 1. Deploy the additive table before rolling out these writes. This PR populates receipt keys alongside the existing projection; it does not switch List reads. 2. Upgrade all writers, then backfill existing List membership from request_summary_by_queue keys. Validate keys against public authoritative summaries, exclude accepting receipts, and verify coverage before cutover. 3. In a follow-up, switch List to scan receipt keys and point-read request_summary. Preserve the API, time bounds, ordering, and page tokens; check page latency and keep old projection writes during the rollback window. 4. Once cutover is stable, stop duplicate projection writes, remove RequestQueueSummary and its store/mapper, and retire request_summary_by_queue after the rollback window. The final model stores one full projection plus immutable lookup keys. Backfill tooling, read cutover, and duplicate-data removal are follow-up work. This preserves currently listed history without reconstructing requests predating the original read-model rollout. Test Plan: Before the internal rollout, provision the new table and confirm newly public requests produce matching receipt keys while List continues serving the old projection. Existing automated tests cover retries, concurrency, hidden receipts, and pagination. Revert Plan: Revert the writer rollout; List still uses the existing projection. Leave the additive table in place until no writers depend on it. API Changes: No RPC/proto changes. Gateway storage implementations must provide the new RequestReceiptStore accessor.
1 parent 7740efe commit 2331282

25 files changed

Lines changed: 853 additions & 17 deletions

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

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ The transport representation of these errors follows the gateway-wide RPC error
7777

7878
## Gateway-Owned Read Model
7979

80-
The gateway owns the append-only request log and three new logical read models. The orchestrator's request and change stores are pipeline working state with different retention semantics, so neither API reads them.
80+
The gateway owns the append-only request log, materialized request summaries, and lookup stores. The orchestrator's request and change stores are pipeline working state with different retention semantics, so neither API reads them.
8181

8282
### Request Summary by Sqid
8383

@@ -91,7 +91,7 @@ The key supports authoritative lookup and conditional status updates for one req
9191

9292
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.
9393

94-
The row duplicates the complete `List` response deliberately. A page is served by one bounded range scan rather than one follow-up authoritative-summary read per result. Status updates propagate from the authoritative sqid summary to this projection.
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.
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

@@ -162,6 +162,20 @@ There is intentionally no backfill from orchestrator working state or historical
162162

163163
The request-summary RPCs replace the former polymorphic Status shape with explicit ID and change-URI lookups. This is a source-breaking API change. The List RPC is additive.
164164

165+
### Receipt-Lookup Transition
166+
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.
170+
171+
Deploy the additive table before enabling these writes. To finish the transition:
172+
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.
176+
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.
178+
165179
## Rejected Alternatives
166180

167181
### Put Request Context in Every Log Entry

‎submitqueue/entity/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ go_library(
2020
"request_batch.go",
2121
"request_history.go",
2222
"request_log.go",
23+
"request_receipt.go",
2324
"request_summary.go",
2425
"speculation.go",
2526
"subject.go",
Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
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 entity
16+
17+
// RequestReceipt is an immutable lookup key for a publicly visible request.
18+
type RequestReceipt struct {
19+
// Queue is the queue supplied at receipt.
20+
Queue string
21+
// ReceivedAtMs is the immutable, positive receipt timestamp in Unix milliseconds.
22+
ReceivedAtMs int64
23+
// RequestID identifies the request within Queue.
24+
RequestID string
25+
}

‎submitqueue/extension/storage/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ go_library(
1212
"request_batch_store.go",
1313
"request_log_store.go",
1414
"request_queue_summary_store.go",
15+
"request_receipt_store.go",
1516
"request_store.go",
1617
"request_summary_store.go",
1718
"request_uri_store.go",

‎submitqueue/extension/storage/mock/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ go_library(
1212
"request_batch_store_mock.go",
1313
"request_log_store_mock.go",
1414
"request_queue_summary_store_mock.go",
15+
"request_receipt_store_mock.go",
1516
"request_store_mock.go",
1617
"request_summary_store_mock.go",
1718
"request_uri_store_mock.go",

‎submitqueue/extension/storage/mock/request_receipt_store_mock.go‎

Lines changed: 72 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.
Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
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 storage
16+
17+
//go:generate mockgen -source=request_receipt_store.go -destination=mock/request_receipt_store_mock.go -package=mock
18+
19+
import (
20+
"context"
21+
22+
"github.com/uber/submitqueue/submitqueue/entity"
23+
)
24+
25+
// RequestReceiptCursor is an exclusive position in descending receipt-time order.
26+
type RequestReceiptCursor struct {
27+
// ReceivedAtMs is the positive receipt timestamp in Unix milliseconds; zero with an empty RequestID starts the range.
28+
ReceivedAtMs int64
29+
// RequestID breaks timestamp ties in descending string order, not numeric order.
30+
RequestID string
31+
}
32+
33+
// RequestReceiptRange bounds a primary-key scan within the store's queue.
34+
type RequestReceiptRange struct {
35+
// ReceivedAtOrAfterMs is the inclusive lower receipt-time bound in Unix milliseconds.
36+
ReceivedAtOrAfterMs int64
37+
// ReceivedBeforeMs is the exclusive upper receipt-time bound and must exceed the lower bound.
38+
ReceivedBeforeMs int64
39+
// Before is the exclusive continuation boundary; its zero value starts the range.
40+
Before RequestReceiptCursor
41+
// Limit is the positive maximum number of mappings to return.
42+
Limit int
43+
}
44+
45+
// RequestReceiptStore retains immutable receipt mappings in its bound queue.
46+
type RequestReceiptStore interface {
47+
// Create rejects queue mismatches, nonpositive receipt times, and empty request IDs.
48+
// An existing composite key returns ErrAlreadyExists.
49+
Create(ctx context.Context, receipt entity.RequestReceipt) error
50+
51+
// List scans the primary-key range in descending (received_at_ms, request_id) order.
52+
// It returns at most Limit mappings, an empty slice when none match, and an error for invalid ranges.
53+
List(ctx context.Context, bounds RequestReceiptRange) ([]entity.RequestReceipt, error)
54+
}

‎submitqueue/gateway/controller/land_test.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -348,6 +348,7 @@ func TestLand_PublishesToQueue(t *testing.T) {
348348
var materializedSummary entity.RequestSummary
349349
var persistedMapping entity.RequestURI
350350
var persistedQueueSummary entity.RequestQueueSummary
351+
var persistedReceipt entity.RequestReceipt
351352
var persistedLog entity.RequestLog
352353

353354
ctrl := gomock.NewController(t)
@@ -359,8 +360,10 @@ func TestLand_PublishesToQueue(t *testing.T) {
359360
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
360361
uriStore := storagemock.NewMockRequestURIStore(ctrl)
361362
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
363+
receiptStore := storagemock.NewMockRequestReceiptStore(ctrl)
362364
logStore := storagemock.NewMockRequestLogStore(ctrl)
363365
store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
366+
store.EXPECT().GetRequestReceiptStore().Return(receiptStore).AnyTimes()
364367
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
365368
store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
366369
store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
@@ -418,6 +421,12 @@ func TestLand_PublishesToQueue(t *testing.T) {
418421
return nil
419422
},
420423
),
424+
receiptStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
425+
func(_ context.Context, receipt entity.RequestReceipt) error {
426+
persistedReceipt = receipt
427+
return nil
428+
},
429+
),
421430
)
422431

423432
controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), registry)
@@ -444,6 +453,7 @@ func TestLand_PublishesToQueue(t *testing.T) {
444453
Metadata: map[string]string{},
445454
}, receiptSummary)
446455
assert.Positive(t, receiptSummary.ReceivedAtMs)
456+
assert.Equal(t, entity.RequestReceipt{Queue: req.Queue, ReceivedAtMs: receiptSummary.ReceivedAtMs, RequestID: result.ID}, persistedReceipt)
447457
assert.Equal(t, entity.RequestLog{
448458
RequestID: "123",
449459
Queue: "test-queue",

‎submitqueue/gateway/controller/log/log_test.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,8 +141,10 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr
141141
logStore := storagemock.NewMockRequestLogStore(ctrl)
142142
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
143143
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
144+
receiptStore := storagemock.NewMockRequestReceiptStore(ctrl)
144145
uriStore := storagemock.NewMockRequestURIStore(ctrl)
145146
store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
147+
store.EXPECT().GetRequestReceiptStore().Return(receiptStore).AnyTimes()
146148
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
147149
store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
148150
store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
@@ -169,6 +171,9 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr
169171
Status: entity.RequestStatusAccepted, Version: 1, Metadata: map[string]string{},
170172
}, nil)
171173
queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(queueErr)
174+
if queueErr == nil {
175+
receiptStore.EXPECT().Create(gomock.Any(), entity.RequestReceipt{Queue: "test-queue", ReceivedAtMs: 1, RequestID: "2"}).Return(nil)
176+
}
172177
return materializer
173178
}
174179

‎submitqueue/gateway/controller/storage_fixture_test.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ type controllerStorageFixture struct {
3333
storage *gwstoragemock.MockStorage
3434
summaryStore *storagemock.MockRequestSummaryStore
3535
queueStore *storagemock.MockRequestQueueSummaryStore
36+
receiptStore *storagemock.MockRequestReceiptStore
3637
uriStore *storagemock.MockRequestURIStore
3738
logStore *storagemock.MockRequestLogStore
3839
mu sync.Mutex
@@ -47,12 +48,14 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt
4748
storage: gwstoragemock.NewMockStorage(ctrl),
4849
summaryStore: storagemock.NewMockRequestSummaryStore(ctrl),
4950
queueStore: storagemock.NewMockRequestQueueSummaryStore(ctrl),
51+
receiptStore: storagemock.NewMockRequestReceiptStore(ctrl),
5052
uriStore: storagemock.NewMockRequestURIStore(ctrl),
5153
logStore: storagemock.NewMockRequestLogStore(ctrl),
5254
summaries: make(map[string]entity.RequestSummary),
5355
queueSummaries: make(map[string]entity.RequestQueueSummary),
5456
}
5557
fixture.storage.EXPECT().GetRequestQueueSummaryStore().Return(fixture.queueStore).AnyTimes()
58+
fixture.storage.EXPECT().GetRequestReceiptStore().Return(fixture.receiptStore).AnyTimes()
5659
fixture.storage.EXPECT().GetRequestSummaryStore().Return(fixture.summaryStore).AnyTimes()
5760
fixture.storage.EXPECT().GetRequestLogStore().Return(fixture.logStore).AnyTimes()
5861
fixture.storage.EXPECT().GetRequestURIStore().Return(fixture.uriStore).AnyTimes()
@@ -127,6 +130,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt
127130
}).AnyTimes()
128131

129132
fixture.uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
133+
fixture.receiptStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
130134
fixture.logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.RequestLog) error {
131135
fixture.mu.Lock()
132136
defer fixture.mu.Unlock()

0 commit comments

Comments
 (0)