feat(stovepipe): report build failure detection latency

platform/metrics/README.md

@@ -77,14 +77,15 @@ defer func() { op.Complete(retErr) }()
 metrics.NamedCounter(c.scope, "publish", "attempts", 1, metrics.NewTag("topic", c.topic))
 ```
 
-## Latency Buckets
+## Duration Buckets
 
-There is no default bucket set. The package exports three common sets:
+There is no default bucket set. The package exports four common sets:
 
 | Set | Range | Use for |
 |-----|-------|---------|
 | `FastLatencyBuckets` | ~100µs – 5s | Fast in-process work such as scoring, cache lookups, and CPU-bound operations |
 | `StorageLatencyBuckets` | ~1ms – 1m | Storage and message-queue round trips such as database reads, writes, publishing, and consuming |
 | `LongLatencyBuckets` | ~5ms – 4h | Long-running pipeline work and external calls such as builds, merges, pushes, and provider calls |
+| `ChangeAgeBuckets` | ~1m – 30d | Elapsed time measured from a source-control change's commit timestamp rather than from work this system started |
 
 Pass one of these sets or a custom `tally.DurationBuckets` to `Begin` or `NamedHistogram`.

platform/metrics/metrics.go

@@ -106,6 +106,28 @@ var (
 		2 * time.Hour,
 		4 * time.Hour,
 	}
+
+	// ChangeAgeBuckets suits durations measured from a source-control change's
+	// commit timestamp. They span minutes to a month because that is the honest
+	// range of such a signal: a break caught in minutes and one that survived a
+	// fortnight are both ordinary observations, and collapsing the tail would hide
+	// exactly the cases worth seeing.
+	ChangeAgeBuckets = tally.DurationBuckets{
+		1 * time.Minute,
+		5 * time.Minute,
+		15 * time.Minute,
+		30 * time.Minute,
+		1 * time.Hour,
+		2 * time.Hour,
+		4 * time.Hour,
+		8 * time.Hour,
+		12 * time.Hour,
+		24 * time.Hour,
+		48 * time.Hour,
+		7 * 24 * time.Hour,
+		14 * 24 * time.Hour,
+		30 * 24 * time.Hour,
+	}
 )
 
 // Op tracks the lifecycle of a named operation. It captures the start time on

platform/metrics/metrics_test.go

@@ -163,6 +163,7 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 		"FastLatencyBuckets":    FastLatencyBuckets,
 		"StorageLatencyBuckets": StorageLatencyBuckets,
 		"LongLatencyBuckets":    LongLatencyBuckets,
+		"ChangeAgeBuckets":      ChangeAgeBuckets,
 	}
 	for name, buckets := range sets {
 		t.Run(name, func(t *testing.T) {

stovepipe/controller/record/record.go

@@ -119,12 +119,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 	switch request.State {
 	case entity.RequestStateSucceeded, entity.RequestStateFailed:
-		fact, err := c.recordFact(ctx, store, request)
+		fact, created, err := c.recordFact(ctx, store, request)
 		if err != nil {
 			return err
 		}
 		if !fact.IsGreen() {
 			metrics.NamedCounter(c.metricsScope, _opName, "not_green", 1)
+			// Only the writer of the fact reports the latency: a redelivery adopts
+			// the stored fact instead, and a second sample would count one break
+			// twice in the distribution.
+			if created {
+				c.reportFailureDetectionLatency(ctx, request)
+			}
 			return nil
 		}
 		if err := c.advanceLastGreen(ctx, store, request); err != nil {
@@ -159,8 +165,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 // 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) {
+// request, so a redelivery cannot reach a different verdict than the original. The
+// second return reports whether this call is the one that wrote the fact, which is
+// how a caller tells the original delivery from a redelivery.
+func (c *Controller) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, bool, error) {
 	factStore := store.GetValidationFactStore()
 
 	fact := entity.ValidationFact{
@@ -181,29 +189,100 @@ func (c *Controller) recordFact(ctx context.Context, store storage.Storage, requ
 			"uri", request.URI,
 			"degree", fact.Degree,
 		)
-		return fact, nil
+		return fact, true, 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)
+			return entity.ValidationFact{}, false, 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(
+			return entity.ValidationFact{}, false, 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
+		return stored, false, 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)
+		return entity.ValidationFact{}, false, fmt.Errorf("failed to create the fact for uri %s: %w", request.URI, err)
+	}
+}
+
+// reportFailureDetectionLatency records how long the break this build failed on went
+// undetected, measured from the commit timestamp of the base it validated against. A
+// histogram rather than a gauge because the distribution over failures is the point:
+// how long a break typically survives, not how long the last one did.
+//
+// Unlike the last-green age, there is no later moment to sample this from — an elapsed
+// time is only meaningful against the failure that just became known — so the
+// source-control lookup cannot be moved off the delivery path onto a clock. It is
+// confined to failures and made once the fact is durable, and every way it can fail is
+// counted and swallowed so a reporting fault cannot retry an outcome already recorded.
+func (c *Controller) reportFailureDetectionLatency(ctx context.Context, request entity.Request) {
+	queueTag := metrics.NewTag("queue", request.Queue)
+	strategyTag := metrics.NewTag("strategy", string(request.BuildStrategy))
+
+	// Only a strategy that validates a delta pins a base commit, so a full build has
+	// no baseline to measure from. Its failures are counted rather than timed: absent
+	// here is the ordinary case, not a fault.
+	if request.BaseURI == "" {
+		metrics.NamedCounter(c.metricsScope, _opName, "failure_detection_missing", 1, queueTag, strategyTag)
+		return
 	}
+
+	sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue})
+	if err != nil {
+		c.failureDetectionUnobserved(request, "resolve_source_control", err)
+		return
+	}
+
+	info, err := sourceControl.ChangeInfo(ctx, request.BaseURI)
+	if err != nil {
+		c.failureDetectionUnobserved(request, "get_change_info", err)
+		return
+	}
+
+	// SourceControl must report a positive creation timestamp, so a missing one is a
+	// broken extension contract rather than a lookup failure. Measuring from 1970
+	// would drop a decades-long sample into the distribution.
+	if info.CreatedAt <= 0 {
+		c.failureDetectionUnobserved(request, "undated_change", nil)
+		return
+	}
+
+	// A base dated in the future means the provider's clock disagrees with ours; a
+	// negative latency would corrupt the distribution rather than describe it.
+	latency := time.Since(time.UnixMilli(info.CreatedAt))
+	if latency < 0 {
+		c.failureDetectionUnobserved(request, "future_change", nil)
+		return
+	}
+
+	metrics.NamedHistogram(c.metricsScope, _opName, "failure_detection_latency", metrics.ChangeAgeBuckets,
+		queueTag, strategyTag,
+	).RecordDuration(latency)
+}
+
+// failureDetectionUnobserved counts a latency that could not be observed, tagged with
+// the step that failed so an unmeasurable failure can be told apart from a broken
+// dependency.
+func (c *Controller) failureDetectionUnobserved(request entity.Request, step string, err error) {
+	metrics.NamedCounter(c.metricsScope, _opName, "failure_detection_errors", 1,
+		metrics.NewTag("queue", request.Queue),
+		metrics.NewTag("step", step),
+	)
+	c.logger.Warnw("failed to observe how long the build failure went undetected",
+		"queue", request.Queue,
+		"base_uri", request.BaseURI,
+		"step", step,
+		"error", err,
+	)
 }
 
 // degreeFor maps a request's build outcome onto a whole-repository degree. Only the

stovepipe/controller/record/record_test.go

@@ -36,9 +36,18 @@ import (
 )
 
 const (
-	testQueue = "monorepo/main"
-	testID    = "request/monorepo/main/7"
-	testURI   = "git://remote/monorepo/main/head-sha"
+	testQueue   = "monorepo/main"
+	testID      = "request/monorepo/main/7"
+	testURI     = "git://remote/monorepo/main/head-sha"
+	testBaseURI = "git://remote/monorepo/main/base-sha"
+)
+
+// Metric names as they appear in a snapshot, so a case asserts on the series an
+// operator queries rather than on how the emit is composed.
+const (
+	failureDetectionLatency = "record_controller.record.failure_detection_latency+queue=monorepo/main,strategy=incremental_since_green"
+	failureDetectionMissing = "record_controller.record.failure_detection_missing+queue=monorepo/main,strategy=full"
+	failureDetectionErrors  = "record_controller.record.failure_detection_errors+queue=monorepo/main,step="
 )
 
 var testChangeTime = time.Unix(1_700_000_000, 0).UTC()
@@ -138,6 +147,24 @@ func requestWithState(state entity.RequestState) entity.Request {
 	}
 }
 
+// failedRequest returns a failed request validated incrementally against
+// testBaseURI — the shape that has a detection latency to report.
+func failedRequest() entity.Request {
+	request := requestWithState(entity.RequestStateFailed)
+	request.BaseURI = testBaseURI
+	request.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen
+	return request
+}
+
+// totalSamples sums the samples across a duration histogram's buckets.
+func totalSamples(buckets map[time.Duration]int64) int64 {
+	var sum int64
+	for _, count := range buckets {
+		sum += count
+	}
+	return sum
+}
+
 // queueRow returns the testQueue's row holding the given bookmark.
 func queueRow(lastGreenURI, lastGreenRequestID string, version int32) entity.Queue {
 	return entity.Queue{
@@ -283,6 +310,126 @@ func TestProcess_RecordsBrokenFactWithoutAdvancing(t *testing.T) {
 	assert.False(t, fact.IsGreen())
 }
 
+func TestProcess_ReportsFailureDetectionLatency(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(failedRequest(), nil)
+	var fact entity.ValidationFact
+	m.expectFactCreated(&fact)
+	m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI).
+		Return(sourcecontrol.ChangeInfo{CreatedAt: time.Now().Add(-time.Hour).UnixMilli()}, nil)
+
+	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+
+	histogram, ok := m.metricsScope.Snapshot().Histograms()[failureDetectionLatency]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, totalSamples(histogram.Durations()))
+}
+
+// TestProcess_FullBuildFailureHasNoBaseline covers a full build: it pins no base
+// commit, so there is nothing to measure the latency from. That is the ordinary case
+// for the strategy, not a fault, so it must not land among the errors.
+func TestProcess_FullBuildFailureHasNoBaseline(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+
+	request := failedRequest()
+	request.BaseURI = ""
+	request.BuildStrategy = entity.BuildStrategyFull
+
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil)
+	var fact entity.ValidationFact
+	m.expectFactCreated(&fact)
+
+	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+
+	snapshot := m.metricsScope.Snapshot()
+	assert.Empty(t, snapshot.Histograms(), "a build with no baseline has no latency to report")
+	counter, ok := snapshot.Counters()[failureDetectionMissing]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, counter.Value())
+}
+
+// TestProcess_RedeliveredFailureIsNotResampled covers a redelivery that adopts a
+// broken fact it already wrote: one break must contribute one sample, or the
+// distribution counts the flakiest deliveries twice.
+func TestProcess_RedeliveredFailureIsNotResampled(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(failedRequest(), 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.DegreeBroken, RequestID: testID}, nil)
+
+	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+	assert.Empty(t, m.metricsScope.Snapshot().Histograms())
+}
+
+// TestProcess_UnobservableDetectionLatencyDoesNotFailRecord covers the observation's
+// error posture: every way it can fail is counted with the step that failed and
+// swallowed, because a reporting fault must not disturb the fact already recorded.
+func TestProcess_UnobservableDetectionLatencyDoesNotFailRecord(t *testing.T) {
+	tests := []struct {
+		name  string
+		step  string
+		setup func(c *Controller, m recordMocks)
+	}{
+		{
+			name: "source control cannot be resolved",
+			step: "resolve_source_control",
+			setup: func(c *Controller, _ recordMocks) {
+				c.sourceControls = failingSourceControlFactory{}
+			},
+		},
+		{
+			name: "the base change cannot be looked up",
+			step: "get_change_info",
+			setup: func(_ *Controller, m recordMocks) {
+				m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI).
+					Return(sourcecontrol.ChangeInfo{}, errors.New("boom"))
+			},
+		},
+		{
+			name: "the base change is undated",
+			step: "undated_change",
+			setup: func(_ *Controller, m recordMocks) {
+				m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI).
+					Return(sourcecontrol.ChangeInfo{CreatedAt: 0}, nil)
+			},
+		},
+		{
+			name: "the base change is dated in the future",
+			step: "future_change",
+			setup: func(_ *Controller, m recordMocks) {
+				m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testBaseURI).
+					Return(sourcecontrol.ChangeInfo{CreatedAt: time.Now().Add(time.Hour).UnixMilli()}, nil)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			c, m := newController(t, ctrl)
+			tt.setup(c, m)
+
+			m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(failedRequest(), nil)
+			var fact entity.ValidationFact
+			m.expectFactCreated(&fact)
+
+			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+
+			snapshot := m.metricsScope.Snapshot()
+			assert.Empty(t, snapshot.Histograms(), "no latency may be reported when it cannot be observed")
+			counter, ok := snapshot.Counters()[failureDetectionErrors+tt.step]
+			require.True(t, ok)
+			assert.EqualValues(t, 1, counter.Value())
+		})
+	}
+}
+
 func TestProcess_AdoptsExistingFactFromSameRequest(t *testing.T) {
 	tests := []struct {
 		name       string