diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index d50eee30..9c96058a 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`. +- `validation_fact_recorded`; +- `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. @@ -98,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. @@ -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 | +| `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. @@ -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 `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 90e91300..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. 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. 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 @@ -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 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/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index ce5d2dcd..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.NewController(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/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..857d4c68 --- /dev/null +++ b/stovepipe/controller/record/dlq.go @@ -0,0 +1,181 @@ +// 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" + "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/metrics" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +const _dlqOpName = "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 { + 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 abandoned record work. +func NewDLQController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + materializer requestlog.Materializer, + topicKey consumer.TopicKey, + consumerGroup string, +) *DLQController { + name := string(topicKey) + "_controller" + return &DLQController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + materializer: materializer, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process retains 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, _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 { + 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 + } + 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 + } + + store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()}) + if err != nil { + 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 + } + + 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) + } + 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 + } + + originalFailure, hasFailure := delivery.Failure() + 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)...) + fields := []any{ + "message_id", msg.ID, + "request_id", request.ID, + "queue", request.Queue, + "request_state", request.State, + "history_event", entity.RequestEventRecordFailed, + "history_metadata", metadata, + "attempt", delivery.Attempt(), + } + 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 recordFailureMetadata(f failure.Failure) map[string]string { + if stage, ok := f.Detail[failureDetailKeyRecordStage].(string); ok && stage == failureRecordStagePromotion { + return map[string]string{requestlog.MetadataKeyRecordStage: stage} + } + return nil +} + +// Name returns the controller's name. +func (c *DLQController) Name() string { return string(c.topicKey) } + +// TopicKey returns the controller's topic key. +func (c *DLQController) TopicKey() consumer.TopicKey { return c.topicKey } + +// 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 new file mode 100644 index 00000000..6a3db2c9 --- /dev/null +++ b/stovepipe/controller/record/dlq_test.go @@ -0,0 +1,250 @@ +// 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-go/tally" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +type dlqMocks struct { + factory *storagemock.MockFactory + store *storagemock.MockStorage + requestStore *storagemock.MockRequestStore + materializer *requestlogmock.MockMaterializer + metricsScope tally.TestScope +} + +func TestDLQControllerIdentity(t *testing.T) { + c, _ := newDLQControllerForTest(t, gomock.NewController(t)) + + 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 TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { + tests := []struct { + name string + failure failure.Failure + hasFailure bool + wantStage string + }{ + { + name: "promotion failure", + failure: failure.Failure{ + Message: "permission denied", + Detail: map[string]any{failureDetailKeyRecordStage: failureRecordStagePromotion}, + }, + hasFailure: true, + wantStage: failureRecordStagePromotion, + }, + { + name: "other record failure", + failure: failure.Failure{Message: "hook publish failed"}, + hasFailure: true, + }, + { + 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", + }, + } + + 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, entity.RequestEventRecordFailed, log.Event) + assert.Equal(t, "event/record_failed/repository", log.ID) + assert.Equal(t, testID, log.RequestID) + assert.Empty(t, log.State) + 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+queue=monorepo/main" + counter, ok := m.metricsScope.Snapshot().Counters()[counterName] + require.True(t, ok) + assert.EqualValues(t, 1, counter.Value()) + }) + } +} + +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) + + err := c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, failure.Failure{}, false)) + require.Error(t, err) + }) + } +} + +func TestDLQControllerAcknowledgesMissingRequest(t *testing.T) { + ctrl := gomock.NewController(t) + 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 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")) + }, + }, + { + 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, dlqMocks) { + t.Helper() + 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(), + scope, + m.factory, + m.materializer, + consumer.TopicKey("record_dlq"), + "stovepipe-record-dlq", + ), m +} + +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, + payload []byte, + tenant string, + originalFailure failure.Failure, + hasFailure bool, +) *consumermock.MockDelivery { + t.Helper() + delivery := consumermock.NewMockDelivery(ctrl) + msg := entityqueue.NewMessage(testID, payload, testID, nil) + msg.Tenant = tenant + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + delivery.EXPECT().Failure().Return(originalFailure, hasFailure).AnyTimes() + return delivery +} 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/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 fe47a9aa..a2d3643a 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" ) // 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, 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 b44d0ed3..62c7f7d2 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: "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},