diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 1aa431836..fe009cab2 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -14,18 +14,18 @@ // Package record holds the record-stage queue controller. It consumes Record // messages (a request id) published by buildsignal once a build reaches a -// terminal status, and turns that outcome into durable validation state. See -// doc/rfc/stovepipe/steps/record.md. +// terminal status, and turns that outcome into durable validation state. // -// Phase 1 records only the queue's last-green bookmark, which process reads to -// choose an incremental build baseline. Validation facts and downstream hooks -// are not implemented yet. +// The durable state is a ValidationFact per validated commit, plus the queue's +// last-green bookmark, which process reads to choose an incremental build +// baseline. Downstream hooks are not implemented yet. package record import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" @@ -37,8 +37,9 @@ import ( "go.uber.org/zap" ) -// Controller consumes Record messages and advances the queue's last-green -// bookmark when the request's build succeeded. Implements consumer.Controller. +// Controller consumes Record messages, records the build's validation fact, and +// advances the queue's last-green bookmark when that fact is green. Implements +// consumer.Controller. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -53,6 +54,11 @@ var _ consumer.Controller = (*Controller)(nil) // _opName is the metric operation name shared by every emit in this file. const _opName = "record" +// wholeRepositoryProject is the project component of a fact covering the whole +// repository rather than one project within it. Per-project facts need target-graph +// attribution that this stage does not do, so every fact it writes is whole-repository. +const wholeRepositoryProject = "" + // NewController creates a new record controller. func NewController( logger *zap.SugaredLogger, @@ -108,17 +114,25 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } switch request.State { - case entity.RequestStateSucceeded: + case entity.RequestStateSucceeded, entity.RequestStateFailed: + fact, err := c.recordFact(ctx, store, request) + if err != nil { + return err + } + if !fact.IsGreen() { + metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1) + return nil + } if err := c.advanceLastGreen(ctx, store, request); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) return err } return nil - case entity.RequestStateFailed, entity.RequestStateCancelled: - // A verdict, but not a green one: nothing to record in phase 1. A - // cancelled build decided nothing about the commit at all. - metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1) + case entity.RequestStateCancelled: + // A cancelled build decided nothing about the commit, so it establishes + // no fact. The identity stays unclaimed; the next commit re-validates. + metrics.NamedCounter(c.metricsScope, _opName, "cancelled", 1) return nil case entity.RequestStateSuperseded: @@ -136,15 +150,77 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } } +// recordFact writes the request's outcome as an immutable whole-repository fact and +// returns the fact that is actually stored, which is not always the one just built: +// facts are first-writer-wins, so an identity already claimed by this same request — +// a redelivery after the write but before the bookmark advanced — yields the stored +// fact instead. Every decision downstream reads that stored fact rather than the +// request, so a redelivery cannot reach a different verdict than the original. +func (c *Controller) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, error) { + factStore := store.GetValidationFactStore() + + fact := entity.ValidationFact{ + URI: request.URI, + Project: wholeRepositoryProject, + Degree: degreeFor(request.State), + RequestID: request.ID, + CreatedAt: time.Now().UnixMilli(), + } + + err := factStore.Create(ctx, fact) + switch { + case err == nil: + metrics.NamedCounter(c.metricsScope, _opName, "fact_created", 1) + c.logger.Infow("recorded validation fact", + "queue", request.Queue, + "request_id", request.ID, + "uri", request.URI, + "degree", fact.Degree, + ) + return fact, nil + + case errors.Is(err, storage.ErrAlreadyExists): + stored, getErr := factStore.Get(ctx, request.URI, wholeRepositoryProject) + if getErr != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return entity.ValidationFact{}, fmt.Errorf("failed to load the existing fact for uri %s: %w", request.URI, getErr) + } + if stored.RequestID != request.ID { + // Two requests validating one URI would break the dedup ingest + // enforces, so this is a broken invariant rather than a race to + // resolve. Non-retryable: the stored fact is immutable. + metrics.NamedCounter(c.metricsScope, _opName, "invariant_errors", 1) + return entity.ValidationFact{}, fmt.Errorf( + "fact for uri %s is owned by request %s, not %s", request.URI, stored.RequestID, request.ID) + } + metrics.NamedCounter(c.metricsScope, _opName, "fact_exists", 1) + return stored, nil + + default: + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return entity.ValidationFact{}, fmt.Errorf("failed to create the fact for uri %s: %w", request.URI, err) + } +} + +// degreeFor maps a request's build outcome onto a whole-repository degree. Only the +// endpoints are produced: a whole-repository build is either clean or it is not, and +// intermediate degrees need per-project attribution that this stage does not do. +func degreeFor(state entity.RequestState) float64 { + if state == entity.RequestStateSucceeded { + return entity.DegreeGreen + } + return entity.DegreeBroken +} + // advanceLastGreen points the queue's bookmark at request, retrying on version // conflicts. The bookmark only moves forward: a candidate whose id is not newer // than the stored one is skipped without a write, which also makes a redelivery // of the same request a no-op. // -// Greenness comes from request.State rather than a persisted validation fact, -// which is what record.md specifies. The two agree — a fact's degree is derived -// from the same immutable state — and this reads the fact once the fact store -// lands. +// The bookmark is a cache of "newest green URI" derived from the facts, so it is +// advanced only after the green fact is durable. Losing the advance to a crash is +// recoverable — the redelivery reloads the same fact and retries — whereas a +// bookmark with no fact behind it would point at greenness nothing recorded. func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) error { queueStore := store.GetQueueStore() diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 34def05e9..777b15deb 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -43,6 +43,17 @@ const ( type recordMocks struct { reqStore *storagemock.MockRequestStore queueStore *storagemock.MockQueueStore + factStore *storagemock.MockValidationFactStore +} + +// expectFactCreated wires a successful fact write and captures it, so a case can +// assert on the recorded degree without pinning the wall-clock CreatedAt. +func (m recordMocks) expectFactCreated(captured *entity.ValidationFact) { + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, fact entity.ValidationFact) error { + *captured = fact + return nil + }) } // staticStorageFactory resolves every queue to one fixed store aggregate. @@ -57,11 +68,13 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMo m := recordMocks{ reqStore: storagemock.NewMockRequestStore(ctrl), queueStore: storagemock.NewMockQueueStore(ctrl), + factStore: storagemock.NewMockValidationFactStore(ctrl), } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes() c := NewController( zap.NewNop().Sugar(), @@ -134,6 +147,8 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(tt.stored, nil) var written entity.Queue @@ -147,10 +162,86 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) { require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) assert.Equal(t, tt.wantURI, written.LastGreenURI) assert.Equal(t, testID, written.LastGreenRequestID) + + // The green fact is what authorises the advance. + assert.Equal(t, entity.DegreeGreen, fact.Degree) + assert.Equal(t, testURI, fact.URI) + assert.Equal(t, testID, fact.RequestID) + assert.Equal(t, wholeRepositoryProject, fact.Project) + assert.Positive(t, fact.CreatedAt) }) } } +func TestProcess_RecordsBrokenFactWithoutAdvancing(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateFailed), nil) + + var fact entity.ValidationFact + m.expectFactCreated(&fact) + // No queue reads or writes: a broken fact never moves the bookmark. + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) + assert.Equal(t, entity.DegreeBroken, fact.Degree) + assert.False(t, fact.IsGreen()) +} + +func TestProcess_AdoptsExistingFactFromSameRequest(t *testing.T) { + tests := []struct { + name string + stored entity.ValidationFact + wantUpdate bool + }{ + { + name: "green fact still advances the bookmark", + stored: entity.ValidationFact{URI: testURI, Degree: entity.DegreeGreen, RequestID: testID}, + wantUpdate: true, + }, + { + name: "broken fact does not", + stored: entity.ValidationFact{URI: testURI, Degree: entity.DegreeBroken, RequestID: testID}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + // A redelivery after the fact was written but before the bookmark + // advanced: the write loses, and the stored fact decides. + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject).Return(tt.stored, nil) + + if tt.wantUpdate { + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) + m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + } + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) + }) + } +} + +func TestProcess_ExistingFactFromDifferentRequestFails(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject). + Return(entity.ValidationFact{URI: testURI, Degree: entity.DegreeGreen, RequestID: "request/monorepo/main/1"}, nil) + // The bookmark must not move on an identity this request does not own. + + require.Error(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) +} + func TestProcess_SkipsBookmarkWhenNotNewer(t *testing.T) { tests := []struct { name string @@ -173,6 +264,8 @@ func TestProcess_SkipsBookmarkWhenNotNewer(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(tt.stored, nil) // No Update: the bookmark only moves forward. @@ -181,12 +274,11 @@ func TestProcess_SkipsBookmarkWhenNotNewer(t *testing.T) { } } -func TestProcess_TerminalWithoutGreenDoesNotTouchQueue(t *testing.T) { +func TestProcess_TerminalWithoutFactDoesNotTouchStores(t *testing.T) { tests := []struct { name string state entity.RequestState }{ - {name: "failed", state: entity.RequestStateFailed}, {name: "cancelled", state: entity.RequestStateCancelled}, {name: "superseded", state: entity.RequestStateSuperseded}, } @@ -197,7 +289,7 @@ func TestProcess_TerminalWithoutGreenDoesNotTouchQueue(t *testing.T) { c, m := newController(t, ctrl) m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(tt.state), nil) - // No queue reads or writes at all. + // Neither a fact nor a queue write: these outcomes decide nothing. require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID)))) }) @@ -235,6 +327,8 @@ func TestProcess_RetriesBookmarkOnVersionMismatch(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) gomock.InOrder( m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(stale, nil), @@ -254,6 +348,8 @@ func TestProcess_MalformedRequestIDFails(t *testing.T) { request := requestWithState(entity.RequestStateSucceeded) m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) m.queueStore.EXPECT().Get(gomock.Any(), testQueue). Return(queueRow("git://remote/monorepo/main/old", "not-a-request-id", 1), nil) @@ -272,11 +368,31 @@ func TestProcess_StorageErrorsPropagate(t *testing.T) { Return(entity.Request{}, errors.New("boom")) }, }, + { + name: "fact create fails", + setup: func(m recordMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(errors.New("boom")) + }, + }, + { + name: "existing fact load fails", + setup: func(m recordMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID). + Return(requestWithState(entity.RequestStateSucceeded), nil) + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject). + Return(entity.ValidationFact{}, errors.New("boom")) + }, + }, { name: "queue load fails", setup: func(m recordMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) m.queueStore.EXPECT().Get(gomock.Any(), testQueue). Return(entity.Queue{}, errors.New("boom")) }, @@ -286,6 +402,8 @@ func TestProcess_StorageErrorsPropagate(t *testing.T) { setup: func(m recordMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) + var fact entity.ValidationFact + m.expectFactCreated(&fact) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil) m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)). Return(errors.New("boom")) diff --git a/stovepipe/entity/validation_fact.go b/stovepipe/entity/validation_fact.go index aff7d8e63..198a6d7e4 100644 --- a/stovepipe/entity/validation_fact.go +++ b/stovepipe/entity/validation_fact.go @@ -15,8 +15,7 @@ package entity // Degree bounds. A degree answers "how broken was this scope?" on a closed -// interval, so the two endpoints are named and the values between them are -// meaningful rather than arbitrary. +// interval, so both endpoints are named rather than spelled as bare literals. const ( // DegreeGreen is a fully healthy scope: nothing broken. DegreeGreen = 0.0 @@ -42,8 +41,10 @@ type ValidationFact struct { // repository; named projects narrow the fact to part of it. Project string `json:"project"` // Degree is how broken the scope was, on the closed interval - // [DegreeGreen, DegreeBroken]. Intermediate values describe partial - // breakage. + // [DegreeGreen, DegreeBroken]. A whole-repository scope only ever takes an + // endpoint: the build either passed or it did not. Intermediate values + // belong to project-level scopes, whose analysis defines how a partial + // degree is derived. Degree float64 `json:"degree"` // RequestID is the request that established the fact. RequestID string `json:"request_id"`