Skip to content

Commit a165683

Browse files
albertywualbertywu
committed
feat(gateway): persist request context on Land
Summary: Add a focused admission writer that creates authoritative request context, change URI mappings, and the initial queue projection before Land appends its accepted log or publishes asynchronous work. Duplicate creates are reconciled in the application layer so identical retries converge and conflicting immutable context fails fast. Test Plan: make fmt make gazelle ./tool/bazel test //submitqueue/core/request:go_default_test //submitqueue/gateway/controller:go_default_test --test_output=errors Revert Plan: Revert this commit to stop writing request context. Existing additive rows may remain unused. API Changes: None. Monitoring and Alerts: Land controller operation metrics report admission write failures through the existing failure path.
1 parent 5d0f839 commit a165683

5 files changed

Lines changed: 366 additions & 23 deletions

File tree

‎submitqueue/core/request/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"admission.go",
67
"log.go",
78
"request.go",
89
],
@@ -20,6 +21,7 @@ go_library(
2021
go_test(
2122
name = "go_default_test",
2223
srcs = [
24+
"admission_test.go",
2325
"log_test.go",
2426
"request_test.go",
2527
],

‎submitqueue/core/request/admission.go‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
// Copyright (c) 2025 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 request
16+
17+
import (
18+
"context"
19+
"errors"
20+
"fmt"
21+
"maps"
22+
"slices"
23+
24+
"github.com/uber/submitqueue/submitqueue/entity"
25+
"github.com/uber/submitqueue/submitqueue/extension/storage"
26+
)
27+
28+
// AdmissionWriter creates immutable request context and initial read-model projections.
29+
// Storage implementations remain mechanical; this type decides whether duplicate creates are identical retries or conflicts.
30+
type AdmissionWriter struct {
31+
store storage.Storage
32+
}
33+
34+
// NewAdmissionWriter creates a request receipt projection writer.
35+
func NewAdmissionWriter(store storage.Storage) *AdmissionWriter {
36+
return &AdmissionWriter{store: store}
37+
}
38+
39+
// Create writes immutable request context and initial accepted projections.
40+
// A duplicate for the same request ID is accepted only when its immutable context matches exactly.
41+
func (m *AdmissionWriter) Create(ctx context.Context, summary entity.RequestSummary) error {
42+
if err := m.createRequestSummary(ctx, summary); err != nil {
43+
return err
44+
}
45+
46+
for _, changeURI := range summary.ChangeURIs {
47+
mapping := entity.RequestURI{
48+
ChangeURI: changeURI,
49+
ReceivedAtMs: summary.ReceivedAtMs,
50+
RequestID: summary.RequestID,
51+
}
52+
if err := m.store.GetRequestURIStore().Create(ctx, mapping); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
53+
return fmt.Errorf("failed to create request URI mapping request_id=%s change_uri=%s: %w", summary.RequestID, changeURI, err)
54+
}
55+
}
56+
57+
queueSummary := queueSummaryFromSummary(summary)
58+
if err := m.store.GetRequestQueueSummaryStore().Create(ctx, queueSummary); err != nil {
59+
if !errors.Is(err, storage.ErrAlreadyExists) {
60+
return fmt.Errorf("failed to create queue summary request_id=%s: %w", summary.RequestID, err)
61+
}
62+
existing, getErr := m.store.GetRequestQueueSummaryStore().Get(ctx, summary.Queue, summary.ReceivedAtMs, summary.RequestID)
63+
if getErr != nil {
64+
return fmt.Errorf("failed to get duplicate queue summary request_id=%s: %w", summary.RequestID, getErr)
65+
}
66+
if !sameQueueSummaryIdentity(existing, queueSummary) {
67+
return fmt.Errorf("conflicting queue summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists)
68+
}
69+
}
70+
71+
return nil
72+
}
73+
74+
func (m *AdmissionWriter) createRequestSummary(ctx context.Context, summary entity.RequestSummary) error {
75+
if err := m.store.GetRequestSummaryStore().Create(ctx, summary); err != nil {
76+
if !errors.Is(err, storage.ErrAlreadyExists) {
77+
return fmt.Errorf("failed to create request summary request_id=%s: %w", summary.RequestID, err)
78+
}
79+
existing, getErr := m.store.GetRequestSummaryStore().Get(ctx, summary.RequestID)
80+
if getErr != nil {
81+
return fmt.Errorf("failed to get duplicate request summary request_id=%s: %w", summary.RequestID, getErr)
82+
}
83+
if !sameRequestSummaryIdentity(existing, summary) {
84+
return fmt.Errorf("conflicting request summary request_id=%s: %w", summary.RequestID, storage.ErrAlreadyExists)
85+
}
86+
}
87+
return nil
88+
}
89+
90+
func queueSummaryFromSummary(summary entity.RequestSummary) entity.RequestQueueSummary {
91+
return entity.RequestQueueSummary{
92+
RequestID: summary.RequestID,
93+
Queue: summary.Queue,
94+
ChangeURIs: slices.Clone(summary.ChangeURIs),
95+
ReceivedAtMs: summary.ReceivedAtMs,
96+
Status: summary.Status,
97+
Version: summary.Version,
98+
LastError: summary.LastError,
99+
Metadata: cloneMetadata(summary.Metadata),
100+
}
101+
}
102+
103+
func sameRequestSummaryIdentity(left, right entity.RequestSummary) bool {
104+
return left.RequestID == right.RequestID &&
105+
left.Queue == right.Queue &&
106+
left.ReceivedAtMs == right.ReceivedAtMs &&
107+
slices.Equal(left.ChangeURIs, right.ChangeURIs)
108+
}
109+
110+
func sameQueueSummaryIdentity(left, right entity.RequestQueueSummary) bool {
111+
return left.RequestID == right.RequestID &&
112+
left.Queue == right.Queue &&
113+
left.ReceivedAtMs == right.ReceivedAtMs &&
114+
slices.Equal(left.ChangeURIs, right.ChangeURIs)
115+
}
116+
117+
func cloneMetadata(metadata map[string]string) map[string]string {
118+
if metadata == nil {
119+
return map[string]string{}
120+
}
121+
return maps.Clone(metadata)
122+
}

‎submitqueue/core/request/admission_test.go‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
// Copyright (c) 2025 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 request
16+
17+
import (
18+
"context"
19+
"fmt"
20+
"testing"
21+
22+
"github.com/stretchr/testify/require"
23+
"github.com/uber/submitqueue/submitqueue/entity"
24+
"github.com/uber/submitqueue/submitqueue/extension/storage"
25+
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
26+
"go.uber.org/mock/gomock"
27+
)
28+
29+
func TestAdmissionWriter_Create(t *testing.T) {
30+
summary := testRequestSummary()
31+
tests := []struct {
32+
name string
33+
setup func(*gomock.Controller, *storagemock.MockStorage)
34+
wantErr error
35+
}{
36+
{
37+
name: "creates all projections",
38+
setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
39+
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
40+
uriStore := storagemock.NewMockRequestURIStore(ctrl)
41+
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
42+
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
43+
store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
44+
store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
45+
summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil)
46+
uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/1", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil)
47+
uriStore.EXPECT().Create(gomock.Any(), entity.RequestURI{ChangeURI: "uri/2", ReceivedAtMs: 10, RequestID: "q/1"}).Return(nil)
48+
queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(nil)
49+
},
50+
},
51+
{
52+
name: "identical retry succeeds",
53+
setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
54+
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
55+
uriStore := storagemock.NewMockRequestURIStore(ctrl)
56+
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
57+
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
58+
store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
59+
store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
60+
summaryStore.EXPECT().Create(gomock.Any(), summary).Return(storage.ErrAlreadyExists)
61+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(summary, nil)
62+
uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists).Times(2)
63+
queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(summary)).Return(storage.ErrAlreadyExists)
64+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(summary), nil)
65+
},
66+
},
67+
{
68+
name: "conflicting summary retry fails",
69+
setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
70+
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
71+
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
72+
summaryStore.EXPECT().Create(gomock.Any(), summary).Return(storage.ErrAlreadyExists)
73+
conflict := summary
74+
conflict.Queue = "other"
75+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(conflict, nil)
76+
},
77+
wantErr: storage.ErrAlreadyExists,
78+
},
79+
{
80+
name: "URI write failure stops queue projection",
81+
setup: func(ctrl *gomock.Controller, store *storagemock.MockStorage) {
82+
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
83+
uriStore := storagemock.NewMockRequestURIStore(ctrl)
84+
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
85+
store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
86+
summaryStore.EXPECT().Create(gomock.Any(), summary).Return(nil)
87+
uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("URI down"))
88+
},
89+
},
90+
}
91+
92+
for _, tt := range tests {
93+
t.Run(tt.name, func(t *testing.T) {
94+
ctrl := gomock.NewController(t)
95+
store := storagemock.NewMockStorage(ctrl)
96+
tt.setup(ctrl, store)
97+
err := NewAdmissionWriter(store).Create(context.Background(), summary)
98+
if tt.wantErr != nil {
99+
require.ErrorIs(t, err, tt.wantErr)
100+
} else if tt.name == "URI write failure stops queue projection" {
101+
require.Error(t, err)
102+
} else {
103+
require.NoError(t, err)
104+
}
105+
})
106+
}
107+
}
108+
109+
func testRequestSummary() entity.RequestSummary {
110+
return entity.RequestSummary{
111+
RequestID: "q/1", Queue: "q", ChangeURIs: []string{"uri/1", "uri/2"}, ReceivedAtMs: 10,
112+
Status: entity.RequestStatusAccepted, StatusTimestampMs: 10, Version: 1, Metadata: map[string]string{},
113+
}
114+
}

