Skip to content

Commit 6ec424a

Browse files
albertywualbertywu
committed
feat(request): materialize request log status
Add projection winner selection, optimistic-concurrency convergence, and queue-summary repair without activating production call sites yet. Validation: make fmt && make build && make test && make e2e-test
1 parent 5062492 commit 6ec424a

3 files changed

Lines changed: 367 additions & 0 deletions

File tree

‎submitqueue/core/request/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ go_library(
55
srcs = [
66
"admission.go",
77
"log.go",
8+
"materializer.go",
89
"request.go",
910
],
1011
importpath = "github.com/uber/submitqueue/submitqueue/core/request",
@@ -23,6 +24,7 @@ go_test(
2324
srcs = [
2425
"admission_test.go",
2526
"log_test.go",
27+
"materializer_test.go",
2628
"request_test.go",
2729
],
2830
embed = [":go_default_library"],

‎submitqueue/core/request/materializer.go‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
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+
22+
"github.com/uber/submitqueue/submitqueue/entity"
23+
"github.com/uber/submitqueue/submitqueue/extension/storage"
24+
)
25+
26+
// Materializer appends request logs and projects the winning current status.
27+
// Storage implementations remain mechanical; this type owns winner selection, optimistic concurrency, and projection repair.
28+
type Materializer struct {
29+
store storage.Storage
30+
}
31+
32+
// NewMaterializer creates a request read-model materializer.
33+
func NewMaterializer(store storage.Storage) *Materializer {
34+
return &Materializer{store: store}
35+
}
36+
37+
// PersistLog appends one audit log and materializes its winning state.
38+
// Projection errors are returned so queue deliveries are retried rather than silently dropping the side write.
39+
func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) error {
40+
if err := m.store.GetRequestLogStore().Insert(ctx, log); err != nil {
41+
return fmt.Errorf("failed to insert request log request_id=%s: %w", log.RequestID, err)
42+
}
43+
44+
for {
45+
summary, err := m.store.GetRequestSummaryStore().Get(ctx, log.RequestID)
46+
if err != nil {
47+
return fmt.Errorf("failed to get request summary request_id=%s: %w", log.RequestID, err)
48+
}
49+
50+
if logWins(log, summary) {
51+
oldVersion := summary.Version
52+
newVersion := oldVersion + 1
53+
updated := summary
54+
updated.Status = log.Status
55+
updated.RequestVersion = log.RequestVersion
56+
updated.StatusTimestampMs = log.TimestampMs
57+
updated.LastError = log.LastError
58+
updated.Metadata = cloneMetadata(log.Metadata)
59+
60+
if err := m.store.GetRequestSummaryStore().Update(ctx, updated, oldVersion, newVersion); err != nil {
61+
if errors.Is(err, storage.ErrVersionMismatch) {
62+
continue
63+
}
64+
return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err)
65+
}
66+
updated.Version = newVersion
67+
summary = updated
68+
}
69+
70+
if err := m.repairQueueSummary(ctx, summary); err != nil {
71+
return err
72+
}
73+
return nil
74+
}
75+
76+
}
77+
78+
func (m *Materializer) repairQueueSummary(ctx context.Context, authoritative entity.RequestSummary) error {
79+
desired := queueSummaryFromSummary(authoritative)
80+
for {
81+
current, err := m.store.GetRequestQueueSummaryStore().Get(ctx, desired.Queue, desired.ReceivedAtMs, desired.RequestID)
82+
if errors.Is(err, storage.ErrNotFound) {
83+
if err := m.store.GetRequestQueueSummaryStore().Create(ctx, desired); err != nil {
84+
if errors.Is(err, storage.ErrAlreadyExists) {
85+
continue
86+
}
87+
return fmt.Errorf("failed to recreate queue summary request_id=%s: %w", desired.RequestID, err)
88+
}
89+
return nil
90+
}
91+
if err != nil {
92+
return fmt.Errorf("failed to get queue summary request_id=%s: %w", desired.RequestID, err)
93+
}
94+
if current.Version == desired.Version {
95+
return nil
96+
}
97+
if current.Version > desired.Version {
98+
return fmt.Errorf("queue summary ahead of authoritative summary request_id=%s queue_version=%d summary_version=%d", desired.RequestID, current.Version, desired.Version)
99+
}
100+
if err := m.store.GetRequestQueueSummaryStore().Update(ctx, desired, current.Version, desired.Version); err != nil {
101+
if errors.Is(err, storage.ErrVersionMismatch) {
102+
continue
103+
}
104+
return fmt.Errorf("failed to update queue summary request_id=%s: %w", desired.RequestID, err)
105+
}
106+
return nil
107+
}
108+
}
109+
110+
func logWins(log entity.RequestLog, summary entity.RequestSummary) bool {
111+
incomingTerminal := isVersionedTerminal(log)
112+
currentTerminal := isVersionedTerminalSummary(summary)
113+
if incomingTerminal != currentTerminal {
114+
return incomingTerminal
115+
}
116+
if incomingTerminal {
117+
if log.RequestVersion != summary.RequestVersion {
118+
return log.RequestVersion > summary.RequestVersion
119+
}
120+
}
121+
return log.TimestampMs > summary.StatusTimestampMs
122+
}
123+
124+
func isVersionedTerminalSummary(summary entity.RequestSummary) bool {
125+
return summary.RequestVersion > 0 && entity.IsRequestStateTerminal(entity.RequestState(summary.Status))
126+
}
127+
128+
func isVersionedTerminal(log entity.RequestLog) bool {
129+
return log.RequestVersion > 0 && entity.IsRequestStateTerminal(entity.RequestState(log.Status))
130+
}

