Skip to content

Commit 7389a33

Browse files
mnoah1mnoah1
committed
feat(stovepipe): record superseded request state
1 parent b4def02 commit 7389a33

2 files changed

Lines changed: 77 additions & 19 deletions

File tree

‎stovepipe/controller/process/process.go‎

Lines changed: 23 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
120120

121121
switch request.State {
122122
case entity.RequestStateProcessing:
123-
if err := c.persistProcessingLog(ctx, store, request); err != nil {
123+
if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil {
124124
return err
125125
}
126126
// Announce here as well as at admit: this is the only path a redelivery
@@ -136,7 +136,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
136136
return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
137137
}
138138
return nil
139-
case entity.RequestStateSuperseded, entity.RequestStateSucceeded, entity.RequestStateFailed, entity.RequestStateCancelled:
139+
case entity.RequestStateSuperseded:
140+
return c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonSupersededByNewerHead)
141+
case entity.RequestStateSucceeded, entity.RequestStateFailed, entity.RequestStateCancelled:
140142
// Terminal: a newer head preempted this request, or its build already finished.
141143
// A stale redelivery has nothing left to do.
142144
return nil
@@ -199,10 +201,17 @@ func (c *Controller) coalesce(ctx context.Context, store storage.Storage, reques
199201
if cmp >= 0 {
200202
return false, nil
201203
}
202-
if err := c.supersedeRequest(ctx, store, request); err != nil {
204+
request, err = c.supersedeRequest(ctx, store, request)
205+
if err != nil {
203206
metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...)
204207
return false, err
205208
}
209+
if request.State != entity.RequestStateSuperseded {
210+
return true, nil
211+
}
212+
if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonSupersededByNewerHead); err != nil {
213+
return false, err
214+
}
206215
metrics.NamedCounter(c.metricsScope, _opName, "superseded", 1, metrics.TagsFromContext(ctx)...)
207216
c.logger.Infow("superseded request for newer head",
208217
"request_id", request.ID,
@@ -270,7 +279,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage,
270279
return nil
271280
}
272281

273-
if err := c.persistProcessingLog(ctx, store, request); err != nil {
282+
if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil {
274283
return err
275284
}
276285

@@ -296,10 +305,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage,
296305
return nil
297306
}
298307

299-
func (c *Controller) persistProcessingLog(ctx context.Context, store storage.Storage, request entity.Request) error {
300-
log := requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)
308+
func (c *Controller) persistStateLog(ctx context.Context, store storage.Storage, request entity.Request, reason entity.RequestOutcomeReason) error {
309+
log := requestlog.NewRequestStateLog(request, reason)
301310
if err := c.materializer.PersistLog(ctx, store, log); err != nil {
302-
return fmt.Errorf("failed to record processing state for request %s: %w", request.ID, err)
311+
return fmt.Errorf("failed to record %s state for request %s: %w", request.State, request.ID, err)
303312
}
304313
return nil
305314
}
@@ -432,13 +441,13 @@ func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage
432441
}
433442
}
434443

435-
// supersedeRequest transitions a request from accepted to superseded, retrying on version conflicts.
436-
func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage, request entity.Request) error {
444+
// supersedeRequest returns the canonical row so the log uses the version that won any CAS race.
445+
func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage, request entity.Request) (entity.Request, error) {
437446
reqStore := store.GetRequestStore()
438447

439448
for {
440449
if request.State != entity.RequestStateAccepted {
441-
return nil
450+
return request, nil
442451
}
443452

444453
updated := request
@@ -448,14 +457,15 @@ func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage
448457
if errors.Is(err, storage.ErrVersionMismatch) {
449458
got, getErr := reqStore.Get(ctx, request.ID)
450459
if getErr != nil {
451-
return fmt.Errorf("failed to reload request %s after version mismatch: %w", request.ID, getErr)
460+
return entity.Request{}, fmt.Errorf("failed to reload request %s after version mismatch: %w", request.ID, getErr)
452461
}
453462
request = got
454463
continue
455464
}
456-
return fmt.Errorf("failed to supersede request %s: %w", request.ID, err)
465+
return entity.Request{}, fmt.Errorf("failed to supersede request %s: %w", request.ID, err)
457466
}
458-
return nil
467+
updated.Version = newVersion
468+
return updated, nil
459469
}
460470
}
461471

‎stovepipe/controller/process/process_test.go‎

