Skip to content

Commit aa93da1

Browse files
mnoah1mnoah1
committed
feat(stovepipe): record accepted request state
Summary: Intent: - Establish every newly admitted request's initial history before dependent processing begins. - Make an Ingest retry repair a missing accepted entry before republishing work. Changes: - Record the durable accepted request through the queue-scoped request log store. - Stop queue-pointer advancement and process publication when recording fails. - Wire one recorder instance into Ingest and cover new, duplicate, repair, and failure paths. This PR builds on #657, which introduces the request state log recorder. --- <sub>Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace</sub>
1 parent 3e2af24 commit aa93da1

5 files changed

Lines changed: 73 additions & 11 deletions

File tree

‎service/stovepipe/server/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ go_library(
2929
"//stovepipe/controller/process:go_default_library",
3030
"//stovepipe/controller/record:go_default_library",
3131
"//stovepipe/core/messagequeue:go_default_library",
32+
"//stovepipe/core/requestlog:go_default_library",
3233
"//stovepipe/extension/buildrunner:go_default_library",
3334
"//stovepipe/extension/buildrunner/fake:go_default_library",
3435
"//stovepipe/extension/queueconfig/default:go_default_library",

‎service/stovepipe/server/main.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ import (
5151
"github.com/uber/submitqueue/stovepipe/controller/process"
5252
"github.com/uber/submitqueue/stovepipe/controller/record"
5353
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
54+
"github.com/uber/submitqueue/stovepipe/core/requestlog"
5455
"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
5556
buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
5657
queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default"
@@ -302,6 +303,7 @@ func run() error {
302303
brf := fakeBuildRunnerFactory{}
303304

304305
storageFty := storageFactory{backend: store}
306+
requestLog := requestlog.NewRecorder(scope)
305307
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf, hookResolver{})
306308
if err != nil {
307309
return err
@@ -336,6 +338,7 @@ func run() error {
336338
newInMemoryCounterFactory(),
337339
sourceControl,
338340
storageFty,
341+
requestLog,
339342
registry,
340343
)
341344
srv := &StovepipeServer{

‎stovepipe/controller/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ go_library(
1717
"//platform/metrics:go_default_library",
1818
"//platform/publish:go_default_library",
1919
"//stovepipe/core/messagequeue:go_default_library",
20+
"//stovepipe/core/requestlog:go_default_library",
2021
"//stovepipe/entity:go_default_library",
2122
"//stovepipe/extension/sourcecontrol:go_default_library",
2223
"//stovepipe/extension/storage:go_default_library",
@@ -40,6 +41,7 @@ go_test(
4041
"//platform/extension/counter/mock:go_default_library",
4142
"//platform/extension/messagequeue/mock:go_default_library",
4243
"//stovepipe/core/messagequeue:go_default_library",
44+
"//stovepipe/core/requestlog/mock:go_default_library",
4345
"//stovepipe/entity:go_default_library",
4446
"//stovepipe/extension/sourcecontrol:go_default_library",
4547
"//stovepipe/extension/sourcecontrol/mock:go_default_library",

‎stovepipe/controller/ingest.go‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"github.com/uber/submitqueue/platform/metrics"
2828
"github.com/uber/submitqueue/platform/publish"
2929
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
30+
"github.com/uber/submitqueue/stovepipe/core/requestlog"
3031
"github.com/uber/submitqueue/stovepipe/entity"
3132
"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
3233
"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -51,15 +52,16 @@ func IsInvalidRequest(err error) bool {
5152
// observed commit into the validation pipeline.
5253
//
5354
// It resolves the queue's head commit URI via the SourceControl extension, dedups on the
54-
// (queue, URI) pair, persists the Request and its URI mapping via storage, and publishes the
55-
// request ID onto the process stage. Ingestion is idempotent: a re-reported head resolves to the
56-
// already-minted request and no new work is published.
55+
// (queue, URI) pair, persists the Request and its initial log entry via storage, and publishes
56+
// the request ID onto the process stage. Ingestion is idempotent: a re-reported head resolves to
57+
// the already-minted request and republishes it only while it remains accepted.
5758
type IngestController struct {
5859
logger *zap.SugaredLogger
5960
metricsScope tally.Scope
6061
counters counter.Factory
6162
sourceControl sourcecontrol.Factory
6263
stores storage.Factory
64+
requestLog requestlog.Recorder
6365
registry consumer.TopicRegistry
6466
}
6567

@@ -71,6 +73,7 @@ func NewIngestController(
7173
counters counter.Factory,
7274
sourceControl sourcecontrol.Factory,
7375
stores storage.Factory,
76+
requestLog requestlog.Recorder,
7477
registry consumer.TopicRegistry,
7578
) *IngestController {
7679
return &IngestController{
@@ -79,6 +82,7 @@ func NewIngestController(
7982
counters: counters,
8083
sourceControl: sourceControl,
8184
stores: stores,
85+
requestLog: requestLog,
8286
registry: registry,
8387
}
8488
}
@@ -87,11 +91,11 @@ func NewIngestController(
8791
// request ID validating it.
8892
//
8993
// It is idempotent and runs to completion on every call, each step tolerant of having already
90-
// run: it resolves (or claims) the (queue, URI) mapping, ensures the Request row exists, and
91-
// publishes the request to the process stage. A retry after a partial failure — for example the
92-
// URI mapping committed but the request write failed — completes the missing steps instead of
93-
// returning a dangling reference. The (queue, URI) mapping is the dedup gate, so concurrent
94-
// ingests of the same head converge on one request.
94+
// run: it resolves (or claims) the (queue, URI) mapping, ensures the Request row and accepted log
95+
// entry exist, and publishes the request to the process stage. A retry after a partial failure —
96+
// for example the URI mapping committed but the request write failed — completes the missing
97+
// steps instead of returning a dangling reference. The (queue, URI) mapping is the dedup gate,
98+
// so concurrent ingests of the same head converge on one request.
9599
func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest) (result entity.IngestResult, retErr error) {
96100
const opName = "ingest"
97101

@@ -136,6 +140,12 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
136140
return entity.IngestResult{}, err
137141
}
138142

143+
if request.State == entity.RequestStateAccepted {
144+
if err := c.requestLog.RecordRequestState(ctx, store.GetRequestLogStore(), request, entity.RequestOutcomeReasonUnknown); err != nil {
145+
return entity.IngestResult{}, fmt.Errorf("failed to record accepted state for request %s: %w", id, err)
146+
}
147+
}
148+
139149
if err := c.advanceQueueLatestRequestID(ctx, store, queue, id); err != nil {
140150
return entity.IngestResult{}, err
141151
}

‎stovepipe/controller/ingest_test.go‎

Lines changed: 49 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
countermock "github.com/uber/submitqueue/platform/extension/counter/mock"
2929
mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
3030
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
31+
requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock"
3132
"github.com/uber/submitqueue/stovepipe/entity"
3233
"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
3334
scmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock"
@@ -49,7 +50,9 @@ type ingestMocks struct {
4950
sc *scmock.MockSourceControl
5051
reqStore *storagemock.MockRequestStore
5152
uriStore *storagemock.MockRequestURIStore
53+
logStore *storagemock.MockRequestLogStore
5254
queueStore *storagemock.MockQueueStore
55+
requestLog *requestlogmock.MockRecorder
5356
publisher *mqmock.MockPublisher
5457
}
5558

@@ -74,13 +77,16 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
7477
sc: scmock.NewMockSourceControl(ctrl),
7578
reqStore: storagemock.NewMockRequestStore(ctrl),
7679
uriStore: storagemock.NewMockRequestURIStore(ctrl),
80+
logStore: storagemock.NewMockRequestLogStore(ctrl),
7781
queueStore: storagemock.NewMockQueueStore(ctrl),
82+
requestLog: requestlogmock.NewMockRecorder(ctrl),
7883
publisher: mqmock.NewMockPublisher(ctrl),
7984
}
8085

8186
store := storagemock.NewMockStorage(ctrl)
8287
store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
8388
store.EXPECT().GetRequestURIStore().Return(m.uriStore).AnyTimes()
89+
store.EXPECT().GetRequestLogStore().Return(m.logStore).AnyTimes()
8490
store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
8591

8692
queue := mqmock.NewMockQueue(ctrl)
@@ -91,10 +97,29 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
9197
})
9298
require.NoError(t, err)
9399

94-
c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, registry)
100+
c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, m.requestLog, registry)
95101
return c, m
96102
}
97103

104+
func acceptedRequest(id string) entity.Request {
105+
return entity.Request{
106+
ID: id,
107+
Queue: testQueue,
108+
URI: testURI,
109+
State: entity.RequestStateAccepted,
110+
Version: 1,
111+
}
112+
}
113+
114+
func expectRecordAccepted(m ingestMocks, id string) {
115+
m.requestLog.EXPECT().RecordRequestState(
116+
gomock.Any(),
117+
m.logStore,
118+
acceptedRequest(id),
119+
entity.RequestOutcomeReasonUnknown,
120+
).Return(nil)
121+
}
122+
98123
// expectResolve wires the SourceControl factory + Latest happy path returning testURI.
99124
func expectResolve(m ingestMocks) {
100125
m.factory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sc, nil)
@@ -153,6 +178,7 @@ func TestIngestController_Ingest(t *testing.T) {
153178
m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(nil)
154179
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/7").Return(entity.Request{}, storage.ErrNotFound)
155180
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
181+
expectRecordAccepted(m, "request/monorepo/main/7")
156182
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
157183
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
158184
},
@@ -164,7 +190,8 @@ func TestIngestController_Ingest(t *testing.T) {
164190
setup: func(m ingestMocks) {
165191
expectResolve(m)
166192
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
167-
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
193+
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
194+
expectRecordAccepted(m, "request/monorepo/main/3")
168195
expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
169196
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
170197
},
@@ -178,6 +205,7 @@ func TestIngestController_Ingest(t *testing.T) {
178205
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
179206
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound)
180207
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
208+
expectRecordAccepted(m, "request/monorepo/main/3")
181209
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/3")
182210
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
183211
},
@@ -192,7 +220,8 @@ func TestIngestController_Ingest(t *testing.T) {
192220
m.counter.EXPECT().Next(gomock.Any(), counterDomainRequest).Return(int64(7), nil)
193221
m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists)
194222
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
195-
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
223+
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
224+
expectRecordAccepted(m, "request/monorepo/main/3")
196225
expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
197226
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
198227
},
@@ -257,11 +286,28 @@ func TestIngestController_Ingest(t *testing.T) {
257286
m.uriStore.EXPECT().Create(gomock.Any(), testURI, gomock.Any()).Return(nil)
258287
m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound)
259288
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
289+
expectRecordAccepted(m, "request/monorepo/main/7")
260290
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
261291
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(errors.New("queue down"))
262292
},
263293
wantErr: true,
264294
},
295+
{
296+
name: "request log error prevents advancement and publish",
297+
queue: testQueue,
298+
setup: func(m ingestMocks) {
299+
expectResolve(m)
300+
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
301+
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
302+
m.requestLog.EXPECT().RecordRequestState(
303+
gomock.Any(),
304+
m.logStore,
305+
acceptedRequest("request/monorepo/main/3"),
306+
entity.RequestOutcomeReasonUnknown,
307+
).Return(errors.New("log unavailable"))
308+
},
309+
wantErr: true,
310+
},
265311
}
266312

267313
for _, tt := range tests {

0 commit comments

Comments
 (0)