diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index d50eee30a..4d866cf88 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_abandoned`. The model deliberately follows SubmitQueue's distinction between statuses describing where a request is and events describing important activity that does not move it. It remains a bounded request-lifecycle log rather than a generic event bus or an audit of every correlated operation. @@ -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_abandoned` | Record-stage work stopped after exhausting primary retries. | 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 `record_abandoned`, then acknowledge 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 90e913005..435cebeb1 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_abandoned` history, 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 not attempted again from the DLQ, and its diagnostic remains in structured logs. ## 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_abandoned` event. 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, or missing Request 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 da69e5405..d6e834e47 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -296,7 +296,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 } @@ -462,8 +462,6 @@ func registerDLQControllers( scope tally.Scope, store storage.Factory, materializer requestlog.Materializer, - registry consumer.TopicRegistry, - sourceControl sourcecontrol.Factory, ) (int, error) { var count int @@ -485,7 +483,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 fc784a878..d8f569ed8 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -61,8 +61,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 d9d67c015..17fbb8b85 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -2,7 +2,10 @@ 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 = [ @@ -25,10 +28,14 @@ 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", diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go new file mode 100644 index 000000000..c3a7fcde2 --- /dev/null +++ b/stovepipe/controller/record/dlq.go @@ -0,0 +1,160 @@ +// 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" + 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) + } + originalFailure, hasFailure := delivery.Failure() + log := requestlog.NewRequestEventLog(request, entity.RequestEventRecordAbandoned, "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) + } + + 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.RequestEventRecordAbandoned, + "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 +} + +// 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 000000000..b326ddad3 --- /dev/null +++ b/stovepipe/controller/record/dlq_test.go @@ -0,0 +1,222 @@ +// 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" + 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 + }{ + { + name: "promotion failure", + failure: failure.Failure{Message: "failed to promote: permission denied"}, + hasFailure: true, + }, + { + name: "other record failure", + failure: failure.Failure{Message: "hook publish failed"}, + 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.RequestEventRecordAbandoned, log.Event) + assert.Equal(t, "event/record_abandoned/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+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")) + }, + }, + } + + 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_test.go b/stovepipe/controller/record/record_test.go index 5f68fc5a4..887f93b1c 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -741,7 +741,8 @@ 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) }) } } diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index fe47a9aa1..43bf4bd9b 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" + // RequestEventRecordAbandoned records that record-stage work stopped after exhausting retries. + RequestEventRecordAbandoned RequestEvent = "record_abandoned" ) // 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, RequestEventRecordAbandoned: 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 b44d0ed3d..4dd57ac9e 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 = RequestEventRecordAbandoned + 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},