From ce2a85696619681a685f86fa300521a42f9b4ce7 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 11 Aug 2026 22:12:46 +0000 Subject: [PATCH] feat(stovepipe): report how long a build failure went undetected MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emit the elapsed time from the commit a failed build validated against, as a histogram tagged with the queue and the build strategy. A histogram because the distribution over failures is the point: how long a break typically survives, not how long the last one did. The observation belongs to record, the stage that turns a build outcome into durable validation state. A break becomes known when its broken fact is written, and an elapsed time is only meaningful against that moment — unlike the last-known-green age, there is no later moment to sample it from, so it cannot be moved off the delivery path onto a clock. It is confined to failures, made once the fact is durable, and swallows every fault: a failed observation is counted with the step that failed and never disturbs the fact already written. Only the writer of a fact reports, so a redelivery that adopts a fact it already wrote cannot count one break twice. A full build pins no base commit, so its failures are counted as unmeasurable rather than timed, which keeps the ordinary case out of the error series. --- platform/metrics/README.md | 5 +- platform/metrics/metrics.go | 22 +++ platform/metrics/metrics_test.go | 1 + stovepipe/controller/record/record.go | 95 +++++++++++-- stovepipe/controller/record/record_test.go | 153 ++++++++++++++++++++- 5 files changed, 263 insertions(+), 13 deletions(-) diff --git a/platform/metrics/README.md b/platform/metrics/README.md index 25108a346..b88cf6146 100644 --- a/platform/metrics/README.md +++ b/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`. diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go index 2f506ca51..20f98c755 100644 --- a/platform/metrics/metrics.go +++ b/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 diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go index f6124d9c4..3dd70c23e 100644 --- a/platform/metrics/metrics_test.go +++ b/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) { diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index afca96cbe..7780646b7 100644 --- a/stovepipe/controller/record/record.go +++ b/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 diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 6d0fc5c09..5562c543a 100644 --- a/stovepipe/controller/record/record_test.go +++ b/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