‎submitqueue/gateway/controller/land.go‎

Lines changed: 37 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"context"
1919
"errors"
2020
"fmt"
21+
"time"
2122

2223
"github.com/uber-go/tally"
2324
mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
@@ -29,6 +30,7 @@ import (
2930
"github.com/uber/submitqueue/platform/errs"
3031
"github.com/uber/submitqueue/platform/extension/counter"
3132
"github.com/uber/submitqueue/platform/metrics"
33+
requestcore "github.com/uber/submitqueue/submitqueue/core/request"
3234
"github.com/uber/submitqueue/submitqueue/core/topickey"
3335
"github.com/uber/submitqueue/submitqueue/entity"
3436
"github.com/uber/submitqueue/submitqueue/extension/queueconfig"
@@ -65,25 +67,27 @@ func IsUnrecognizedQueue(err error) bool {
6567

6668
// LandController handles land business logic for the gateway
6769
type LandController struct {
68-
logger *zap.SugaredLogger
69-
metricsScope tally.Scope
70-
counter counter.Counter
71-
store storage.Storage
72-
queueConfigs queueconfig.Store
73-
registry consumer.TopicRegistry
70+
logger *zap.SugaredLogger
71+
metricsScope tally.Scope
72+
counter counter.Counter
73+
store storage.Storage
74+
admissionWriter *requestcore.AdmissionWriter
75+
queueConfigs queueconfig.Store
76+
registry consumer.TopicRegistry
7477
}
7578

7679
// NewLandController creates a new instance of the gateway land controller.
7780
// The controller publishes land requests to the topic registered under
7881
// topickey.TopicKeyStart in the registry.
7982
func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter counter.Counter, store storage.Storage, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) *LandController {
8083
return &LandController{
81-
logger: logger,
82-
metricsScope: scope.SubScope("land_controller"),
83-
counter: counter,
84-
store: store,
85-
queueConfigs: queueConfigs,
86-
registry: registry,
84+
logger: logger,
85+
metricsScope: scope.SubScope("land_controller"),
86+
counter: counter,
87+
store: store,
88+
admissionWriter: requestcore.NewAdmissionWriter(store),
89+
queueConfigs: queueConfigs,
90+
registry: registry,
8791
}
8892
}
8993

@@ -132,13 +136,32 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
132136
Change: change,
133137
LandStrategy: strategy,
134138
}
139+
receivedAtMs := time.Now().UnixMilli()
140+
summary := entity.RequestSummary{
141+
RequestID: landRequest.ID,
142+
Queue: landRequest.Queue,
143+
ChangeURIs: append([]string{}, landRequest.Change.URIs...),
144+
ReceivedAtMs: receivedAtMs,
145+
Status: entity.RequestStatusAccepted,
146+
StatusTimestampMs: receivedAtMs,
147+
Version: 1,
148+
Metadata: map[string]string{},
149+
}
150+
if err := c.admissionWriter.Create(ctx, summary); err != nil {
151+
return nil, fmt.Errorf("LandController failed to create request receipt sqid=%s: %w", landRequest.ID, err)
152+
}
135153

136154
// Record the accepted status in the request log for reconciliation. Once the request materializes as a Request entity, the status might be updated to "new".
137155
// It is important to record the status before publishing to the queue for processing. It is important to publish straight to the database and not via a entityqueue.
138156
// Gateway has to stay consistent with the request log.
139-
logEntry := entity.NewRequestLog(landRequest.ID, entity.RequestStatusAccepted, 0, "", nil)
157+
logEntry := entity.RequestLog{
158+
RequestID: landRequest.ID,
159+
TimestampMs: receivedAtMs,
160+
Status: entity.RequestStatusAccepted,
161+
Metadata: map[string]string{},
162+
}
140163
if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
141-
return nil, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", landRequest.ID, err)
164+
return nil, fmt.Errorf("LandController failed to insert accepted request log for sqid=%s: %w", landRequest.ID, err)
142165
}
143166

144167
c.logger.Debugw("land request created",

0 commit comments

Comments
 (0)