From 4635ad3f4f0496a6eeaabaeba4e6ffaff64cf0de Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 18 Sep 2026 20:34:46 +0000 Subject: [PATCH 1/4] fix(stovepipe): bound record promotion retries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Prevent permanent promotion failures from occupying the record DLQ indefinitely. - Preserve successful validation state while making abandoned promotion visible in request history. Changes: - Add record-specific DLQ handling that does not repeat known promotion failures. - Retain an idempotent promotion_failed history event before acknowledging abandoned promotion work. - Keep non-promotion reconciliation under the existing DLQ retry policy. --- Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace --- doc/rfc/stovepipe/request-log.md | 5 +- doc/rfc/stovepipe/steps/record.md | 8 +- service/stovepipe/server/main.go | 2 +- stovepipe/controller/record/BUILD.bazel | 14 +- stovepipe/controller/record/dlq.go | 120 ++++++++++++++++ stovepipe/controller/record/dlq_test.go | 151 +++++++++++++++++++++ stovepipe/controller/record/record.go | 14 +- stovepipe/controller/record/record_test.go | 5 +- stovepipe/entity/request_log.go | 4 +- stovepipe/entity/request_log_test.go | 9 ++ 10 files changed, 320 insertions(+), 12 deletions(-) create mode 100644 stovepipe/controller/record/dlq.go create mode 100644 stovepipe/controller/record/dlq_test.go diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index d50eee30..a3b9fe1d 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -6,7 +6,8 @@ Stovepipe retains an append-only request log for each validation request. Its in - `build_triggered`; - `build_finished`; -- `validation_fact_recorded`. +- `validation_fact_recorded`; +- `promotion_failed`. 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. @@ -122,6 +123,7 @@ Immutable Request context such as URI, build strategy, and base URI remains on ` | `build_triggered` | A runner accepted a build and its Build row became durable. | Build ID metadata and creation time | | `build_finished` | The Build first reached a write-once terminal status. | Build ID metadata and status-change time | | `validation_fact_recorded` | The immutable whole-repository fact became durable. | Degree metadata and fact creation time | +| `promotion_failed` | The green commit could not be promoted and the attempt was abandoned. | 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. Project facts remain outside the initial vocabulary. @@ -208,6 +210,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo | Build | Create Build after runner acceptance, then retain `build_triggered`. | An identical existing Build ensures the event before buildsignal publication. | | 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`. | An identical fact owned by the Request ensures the event before bookmark or promotion work. | +| Record DLQ | Retain `promotion_failed` before abandoning a failed promotion. | The stable event ID makes history retention idempotent without repeating the promotion. | | 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/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 90e91300..6ec179af 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -202,7 +202,7 @@ Ordering is per-subject only and the subject is the Request, so events for *diff Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. -Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The `record_dlq` consumer is this same controller on the dead-letter topic, so it re-runs this identical idempotent algorithm and the republish is its own recovery path: the fact is already durable, and only the notification was outstanding. +Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The `record_dlq` consumer wraps this same controller and normally re-runs the idempotent algorithm, so the fact remains durable while an outstanding notification can recover. Promotion is the exception: once a promotion failure reaches `record_dlq`, it retains a `promotion_failed` request-history event, logs and counts the abandonment, and acknowledges without another outbound call. The durable green fact and bookmark remain authoritative, the recorded hook is not published, and a later green request or operator action may heal the promotion ref. ## Request lifecycle @@ -219,7 +219,7 @@ Phase 2 broadens "complete" to "all planned facts recorded", which needs a marke There is no `Update`. The first fact written for an identity is the permanent answer, and a caller that needs to know whether it won the race reads `ErrAlreadyExists` and then loads the winner. -The topic key, the message, and the consumer all exist. The DLQ consumer does not (see [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior)). +The topic key, message, primary consumer, and DLQ consumer all exist. | Topic key | Message | Producer | Consumer | Partition key | Message id | @@ -271,13 +271,13 @@ Every effect is recognize-and-skip, so a redelivery after a complete run re-runs ## DLQ and fail-closed behavior -**Neither** `record_dlq` **nor** `build_dlq` **has a consumer today, and both topics are already receiving messages.** Every primary subscription comes from `DefaultSubscriptionConfig`, which enables dead-lettering with the `_dlq` suffix, so a message that is rejected outright *or* runs out of retries moves to its stage's dead-letter topic. The wiring registers only `process_dlq` and `buildsignal_dlq`, so messages pile up unread on the other two. +Every primary subscription comes from `DefaultSubscriptionConfig`, which enables dead-lettering with the `_dlq` suffix, so a message that is rejected outright or runs out of retries moves to its stage's dead-letter topic. Stovepipe registers a reconciler for every pipeline DLQ, including `record_dlq`. Two different things put a message there, and only one is a poison payload. A delivery that fails with its retry budget spent is dead-lettered by the nack itself, carrying the reason it actually failed. A delivery that never reaches a nack, because it crashed or because its **ack failed** and the visibility timeout redelivered it, is dead-lettered by the poll loop once `retry_count` reaches `MaxAttempts` (3 by default), without the controller running on that final attempt and with only a generic reason recorded. So a missing reconciler exposes more than malformed messages: a fact can be lost to a storage failure that would have succeeded on a later retry, or to an ack that never landed even though the write did. Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, which costs the queue an incremental baseline and forces a full build at the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit. -This is the same failure shape [buildsignal.md](buildsignal.md#what-it-costs-when-a-backend-does-not-classify-status-errors) describes for a deployment that registers primary consumers without their reconciler. When the reconciler is built it should re-run this same idempotent algorithm from the request id, under `errs.AlwaysRetryableProcessor`: write and publish the immutable fact as usual if the Request carries a build outcome, keep retrying if Request storage is temporarily unavailable, and treat a malformed payload or a permanently missing Request as poison, which needs an operational alert rather than more retries. +The reconciler re-runs the same idempotent algorithm from the request id under `errs.AlwaysRetryableProcessor`: it writes and publishes the immutable fact as usual if the Request carries a build outcome and keeps retrying failures outside promotion. Promotion carries narrower policy because it is an outbound derived-cache update rather than durable validation state. `record_dlq` uses the promotion-stage attribution already stored in the failure to acknowledge a known promotion failure without calling the backend again. If reconciliation began for another failure and newly reaches a promotion error, it acknowledges that delivery too. This gives promotion only the primary consumer's retry policy and prevents one external failure from occupying the DLQ indefinitely. ## Future Items diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index ce5d2dcd..b1c472f0 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -530,7 +530,7 @@ func registerDLQControllers( } count++ - recordDLQController := record.NewController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") + recordDLQController := record.NewDLQController(record.NewController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")) if err := c.Register(recordDLQController); err != nil { return count, fmt.Errorf("failed to register record dlq controller: %w", err) } diff --git a/stovepipe/controller/record/BUILD.bazel b/stovepipe/controller/record/BUILD.bazel index d9d67c01..38f38982 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -2,13 +2,18 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["record.go"], + srcs = [ + "dlq.go", + "record.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/controller/record", visibility = ["//visibility:public"], deps = [ "//api/base/hook:go_default_library", + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", "//platform/hook:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/hookevent:go_default_library", @@ -25,13 +30,18 @@ go_library( go_test( name = "go_default_test", - srcs = ["record_test.go"], + srcs = [ + "dlq_test.go", + "record_test.go", + ], embed = [":go_default_library"], deps = [ "//api/base/hook:go_default_library", + "//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/extension/messagequeue/mock:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/hookevent:go_default_library", diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go new file mode 100644 index 00000000..343e23dc --- /dev/null +++ b/stovepipe/controller/record/dlq.go @@ -0,0 +1,120 @@ +// 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 record + +import ( + "context" + "fmt" + + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + "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" +) + +// DLQController abandons promotion failures while leaving all other record +// reconciliation failures under the DLQ consumer's normal retry policy. +type DLQController struct { + controller *Controller +} + +var _ consumer.Controller = (*DLQController)(nil) + +// NewDLQController wraps a record controller with promotion-specific DLQ policy. +func NewDLQController(controller *Controller) *DLQController { + return &DLQController{controller: controller} +} + +// Process never replays a known promotion failure. If reconciliation reaches +// promotion for a failure originally raised by another stage, it makes that one +// attempt and acknowledges a promotion failure rather than retrying it on the DLQ. +func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { + if originalFailure, failed := delivery.Failure(); failed && isPromotionFailure(originalFailure) { + return c.abandonPromotion(ctx, delivery, originalFailure, "dead_lettered") + } + + err := c.controller.Process(ctx, delivery) + if err == nil { + return nil + } + + currentFailure := errs.Attribution(err) + if !isPromotionFailure(currentFailure) { + return err + } + return c.abandonPromotion(ctx, delivery, currentFailure, "reconciliation_failed") +} + +func isPromotionFailure(f failure.Failure) bool { + stage, ok := f.Detail[failureDetailKeyRecordStage].(string) + return ok && stage == failureRecordStagePromotion +} + +func (c *DLQController) abandonPromotion(ctx context.Context, delivery consumer.Delivery, f failure.Failure, reason string) error { + if err := c.persistPromotionFailedHistory(ctx, delivery); err != nil { + return promotionFailure(fmt.Errorf("failed to persist promotion failure history: %w", err)) + } + + msg := delivery.Message() + metrics.NamedCounter(c.controller.metricsScope, _opName, "promotions_abandoned", 1, + metrics.TagsFromContext(ctx, metrics.NewTag("reason", reason))..., + ) + c.controller.logger.Errorw("abandoned promotion from record dlq", + "queue", msg.Tenant, + "message_id", msg.ID, + "attempt", delivery.Attempt(), + "reason", reason, + "error", f.Message, + ) + return nil +} + +func (c *DLQController) persistPromotionFailedHistory(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + rec := &stovepipemq.Record{} + if err := stovepipemq.Unmarshal(msg.Payload, rec); err != nil { + return fmt.Errorf("failed to deserialize record: %w", err) + } + if err := entityqueue.ValidatePayloadQueue(msg, rec.GetQueueName()); err != nil { + return fmt.Errorf("invalid message identity: %w", err) + } + store, err := c.controller.stores.For(storage.Config{QueueName: rec.GetQueueName()}) + if err != nil { + return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) + } + request, err := c.controller.loadRequest(ctx, store, rec.GetId()) + if err != nil { + return err + } + log := requestlog.NewRequestEventLog(request, entity.RequestEventPromotionFailed, "repository", nil) + if err := c.controller.materializer.PersistLog(ctx, store, log); err != nil { + return fmt.Errorf("failed to record promotion failure for request %s: %w", request.ID, err) + } + return nil +} + +// Name returns the wrapped record controller's name. +func (c *DLQController) Name() string { return c.controller.Name() } + +// TopicKey returns the wrapped record controller's topic key. +func (c *DLQController) TopicKey() consumer.TopicKey { return c.controller.TopicKey() } + +// ConsumerGroup returns the wrapped record controller's consumer group. +func (c *DLQController) ConsumerGroup() string { return c.controller.ConsumerGroup() } diff --git a/stovepipe/controller/record/dlq_test.go b/stovepipe/controller/record/dlq_test.go new file mode 100644 index 00000000..1a20d7e3 --- /dev/null +++ b/stovepipe/controller/record/dlq_test.go @@ -0,0 +1,151 @@ +// 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 record + +import ( + "context" + "errors" + "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" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "github.com/uber/submitqueue/platform/errs" + requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/mock/gomock" +) + +func TestDLQControllerSkipsDeadLetteredPromotion(t *testing.T) { + ctrl := gomock.NewController(t) + inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + c := NewDLQController(inner) + request := requestWithState(entity.RequestStateSucceeded) + mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + expectPromotionFailedHistory(t, ctrl, inner, mocks, false) + + delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ + Message: "permission denied", + Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, + }) + + require.NoError(t, c.Process(queueContext(), delivery)) + assert.NotContains(t, mocks.metricsScope.Snapshot().Counters(), "record_dlq_controller.record.promotions+queue=monorepo/main") + assertAbandonedPromotionCount(t, mocks, "dead_lettered") +} + +func TestDLQControllerAcknowledgesNewPromotionFailure(t *testing.T) { + ctrl := gomock.NewController(t) + inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + c := NewDLQController(inner) + expectGreenPromotionReplay(mocks, errors.New("unavailable")) + mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) + expectPromotionFailedHistory(t, ctrl, inner, mocks, true) + delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{}) + + require.NoError(t, c.Process(queueContext(), delivery)) + assertAbandonedPromotionCount(t, mocks, "reconciliation_failed") +} + +func TestDLQControllerLeavesOtherFailuresToDLQPolicy(t *testing.T) { + ctrl := gomock.NewController(t) + inner, _ := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + c := NewDLQController(inner) + delivery := newDLQDeliveryWithPayload(ctrl, 1, []byte("not protobuf json"), failure.Failure{}) + + require.Error(t, c.Process(queueContext(), delivery)) +} + +func TestDLQControllerPreservesPromotionAttributionWhenHistoryFails(t *testing.T) { + ctrl := gomock.NewController(t) + inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + c := NewDLQController(inner) + mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) + materializer := requestlogmock.NewMockMaterializer(ctrl) + inner.materializer = materializer + materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).Return(errors.New("db down")) + delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ + Message: "permission denied", + Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, + }) + + err := c.Process(queueContext(), delivery) + require.Error(t, err) + assert.Equal(t, failureRecordStagePromotion, errs.Attribution(err).Detail[failureDetailKeyRecordStage]) +} + +func expectPromotionFailedHistory(t *testing.T, ctrl *gomock.Controller, controller *Controller, mocks recordMocks, includeValidationFact bool) { + t.Helper() + materializer := requestlogmock.NewMockMaterializer(ctrl) + controller.materializer = materializer + var calls []any + if includeValidationFact { + calls = append(calls, materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event) + return nil + }, + )) + } + calls = append(calls, materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, entity.RequestEventPromotionFailed, log.Event) + assert.Equal(t, "event/promotion_failed/repository", log.ID) + assert.Equal(t, testID, log.RequestID) + assert.Empty(t, log.State) + assert.Empty(t, log.Metadata) + return nil + }, + )) + gomock.InOrder(calls...) +} + +func expectGreenPromotionReplay(mocks recordMocks, promotionErr error) { + mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) + mocks.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + mocks.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject).Return(entity.ValidationFact{ + URI: testURI, + Degree: entity.DegreeGreen, + RequestID: testID, + }, nil) + mocks.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(testURI, testID, 3), nil) + mocks.sourceControl.EXPECT().Promote(gomock.Any(), testURI).Return(promotionErr) +} + +func newDLQDelivery(t *testing.T, ctrl *gomock.Controller, attempt int, originalFailure failure.Failure) *consumermock.MockDelivery { + t.Helper() + return newDLQDeliveryWithPayload(ctrl, attempt, recordPayload(t, testID), originalFailure) +} + +func newDLQDeliveryWithPayload(ctrl *gomock.Controller, attempt int, payload []byte, originalFailure failure.Failure) *consumermock.MockDelivery { + delivery := consumermock.NewMockDelivery(ctrl) + msg := entityqueue.NewMessage(testID, payload, testID, nil) + msg.Tenant = testQueue + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(attempt).AnyTimes() + delivery.EXPECT().Failure().Return(originalFailure, true).AnyTimes() + return delivery +} + +func assertAbandonedPromotionCount(t *testing.T, mocks recordMocks, reason string) { + t.Helper() + counter, ok := mocks.metricsScope.Snapshot().Counters()["record_dlq_controller.record.promotions_abandoned+queue=monorepo/main,reason="+reason] + require.True(t, ok) + assert.EqualValues(t, 1, counter.Value()) +} diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 38fc8e71..9ab44922 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -40,6 +40,7 @@ import ( basehook "github.com/uber/submitqueue/api/base/hook" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + platformerrs "github.com/uber/submitqueue/platform/errs" platformhook "github.com/uber/submitqueue/platform/hook" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/stovepipe/core/hookevent" @@ -72,6 +73,11 @@ var _ consumer.Controller = (*Controller)(nil) // _opName is the metric operation name shared by every emit in this file. const _opName = "record" +const ( + failureDetailKeyRecordStage = "stovepipe.record.stage" + failureRecordStagePromotion = "promotion" +) + // 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. @@ -483,7 +489,7 @@ func (c *Controller) promote(ctx context.Context, request entity.Request) error metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, metrics.TagsFromContext(ctx, metrics.NewTag("stage", "resolve"))..., ) - return fmt.Errorf("failed to resolve source control for queue %s: %w", request.Queue, err) + return promotionFailure(fmt.Errorf("failed to resolve source control for queue %s: %w", request.Queue, err)) } if err := sc.Promote(ctx, request.URI); err != nil { @@ -502,7 +508,7 @@ func (c *Controller) promote(ctx context.Context, request entity.Request) error metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, metrics.TagsFromContext(ctx, metrics.NewTag("stage", "promote"))..., ) - return fmt.Errorf("failed to promote uri %s of queue %s: %w", request.URI, request.Queue, err) + return promotionFailure(fmt.Errorf("failed to promote uri %s of queue %s: %w", request.URI, request.Queue, err)) } metrics.NamedCounter(c.metricsScope, _opName, "promotions", 1, metrics.TagsFromContext(ctx)...) @@ -514,6 +520,10 @@ func (c *Controller) promote(ctx context.Context, request entity.Request) error return nil } +func promotionFailure(err error) error { + return platformerrs.Detail(err, map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}) +} + // publishHookEvent sends one event about request to the domain's hook topic. // // Called last, after the fact write and after both caches derived from it have diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 5f68fc5a..c789411c 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -27,6 +27,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "github.com/uber/submitqueue/platform/errs" mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/stovepipe/core/hookevent" @@ -741,7 +742,9 @@ func TestProcess_PromotionErrorsPropagate(t *testing.T) { // The bookmark already advanced, so the redelivery re-promotes the // same commit; failing here is what makes that retry happen. - require.Error(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID)))) + err := c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))) + require.Error(t, err) + assert.Equal(t, failureRecordStagePromotion, errs.Attribution(err).Detail[failureDetailKeyRecordStage]) }) } } diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index fe47a9aa..cf6087de 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -28,6 +28,8 @@ const ( RequestEventBuildFinished RequestEvent = "build_finished" // RequestEventValidationFactRecorded records that an immutable validation verdict was established. RequestEventValidationFactRecorded RequestEvent = "validation_fact_recorded" + // RequestEventPromotionFailed records that the request's green commit could not be promoted. + RequestEventPromotionFailed RequestEvent = "promotion_failed" ) // RequestOutcomeReason identifies the durable domain reason for a terminal request state. @@ -133,7 +135,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded: + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventPromotionFailed: 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 b44d0ed3..35c331f3 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: "promotion failed event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventPromotionFailed + entry.RequestVersion = 0 + 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}, From af6998244ef450b101145a926353d62d1b0e14ab Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 18 Sep 2026 20:44:05 +0000 Subject: [PATCH 2/4] refactor(stovepipe): make record DLQ standalone --- doc/rfc/stovepipe/steps/record.md | 2 +- service/stovepipe/server/main.go | 2 +- stovepipe/controller/record/dlq.go | 115 +++++++++++++++--------- stovepipe/controller/record/dlq_test.go | 51 ++++++++--- stovepipe/controller/record/record.go | 55 ++++++++---- 5 files changed, 153 insertions(+), 72 deletions(-) diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 6ec179af..6475ee97 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -202,7 +202,7 @@ Ordering is per-subject only and the subject is the Request, so events for *diff Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. -Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The `record_dlq` consumer wraps this same controller and normally re-runs the idempotent algorithm, so the fact remains durable while an outstanding notification can recover. Promotion is the exception: once a promotion failure reaches `record_dlq`, it retains a `promotion_failed` request-history event, logs and counts the abandonment, and acknowledges without another outbound call. The durable green fact and bookmark remain authoritative, the recorded hook is not published, and a later green request or operator action may heal the promotion ref. +Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The standalone `record_dlq` controller reconciles the same durable effects, so the fact remains durable while an outstanding notification can recover. Promotion is the exception: once a promotion failure reaches `record_dlq`, it retains a `promotion_failed` request-history event, logs and counts the abandonment, and acknowledges without another outbound call. The durable green fact and bookmark remain authoritative, the recorded hook is not published, and a later green request or operator action may heal the promotion ref. ## Request lifecycle diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index b1c472f0..a6d6b522 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -530,7 +530,7 @@ func registerDLQControllers( } count++ - recordDLQController := record.NewDLQController(record.NewController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")) + recordDLQController := record.NewDLQController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") if err := c.Register(recordDLQController); err != nil { return count, fmt.Errorf("failed to register record dlq controller: %w", err) } diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go index 343e23dc..f785f01b 100644 --- a/stovepipe/controller/record/dlq.go +++ b/stovepipe/controller/record/dlq.go @@ -18,6 +18,7 @@ import ( "context" "fmt" + "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" @@ -26,31 +27,74 @@ import ( 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/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" ) -// DLQController abandons promotion failures while leaving all other record -// reconciliation failures under the DLQ consumer's normal retry policy. +// DLQController reconciles record work without replaying known promotion failures. type DLQController struct { - controller *Controller + requestRecorder + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string } var _ consumer.Controller = (*DLQController)(nil) -// NewDLQController wraps a record controller with promotion-specific DLQ policy. -func NewDLQController(controller *Controller) *DLQController { - return &DLQController{controller: controller} +// NewDLQController creates a controller for record dead-letter reconciliation. +func NewDLQController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + materializer requestlog.Materializer, + sourceControl sourcecontrol.Factory, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) *DLQController { + name := string(topicKey) + "_controller" + return &DLQController{ + requestRecorder: newRequestRecorder(logger, scope, materializer, sourceControl, registry, name), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + } } -// Process never replays a known promotion failure. If reconciliation reaches -// promotion for a failure originally raised by another stage, it makes that one -// attempt and acknowledges a promotion failure rather than retrying it on the DLQ. +// Process reconstructs the record stage's durable effects. It never replays a +// known promotion failure; a newly encountered promotion failure is retained in +// request history and acknowledged rather than retried on the DLQ. func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + rec := &stovepipemq.Record{} + if err := stovepipemq.Unmarshal(msg.Payload, rec); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to deserialize record: %w", err) + } + if err := entityqueue.ValidatePayloadQueue(msg, rec.GetQueueName()); err != nil { + return fmt.Errorf("invalid message identity: %w", err) + } + store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) + } + request, err := loadRequest(ctx, store, rec.GetId()) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return err + } + if rec.GetQueueName() != "" && rec.GetQueueName() != request.Queue { + metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID) + } + if originalFailure, failed := delivery.Failure(); failed && isPromotionFailure(originalFailure) { - return c.abandonPromotion(ctx, delivery, originalFailure, "dead_lettered") + return c.abandonPromotion(ctx, delivery, store, request, originalFailure, "dead_lettered") } - err := c.controller.Process(ctx, delivery) + err = c.recordRequest(ctx, store, request) if err == nil { return nil } @@ -59,7 +103,7 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) if !isPromotionFailure(currentFailure) { return err } - return c.abandonPromotion(ctx, delivery, currentFailure, "reconciliation_failed") + return c.abandonPromotion(ctx, delivery, store, request, currentFailure, "reconciliation_failed") } func isPromotionFailure(f failure.Failure) bool { @@ -67,16 +111,23 @@ func isPromotionFailure(f failure.Failure) bool { return ok && stage == failureRecordStagePromotion } -func (c *DLQController) abandonPromotion(ctx context.Context, delivery consumer.Delivery, f failure.Failure, reason string) error { - if err := c.persistPromotionFailedHistory(ctx, delivery); err != nil { +func (c *DLQController) abandonPromotion( + ctx context.Context, + delivery consumer.Delivery, + store storage.Storage, + request entity.Request, + f failure.Failure, + reason string, +) error { + if err := c.persistPromotionFailedHistory(ctx, store, request); err != nil { return promotionFailure(fmt.Errorf("failed to persist promotion failure history: %w", err)) } msg := delivery.Message() - metrics.NamedCounter(c.controller.metricsScope, _opName, "promotions_abandoned", 1, + metrics.NamedCounter(c.metricsScope, _opName, "promotions_abandoned", 1, metrics.TagsFromContext(ctx, metrics.NewTag("reason", reason))..., ) - c.controller.logger.Errorw("abandoned promotion from record dlq", + c.logger.Errorw("abandoned promotion from record dlq", "queue", msg.Tenant, "message_id", msg.ID, "attempt", delivery.Attempt(), @@ -86,35 +137,19 @@ func (c *DLQController) abandonPromotion(ctx context.Context, delivery consumer. return nil } -func (c *DLQController) persistPromotionFailedHistory(ctx context.Context, delivery consumer.Delivery) error { - msg := delivery.Message() - rec := &stovepipemq.Record{} - if err := stovepipemq.Unmarshal(msg.Payload, rec); err != nil { - return fmt.Errorf("failed to deserialize record: %w", err) - } - if err := entityqueue.ValidatePayloadQueue(msg, rec.GetQueueName()); err != nil { - return fmt.Errorf("invalid message identity: %w", err) - } - store, err := c.controller.stores.For(storage.Config{QueueName: rec.GetQueueName()}) - if err != nil { - return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) - } - request, err := c.controller.loadRequest(ctx, store, rec.GetId()) - if err != nil { - return err - } +func (c *DLQController) persistPromotionFailedHistory(ctx context.Context, store storage.Storage, request entity.Request) error { log := requestlog.NewRequestEventLog(request, entity.RequestEventPromotionFailed, "repository", nil) - if err := c.controller.materializer.PersistLog(ctx, store, log); err != nil { + if err := c.materializer.PersistLog(ctx, store, log); err != nil { return fmt.Errorf("failed to record promotion failure for request %s: %w", request.ID, err) } return nil } -// Name returns the wrapped record controller's name. -func (c *DLQController) Name() string { return c.controller.Name() } +// Name returns the controller's name. +func (c *DLQController) Name() string { return string(c.topicKey) } -// TopicKey returns the wrapped record controller's topic key. -func (c *DLQController) TopicKey() consumer.TopicKey { return c.controller.TopicKey() } +// TopicKey returns the controller's topic key. +func (c *DLQController) TopicKey() consumer.TopicKey { return c.topicKey } -// ConsumerGroup returns the wrapped record controller's consumer group. -func (c *DLQController) ConsumerGroup() string { return c.controller.ConsumerGroup() } +// ConsumerGroup returns the controller's consumer group. +func (c *DLQController) ConsumerGroup() string { return c.consumerGroup } diff --git a/stovepipe/controller/record/dlq_test.go b/stovepipe/controller/record/dlq_test.go index 1a20d7e3..89fb62e9 100644 --- a/stovepipe/controller/record/dlq_test.go +++ b/stovepipe/controller/record/dlq_test.go @@ -30,15 +30,15 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/mock/gomock" + "go.uber.org/zap" ) func TestDLQControllerSkipsDeadLetteredPromotion(t *testing.T) { ctrl := gomock.NewController(t) - inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") - c := NewDLQController(inner) + c, mocks := newDLQControllerForTest(t, ctrl) request := requestWithState(entity.RequestStateSucceeded) mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) - expectPromotionFailedHistory(t, ctrl, inner, mocks, false) + expectPromotionFailedHistory(t, ctrl, c, mocks, false) delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ Message: "permission denied", @@ -52,21 +52,32 @@ func TestDLQControllerSkipsDeadLetteredPromotion(t *testing.T) { func TestDLQControllerAcknowledgesNewPromotionFailure(t *testing.T) { ctrl := gomock.NewController(t) - inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") - c := NewDLQController(inner) + c, mocks := newDLQControllerForTest(t, ctrl) expectGreenPromotionReplay(mocks, errors.New("unavailable")) - mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) - expectPromotionFailedHistory(t, ctrl, inner, mocks, true) + expectPromotionFailedHistory(t, ctrl, c, mocks, true) delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{}) require.NoError(t, c.Process(queueContext(), delivery)) assertAbandonedPromotionCount(t, mocks, "reconciliation_failed") } +func TestDLQControllerReconcilesDurableRecordEffects(t *testing.T) { + ctrl := gomock.NewController(t) + c, mocks := newDLQControllerForTest(t, ctrl) + mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateFailed), nil) + var fact entity.ValidationFact + mocks.expectFactCreated(&fact) + delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{}) + + require.NoError(t, c.Process(queueContext(), delivery)) + assert.Equal(t, entity.DegreeBroken, fact.Degree) + assert.Equal(t, testID, fact.RequestID) + assert.Len(t, mocks.hooks.events, 1) +} + func TestDLQControllerLeavesOtherFailuresToDLQPolicy(t *testing.T) { ctrl := gomock.NewController(t) - inner, _ := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") - c := NewDLQController(inner) + c, _ := newDLQControllerForTest(t, ctrl) delivery := newDLQDeliveryWithPayload(ctrl, 1, []byte("not protobuf json"), failure.Failure{}) require.Error(t, c.Process(queueContext(), delivery)) @@ -74,11 +85,10 @@ func TestDLQControllerLeavesOtherFailuresToDLQPolicy(t *testing.T) { func TestDLQControllerPreservesPromotionAttributionWhenHistoryFails(t *testing.T) { ctrl := gomock.NewController(t) - inner, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") - c := NewDLQController(inner) + c, mocks := newDLQControllerForTest(t, ctrl) mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) materializer := requestlogmock.NewMockMaterializer(ctrl) - inner.materializer = materializer + c.materializer = materializer materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).Return(errors.New("db down")) delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ Message: "permission denied", @@ -90,7 +100,7 @@ func TestDLQControllerPreservesPromotionAttributionWhenHistoryFails(t *testing.T assert.Equal(t, failureRecordStagePromotion, errs.Attribution(err).Detail[failureDetailKeyRecordStage]) } -func expectPromotionFailedHistory(t *testing.T, ctrl *gomock.Controller, controller *Controller, mocks recordMocks, includeValidationFact bool) { +func expectPromotionFailedHistory(t *testing.T, ctrl *gomock.Controller, controller *DLQController, mocks recordMocks, includeValidationFact bool) { t.Helper() materializer := requestlogmock.NewMockMaterializer(ctrl) controller.materializer = materializer @@ -116,6 +126,21 @@ func expectPromotionFailedHistory(t *testing.T, ctrl *gomock.Controller, control gomock.InOrder(calls...) } +func newDLQControllerForTest(t *testing.T, ctrl *gomock.Controller) (*DLQController, recordMocks) { + t.Helper() + fixture, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + return NewDLQController( + zap.NewNop().Sugar(), + mocks.metricsScope, + fixture.stores, + fixture.materializer, + fixture.sourceControl, + fixture.registry, + consumer.TopicKey("record_dlq"), + "stovepipe-record-dlq", + ), mocks +} + func expectGreenPromotionReplay(mocks recordMocks, promotionErr error) { mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) mocks.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 9ab44922..aaa2354b 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -57,14 +57,18 @@ import ( // when that fact is green advances the queue's last-green bookmark and promotes // the commit. Implements consumer.Controller. type Controller struct { + requestRecorder + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string +} + +type requestRecorder struct { logger *zap.SugaredLogger metricsScope tally.Scope - stores storage.Factory materializer requestlog.Materializer sourceControl sourcecontrol.Factory registry consumer.TopicRegistry - topicKey consumer.TopicKey - consumerGroup string } // Verify Controller implements consumer.Controller interface at compile time. @@ -96,14 +100,27 @@ func NewController( ) *Controller { name := string(topicKey) + "_controller" return &Controller{ + requestRecorder: newRequestRecorder(logger, scope, materializer, sourceControl, registry, name), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +func newRequestRecorder( + logger *zap.SugaredLogger, + scope tally.Scope, + materializer requestlog.Materializer, + sourceControl sourcecontrol.Factory, + registry consumer.TopicRegistry, + name string, +) requestRecorder { + return requestRecorder{ logger: logger.Named(name), metricsScope: scope.SubScope(name), - stores: stores, materializer: materializer, sourceControl: sourceControl, registry: registry, - topicKey: topicKey, - consumerGroup: consumerGroup, } } @@ -134,7 +151,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) } - request, err := c.loadRequest(ctx, store, rec.Id) + request, err := loadRequest(ctx, store, rec.Id) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return err @@ -147,6 +164,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID) } + return c.recordRequest(ctx, store, request) +} + +func (c *requestRecorder) recordRequest(ctx context.Context, store storage.Storage, request entity.Request) error { switch request.State { case entity.RequestStateSucceeded, entity.RequestStateFailed: fact, created, err := c.recordFact(ctx, store, request) @@ -182,7 +203,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } } -func (c *Controller) persistValidationFactRecordedLog( +func (c *requestRecorder) persistValidationFactRecordedLog( ctx context.Context, store storage.Storage, request entity.Request, @@ -207,7 +228,7 @@ func (c *Controller) persistValidationFactRecordedLog( // green fact advances the queue's bookmark and, when this request ends up holding // it, promotes the commit. A broken fact moves neither, and instead reports how // long the break it names went undetected. -func (c *Controller) applyFactToDerivedCaches( +func (c *requestRecorder) applyFactToDerivedCaches( ctx context.Context, store storage.Storage, request entity.Request, @@ -246,7 +267,7 @@ func (c *Controller) applyFactToDerivedCaches( // 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) { +func (c *requestRecorder) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, bool, error) { factStore := store.GetValidationFactStore() fact := entity.ValidationFact{ @@ -302,7 +323,7 @@ func (c *Controller) recordFact(ctx context.Context, store storage.Storage, requ // 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) { +func (c *requestRecorder) reportFailureDetectionLatency(ctx context.Context, request entity.Request) { strategyTag := metrics.NewTag("strategy", string(request.BuildStrategy)) // Only a strategy that validates a delta pins a base commit, so a full build has @@ -349,7 +370,7 @@ func (c *Controller) reportFailureDetectionLatency(ctx context.Context, request // 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(ctx context.Context, request entity.Request, step string, err error) { +func (c *requestRecorder) failureDetectionUnobserved(ctx context.Context, request entity.Request, step string, err error) { metrics.NamedCounter(c.metricsScope, _opName, "failure_detection_errors", 1, metrics.TagsFromContext(ctx, metrics.NewTag("step", step))..., ) @@ -381,7 +402,7 @@ func degreeFor(state entity.RequestState) float64 { // 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) (bool, error) { +func (c *requestRecorder) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) (bool, error) { queueStore := store.GetQueueStore() for { @@ -428,7 +449,7 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage // points at, once that bookmark is durable. Reporting is best-effort so an // observability failure cannot turn a successful record operation into a retry, // which is why each cause is counted and logged separately instead of returned. -func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { +func (c *requestRecorder) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, metrics.TagsFromContext(ctx)...) @@ -483,7 +504,7 @@ func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity. // green fact is durable. Promotion is idempotent, so a redelivery repeats it // harmlessly. A commit that a rewritten history dropped from the ref cannot be // promoted by any retry, so that case is counted and skipped rather than failed. -func (c *Controller) promote(ctx context.Context, request entity.Request) error { +func (c *requestRecorder) promote(ctx context.Context, request entity.Request) error { sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, @@ -533,7 +554,7 @@ func promotionFailure(err error) error { // // Partitioning by request id matches the record topic's own, carrying // per-request ordering across the seam. -func (c *Controller) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error { +func (c *requestRecorder) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error { if err := platformhook.Publish(ctx, c.registry, request.Queue, event, request.ID); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "hook_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("failed to announce %s for request %s: %w", event.GetType(), request.ID, err) @@ -567,7 +588,7 @@ func compareToBookmark(queue, candidate, current string) (int, error) { } // loadRequest loads the request by id. -func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { +func loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { return loader.ByID(ctx, id, store.GetRequestStore().Get, "request") } From 914032c6148e4a1e1fad1ecb5556ff33911a4265 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 18 Sep 2026 20:59:15 +0000 Subject: [PATCH 3/4] refactor(stovepipe): abandon failed record work --- doc/rfc/stovepipe/request-log.md | 6 +- doc/rfc/stovepipe/steps/record.md | 4 +- service/stovepipe/server/main.go | 6 +- service/stovepipe/server/main_test.go | 3 +- stovepipe/controller/record/dlq.go | 163 ++++++++------ stovepipe/controller/record/dlq_test.go | 283 +++++++++++++++--------- stovepipe/controller/record/record.go | 55 ++--- stovepipe/entity/request_log.go | 4 +- stovepipe/entity/request_log_test.go | 9 + 9 files changed, 306 insertions(+), 227 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index a3b9fe1d..4367c1e2 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -2,11 +2,12 @@ ## 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 three asynchronous milestones needed to explain those transitions and the public verdict: +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: - `build_triggered`; - `build_finished`; - `validation_fact_recorded`; +- `record_failed`; - `promotion_failed`. 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. @@ -123,6 +124,7 @@ Immutable Request context such as URI, build strategy, and base URI remains on ` | `build_triggered` | A runner accepted a build and its Build row became durable. | Build ID metadata and creation time | | `build_finished` | The Build first reached a write-once terminal status. | Build ID metadata and status-change time | | `validation_fact_recorded` | The immutable whole-repository fact became durable. | Degree metadata and fact creation time | +| `record_failed` | Record-stage work could not be completed and was abandoned after exhausting primary retries. | Event retention time | | `promotion_failed` | The green commit could not be promoted and the attempt was abandoned. | 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. Project facts remain outside the initial vocabulary. @@ -210,7 +212,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo | Build | Create Build after runner acceptance, then retain `build_triggered`. | An identical existing Build ensures the event before buildsignal publication. | | 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`. | An identical fact owned by the Request ensures the event before bookmark or promotion work. | -| Record DLQ | Retain `promotion_failed` before abandoning a failed promotion. | The stable event ID makes history retention idempotent without repeating the promotion. | +| Record DLQ | Retain `record_failed`, or the more specific `promotion_failed` when failure attribution identifies promotion, before abandoning the remaining record work. | Stable event IDs make history retention idempotent without replaying facts, bookmarks, promotion, or hooks. | | 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/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 6475ee97..dac90634 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -202,7 +202,7 @@ Ordering is per-subject only and the subject is the Request, so events for *diff Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. -Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. The standalone `record_dlq` controller reconciles the same durable effects, so the fact remains durable while an outstanding notification can recover. Promotion is the exception: once a promotion failure reaches `record_dlq`, it retains a `promotion_failed` request-history event, logs and counts the abandonment, and acknowledges without another outbound call. The durable green fact and bookmark remain authoritative, the recorded hook is not published, and a later green request or operator action may heal the promotion ref. +Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. Once primary retries are exhausted, `record_dlq` stops trying to complete the original work: it reloads the already-terminal Request, retains `record_failed` history (or `promotion_failed` when the failure was attributed to promotion), logs and counts the abandonment, and acknowledges. It does not create a fact, advance the bookmark, promote, or publish a hook. Any partial durable effects already written remain authoritative; absence of a fact remains fail-closed, and a later request or operator action may repair external state. Promotion is one example of this case: an outbound call rejected for persistent permissions is recorded and not attempted again from the DLQ. ## Request lifecycle @@ -277,7 +277,7 @@ Two different things put a message there, and only one is a poison payload. A de Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, which costs the queue an incremental baseline and forces a full build at the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit. -The reconciler re-runs the same idempotent algorithm from the request id under `errs.AlwaysRetryableProcessor`: it writes and publishes the immutable fact as usual if the Request carries a build outcome and keeps retrying failures outside promotion. Promotion carries narrower policy because it is an outbound derived-cache update rather than durable validation state. `record_dlq` uses the promotion-stage attribution already stored in the failure to acknowledge a known promotion failure without calling the backend again. If reconciliation began for another failure and newly reaches a promotion error, it acknowledges that delivery too. This gives promotion only the primary consumer's retry policy and prevents one external failure from occupying the DLQ indefinitely. +The reconciler does not re-run the stage. It loads the Request only to identify existing durable state and retain a stable failure event: `promotion_failed` when the delivery's structured attribution identifies promotion, otherwise `record_failed`. The DLQ consumer uses `errs.AlwaysRetryableProcessor`, so a transient Request read or history persistence failure keeps retrying until that observable abandonment is durable. A malformed payload, invalid or unresolvable queue identity, missing Request, or Request from another queue cannot be repaired by redelivery; those cases are logged, counted, and acknowledged so poison cannot occupy the DLQ indefinitely. No DLQ path creates a fact, advances a bookmark, invokes source control, or publishes a hook. ## Future Items diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index a6d6b522..be4bf2ed 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -337,7 +337,7 @@ func run() error { if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer) if err != nil { return err } @@ -507,8 +507,6 @@ func registerDLQControllers( scope tally.Scope, store storage.Factory, materializer requestlog.Materializer, - registry consumer.TopicRegistry, - sourceControl sourcecontrol.Factory, ) (int, error) { var count int @@ -530,7 +528,7 @@ func registerDLQControllers( } count++ - recordDLQController := record.NewDLQController(logger, scope, store, materializer, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") + recordDLQController := record.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") if err := c.Register(recordDLQController); err != nil { return count, fmt.Errorf("failed to register record dlq controller: %w", err) } diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index e8f1d606..a55b61a2 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -177,8 +177,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}) require.NoError(t, err) - _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry, - fakeSourceControlFactory{}) + _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope)) require.NoError(t, err) return registry, primary.controllers, deadLetter.controllers diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go index f785f01b..516877a6 100644 --- a/stovepipe/controller/record/dlq.go +++ b/stovepipe/controller/record/dlq.go @@ -16,133 +16,160 @@ package record import ( "context" + "errors" "fmt" "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" - "github.com/uber/submitqueue/platform/errs" "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/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) -// DLQController reconciles record work without replaying known promotion failures. +const _dlqOpName = "record_dlq" + +// DLQController records that record-stage work was abandoned without replaying +// any of the stage's durable writes or outbound calls. type DLQController struct { - requestRecorder + logger *zap.SugaredLogger + metricsScope tally.Scope stores storage.Factory + materializer requestlog.Materializer topicKey consumer.TopicKey consumerGroup string } var _ consumer.Controller = (*DLQController)(nil) -// NewDLQController creates a controller for record dead-letter reconciliation. +// NewDLQController creates a controller for abandoned record work. func NewDLQController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, - sourceControl sourcecontrol.Factory, - registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *DLQController { name := string(topicKey) + "_controller" return &DLQController{ - requestRecorder: newRequestRecorder(logger, scope, materializer, sourceControl, registry, name), - stores: stores, - topicKey: topicKey, - consumerGroup: consumerGroup, + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + materializer: materializer, + topicKey: topicKey, + consumerGroup: consumerGroup, } } -// Process reconstructs the record stage's durable effects. It never replays a -// known promotion failure; a newly encountered promotion failure is retained in -// request history and acknowledged rather than retried on the DLQ. +// Process retains an observable abandonment occurrence for record work that the +// primary consumer could not finish. Deterministic poison is acknowledged; +// failures reading durable state or retaining history are returned for retry. func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() rec := &stovepipemq.Record{} if err := stovepipemq.Unmarshal(msg.Payload, rec); err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("failed to deserialize record: %w", err) + metrics.NamedCounter(c.metricsScope, _dlqOpName, "deserialize_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding malformed record dlq message", + "message_id", msg.ID, + "error", err, + ) + return nil } if err := entityqueue.ValidatePayloadQueue(msg, rec.GetQueueName()); err != nil { - return fmt.Errorf("invalid message identity: %w", err) + metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_identity_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding record dlq message with invalid queue identity", + "message_id", msg.ID, + "request_id", rec.GetId(), + "error", err, + ) + return nil } - store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()}) - if err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) + if rec.GetId() == "" { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "empty_id_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding record dlq message with empty request id", + "message_id", msg.ID, + "queue", rec.GetQueueName(), + ) + return nil } - request, err := loadRequest(ctx, store, rec.GetId()) + + store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()}) if err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) - return err - } - if rec.GetQueueName() != "" && rec.GetQueueName() != request.Queue { - metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID) + metrics.NamedCounter(c.metricsScope, _dlqOpName, "storage_resolve_errors", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding record dlq message for unresolvable queue", + "message_id", msg.ID, + "request_id", rec.GetId(), + "queue", rec.GetQueueName(), + "error", err, + ) + return nil } - if originalFailure, failed := delivery.Failure(); failed && isPromotionFailure(originalFailure) { - return c.abandonPromotion(ctx, delivery, store, request, originalFailure, "dead_lettered") + request, err := store.GetRequestStore().Get(ctx, rec.GetId()) + 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 record dlq message for missing request", + "message_id", msg.ID, + "request_id", rec.GetId(), + "queue", rec.GetQueueName(), + ) + return nil + } + metrics.NamedCounter(c.metricsScope, _dlqOpName, "request_store_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to load request %s for record dlq: %w", rec.GetId(), err) } - - err = c.recordRequest(ctx, store, request) - if err == nil { + if request.Queue != rec.GetQueueName() { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_mismatch", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding record dlq message whose request belongs to another queue", + "message_id", msg.ID, + "request_id", request.ID, + "payload_queue", rec.GetQueueName(), + "request_queue", request.Queue, + ) return nil } - currentFailure := errs.Attribution(err) - if !isPromotionFailure(currentFailure) { - return err - } - return c.abandonPromotion(ctx, delivery, store, request, currentFailure, "reconciliation_failed") -} - -func isPromotionFailure(f failure.Failure) bool { - stage, ok := f.Detail[failureDetailKeyRecordStage].(string) - return ok && stage == failureRecordStagePromotion -} - -func (c *DLQController) abandonPromotion( - ctx context.Context, - delivery consumer.Delivery, - store storage.Storage, - request entity.Request, - f failure.Failure, - reason string, -) error { - if err := c.persistPromotionFailedHistory(ctx, store, request); err != nil { - return promotionFailure(fmt.Errorf("failed to persist promotion failure history: %w", err)) + originalFailure, hasFailure := delivery.Failure() + event := recordFailureEvent(originalFailure) + log := requestlog.NewRequestEventLog(request, event, "repository", nil) + 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 abandoned record work for request %s: %w", request.ID, err) } - msg := delivery.Message() - metrics.NamedCounter(c.metricsScope, _opName, "promotions_abandoned", 1, - metrics.TagsFromContext(ctx, metrics.NewTag("reason", reason))..., + metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, + metrics.TagsFromContext(ctx, metrics.NewTag("event", string(event)))..., ) - c.logger.Errorw("abandoned promotion from record dlq", - "queue", msg.Tenant, + fields := []any{ "message_id", msg.ID, + "request_id", request.ID, + "queue", request.Queue, + "request_state", request.State, + "history_event", event, "attempt", delivery.Attempt(), - "reason", reason, - "error", f.Message, - ) + } + if hasFailure { + fields = append(fields, + "failure", originalFailure.Message, + "failure_subjects", originalFailure.Subjects, + "failure_detail", originalFailure.Detail, + ) + } + c.logger.Errorw("abandoned record work after retaining failure history", fields...) return nil } -func (c *DLQController) persistPromotionFailedHistory(ctx context.Context, store storage.Storage, request entity.Request) error { - log := requestlog.NewRequestEventLog(request, entity.RequestEventPromotionFailed, "repository", nil) - if err := c.materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("failed to record promotion failure for request %s: %w", request.ID, err) +func recordFailureEvent(f failure.Failure) entity.RequestEvent { + if stage, ok := f.Detail[failureDetailKeyRecordStage].(string); ok && stage == failureRecordStagePromotion { + return entity.RequestEventPromotionFailed } - return nil + return entity.RequestEventRecordFailed } // Name returns the controller's name. diff --git a/stovepipe/controller/record/dlq_test.go b/stovepipe/controller/record/dlq_test.go index 89fb62e9..ef5810ca 100644 --- a/stovepipe/controller/record/dlq_test.go +++ b/stovepipe/controller/record/dlq_test.go @@ -21,156 +21,219 @@ import ( "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" - "github.com/uber/submitqueue/platform/errs" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" 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" ) -func TestDLQControllerSkipsDeadLetteredPromotion(t *testing.T) { - ctrl := gomock.NewController(t) - c, mocks := newDLQControllerForTest(t, ctrl) - request := requestWithState(entity.RequestStateSucceeded) - mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) - expectPromotionFailedHistory(t, ctrl, c, mocks, false) - - delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ - Message: "permission denied", - Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, - }) - - require.NoError(t, c.Process(queueContext(), delivery)) - assert.NotContains(t, mocks.metricsScope.Snapshot().Counters(), "record_dlq_controller.record.promotions+queue=monorepo/main") - assertAbandonedPromotionCount(t, mocks, "dead_lettered") +type dlqMocks struct { + factory *storagemock.MockFactory + store *storagemock.MockStorage + requestStore *storagemock.MockRequestStore + materializer *requestlogmock.MockMaterializer + metricsScope tally.TestScope } -func TestDLQControllerAcknowledgesNewPromotionFailure(t *testing.T) { - ctrl := gomock.NewController(t) - c, mocks := newDLQControllerForTest(t, ctrl) - expectGreenPromotionReplay(mocks, errors.New("unavailable")) - expectPromotionFailedHistory(t, ctrl, c, mocks, true) - delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{}) +func TestDLQControllerIdentity(t *testing.T) { + c, _ := newDLQControllerForTest(t, gomock.NewController(t)) - require.NoError(t, c.Process(queueContext(), delivery)) - assertAbandonedPromotionCount(t, mocks, "reconciliation_failed") + assert.Equal(t, "record_dlq", c.Name()) + assert.Equal(t, consumer.TopicKey("record_dlq"), c.TopicKey()) + assert.Equal(t, "stovepipe-record-dlq", c.ConsumerGroup()) } -func TestDLQControllerReconcilesDurableRecordEffects(t *testing.T) { - ctrl := gomock.NewController(t) - c, mocks := newDLQControllerForTest(t, ctrl) - mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateFailed), nil) - var fact entity.ValidationFact - mocks.expectFactCreated(&fact) - delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{}) - - require.NoError(t, c.Process(queueContext(), delivery)) - assert.Equal(t, entity.DegreeBroken, fact.Degree) - assert.Equal(t, testID, fact.RequestID) - assert.Len(t, mocks.hooks.events, 1) +func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { + tests := []struct { + name string + failure failure.Failure + hasFailure bool + wantEvent entity.RequestEvent + }{ + { + name: "promotion failure", + failure: failure.Failure{ + Message: "permission denied", + Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, + }, + hasFailure: true, + wantEvent: entity.RequestEventPromotionFailed, + }, + { + name: "other record failure", + failure: failure.Failure{Message: "hook publish failed"}, + hasFailure: true, + wantEvent: entity.RequestEventRecordFailed, + }, + { + name: "missing failure attribution", + wantEvent: entity.RequestEventRecordFailed, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newDLQControllerForTest(t, ctrl) + expectDLQRequestLoad(m, requestWithState(entity.RequestStateSucceeded), nil) + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Equal(t, tt.wantEvent, log.Event) + assert.Equal(t, "event/"+string(tt.wantEvent)+"/repository", log.ID) + assert.Equal(t, testID, log.RequestID) + assert.Empty(t, log.State) + assert.Empty(t, log.Metadata) + return nil + }, + ) + + require.NoError(t, c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, tt.failure, tt.hasFailure))) + + counterName := "record_dlq_controller.record_dlq.requests_abandoned+event=" + string(tt.wantEvent) + ",queue=monorepo/main" + counter, ok := m.metricsScope.Snapshot().Counters()[counterName] + require.True(t, ok) + assert.EqualValues(t, 1, counter.Value()) + }) + } } -func TestDLQControllerLeavesOtherFailuresToDLQPolicy(t *testing.T) { - ctrl := gomock.NewController(t) - c, _ := newDLQControllerForTest(t, ctrl) - delivery := newDLQDeliveryWithPayload(ctrl, 1, []byte("not protobuf json"), failure.Failure{}) +func TestDLQControllerRetriesDurableStateFailures(t *testing.T) { + tests := []struct { + name string + setup func(dlqMocks) + }{ + { + name: "request load", + setup: func(m dlqMocks) { + expectDLQRequestLoad(m, entity.Request{}, errors.New("db down")) + }, + }, + { + name: "history persistence", + setup: func(m dlqMocks) { + expectDLQRequestLoad(m, requestWithState(entity.RequestStateSucceeded), nil) + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(errors.New("db down")) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newDLQControllerForTest(t, ctrl) + tt.setup(m) - require.Error(t, c.Process(queueContext(), delivery)) + err := c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, failure.Failure{}, false)) + require.Error(t, err) + }) + } } -func TestDLQControllerPreservesPromotionAttributionWhenHistoryFails(t *testing.T) { +func TestDLQControllerAcknowledgesMissingRequest(t *testing.T) { ctrl := gomock.NewController(t) - c, mocks := newDLQControllerForTest(t, ctrl) - mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) - materializer := requestlogmock.NewMockMaterializer(ctrl) - c.materializer = materializer - materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).Return(errors.New("db down")) - delivery := newDLQDelivery(t, ctrl, 1, failure.Failure{ - Message: "permission denied", - Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, - }) - - err := c.Process(queueContext(), delivery) - require.Error(t, err) - assert.Equal(t, failureRecordStagePromotion, errs.Attribution(err).Detail[failureDetailKeyRecordStage]) + c, m := newDLQControllerForTest(t, ctrl) + expectDLQRequestLoad(m, entity.Request{}, storage.ErrNotFound) + + require.NoError(t, c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, failure.Failure{}, false))) } -func expectPromotionFailedHistory(t *testing.T, ctrl *gomock.Controller, controller *DLQController, mocks recordMocks, includeValidationFact bool) { - t.Helper() - materializer := requestlogmock.NewMockMaterializer(ctrl) - controller.materializer = materializer - var calls []any - if includeValidationFact { - calls = append(calls, materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).DoAndReturn( - func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - assert.Equal(t, entity.RequestEventValidationFactRecorded, log.Event) - return nil +func TestDLQControllerAcknowledgesUnresolvableMessages(t *testing.T) { + otherQueuePayload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: testID, QueueName: "monorepo/other"}) + require.NoError(t, err) + emptyIDPayload, err := stovepipemq.Marshal(&stovepipemq.Record{QueueName: testQueue}) + require.NoError(t, err) + + tests := []struct { + name string + payload []byte + tenant string + setup func(dlqMocks) + }{ + {name: "malformed payload", payload: []byte("not protobuf json"), tenant: testQueue}, + {name: "queue identity mismatch", payload: otherQueuePayload, tenant: testQueue}, + {name: "empty request id", payload: emptyIDPayload, tenant: testQueue}, + { + name: "unresolvable queue", + payload: recordPayload(t, testID), + tenant: testQueue, + setup: func(m dlqMocks) { + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(nil, errors.New("unknown queue")) }, - )) - } - calls = append(calls, materializer.EXPECT().PersistLog(gomock.Any(), mocks.store, gomock.Any()).DoAndReturn( - func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - assert.Equal(t, entity.RequestEventPromotionFailed, log.Event) - assert.Equal(t, "event/promotion_failed/repository", log.ID) - assert.Equal(t, testID, log.RequestID) - assert.Empty(t, log.State) - assert.Empty(t, log.Metadata) - return nil }, - )) - gomock.InOrder(calls...) + { + name: "stored request queue mismatch", + payload: recordPayload(t, testID), + tenant: testQueue, + setup: func(m dlqMocks) { + request := requestWithState(entity.RequestStateSucceeded) + request.Queue = "monorepo/other" + expectDLQRequestLoad(m, request, nil) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newDLQControllerForTest(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } + + require.NoError(t, c.Process(queueContext(), newDLQDelivery(t, ctrl, tt.payload, tt.tenant, failure.Failure{}, false))) + }) + } } -func newDLQControllerForTest(t *testing.T, ctrl *gomock.Controller) (*DLQController, recordMocks) { +func newDLQControllerForTest(t *testing.T, ctrl *gomock.Controller) (*DLQController, dlqMocks) { t.Helper() - fixture, mocks := newControllerForTopic(t, ctrl, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + scope := tally.NewTestScope("", nil) + m := dlqMocks{ + factory: storagemock.NewMockFactory(ctrl), + store: storagemock.NewMockStorage(ctrl), + requestStore: storagemock.NewMockRequestStore(ctrl), + materializer: requestlogmock.NewMockMaterializer(ctrl), + metricsScope: scope, + } + m.store.EXPECT().GetRequestStore().Return(m.requestStore).AnyTimes() + return NewDLQController( zap.NewNop().Sugar(), - mocks.metricsScope, - fixture.stores, - fixture.materializer, - fixture.sourceControl, - fixture.registry, + scope, + m.factory, + m.materializer, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq", - ), mocks + ), m } -func expectGreenPromotionReplay(mocks recordMocks, promotionErr error) { - mocks.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSucceeded), nil) - mocks.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) - mocks.factStore.EXPECT().Get(gomock.Any(), testURI, wholeRepositoryProject).Return(entity.ValidationFact{ - URI: testURI, - Degree: entity.DegreeGreen, - RequestID: testID, - }, nil) - mocks.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(testURI, testID, 3), nil) - mocks.sourceControl.EXPECT().Promote(gomock.Any(), testURI).Return(promotionErr) +func expectDLQRequestLoad(m dlqMocks, request entity.Request, err error) { + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + m.requestStore.EXPECT().Get(gomock.Any(), testID).Return(request, err) } -func newDLQDelivery(t *testing.T, ctrl *gomock.Controller, attempt int, originalFailure failure.Failure) *consumermock.MockDelivery { +func newDLQDelivery( + t *testing.T, + ctrl *gomock.Controller, + payload []byte, + tenant string, + originalFailure failure.Failure, + hasFailure bool, +) *consumermock.MockDelivery { t.Helper() - return newDLQDeliveryWithPayload(ctrl, attempt, recordPayload(t, testID), originalFailure) -} - -func newDLQDeliveryWithPayload(ctrl *gomock.Controller, attempt int, payload []byte, originalFailure failure.Failure) *consumermock.MockDelivery { delivery := consumermock.NewMockDelivery(ctrl) msg := entityqueue.NewMessage(testID, payload, testID, nil) - msg.Tenant = testQueue + msg.Tenant = tenant delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Attempt().Return(attempt).AnyTimes() - delivery.EXPECT().Failure().Return(originalFailure, true).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + delivery.EXPECT().Failure().Return(originalFailure, hasFailure).AnyTimes() return delivery } - -func assertAbandonedPromotionCount(t *testing.T, mocks recordMocks, reason string) { - t.Helper() - counter, ok := mocks.metricsScope.Snapshot().Counters()["record_dlq_controller.record.promotions_abandoned+queue=monorepo/main,reason="+reason] - require.True(t, ok) - assert.EqualValues(t, 1, counter.Value()) -} diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index aaa2354b..9ab44922 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -57,18 +57,14 @@ import ( // when that fact is green advances the queue's last-green bookmark and promotes // the commit. Implements consumer.Controller. type Controller struct { - requestRecorder - stores storage.Factory - topicKey consumer.TopicKey - consumerGroup string -} - -type requestRecorder struct { logger *zap.SugaredLogger metricsScope tally.Scope + stores storage.Factory materializer requestlog.Materializer sourceControl sourcecontrol.Factory registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string } // Verify Controller implements consumer.Controller interface at compile time. @@ -100,27 +96,14 @@ func NewController( ) *Controller { name := string(topicKey) + "_controller" return &Controller{ - requestRecorder: newRequestRecorder(logger, scope, materializer, sourceControl, registry, name), - stores: stores, - topicKey: topicKey, - consumerGroup: consumerGroup, - } -} - -func newRequestRecorder( - logger *zap.SugaredLogger, - scope tally.Scope, - materializer requestlog.Materializer, - sourceControl sourcecontrol.Factory, - registry consumer.TopicRegistry, - name string, -) requestRecorder { - return requestRecorder{ logger: logger.Named(name), metricsScope: scope.SubScope(name), + stores: stores, materializer: materializer, sourceControl: sourceControl, registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, } } @@ -151,7 +134,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("failed to resolve storage for queue %q: %w", rec.GetQueueName(), err) } - request, err := loadRequest(ctx, store, rec.Id) + request, err := c.loadRequest(ctx, store, rec.Id) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return err @@ -164,10 +147,6 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID) } - return c.recordRequest(ctx, store, request) -} - -func (c *requestRecorder) recordRequest(ctx context.Context, store storage.Storage, request entity.Request) error { switch request.State { case entity.RequestStateSucceeded, entity.RequestStateFailed: fact, created, err := c.recordFact(ctx, store, request) @@ -203,7 +182,7 @@ func (c *requestRecorder) recordRequest(ctx context.Context, store storage.Stora } } -func (c *requestRecorder) persistValidationFactRecordedLog( +func (c *Controller) persistValidationFactRecordedLog( ctx context.Context, store storage.Storage, request entity.Request, @@ -228,7 +207,7 @@ func (c *requestRecorder) persistValidationFactRecordedLog( // green fact advances the queue's bookmark and, when this request ends up holding // it, promotes the commit. A broken fact moves neither, and instead reports how // long the break it names went undetected. -func (c *requestRecorder) applyFactToDerivedCaches( +func (c *Controller) applyFactToDerivedCaches( ctx context.Context, store storage.Storage, request entity.Request, @@ -267,7 +246,7 @@ func (c *requestRecorder) applyFactToDerivedCaches( // 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 *requestRecorder) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, bool, error) { +func (c *Controller) recordFact(ctx context.Context, store storage.Storage, request entity.Request) (entity.ValidationFact, bool, error) { factStore := store.GetValidationFactStore() fact := entity.ValidationFact{ @@ -323,7 +302,7 @@ func (c *requestRecorder) recordFact(ctx context.Context, store storage.Storage, // 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 *requestRecorder) reportFailureDetectionLatency(ctx context.Context, request entity.Request) { +func (c *Controller) reportFailureDetectionLatency(ctx context.Context, request entity.Request) { strategyTag := metrics.NewTag("strategy", string(request.BuildStrategy)) // Only a strategy that validates a delta pins a base commit, so a full build has @@ -370,7 +349,7 @@ func (c *requestRecorder) reportFailureDetectionLatency(ctx context.Context, req // 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 *requestRecorder) failureDetectionUnobserved(ctx context.Context, request entity.Request, step string, err error) { +func (c *Controller) failureDetectionUnobserved(ctx context.Context, request entity.Request, step string, err error) { metrics.NamedCounter(c.metricsScope, _opName, "failure_detection_errors", 1, metrics.TagsFromContext(ctx, metrics.NewTag("step", step))..., ) @@ -402,7 +381,7 @@ func degreeFor(state entity.RequestState) float64 { // 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 *requestRecorder) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) (bool, error) { +func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage, request entity.Request) (bool, error) { queueStore := store.GetQueueStore() for { @@ -449,7 +428,7 @@ func (c *requestRecorder) advanceLastGreen(ctx context.Context, store storage.St // points at, once that bookmark is durable. Reporting is best-effort so an // observability failure cannot turn a successful record operation into a retry, // which is why each cause is counted and logged separately instead of returned. -func (c *requestRecorder) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { +func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, metrics.TagsFromContext(ctx)...) @@ -504,7 +483,7 @@ func (c *requestRecorder) emitLastGreenTimestamp(ctx context.Context, request en // green fact is durable. Promotion is idempotent, so a redelivery repeats it // harmlessly. A commit that a rewritten history dropped from the ref cannot be // promoted by any retry, so that case is counted and skipped rather than failed. -func (c *requestRecorder) promote(ctx context.Context, request entity.Request) error { +func (c *Controller) promote(ctx context.Context, request entity.Request) error { sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, @@ -554,7 +533,7 @@ func promotionFailure(err error) error { // // Partitioning by request id matches the record topic's own, carrying // per-request ordering across the seam. -func (c *requestRecorder) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error { +func (c *Controller) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error { if err := platformhook.Publish(ctx, c.registry, request.Queue, event, request.ID); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "hook_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("failed to announce %s for request %s: %w", event.GetType(), request.ID, err) @@ -588,7 +567,7 @@ func compareToBookmark(queue, candidate, current string) (int, error) { } // loadRequest loads the request by id. -func loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { +func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { return loader.ByID(ctx, id, store.GetRequestStore().Get, "request") } diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index cf6087de..b88dca58 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -28,6 +28,8 @@ const ( RequestEventBuildFinished RequestEvent = "build_finished" // RequestEventValidationFactRecorded records that an immutable validation verdict was established. RequestEventValidationFactRecorded RequestEvent = "validation_fact_recorded" + // RequestEventRecordFailed records that record-stage work could not be completed. + RequestEventRecordFailed RequestEvent = "record_failed" // RequestEventPromotionFailed records that the request's green commit could not be promoted. RequestEventPromotionFailed RequestEvent = "promotion_failed" ) @@ -135,7 +137,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventPromotionFailed: + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventRecordFailed, RequestEventPromotionFailed: 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 35c331f3..beb80b12 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: "record failed event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventRecordFailed + entry.RequestVersion = 0 + return entry + }, + }, { name: "promotion failed event", mutate: func(entry RequestLog) RequestLog { From a277228fe0767eec1282941ee9ade0691a799a2b Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 18 Sep 2026 21:22:29 +0000 Subject: [PATCH 4/4] refactor(stovepipe): consolidate record failure history --- doc/rfc/stovepipe/request-log.md | 10 ++++---- doc/rfc/stovepipe/steps/record.md | 4 ++-- stovepipe/controller/record/dlq.go | 17 +++++++------ stovepipe/controller/record/dlq_test.go | 29 ++++++++++++++++------- stovepipe/core/requestlog/materializer.go | 2 ++ stovepipe/entity/request_log.go | 4 +--- stovepipe/entity/request_log_test.go | 9 ------- 7 files changed, 37 insertions(+), 38 deletions(-) diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 4367c1e2..9c96058a 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -7,8 +7,7 @@ Stovepipe retains an append-only request log for each validation request. Its in - `build_triggered`; - `build_finished`; - `validation_fact_recorded`; -- `record_failed`; -- `promotion_failed`. +- `record_failed`. 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. @@ -100,7 +99,7 @@ Enums are strings with unknown sentinels, and the entity has no storage or trans The core column shape follows SubmitQueue's `RequestLog`, except Stovepipe omits SubmitQueue's redundant `Type` column. Context that is meaningful only for one occurrence kind remains in the JSON metadata map instead of adding sparse columns. -Metadata is never used for occurrence identity, filtering, or control flow. Nil and empty maps are equivalent. The initial entity and storage contract treats the map as opaque JSON; writer and projection work may later define and enforce keys such as `superseded_by_request_id`, `build_id`, and `fact_degree`. Producers must not store credentials, raw dependency errors, stack traces, or unbounded payloads. The initial public history API does not expose the raw map. +Metadata is never used for occurrence identity, filtering, or control flow. Nil and empty maps are equivalent. The initial entity and storage contract treats the map as opaque JSON; writer and projection work may later define and enforce keys such as `superseded_by_request_id`, `build_id`, `fact_degree`, and `record_stage`. Producers must not store credentials, raw dependency errors, stack traces, or unbounded payloads. The initial public history API does not expose the raw map. Immutable Request context such as URI, build strategy, and base URI remains on `Request` and is resolved there rather than copied into log records or history responses. Build status and version remain on `Build`; the triggered and finished event kinds plus the terminal Request state describe the lifecycle without duplicating Build snapshots. Diagnostic error codes remain in structured logs until a concrete public vocabulary is required. @@ -124,8 +123,7 @@ Immutable Request context such as URI, build strategy, and base URI remains on ` | `build_triggered` | A runner accepted a build and its Build row became durable. | Build ID metadata and creation time | | `build_finished` | The Build first reached a write-once terminal status. | Build ID metadata and status-change time | | `validation_fact_recorded` | The immutable whole-repository fact became durable. | Degree metadata and fact creation time | -| `record_failed` | Record-stage work could not be completed and was abandoned after exhausting primary retries. | Event retention time | -| `promotion_failed` | The green commit could not be promoted and the attempt was abandoned. | Event retention time | +| `record_failed` | Record-stage work could not be completed and was abandoned after exhausting primary retries. | Event retention time and recognized record substage metadata when available | Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Project facts remain outside the initial vocabulary. @@ -212,7 +210,7 @@ Request creation, Build changes, and fact creation use the same source-write, lo | Build | Create Build after runner acceptance, then retain `build_triggered`. | An identical existing Build ensures the event before buildsignal publication. | | 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`. | An identical fact owned by the Request ensures the event before bookmark or promotion work. | -| Record DLQ | Retain `record_failed`, or the more specific `promotion_failed` when failure attribution identifies promotion, before abandoning the remaining record work. | Stable event IDs make history retention idempotent without replaying facts, bookmarks, promotion, or hooks. | +| Record DLQ | Retain `record_failed` with recognized substage metadata when failure attribution provides it, then abandon the remaining record work. | The stable event ID makes history retention idempotent without replaying facts, bookmarks, promotion, or hooks. | | 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/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index dac90634..f5adfad1 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -202,7 +202,7 @@ Ordering is per-subject only and the subject is the Request, so events for *diff Absence of an event is not a signal. A Request abandoned before any build went terminal never reaches this stage, and a superseded one publishes nothing, so a consumer waiting for one event per ingested commit waits forever on those. Gating keeps treating "no recorded fact" as not green. The converse holds too — an event is not proof the code was tested, since a fail-closed Request can produce a broken fact without a build having failed. -Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. Once primary retries are exhausted, `record_dlq` stops trying to complete the original work: it reloads the already-terminal Request, retains `record_failed` history (or `promotion_failed` when the failure was attributed to promotion), logs and counts the abandonment, and acknowledges. It does not create a fact, advance the bookmark, promote, or publish a hook. Any partial durable effects already written remain authoritative; absence of a fact remains fail-closed, and a later request or operator action may repair external state. Promotion is one example of this case: an outbound call rejected for persistent permissions is recorded and not attempted again from the DLQ. +Hooks here must be idempotent on `id`, as everywhere. "Fire-and-forget" describes downstream consumption, not the publish: `record` never waits for a hook, but a failed *publish* fails the delivery. Per `[platform/errs](../../../../platform/errs/README.md)` rule 4 it is not wrapped retryable just because replaying it is convenient, so it dead-letters. Once primary retries are exhausted, `record_dlq` stops trying to complete the original work: it reloads the already-terminal Request, retains `record_failed` history with recognized substage metadata when available, logs and counts the abandonment, and acknowledges. It does not create a fact, advance the bookmark, promote, or publish a hook. Any partial durable effects already written remain authoritative; absence of a fact remains fail-closed, and a later request or operator action may repair external state. Promotion is one example of this case: an outbound call rejected for persistent permissions is retained as `record_failed` with `record_stage=promotion` and is not attempted again from the DLQ. ## Request lifecycle @@ -277,7 +277,7 @@ Two different things put a message there, and only one is a poison payload. A de Gating stays safe, because everything this stage can lose reads as not-green: a Request with no fact is indistinguishable from one not yet validated. What is lost is the *fact*. A green build whose fact write permanently failed leaves the URI looking unvalidated, which costs the queue an incremental baseline and forces a full build at the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit. -The reconciler does not re-run the stage. It loads the Request only to identify existing durable state and retain a stable failure event: `promotion_failed` when the delivery's structured attribution identifies promotion, otherwise `record_failed`. The DLQ consumer uses `errs.AlwaysRetryableProcessor`, so a transient Request read or history persistence failure keeps retrying until that observable abandonment is durable. A malformed payload, invalid or unresolvable queue identity, missing Request, or Request from another queue cannot be repaired by redelivery; those cases are logged, counted, and acknowledged so poison cannot occupy the DLQ indefinitely. No DLQ path creates a fact, advances a bookmark, invokes source control, or publishes a hook. +The reconciler does not re-run the stage. It loads the Request only to identify existing durable state and retain a stable `record_failed` event. When the delivery's structured attribution identifies a recognized substage, the event retains that bounded context as metadata; promotion is currently recorded as `record_stage=promotion`. The DLQ consumer uses `errs.AlwaysRetryableProcessor`, so a transient Request read or history persistence failure keeps retrying until that observable abandonment is durable. A malformed payload, invalid or unresolvable queue identity, missing Request, or Request from another queue cannot be repaired by redelivery; those cases are logged, counted, and acknowledged so poison cannot occupy the DLQ indefinitely. No DLQ path creates a fact, advances a bookmark, invokes source control, or publishes a hook. ## Future Items diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go index 516877a6..857d4c68 100644 --- a/stovepipe/controller/record/dlq.go +++ b/stovepipe/controller/record/dlq.go @@ -136,22 +136,21 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) } originalFailure, hasFailure := delivery.Failure() - event := recordFailureEvent(originalFailure) - log := requestlog.NewRequestEventLog(request, event, "repository", nil) + metadata := recordFailureMetadata(originalFailure) + log := requestlog.NewRequestEventLog(request, entity.RequestEventRecordFailed, "repository", metadata) 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 abandoned record work for request %s: %w", request.ID, err) } - metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, - metrics.TagsFromContext(ctx, metrics.NewTag("event", string(event)))..., - ) + metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, metrics.TagsFromContext(ctx)...) fields := []any{ "message_id", msg.ID, "request_id", request.ID, "queue", request.Queue, "request_state", request.State, - "history_event", event, + "history_event", entity.RequestEventRecordFailed, + "history_metadata", metadata, "attempt", delivery.Attempt(), } if hasFailure { @@ -165,11 +164,11 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) return nil } -func recordFailureEvent(f failure.Failure) entity.RequestEvent { +func recordFailureMetadata(f failure.Failure) map[string]string { if stage, ok := f.Detail[failureDetailKeyRecordStage].(string); ok && stage == failureRecordStagePromotion { - return entity.RequestEventPromotionFailed + return map[string]string{requestlog.MetadataKeyRecordStage: stage} } - return entity.RequestEventRecordFailed + return nil } // Name returns the controller's name. diff --git a/stovepipe/controller/record/dlq_test.go b/stovepipe/controller/record/dlq_test.go index ef5810ca..6a3db2c9 100644 --- a/stovepipe/controller/record/dlq_test.go +++ b/stovepipe/controller/record/dlq_test.go @@ -27,6 +27,7 @@ import ( "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" @@ -56,7 +57,7 @@ func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { name string failure failure.Failure hasFailure bool - wantEvent entity.RequestEvent + wantStage string }{ { name: "promotion failure", @@ -65,17 +66,23 @@ func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, }, hasFailure: true, - wantEvent: entity.RequestEventPromotionFailed, + wantStage: failureRecordStagePromotion, }, { name: "other record failure", failure: failure.Failure{Message: "hook publish failed"}, hasFailure: true, - wantEvent: entity.RequestEventRecordFailed, }, { - name: "missing failure attribution", - wantEvent: entity.RequestEventRecordFailed, + name: "unrecognized stage is omitted", + failure: failure.Failure{ + Message: "fact write failed", + Detail: map[string]any{failureDetailKeyRecordStage: "future_stage"}, + }, + hasFailure: true, + }, + { + name: "missing failure attribution", }, } @@ -86,18 +93,22 @@ func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { expectDLQRequestLoad(m, requestWithState(entity.RequestStateSucceeded), nil) m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { - assert.Equal(t, tt.wantEvent, log.Event) - assert.Equal(t, "event/"+string(tt.wantEvent)+"/repository", log.ID) + assert.Equal(t, entity.RequestEventRecordFailed, log.Event) + assert.Equal(t, "event/record_failed/repository", log.ID) assert.Equal(t, testID, log.RequestID) assert.Empty(t, log.State) - assert.Empty(t, log.Metadata) + if tt.wantStage == "" { + assert.Empty(t, log.Metadata) + } else { + assert.Equal(t, map[string]string{requestlog.MetadataKeyRecordStage: tt.wantStage}, log.Metadata) + } return nil }, ) require.NoError(t, c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, tt.failure, tt.hasFailure))) - counterName := "record_dlq_controller.record_dlq.requests_abandoned+event=" + string(tt.wantEvent) + ",queue=monorepo/main" + counterName := "record_dlq_controller.record_dlq.requests_abandoned+queue=monorepo/main" counter, ok := m.metricsScope.Snapshot().Counters()[counterName] require.True(t, ok) assert.EqualValues(t, 1, counter.Value()) diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go index 578aa603..2355e1fc 100644 --- a/stovepipe/core/requestlog/materializer.go +++ b/stovepipe/core/requestlog/materializer.go @@ -40,6 +40,8 @@ const ( MetadataKeyBuildID = "build_id" // MetadataKeyFactDegree records the degree established by a validation fact. MetadataKeyFactDegree = "fact_degree" + // MetadataKeyRecordStage identifies the record substage that failed when known. + MetadataKeyRecordStage = "record_stage" ) // Materializer persists request-log occurrences into their queue-scoped read model. diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index b88dca58..a2d3643a 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -30,8 +30,6 @@ const ( RequestEventValidationFactRecorded RequestEvent = "validation_fact_recorded" // RequestEventRecordFailed records that record-stage work could not be completed. RequestEventRecordFailed RequestEvent = "record_failed" - // RequestEventPromotionFailed records that the request's green commit could not be promoted. - RequestEventPromotionFailed RequestEvent = "promotion_failed" ) // RequestOutcomeReason identifies the durable domain reason for a terminal request state. @@ -137,7 +135,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventRecordFailed, RequestEventPromotionFailed: + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventRecordFailed: 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 beb80b12..62c7f7d2 100644 --- a/stovepipe/entity/request_log_test.go +++ b/stovepipe/entity/request_log_test.go @@ -135,15 +135,6 @@ func TestRequestLogValidate(t *testing.T) { return entry }, }, - { - name: "promotion failed event", - mutate: func(entry RequestLog) RequestLog { - entry.State = RequestStateUnknown - entry.Event = RequestEventPromotionFailed - entry.RequestVersion = 0 - 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},