‎submitqueue/core/request/materializer_test.go‎

Lines changed: 235 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,235 @@
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+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/stretchr/testify/require"
24+
"github.com/uber/submitqueue/submitqueue/entity"
25+
"github.com/uber/submitqueue/submitqueue/extension/storage"
26+
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
27+
"go.uber.org/mock/gomock"
28+
)
29+
30+
func TestMaterializer_PersistLog(t *testing.T) {
31+
base := testRequestSummary()
32+
log := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusLanded, RequestVersion: 2, Metadata: map[string]string{}}
33+
t.Run("winning log updates both projections", func(t *testing.T) {
34+
ctrl := gomock.NewController(t)
35+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
36+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
37+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(base, nil)
38+
summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, updated entity.RequestSummary, _, _ int32) error {
39+
assert.Equal(t, entity.RequestStatusLanded, updated.Status)
40+
assert.Equal(t, int32(2), updated.RequestVersion)
41+
return nil
42+
})
43+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil)
44+
queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
45+
require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log))
46+
})
47+
48+
t.Run("unversioned terminal status does not receive terminal precedence", func(t *testing.T) {
49+
ctrl := gomock.NewController(t)
50+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
51+
current := base
52+
current.Status = entity.RequestStatusLanded
53+
current.RequestVersion = 0
54+
incoming := entity.RequestLog{RequestID: "q/1", TimestampMs: 20, Status: entity.RequestStatusProcessing, RequestVersion: 0, Metadata: map[string]string{}}
55+
logStore.EXPECT().Insert(gomock.Any(), incoming).Return(nil)
56+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(current, nil)
57+
summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn(func(_ context.Context, updated entity.RequestSummary, _, _ int32) error {
58+
assert.Equal(t, entity.RequestStatusProcessing, updated.Status)
59+
return nil
60+
})
61+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(current), nil)
62+
queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
63+
require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), incoming))
64+
})
65+
66+
t.Run("CAS conflict reloads and repairs winner", func(t *testing.T) {
67+
ctrl := gomock.NewController(t)
68+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
69+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
70+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(base, nil)
71+
summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(storage.ErrVersionMismatch)
72+
advanced := base
73+
advanced.Status = entity.RequestStatusLanded
74+
advanced.RequestVersion = 2
75+
advanced.StatusTimestampMs = 20
76+
advanced.Version = 2
77+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil)
78+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil)
79+
queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
80+
require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log))
81+
})
82+
83+
t.Run("non-winning redelivery repairs stale queue projection", func(t *testing.T) {
84+
ctrl := gomock.NewController(t)
85+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
86+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
87+
advanced := base
88+
advanced.Status = entity.RequestStatusLanded
89+
advanced.RequestVersion = 2
90+
advanced.StatusTimestampMs = 20
91+
advanced.Version = 2
92+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil)
93+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(base), nil)
94+
queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
95+
require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log))
96+
})
97+
98+
t.Run("missing queue projection is recreated", func(t *testing.T) {
99+
ctrl := gomock.NewController(t)
100+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
101+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
102+
advanced := base
103+
advanced.Status = entity.RequestStatusLanded
104+
advanced.RequestVersion = 2
105+
advanced.StatusTimestampMs = 20
106+
advanced.Version = 2
107+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil)
108+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(entity.RequestQueueSummary{}, storage.ErrNotFound)
109+
queueStore.EXPECT().Create(gomock.Any(), queueSummaryFromSummary(advanced)).Return(nil)
110+
require.NoError(t, NewMaterializer(store).PersistLog(context.Background(), log))
111+
})
112+
113+
t.Run("retry after projection failure appends another audit row", func(t *testing.T) {
114+
ctrl := gomock.NewController(t)
115+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
116+
materializer := NewMaterializer(store)
117+
advanced := base
118+
advanced.Status = entity.RequestStatusLanded
119+
advanced.RequestVersion = 2
120+
advanced.StatusTimestampMs = 20
121+
advanced.Version = 2
122+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil).Times(2)
123+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil).Times(2)
124+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(entity.RequestQueueSummary{}, errors.New("queue store down"))
125+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueSummaryFromSummary(advanced), nil)
126+
127+
require.Error(t, materializer.PersistLog(context.Background(), log))
128+
require.NoError(t, materializer.PersistLog(context.Background(), log))
129+
})
130+
131+
t.Run("missing authoritative summary fails", func(t *testing.T) {
132+
ctrl := gomock.NewController(t)
133+
store, summaryStore, _, logStore := materializerStores(ctrl)
134+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
135+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.RequestSummary{}, storage.ErrNotFound)
136+
require.Error(t, NewMaterializer(store).PersistLog(context.Background(), log))
137+
})
138+
139+
t.Run("queue projection ahead fails fast", func(t *testing.T) {
140+
ctrl := gomock.NewController(t)
141+
store, summaryStore, queueStore, logStore := materializerStores(ctrl)
142+
logStore.EXPECT().Insert(gomock.Any(), log).Return(nil)
143+
advanced := base
144+
advanced.Status = entity.RequestStatusLanded
145+
advanced.RequestVersion = 2
146+
advanced.StatusTimestampMs = 20
147+
advanced.Version = 2
148+
summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(advanced, nil)
149+
queueAhead := queueSummaryFromSummary(advanced)
150+
queueAhead.Version = 3
151+
queueStore.EXPECT().Get(gomock.Any(), "q", int64(10), "q/1").Return(queueAhead, nil)
152+
require.Error(t, NewMaterializer(store).PersistLog(context.Background(), log))
153+
})
154+
}
155+
156+
func TestLogWins(t *testing.T) {
157+
base := testRequestSummary()
158+
tests := []struct {
159+
name string
160+
current entity.RequestSummary
161+
incoming entity.RequestLog
162+
want bool
163+
}{
164+
{
165+
name: "versioned terminal beats newer unversioned status",
166+
current: entity.RequestSummary{Status: entity.RequestStatusProcessing, StatusTimestampMs: 200},
167+
incoming: entity.RequestLog{
168+
Status: entity.RequestStatusLanded, RequestVersion: 1, TimestampMs: 100,
169+
},
170+
want: true,
171+
},
172+
{
173+
name: "nonterminal cannot replace versioned terminal",
174+
current: entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 1, StatusTimestampMs: 100},
175+
incoming: entity.RequestLog{
176+
Status: entity.RequestStatusProcessing, TimestampMs: 200,
177+
},
178+
},
179+
{
180+
name: "higher terminal request version wins",
181+
current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 1, StatusTimestampMs: 200},
182+
incoming: entity.RequestLog{
183+
Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 100,
184+
},
185+
want: true,
186+
},
187+
{
188+
name: "lower terminal request version loses",
189+
current: entity.RequestSummary{Status: entity.RequestStatusLanded, RequestVersion: 2, StatusTimestampMs: 100},
190+
incoming: entity.RequestLog{
191+
Status: entity.RequestStatusError, RequestVersion: 1, TimestampMs: 200,
192+
},
193+
},
194+
{
195+
name: "equal terminal version uses later timestamp",
196+
current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 100},
197+
incoming: entity.RequestLog{
198+
Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 200,
199+
},
200+
want: true,
201+
},
202+
{
203+
name: "without terminal winner later timestamp wins",
204+
current: base,
205+
incoming: entity.RequestLog{
206+
Status: entity.RequestStatusStarted, TimestampMs: base.StatusTimestampMs + 1,
207+
},
208+
want: true,
209+
},
210+
{
211+
name: "exact version and timestamp tie keeps current winner",
212+
current: entity.RequestSummary{Status: entity.RequestStatusError, RequestVersion: 2, StatusTimestampMs: 200},
213+
incoming: entity.RequestLog{
214+
Status: entity.RequestStatusLanded, RequestVersion: 2, TimestampMs: 200,
215+
},
216+
},
217+
}
218+
219+
for _, tt := range tests {
220+
t.Run(tt.name, func(t *testing.T) {
221+
assert.Equal(t, tt.want, logWins(tt.incoming, tt.current))
222+
})
223+
}
224+
}
225+
226+
func materializerStores(ctrl *gomock.Controller) (*storagemock.MockStorage, *storagemock.MockRequestSummaryStore, *storagemock.MockRequestQueueSummaryStore, *storagemock.MockRequestLogStore) {
227+
store := storagemock.NewMockStorage(ctrl)
228+
summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
229+
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
230+
logStore := storagemock.NewMockRequestLogStore(ctrl)
231+
store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
232+
store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
233+
store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
234+
return store, summaryStore, queueStore, logStore
235+
}

0 commit comments

Comments
 (0)