feat(stovepipe): record the validation fact and decide greenness from it

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()
 

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"))