From d6d6ae32ebe92fd595d5507262144391655a1c30 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Tue, 10 Nov 2026 21:50:14 +0000 Subject: [PATCH] feat(stovepipe): Recover project analysis failures **What**: - Add dead-letter recovery that records unfinished project analysis and hands the request to repository recording. - Preserve partial results and skip provider calls on subsequent deliveries after abandonment. **Why**: - Keep exhausted analysis failures from stranding repository recording or repeating the same provider failure indefinitely. Results already written remain valid, but incomplete attribution must not be reported as a completed empty result. - [x] Add unit tests. - Revert this PR. The recovery controller is not registered and no schema changes are required. - [CODEM-632](https://linear.app/uber/issue/CODEM-632/projects-issues) --- .../controller/projectanalysis/BUILD.bazel | 12 +- .../controller/projectanalysis/README.md | 12 + stovepipe/controller/projectanalysis/dlq.go | 149 ++++++++++++ .../controller/projectanalysis/dlq_test.go | 226 ++++++++++++++++++ .../projectanalysis/projectanalysis.go | 10 +- .../projectanalysis/projectanalysis_test.go | 18 +- stovepipe/entity/request_log.go | 5 +- stovepipe/entity/request_log_test.go | 9 + 8 files changed, 433 insertions(+), 8 deletions(-) create mode 100644 stovepipe/controller/projectanalysis/dlq.go create mode 100644 stovepipe/controller/projectanalysis/dlq_test.go diff --git a/stovepipe/controller/projectanalysis/BUILD.bazel b/stovepipe/controller/projectanalysis/BUILD.bazel index 28f2f0909..65915d067 100644 --- a/stovepipe/controller/projectanalysis/BUILD.bazel +++ b/stovepipe/controller/projectanalysis/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["projectanalysis.go"], + srcs = [ + "dlq.go", + "projectanalysis.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/controller/projectanalysis", visibility = ["//visibility:public"], deps = [ @@ -27,9 +30,13 @@ go_library( go_test( name = "go_default_test", - srcs = ["projectanalysis_test.go"], + srcs = [ + "dlq_test.go", + "projectanalysis_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", @@ -50,5 +57,6 @@ go_test( "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//:go_default_library", + "@org_uber_go_zap//zaptest/observer:go_default_library", ], ) diff --git a/stovepipe/controller/projectanalysis/README.md b/stovepipe/controller/projectanalysis/README.md index 45e01bc1b..e6042fc3b 100644 --- a/stovepipe/controller/projectanalysis/README.md +++ b/stovepipe/controller/projectanalysis/README.md @@ -10,3 +10,15 @@ Retries adopt immutable facts and repair missing lookup entries. A completed request skips provider calls and retries only the recording handoff. Empty results still record completion. Cancelled requests bypass project analysis; superseded requests require no work. + +## Recovery + +DLQ recovery leaves existing facts and association rows untouched. If completion +was already recorded, it retries only the repository-recording handoff. Otherwise +it records `project_analysis_abandoned` before handing off, without calling either +provider or marking attribution complete. History writes and handoff publication +are safe to retry. + +Normal redeliveries also recognize abandonment and skip provider calls. An +already-running delivery may still finish immutable writes; abandonment does not +roll them back or make a partial association set complete. diff --git a/stovepipe/controller/projectanalysis/dlq.go b/stovepipe/controller/projectanalysis/dlq.go new file mode 100644 index 000000000..928cd8862 --- /dev/null +++ b/stovepipe/controller/projectanalysis/dlq.go @@ -0,0 +1,149 @@ +// 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 projectanalysis + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/id" + 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" +) + +// DLQController retains incomplete project analysis without repeating provider calls. +type DLQController struct { + logger *zap.SugaredLogger + scope tally.Scope + stores storage.Factory + requestLogMaterializer requestlog.Materializer + registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string +} + +func NewDLQController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + requestLogMaterializer requestlog.Materializer, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) *DLQController { + name := string(topicKey) + "_controller" + return &DLQController{ + logger: logger.Named(name), + scope: scope.SubScope(name), + stores: stores, + requestLogMaterializer: requestLogMaterializer, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process acknowledges unusable messages and retries failures retaining history or publishing. +func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { + op := metrics.Begin(c.scope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + msg := delivery.Message() + analysis := &stovepipemq.ProjectAnalysis{} + if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil { + c.logger.Errorw("discarding malformed project analysis dlq message", "message_id", msg.ID, "error", err) + return nil + } + if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil { + c.logger.Errorw("discarding project analysis dlq message with invalid queue identity", "message_id", msg.ID, "error", err) + return nil + } + if err := id.Validate(analysis.GetId()); err != nil { + c.logger.Errorw("discarding project analysis dlq message with invalid request ID", "message_id", msg.ID, "error", err) + return nil + } + if msg.PartitionKey != analysis.GetId() { + c.logger.Errorw("discarding project analysis dlq message from the wrong partition", "message_id", msg.ID, "partition", msg.PartitionKey) + return nil + } + store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()}) + if err != nil { + c.logger.Errorw("discarding project analysis dlq message for an unresolvable queue", + "message_id", msg.ID, "queue", analysis.GetQueueName(), "error", err) + return nil + } + request, err := store.GetRequestStore().Get(ctx, analysis.GetId()) + if errors.Is(err, storage.ErrNotFound) { + c.logger.Errorw("discarding project analysis dlq message for a missing request", + "message_id", msg.ID, "queue", analysis.GetQueueName(), "request_id", analysis.GetId()) + return nil + } + if err != nil { + return fmt.Errorf("load project analysis dlq request: %w", err) + } + if request.Queue != analysis.GetQueueName() { + c.logger.Errorw("discarding project analysis dlq message for another queue", + "message_id", msg.ID, "queue", analysis.GetQueueName(), "request_id", request.ID) + return nil + } + switch request.State { + case entity.RequestStateSucceeded, entity.RequestStateFailed: + case entity.RequestStateCancelled: + return publishRecord(ctx, c.registry, request) + case entity.RequestStateSuperseded: + return nil + default: + c.logger.Errorw("discarding project analysis dlq message for a nonterminal request", + "message_id", msg.ID, "queue", request.Queue, "request_id", request.ID, "state", request.State) + return nil + } + + if request.TerminalBuildID != "" { + projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) + _, err := store.GetRequestLogStore().Get(ctx, request.ID, projectIssuesRecordedLog.ID) + if err == nil { + return publishRecord(ctx, c.registry, request) + } + if !errors.Is(err, storage.ErrNotFound) { + return fmt.Errorf("read completed project analysis during recovery: %w", err) + } + } + + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) + if err := c.requestLogMaterializer.PersistLog(ctx, store, projectAnalysisAbandonedLog); err != nil { + return fmt.Errorf("retain abandoned project analysis: %w", err) + } + fields := []any{"message_id", msg.ID, "queue", request.Queue, "request_id", request.ID, "build_id", request.TerminalBuildID} + if original, ok := delivery.Failure(); ok { + fields = append(fields, "failure", original.Message, "failure_subjects", original.Subjects, "failure_detail", original.Detail) + } + c.logger.Errorw("project analysis abandoned; retaining partial results", fields...) + metrics.NamedCounter(c.scope, "process", "analysis_abandoned", 1, metrics.TagsFromContext(ctx)...) + return publishRecord(ctx, c.registry, request) +} + +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/projectanalysis/dlq_test.go b/stovepipe/controller/projectanalysis/dlq_test.go new file mode 100644 index 000000000..22d15626c --- /dev/null +++ b/stovepipe/controller/projectanalysis/dlq_test.go @@ -0,0 +1,226 @@ +// 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 projectanalysis + +import ( + "context" + "maps" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + 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" + "go.uber.org/zap/zaptest/observer" +) + +func (f *fixture) recover(ctx context.Context, msg entityqueue.Message) error { + delivery := consumermock.NewMockDelivery(f.ctrl) + delivery.EXPECT().Message().Return(msg) + delivery.EXPECT().Failure().Return(failure.Failure{Message: "project analysis failed"}, true).AnyTimes() + return f.dlq.Process(ctx, delivery) +} + +func TestDLQRetainsPartialResults(t *testing.T) { + for _, operation := range []string{ + "resolve", "attribute", "reverse/create", "fact/create", "index/create", + "log/project_facts_recorded", "log/project_issues_recorded", + } { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + request := f.request + f.failures[operation] = assert.AnError + require.ErrorIs(t, f.deliver(context.Background(), msg), assert.AnError) + facts, indexes := maps.Clone(f.facts), maps.Clone(f.indexes) + source, reverse := maps.Clone(f.source), maps.Clone(f.reverse) + resolverCalls, attributerCalls := f.resolverCalls, f.attributerCalls + + require.NoError(t, f.recover(context.Background(), msg)) + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) + require.Equal(t, projectAnalysisAbandonedLog, f.logs[projectAnalysisAbandonedLog.ID]) + require.Empty(t, f.logs[projectAnalysisAbandonedLog.ID].Metadata) + require.NotContains(t, f.logs, f.completion().ID) + require.Equal(t, request, f.request) + require.Equal(t, facts, f.facts) + require.Equal(t, indexes, f.indexes) + require.Equal(t, source, f.source) + require.Equal(t, reverse, f.reverse) + require.Equal(t, resolverCalls, f.resolverCalls) + require.Equal(t, attributerCalls, f.attributerCalls) + require.Len(t, f.records, 1) + + logCount := len(f.logs) + require.NoError(t, f.recover(context.Background(), msg)) + require.NoError(t, f.deliver(context.Background(), msg)) + require.Len(t, f.logs, logCount) + require.NotContains(t, f.logs, f.completion().ID) + require.Equal(t, facts, f.facts) + require.Equal(t, source, f.source) + require.Equal(t, reverse, f.reverse) + require.Equal(t, resolverCalls, f.resolverCalls) + require.Equal(t, attributerCalls, f.attributerCalls) + require.Len(t, f.records, 3) + for _, record := range f.records { + require.Equal(t, f.records[0].ID, record.ID) + require.Equal(t, f.records[0].Payload, record.Payload) + } + }) + } +} + +func TestDLQPreservesCompletedAnalysis(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + f.failures["record/publish"] = assert.AnError + require.ErrorIs(t, f.deliver(context.Background(), msg), assert.AnError) + facts, source, reverse := maps.Clone(f.facts), maps.Clone(f.source), maps.Clone(f.reverse) + require.Contains(t, f.logs, f.completion().ID) + + require.NoError(t, f.recover(context.Background(), msg)) + require.NoError(t, f.recover(context.Background(), msg)) + require.NotContains(t, f.logs, requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil).ID) + require.Equal(t, facts, f.facts) + require.Equal(t, source, f.source) + require.Equal(t, reverse, f.reverse) + require.Equal(t, 1, f.resolverCalls) + require.Equal(t, 1, f.attributerCalls) + require.Len(t, f.records, 2) + require.Equal(t, f.publishAttempts[0].ID, f.records[0].ID) +} + +func TestDLQRetriesRecoveryFailures(t *testing.T) { + for _, tt := range []struct { + operation string + err error + }{ + {operation: "request/get", err: context.Canceled}, + {operation: "completion/get", err: assert.AnError}, + {operation: "log/project_analysis_abandoned", err: assert.AnError}, + {operation: "record/publish", err: assert.AnError}, + } { + t.Run(tt.operation, func(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + f.failures[tt.operation] = tt.err + require.ErrorIs(t, f.recover(context.Background(), msg), tt.err) + require.Empty(t, f.records) + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + if tt.operation == "record/publish" { + require.Contains(t, f.logs, requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil).ID) + } + + require.NoError(t, f.recover(context.Background(), msg)) + require.Contains(t, f.logs, requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil).ID) + require.Len(t, f.records, 1) + require.Empty(t, f.facts) + require.Empty(t, f.source) + require.NotContains(t, f.logs, f.completion().ID) + }) + } +} + +func TestDLQHandlesTerminalOutcomes(t *testing.T) { + for _, tt := range []struct { + name string + state entity.RequestState + build string + abandoned bool + recordCount int + }{ + {name: "failed", state: entity.RequestStateFailed, build: "build/7", abandoned: true, recordCount: 1}, + {name: "succeeded", state: entity.RequestStateSucceeded, build: "build/7", abandoned: true, recordCount: 1}, + {name: "failed without a build outcome", state: entity.RequestStateFailed, abandoned: true, recordCount: 1}, + {name: "cancelled", state: entity.RequestStateCancelled, recordCount: 1}, + {name: "superseded", state: entity.RequestStateSuperseded}, + {name: "nonterminal", state: entity.RequestStateProcessing}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.request.State, f.request.TerminalBuildID = tt.state, tt.build + f.logs = map[string]entity.RequestLog{} + msg := f.message(t) + require.NoError(t, f.recover(context.Background(), msg)) + require.Len(t, f.records, tt.recordCount) + if tt.abandoned { + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) + require.Equal(t, projectAnalysisAbandonedLog, f.logs[projectAnalysisAbandonedLog.ID]) + require.NoError(t, f.deliver(context.Background(), msg)) + require.Len(t, f.records, tt.recordCount+1) + } else { + require.Empty(t, f.logs) + } + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.Empty(t, f.facts) + require.Empty(t, f.source) + }) + } +} + +func TestDLQDiscardsUnusableMessages(t *testing.T) { + for _, scenario := range []string{ + "malformed", "invalid request ID", "tenant mismatch", "partition mismatch", + "stored queue mismatch", "missing request", "unknown queue", + } { + t.Run(scenario, func(t *testing.T) { + f := newFixture(t) + msg := f.message(t) + switch scenario { + case "malformed": + msg.Payload = []byte("not json") + case "invalid request ID": + var err error + msg.Payload, err = stovepipemq.Marshal(&stovepipemq.ProjectAnalysis{Id: "invalid", QueueName: testQueue}) + require.NoError(t, err) + case "tenant mismatch": + msg.Tenant = "another/queue" + case "partition mismatch": + msg.PartitionKey = "another-request" + case "stored queue mismatch": + f.request.Queue = "another/queue" + case "missing request": + f.failures["request/get"] = storage.ErrNotFound + case "unknown queue": + f.failures["storage/for"] = assert.AnError + } + require.NoError(t, f.recover(context.Background(), msg)) + require.Empty(t, f.records) + require.Zero(t, f.resolverCalls) + require.Zero(t, f.attributerCalls) + require.NotContains(t, f.logs, requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil).ID) + }) + } +} + +func TestDLQReportsOriginalFailure(t *testing.T) { + f := newFixture(t) + core, logs := observer.New(zap.ErrorLevel) + f.dlq.logger = zap.New(core).Sugar() + require.NoError(t, f.recover(context.Background(), f.message(t))) + require.Len(t, logs.All(), 1) + fields := logs.All()[0].ContextMap() + require.Equal(t, "project analysis failed", fields["failure"]) + require.Equal(t, f.request.Queue, fields["queue"]) + require.Equal(t, f.request.ID, fields["request_id"]) + require.Equal(t, f.request.TerminalBuildID, fields["build_id"]) +} diff --git a/stovepipe/controller/projectanalysis/projectanalysis.go b/stovepipe/controller/projectanalysis/projectanalysis.go index 75742904d..a38bf627b 100644 --- a/stovepipe/controller/projectanalysis/projectanalysis.go +++ b/stovepipe/controller/projectanalysis/projectanalysis.go @@ -117,12 +117,20 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r default: return fmt.Errorf("project analysis requires a terminal request") } + history := store.GetRequestLogStore() + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) + _, err = history.Get(ctx, request.ID, projectAnalysisAbandonedLog.ID) + if err == nil { + return publishRecord(ctx, c.registry, request) + } + if !errors.Is(err, storage.ErrNotFound) { + return fmt.Errorf("read project analysis abandonment: %w", err) + } if request.TerminalBuildID == "" { return fmt.Errorf("project analysis requires the request's terminal build") } projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil) - history := store.GetRequestLogStore() _, err = history.Get(ctx, request.ID, projectIssuesRecordedLog.ID) if err == nil { return publishRecord(ctx, c.registry, request) diff --git a/stovepipe/controller/projectanalysis/projectanalysis_test.go b/stovepipe/controller/projectanalysis/projectanalysis_test.go index 09d9bd060..084051c4b 100644 --- a/stovepipe/controller/projectanalysis/projectanalysis_test.go +++ b/stovepipe/controller/projectanalysis/projectanalysis_test.go @@ -47,6 +47,7 @@ const testQueue = "validation/main" type fixture struct { controller *Controller + dlq *DLQController ctrl *gomock.Controller request entity.Request projectStatuses []projectstatus.Result @@ -104,9 +105,12 @@ func newFixture(t *testing.T) *fixture { return f.request, f.fail("request/get") }).AnyTimes() history.EXPECT().Get(gomock.Any(), "7", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, id string) (entity.RequestLog, error) { + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) operation := "analysis/get" if id == f.completion().ID { operation = "completion/get" + } else if id == projectAnalysisAbandonedLog.ID { + operation = "abandonment/get" } if err := f.fail(operation); err != nil { return entity.RequestLog{}, err @@ -143,8 +147,10 @@ func newFixture(t *testing.T) *fixture { reverse.EXPECT().Create(gomock.Any(), "7", gomock.Any()).DoAndReturn(f.retainAttribution(f.reverse, "reverse/create")).AnyTimes() requestLogMaterializer := requestlogmock.NewMockMaterializer(ctrl) requestLogMaterializer.EXPECT().PersistLog(gomock.Any(), store, gomock.Any()).DoAndReturn(func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - require.Len(t, f.facts, len(f.projectStatuses)) - require.Len(t, f.indexes, len(f.projectStatuses)) + if log.Event != entity.RequestEventProjectAnalysisAbandoned { + require.Len(t, f.facts, len(f.projectStatuses)) + require.Len(t, f.indexes, len(f.projectStatuses)) + } if log.Event == entity.RequestEventProjectIssuesRecorded { require.Equal(t, f.source, f.reverse) factsLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectFactsRecorded, "all", nil) @@ -185,7 +191,10 @@ func newFixture(t *testing.T) *fixture { publisher := queuemock.NewMockPublisher(ctrl) publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { if f.request.State != entity.RequestStateCancelled { - require.Contains(t, f.logs, f.completion().ID) + projectAnalysisAbandonedLog := requestlog.NewRequestEventLog(f.request, entity.RequestEventProjectAnalysisAbandoned, "projects", nil) + _, complete := f.logs[f.completion().ID] + _, abandoned := f.logs[projectAnalysisAbandonedLog.ID] + require.True(t, complete || abandoned) } f.publishAttempts = append(f.publishAttempts, msg) if err := f.fail("record/publish"); err != nil { @@ -199,6 +208,7 @@ func newFixture(t *testing.T) *fixture { registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: stovepipemq.TopicKeyRecord, Name: "record", Queue: queue}}) require.NoError(t, err) f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, stores, requestLogMaterializer, resolvers, attributers, registry, stovepipemq.TopicKeyProjectAnalysis, "stovepipe-projectanalysis") + f.dlq = NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, stores, requestLogMaterializer, registry, "projectanalysis_dlq", "stovepipe-projectanalysis-dlq") return f } @@ -301,7 +311,7 @@ func TestProcessRecordsProjectStatuses(t *testing.T) { func TestProcessRetriesPartialWork(t *testing.T) { for _, operation := range []string{ - "storage/for", "request/get", "completion/get", "analysis/get", "resolver/for", "resolve", "attributer/for", "attribute", + "storage/for", "request/get", "abandonment/get", "completion/get", "analysis/get", "resolver/for", "resolve", "attributer/for", "attribute", "source/create", "reverse/create", "fact/create", "index/create", "log/project_facts_recorded", "log/project_issues_recorded", "record/publish", } { t.Run(operation, func(t *testing.T) { diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index ceaf1263d..fe6c8ce26 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -32,6 +32,8 @@ const ( RequestEventProjectFactsRecorded RequestEvent = "project_facts_recorded" // RequestEventProjectIssuesRecorded records completed project-issue attribution for a request. RequestEventProjectIssuesRecorded RequestEvent = "project_issues_recorded" + // RequestEventProjectAnalysisAbandoned records unfinished project analysis after primary processing stopped. + RequestEventProjectAnalysisAbandoned RequestEvent = "project_analysis_abandoned" // RequestEventRecordAbandoned records that record-stage work stopped after exhausting retries. RequestEventRecordAbandoned RequestEvent = "record_abandoned" // RequestEventBuildAnalysisAbandoned records incomplete build analysis, not a validation outcome. @@ -155,7 +157,8 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventProjectIssuesRecorded, + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, + RequestEventProjectFactsRecorded, RequestEventProjectIssuesRecorded, RequestEventProjectAnalysisAbandoned, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned, RequestEventBuildAnalysisCompleted, RequestEventIssueTrackingStarted, RequestEventIssuesReconciled, RequestEventIssueTrackingAbandoned, RequestEventFixSubmitted, RequestEventFixSubmissionSkipped, RequestEventFixSubmissionAbandoned: diff --git a/stovepipe/entity/request_log_test.go b/stovepipe/entity/request_log_test.go index d31334b92..99aa17aa7 100644 --- a/stovepipe/entity/request_log_test.go +++ b/stovepipe/entity/request_log_test.go @@ -126,6 +126,15 @@ func TestRequestLogValidate(t *testing.T) { return entry }, }, + { + name: "project analysis abandoned", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventProjectAnalysisAbandoned + entry.RequestVersion = 0 + return entry + }, + }, { name: "build event", mutate: func(entry RequestLog) RequestLog {