From 8394d14ece605377c8cb03eb6dcb5b4b69b0de1d Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 22 Oct 2026 04:04:55 +0000 Subject: [PATCH] feat(stovepipe): Record incomplete build analysis --- doc/rfc/stovepipe/request-log.md | 7 +- .../controller/buildanalysis/BUILD.bazel | 15 +- stovepipe/controller/buildanalysis/README.md | 10 +- stovepipe/controller/buildanalysis/dlq.go | 185 +++++++++++++ .../controller/buildanalysis/dlq_test.go | 248 ++++++++++++++++++ stovepipe/entity/request_log.go | 4 +- stovepipe/entity/request_log_test.go | 10 + 7 files changed, 473 insertions(+), 6 deletions(-) create mode 100644 stovepipe/controller/buildanalysis/dlq.go create mode 100644 stovepipe/controller/buildanalysis/dlq_test.go diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index a32aac6de..83d7c04d5 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -2,13 +2,14 @@ ## Summary -Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus asynchronous milestones needed to explain those transitions, the public verdict, and abandoned terminal-stage work: +Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus asynchronous milestones needed to explain those transitions, the public verdict, and abandoned stage work: - `build_triggered`; - `build_finished`; - `validation_fact_recorded`; - `project_facts_recorded`; -- `record_abandoned`. +- `record_abandoned`; +- `build_analysis_abandoned`. The model deliberately follows SubmitQueue's distinction between statuses describing where a request is and events describing important activity that does not move it. It remains a bounded request-lifecycle log rather than a generic event bus or an audit of every correlated operation. @@ -126,6 +127,7 @@ Immutable Request context such as URI, build strategy, base URI, and terminal bu | `validation_fact_recorded` | The immutable whole-repository fact became durable. | Degree metadata and fact creation time | | `project_facts_recorded` | All project validation facts returned for a request became durable. | Project-fact count metadata and completion time | | `record_abandoned` | Record-stage work stopped after exhausting primary retries. | Event retention time | +| `build_analysis_abandoned` | Build analysis could not complete; issue observations may be incomplete. This is not a validation verdict. | Build ID metadata and event retention time | Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. @@ -214,6 +216,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo | Buildsignal | Persist terminal Build and retain `build_finished`; CAS the Request outcome and retain its terminal state. | Existing terminal Build and Request outcome each ensure their own entry before record publication. | | Record | Create or verify the whole-repository fact, then retain `validation_fact_recorded`; after recording project facts, retain one `project_facts_recorded` event. | An identical whole-repository fact and project-fact batch ensure their events before bookmark or promotion work. | | Record DLQ | Retain `record_abandoned`, then acknowledge the remaining record work. | The stable event ID makes history retention idempotent without replaying facts, bookmarks, promotion, or hooks. | +| Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. | | Reconciler | CAS an unrecoverable non-terminal Request to failed, then retain failed. | An existing terminal Request is repaired from its persisted outcome without relabeling it. | Build running and unchanged polls create no entry. A failed runner trigger that creates no Build creates no event. Cancelled and superseded requests create no validation fact. diff --git a/stovepipe/controller/buildanalysis/BUILD.bazel b/stovepipe/controller/buildanalysis/BUILD.bazel index d83fe9cc7..37afa9972 100644 --- a/stovepipe/controller/buildanalysis/BUILD.bazel +++ b/stovepipe/controller/buildanalysis/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["buildanalysis.go"], + srcs = [ + "buildanalysis.go", + "dlq.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/controller/buildanalysis", visibility = ["//visibility:public"], deps = [ @@ -11,6 +14,7 @@ go_library( "//platform/metrics:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildissue:go_default_library", "//stovepipe/extension/storage:go_default_library", @@ -21,14 +25,21 @@ go_library( go_test( name = "go_default_test", - srcs = ["buildanalysis_test.go"], + srcs = [ + "buildanalysis_test.go", + "dlq_test.go", + ], embed = [":go_default_library"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/requestlog:go_default_library", + "//stovepipe/core/requestlog/mock:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildissue:go_default_library", "//stovepipe/extension/buildissue/mock:go_default_library", diff --git a/stovepipe/controller/buildanalysis/README.md b/stovepipe/controller/buildanalysis/README.md index aac7cde7b..95c78104d 100644 --- a/stovepipe/controller/buildanalysis/README.md +++ b/stovepipe/controller/buildanalysis/README.md @@ -18,4 +18,12 @@ Status polling and updates remain the responsibility of buildsignal. Analysis ne Detection, persistence, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter. -The message and controller are defined without runtime producers or consumer registration. Host integration and detector implementations are separate work. +## DLQ recovery + +The recovery controller records a `build_analysis_abandoned` event in the Request's history, with the build ID, to indicate that analysis could not complete and issue observations may be incomplete. It does not rerun detection, delete earlier observations, change Request or Build state, or release a build slot. Original failure details are logged rather than included in the immutable history event. + +History identity is stable per build, so repeated recovery deliveries and separate analysis notices for the same build retain one event. Different builds retain different events, even for the same Request. A later successful analysis does not erase the historical occurrence. + +State-read and history-write errors are returned for retry. Hosts must run recovery with the DLQ consumer's always-retryable processor and unlimited-retry subscription; only recovery work is retried, not analysis. Unusable payloads, invalid or unresolvable queues, missing Build or Request records, and queue mismatches are logged and acknowledged rather than retried indefinitely. + +The controllers are defined without runtime producers or consumer registration. Host integration and detector implementations are separate work. diff --git a/stovepipe/controller/buildanalysis/dlq.go b/stovepipe/controller/buildanalysis/dlq.go new file mode 100644 index 000000000..36dad5636 --- /dev/null +++ b/stovepipe/controller/buildanalysis/dlq.go @@ -0,0 +1,185 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package buildanalysis + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +const _dlqOpName = "buildanalysis_dlq" + +// DLQController records incomplete analysis without rerunning detection or +// changing the validation outcome. +type DLQController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + materializer requestlog.Materializer + topicKey consumer.TopicKey + consumerGroup string +} + +// NewDLQController creates a controller for incomplete build analysis. +func NewDLQController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + materializer requestlog.Materializer, + topicKey consumer.TopicKey, + consumerGroup string, +) *DLQController { + name := string(topicKey) + "_controller" + return &DLQController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + materializer: materializer, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process retains incomplete-analysis history and acknowledges unusable messages. +// State-read and history-write failures are returned for DLQ recovery retry. +func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + analysis := &stovepipemq.BuildAnalysis{} + if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding malformed analysis dlq message", + "message_id", msg.ID, + "error", err, + ) + return nil + } + + if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message with invalid queue identity", + "message_id", msg.ID, + "build_id", analysis.GetId(), + "error", err, + ) + return nil + } + + if analysis.GetId() == "" { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_id_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message with empty build id", + "message_id", msg.ID, + "queue", analysis.GetQueueName(), + ) + return nil + } + + store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message for unresolvable queue", + "message_id", msg.ID, + "build_id", analysis.GetId(), + "queue", analysis.GetQueueName(), + "error", err, + ) + return nil + } + + build, err := store.GetBuildStore().Get(ctx, analysis.GetId()) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "build_not_found", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message for missing build", + "message_id", msg.ID, "build_id", analysis.GetId(), "queue", analysis.GetQueueName()) + return nil + } + + return fmt.Errorf("failed to load build %s for analysis dlq: %w", analysis.GetId(), err) + } + + if build.RequestID == "" { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_request_id", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message for build without a request", + "message_id", msg.ID, "build_id", analysis.GetId()) + return nil + } + + request, err := store.GetRequestStore().Get(ctx, build.RequestID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "request_not_found", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message for missing request", + "message_id", msg.ID, "build_id", analysis.GetId(), "request_id", build.RequestID) + return nil + } + + return fmt.Errorf("failed to load request %s for analysis dlq: %w", build.RequestID, err) + } + + if request.Queue != analysis.GetQueueName() { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding analysis dlq message for request in another queue", + "message_id", msg.ID, "request_id", request.ID, "queue", request.Queue, + "message_queue", analysis.GetQueueName()) + return nil + } + + log := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, analysis.GetId(), + map[string]string{requestlog.MetadataKeyBuildID: analysis.GetId()}) + if err := c.materializer.PersistLog(ctx, store, log); err != nil { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "history_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to retain incomplete build analysis for request %s: %w", request.ID, err) + } + + originalFailure, hasFailure := delivery.Failure() + metrics.NamedCounter(c.metricsScope, _dlqOpName, "analysis_abandoned", 1, metrics.TagsFromContext(ctx)...) + fields := []any{ + "message_id", msg.ID, + "request_id", request.ID, + "build_id", analysis.GetId(), + "queue", request.Queue, + "request_state", request.State, + "history_event", entity.RequestEventBuildAnalysisAbandoned, + "attempt", delivery.Attempt(), + } + if hasFailure { + fields = append(fields, + "failure", originalFailure.Message, + "failure_subjects", originalFailure.Subjects, + "failure_detail", originalFailure.Detail, + ) + } + c.logger.Errorw("build analysis could not complete; issue observations may be incomplete", fields...) + return nil +} + +func (c *DLQController) Name() string { return string(c.topicKey) } + +func (c *DLQController) TopicKey() consumer.TopicKey { return c.topicKey } + +func (c *DLQController) ConsumerGroup() string { return c.consumerGroup } + +var _ consumer.Controller = (*DLQController)(nil) diff --git a/stovepipe/controller/buildanalysis/dlq_test.go b/stovepipe/controller/buildanalysis/dlq_test.go new file mode 100644 index 000000000..6280d71b4 --- /dev/null +++ b/stovepipe/controller/buildanalysis/dlq_test.go @@ -0,0 +1,248 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package buildanalysis + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +type dlqMocks struct { + ctrl *gomock.Controller + factory *storagemock.MockFactory + store *storagemock.MockStorage + builds *storagemock.MockBuildStore + requests *storagemock.MockRequestStore + materializer *requestlogmock.MockMaterializer +} + +func newDLQTestController(t *testing.T) (*DLQController, dlqMocks) { + t.Helper() + + ctrl := gomock.NewController(t) + m := dlqMocks{ + ctrl: ctrl, + factory: storagemock.NewMockFactory(ctrl), + store: storagemock.NewMockStorage(ctrl), + builds: storagemock.NewMockBuildStore(ctrl), + requests: storagemock.NewMockRequestStore(ctrl), + materializer: requestlogmock.NewMockMaterializer(ctrl), + } + m.store.EXPECT().GetBuildStore().Return(m.builds).AnyTimes() + m.store.EXPECT().GetRequestStore().Return(m.requests).AnyTimes() + + c := NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, m.factory, m.materializer, + consumer.TopicKey("buildanalysis_dlq"), "stovepipe-buildanalysis-dlq") + + return c, m +} + +func (m dlqMocks) expectLoad(request entity.Request) { + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + m.builds.EXPECT().Get(gomock.Any(), testBuildID). + Return(entity.Build{ID: testBuildID, RequestID: request.ID, Status: entity.BuildStatusRunning}, nil) + m.requests.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) +} + +func (m dlqMocks) delivery(msg entityqueue.Message, hasFailure bool) *consumermock.MockDelivery { + d := consumermock.NewMockDelivery(m.ctrl) + d.EXPECT().Message().Return(msg) + d.EXPECT().Failure().Return(failure.Failure{Message: "detector unavailable"}, hasFailure).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + + return d +} + +func TestDLQRetainsHistoryWithoutChangingValidation(t *testing.T) { + for _, state := range []entity.RequestState{ + entity.RequestStateAccepted, entity.RequestStateProcessing, entity.RequestStateSucceeded, + entity.RequestStateFailed, entity.RequestStateCancelled, entity.RequestStateSuperseded, + } { + t.Run(string(state), func(t *testing.T) { + c, m := newDLQTestController(t) + request := entity.Request{ID: "request/1", Queue: testQueue, State: state, Version: 3} + m.expectLoad(request) + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, "event/build_analysis_abandoned/"+testBuildID, log.ID) + assert.Equal(t, request.ID, log.RequestID) + assert.Equal(t, testQueue, log.Queue) + assert.Equal(t, entity.RequestEventBuildAnalysisAbandoned, log.Event) + assert.Equal(t, map[string]string{requestlog.MetadataKeyBuildID: testBuildID}, log.Metadata) + assert.Empty(t, log.State) + assert.Zero(t, log.RequestVersion) + assert.Empty(t, log.OutcomeReason) + return nil + }, + ) + + require.NoError(t, c.Process(context.Background(), m.delivery(analysisMessage(t), state.IsTerminal()))) + }) + } +} + +func TestDLQReturnsStateAndHistoryErrorsForRetry(t *testing.T) { + for _, step := range []string{"load build", "load request", "retain history"} { + t.Run(step, func(t *testing.T) { + c, m := newDLQTestController(t) + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + + if step == "load build" { + m.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, assert.AnError) + } else { + m.builds.EXPECT().Get(gomock.Any(), testBuildID). + Return(entity.Build{ID: testBuildID, RequestID: "request/1", Status: entity.BuildStatusFailed}, nil) + + if step == "load request" { + m.requests.EXPECT().Get(gomock.Any(), "request/1").Return(entity.Request{}, assert.AnError) + } else { + m.requests.EXPECT().Get(gomock.Any(), "request/1"). + Return(entity.Request{ID: "request/1", Queue: testQueue}, nil) + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(assert.AnError) + } + } + + err := c.Process(context.Background(), m.delivery(analysisMessage(t), false)) + require.ErrorIs(t, err, assert.AnError) + }) + } +} + +func TestDLQAcknowledgesUnusableMessages(t *testing.T) { + for _, kind := range []string{ + "malformed payload", "empty queue", "empty tenant", "tenant mismatch", "metadata mismatch", "empty build ID", + "unresolvable queue", "missing build", "build without request", "missing request", "request queue mismatch", + } { + t.Run(kind, func(t *testing.T) { + c, m := newDLQTestController(t) + msg := analysisMessage(t) + + switch kind { + case "malformed payload": + msg.Payload = []byte("{") + case "empty queue": + msg.Payload = []byte(`{"id":"build/42"}`) + case "empty tenant": + msg.Tenant = "" + case "tenant mismatch": + msg.Tenant = "other/main" + case "metadata mismatch": + msg.Metadata[entityqueue.MetadataKeyQueueName] = "other/main" + case "empty build ID": + msg.Payload = []byte(`{"queue_name":"validation/main"}`) + case "unresolvable queue": + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(nil, assert.AnError) + default: + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + switch kind { + case "missing build": + m.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, storage.ErrNotFound) + case "build without request": + m.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID}, nil) + case "missing request": + m.builds.EXPECT().Get(gomock.Any(), testBuildID). + Return(entity.Build{ID: testBuildID, RequestID: "request/1"}, nil) + m.requests.EXPECT().Get(gomock.Any(), "request/1").Return(entity.Request{}, storage.ErrNotFound) + case "request queue mismatch": + m.builds.EXPECT().Get(gomock.Any(), testBuildID). + Return(entity.Build{ID: testBuildID, RequestID: "request/1"}, nil) + m.requests.EXPECT().Get(gomock.Any(), "request/1"). + Return(entity.Request{ID: "request/1", Queue: "other/main"}, nil) + } + } + + require.NoError(t, c.Process(context.Background(), m.delivery(msg, false))) + }) + } +} + +func TestDLQDeduplicatesHistoryAcrossRetriesAndNotices(t *testing.T) { + c, m := newDLQTestController(t) + c.materializer = requestlog.NewMaterializer(tally.NoopScope) + logStore := storagemock.NewMockRequestLogStore(m.ctrl) + m.store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + retained := make(map[string]entity.RequestLog) + failFirstWrite := true + + logStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, log entity.RequestLog) error { + require.NoError(t, log.Validate()) + if _, exists := retained[log.ID]; exists { + return storage.ErrAlreadyExists + } + + retained[log.ID] = log + if failFirstWrite { + failFirstWrite = false + return assert.AnError + } + + return nil + }, + ).Times(4) + logStore.EXPECT().Get(gomock.Any(), "request/1", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, id string) (entity.RequestLog, error) { + return retained[id], nil + }, + ).Times(2) + + for i, buildID := range []string{testBuildID, testBuildID, testBuildID, "build/43"} { + request := entity.Request{ID: "request/1", Queue: testQueue, State: entity.RequestStateProcessing, Version: int32(i + 1)} + if i > 0 { + request.State = entity.RequestStateSucceeded + } + + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + m.builds.EXPECT().Get(gomock.Any(), buildID).Return(entity.Build{ID: buildID, RequestID: request.ID}, nil) + m.requests.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + msg := analysisMessage(t) + msg.ID = fmt.Sprintf("notice-%d", i) + msg.PartitionKey = buildID + payload, err := stovepipemq.Marshal(&stovepipemq.BuildAnalysis{Id: buildID, QueueName: testQueue}) + require.NoError(t, err) + msg.Payload = payload + + err = c.Process(context.Background(), m.delivery(msg, i != 0)) + if i == 0 { + require.ErrorIs(t, err, assert.AnError) + } else { + require.NoError(t, err) + } + } + + require.Len(t, retained, 2) + for _, buildID := range []string{testBuildID, "build/43"} { + log, ok := retained["event/build_analysis_abandoned/"+buildID] + require.True(t, ok) + assert.Equal(t, buildID, log.Metadata[requestlog.MetadataKeyBuildID]) + } +} diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index 321a3061a..7c53434c4 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -32,6 +32,8 @@ const ( RequestEventProjectFactsRecorded RequestEvent = "project_facts_recorded" // RequestEventRecordAbandoned records that record-stage work stopped after exhausting retries. RequestEventRecordAbandoned RequestEvent = "record_abandoned" + // RequestEventBuildAnalysisAbandoned records incomplete build analysis, not a validation outcome. + RequestEventBuildAnalysisAbandoned RequestEvent = "build_analysis_abandoned" ) // RequestOutcomeReason identifies the durable domain reason for a terminal request state. @@ -137,7 +139,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventRecordAbandoned: + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned: default: return fmt.Errorf("unknown request event %q", e.Event) } diff --git a/stovepipe/entity/request_log_test.go b/stovepipe/entity/request_log_test.go index 67954c209..dc6a82975 100644 --- a/stovepipe/entity/request_log_test.go +++ b/stovepipe/entity/request_log_test.go @@ -145,6 +145,16 @@ func TestRequestLogValidate(t *testing.T) { return entry }, }, + { + name: "incomplete build analysis event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildAnalysisAbandoned + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + }, {name: "missing ID", mutate: func(entry RequestLog) RequestLog { entry.ID = ""; return entry }, wantErr: true}, {name: "missing queue", mutate: func(entry RequestLog) RequestLog { entry.Queue = ""; return entry }, wantErr: true}, {name: "missing request ID", mutate: func(entry RequestLog) RequestLog { entry.RequestID = ""; return entry }, wantErr: true},