Lines changed: 54 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -191,17 +191,22 @@ func expectProcessingLogAndHandoff(t *testing.T, m processMocks, id string, vers
191191

192192
func expectProcessingLog(t *testing.T, m processMocks, id string, version int32) *gomock.Call {
193193
t.Helper()
194+
return expectStateLog(t, m, id, entity.RequestStateProcessing, version, entity.RequestOutcomeReasonUnknown)
195+
}
196+
197+
func expectStateLog(t *testing.T, m processMocks, id string, state entity.RequestState, version int32, reason entity.RequestOutcomeReason) *gomock.Call {
198+
t.Helper()
194199

195200
request := entity.Request{
196201
ID: id,
197202
Queue: testQueue,
198-
State: entity.RequestStateProcessing,
203+
State: state,
199204
Version: version,
200205
}
201206
return m.materializer.EXPECT().PersistLog(
202207
gomock.Any(),
203208
m.store,
204-
requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown),
209+
requestlog.NewRequestStateLog(request, reason),
205210
).Return(nil)
206211
}
207212

@@ -487,11 +492,27 @@ func TestProcess(t *testing.T) {
487492
wantRetry bool
488493
}{
489494
{
490-
name: "superseded is no-op",
495+
name: "superseded redelivery repairs its state log",
491496
setup: func(m processMocks) {
492497
m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{
493498
ID: testID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2,
494499
}, nil)
500+
expectStateLog(t, m, testID, entity.RequestStateSuperseded, 2, entity.RequestOutcomeReasonSupersededByNewerHead)
501+
},
502+
},
503+
{
504+
name: "superseded redelivery fails when log persistence fails",
505+
wantErr: true,
506+
setup: func(m processMocks) {
507+
request := entity.Request{
508+
ID: testID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2,
509+
}
510+
m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil)
511+
m.materializer.EXPECT().PersistLog(
512+
gomock.Any(),
513+
m.store,
514+
requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonSupersededByNewerHead),
515+
).Return(errors.New("db down"))
495516
},
496517
},
497518
{
@@ -757,7 +778,8 @@ func TestProcess(t *testing.T) {
757778
}, nil)
758779
superseded := acceptedRequest(testID)
759780
superseded.State = entity.RequestStateSuperseded
760-
m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil)
781+
updateCall := m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil)
782+
expectStateLog(t, m, testID, entity.RequestStateSuperseded, 2, entity.RequestOutcomeReasonSupersededByNewerHead).After(updateCall)
761783
},
762784
},
763785
{
@@ -857,7 +879,32 @@ func TestProcess(t *testing.T) {
857879
}, nil)
858880
updated := acceptedRequest(testOlderID)
859881
updated.State = entity.RequestStateSuperseded
860-
m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil)
882+
updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil)
883+
expectStateLog(t, m, testOlderID, entity.RequestStateSuperseded, 2, entity.RequestOutcomeReasonSupersededByNewerHead).After(updateCall)
884+
},
885+
},
886+
{
887+
name: "superseded log failure retries the coalesce step",
888+
id: testOlderID,
889+
wantErr: true,
890+
setup: func(m processMocks) {
891+
m.reqStore.EXPECT().Get(gomock.Any(), testOlderID).Return(acceptedRequest(testOlderID), nil)
892+
m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
893+
Name: testQueue,
894+
LatestRequestID: testID,
895+
Version: 1,
896+
}, nil)
897+
updated := acceptedRequest(testOlderID)
898+
updated.State = entity.RequestStateSuperseded
899+
updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil)
900+
request := entity.Request{
901+
ID: testOlderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2,
902+
}
903+
m.materializer.EXPECT().PersistLog(
904+
gomock.Any(),
905+
m.store,
906+
requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonSupersededByNewerHead),
907+
).Return(errors.New("db down")).After(updateCall)
861908
},
862909
},
863910
{
@@ -873,9 +920,10 @@ func TestProcess(t *testing.T) {
873920
updated := acceptedRequest(testOlderID)
874921
updated.State = entity.RequestStateSuperseded
875922
m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(storage.ErrVersionMismatch)
876-
m.reqStore.EXPECT().Get(gomock.Any(), testOlderID).Return(entity.Request{
923+
reloadCall := m.reqStore.EXPECT().Get(gomock.Any(), testOlderID).Return(entity.Request{
877924
ID: testOlderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2,
878925
}, nil)
926+
expectStateLog(t, m, testOlderID, entity.RequestStateSuperseded, 2, entity.RequestOutcomeReasonSupersededByNewerHead).After(reloadCall)
879927
},
880928
},
881929
{

0 commit comments

Comments
 (0)