diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index a16820a3b..a4b3dc2ea 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -217,13 +217,13 @@ The `project_facts_recorded` event also projects its retained timestamp into `Re | 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`; after recording project facts, retain one `project_facts_recorded` event. | An identical whole-repository fact and project-fact batch ensure their events before bookmark or promotion work. | -| Record DLQ | Retain `record_abandoned`, then acknowledge the remaining record work. | The stable event ID makes history retention idempotent without replaying facts, bookmarks, promotion, or hooks. | +| Record DLQ | Retain `record_abandoned`; for failed Requests, create or verify the repository fact and retain `validation_fact_recorded` before publishing its hook. | Existing facts remain authoritative; stable log and hook IDs make replay idempotent without resolving projects or replaying promotion. | | Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. | | Reconciler | CAS an unrecoverable non-terminal Request to failed, then retain failed. | An existing terminal Request is repaired from its persisted outcome without relabeling it. | Build running and unchanged polls create no entry. A failed runner trigger that creates no Build creates no event. Cancelled and superseded requests create no validation fact. -No controller infers an outcome reason from `RequestStateFailed` alone, invents a broken fact, or records DLQ as the domain reason. +No controller infers an outcome reason from `RequestStateFailed` alone or records DLQ as the domain reason. Record DLQ uses that persisted state to recover a missing broken repository fact; it does not infer project verdicts. ### Repair trigger requirement diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index c8170d049..00df9f18e 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -207,7 +207,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_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. +Hooks here must be idempotent on `id`, as everywhere. A failed publish fails the primary delivery and follows the normal classifier and retry policy. After dead-lettering, `record_dlq` retains `record_abandoned` and ensures a failed Request has a repository fact and a `validation.repository.recorded` notification. Existing immutable facts remain authoritative. See [DLQ and fail-closed behavior](#dlq-and-fail-closed-behavior) for the recovery boundary. ## Request lifecycle @@ -280,9 +280,11 @@ Every primary subscription comes from `DefaultSubscriptionConfig`, which enables 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, while its successful-build baseline remains available to the next head. A lost notification joins that list, and unlike the fact it gets no second chance from a later commit. +The reconciler retains a stable `record_abandoned` event. For a Request already in `failed`, it creates a `DegreeBroken` whole-repository fact if absent, retains `validation_fact_recorded` history, and publishes `validation.repository.recorded` after both writes succeed. An existing fact owned by the same Request is adopted unchanged, including its degree and timestamp; a fact owned by another Request is an invariant error. Replays repair missing history and repeat the notification with the same event ID. -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. +This fallback does not resolve project attribution or mark project recording complete. A failed build establishes a repository failure even when no individual project verdict is known. Build-issue analysis and detectors remain independent. The reconciler does not change the Request state, advance the bookmark, promote, or release build slots. Other Request states retain abandonment history only, so a succeeded build whose fact could not be recorded still appears unvalidated. + +The DLQ consumer uses `errs.AlwaysRetryableProcessor`: Request reads, fact writes, history persistence, and notification publication retry until recovery succeeds. Malformed payloads, invalid or unresolvable queue identities, queue mismatches, and missing Requests are logged, counted, and acknowledged. ## Future Items diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index f13f8b8cb..8a9279fe3 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -262,7 +262,7 @@ func run() error { if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, materializer, registry) if err != nil { return err } @@ -438,6 +438,7 @@ func registerDLQControllers( scope tally.Scope, store storage.Factory, materializer requestlog.Materializer, + registry consumer.TopicRegistry, ) (int, error) { var count int @@ -465,7 +466,7 @@ func registerDLQControllers( } count++ - recordDLQController := record.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") + recordDLQController := record.NewDLQController(logger, scope, store, materializer, 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/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index a864d2305..8387c21ed 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -64,7 +64,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New()) require.NoError(t, err) - _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope)) + _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry) 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 edb3275ae..1f4ee39f6 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "dlq.go", + "facts.go", "record.go", ], importpath = "github.com/uber/submitqueue/stovepipe/controller/record", diff --git a/stovepipe/controller/record/dlq.go b/stovepipe/controller/record/dlq.go index c3a7fcde2..ea605edf9 100644 --- a/stovepipe/controller/record/dlq.go +++ b/stovepipe/controller/record/dlq.go @@ -23,6 +23,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/hookevent" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" @@ -32,13 +33,14 @@ import ( const _dlqOpName = "record_dlq" -// DLQController records that record-stage work was abandoned without replaying -// any of the stage's durable writes or outbound calls. +// DLQController retains abandoned record work and ensures failed requests have +// a repository fact without requiring project attribution. type DLQController struct { logger *zap.SugaredLogger metricsScope tally.Scope - stores storage.Factory materializer requestlog.Materializer + registry consumer.TopicRegistry + stores storage.Factory topicKey consumer.TopicKey consumerGroup string } @@ -51,6 +53,7 @@ func NewDLQController( scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *DLQController { @@ -58,8 +61,9 @@ func NewDLQController( return &DLQController{ logger: logger.Named(name), metricsScope: scope.SubScope(name), - stores: stores, materializer: materializer, + registry: registry, + stores: stores, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -67,7 +71,7 @@ func NewDLQController( // 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. +// failures recording facts, retaining history, or publishing are retried. func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() rec := &stovepipemq.Record{} @@ -123,6 +127,13 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) 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 rec.GetQueueName() != "" && rec.GetQueueName() != request.Queue { + metrics.NamedCounter(c.metricsScope, _dlqOpName, "queue_mismatch", 1, metrics.TagsFromContext(ctx)...) + c.logger.Errorw("discarding record dlq message with mismatched request queue", + "message_id", msg.ID, "request_id", request.ID, + "payload_queue", rec.GetQueueName(), "request_queue", request.Queue) + return nil + } originalFailure, hasFailure := delivery.Failure() log := requestlog.NewRequestEventLog(request, entity.RequestEventRecordAbandoned, "repository", nil) if err := c.materializer.PersistLog(ctx, store, log); err != nil { @@ -130,6 +141,19 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) return fmt.Errorf("failed to retain abandoned record work for request %s: %w", request.ID, err) } + if request.State == entity.RequestStateFailed { + fact, _, err := recordFact(ctx, c.logger, c.metricsScope, store, request, nil) + if err != nil { + return err + } + if err := persistValidationFactRecordedLog(ctx, c.materializer, store, request, fact); err != nil { + return err + } + if err := publishHookEvent(ctx, c.logger, c.metricsScope, c.registry, request, hookevent.NewValidationRepositoryRecorded(request)); err != nil { + return err + } + } + metrics.NamedCounter(c.metricsScope, _dlqOpName, "requests_abandoned", 1, metrics.TagsFromContext(ctx)...) fields := []any{ "message_id", msg.ID, diff --git a/stovepipe/controller/record/dlq_test.go b/stovepipe/controller/record/dlq_test.go index b326ddad3..84e6733ea 100644 --- a/stovepipe/controller/record/dlq_test.go +++ b/stovepipe/controller/record/dlq_test.go @@ -17,16 +17,21 @@ package record import ( "context" "errors" + "strconv" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" "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" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/stovepipe/core/hookevent" 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" @@ -39,6 +44,8 @@ type dlqMocks struct { factory *storagemock.MockFactory store *storagemock.MockStorage requestStore *storagemock.MockRequestStore + factStore *storagemock.MockValidationFactStore + hooks *hookRecorder materializer *requestlogmock.MockMaterializer metricsScope tally.TestScope } @@ -54,29 +61,36 @@ func TestDLQControllerIdentity(t *testing.T) { func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { tests := []struct { name string + state entity.RequestState failure failure.Failure hasFailure bool }{ { name: "promotion failure", + state: entity.RequestStateSucceeded, failure: failure.Failure{Message: "failed to promote: permission denied"}, hasFailure: true, }, { name: "other record failure", + state: entity.RequestStateSucceeded, failure: failure.Failure{Message: "hook publish failed"}, hasFailure: true, }, { - name: "missing failure attribution", + name: "missing failure attribution", + state: entity.RequestStateSucceeded, }, + {name: "cancelled request", state: entity.RequestStateCancelled}, + {name: "superseded request", state: entity.RequestStateSuperseded}, + {name: "nonterminal request", state: entity.RequestStateProcessing}, } 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) + expectDLQRequestLoad(m, requestWithState(tt.state), 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) @@ -90,6 +104,8 @@ func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { require.NoError(t, c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, tt.failure, tt.hasFailure))) + assert.Empty(t, m.hooks.events) + counterName := "record_dlq_controller.record_dlq.requests_abandoned+queue=monorepo/main" counter, ok := m.metricsScope.Snapshot().Counters()[counterName] require.True(t, ok) @@ -98,6 +114,140 @@ func TestDLQControllerRetainsAbandonedRecordHistory(t *testing.T) { } } +func TestDLQControllerRecordsFailedRepository(t *testing.T) { + for _, test := range []struct { + name string + hookFailure bool + existing *entity.ValidationFact + }{ + {name: "missing fact"}, + {name: "retry after hook publish failure", hookFailure: true}, + {name: "existing partial fact", existing: &entity.ValidationFact{Degree: entity.DegreePartial}}, + {name: "existing green fact", existing: &entity.ValidationFact{Degree: entity.DegreeGreen}}, + {name: "existing broken fact", existing: &entity.ValidationFact{Degree: entity.DegreeBroken}}, + } { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newDLQControllerForTest(t, ctrl) + request := requestWithState(entity.RequestStateFailed) + m.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil).Times(2) + m.requestStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil).Times(2) + var stored *entity.ValidationFact + if test.existing != nil { + fact := *test.existing + fact.URI, fact.RequestID, fact.CreatedAt = testURI, testID, 1234 + stored = &fact + } + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, fact entity.ValidationFact) error { + assert.Equal(t, testURI, fact.URI) + assert.Empty(t, fact.Project) + assert.Equal(t, testID, fact.RequestID) + assert.Equal(t, entity.DegreeBroken, fact.Degree) + if stored != nil { + return storage.ErrAlreadyExists + } + stored = &fact + return nil + }).Times(2) + m.factStore.EXPECT().Get(gomock.Any(), testURI, "").DoAndReturn( + func(context.Context, string, string) (entity.ValidationFact, error) { + require.NotNil(t, stored) + return *stored, nil + }).AnyTimes() + var logs []entity.RequestLog + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + assert.Empty(t, log.State) + logs = append(logs, log) + return nil + }).Times(4) + + for attempt := range 2 { + m.hooks.err = nil + if test.hookFailure && attempt == 0 { + m.hooks.err = errors.New("publish unavailable") + } + err := c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, failure.Failure{}, false)) + if m.hooks.err != nil { + require.ErrorIs(t, err, m.hooks.err) + } else { + require.NoError(t, err) + } + } + + require.NotNil(t, stored) + if test.existing == nil { + assert.Equal(t, entity.DegreeBroken, stored.Degree) + } else { + assert.Equal(t, test.existing.Degree, stored.Degree) + assert.EqualValues(t, 1234, stored.CreatedAt) + } + require.Len(t, logs, 4) + assert.Equal(t, entity.RequestEventRecordAbandoned, logs[0].Event) + assert.Equal(t, entity.RequestEventValidationFactRecorded, logs[1].Event) + assert.Equal(t, logs[0].ID, logs[2].ID) + assert.Equal(t, logs[1], logs[3]) + assert.Equal(t, stored.CreatedAt, logs[1].TimestampMs) + assert.Equal(t, strconv.FormatFloat(stored.Degree, 'g', -1, 64), logs[1].Metadata[requestlog.MetadataKeyFactDegree]) + if test.hookFailure { + require.Len(t, m.hooks.events, 1) + } else { + require.Len(t, m.hooks.events, 2) + } + for _, event := range m.hooks.events { + assert.Equal(t, string(hookevent.TypeValidationRepositoryRecorded), event.GetType()) + assert.Equal(t, hookevent.NewValidationRepositoryRecorded(request).GetId(), event.GetId()) + } + assert.Equal(t, testQueue, m.hooks.tenant) + }) + } +} + +func TestDLQControllerRetriesFailedRepositoryRecording(t *testing.T) { + for _, stage := range []string{"abandonment history", "fact create", "existing fact read", "fact history", "hook publish", "different fact owner"} { + t.Run(stage, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newDLQControllerForTest(t, ctrl) + expectDLQRequestLoad(m, requestWithState(entity.RequestStateFailed), nil) + sentinel := errors.New("unavailable") + if stage == "abandonment history" { + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(sentinel) + } else { + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil) + switch stage { + case "fact create": + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(sentinel) + case "existing fact read", "different fact owner": + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + if stage == "existing fact read" { + m.factStore.EXPECT().Get(gomock.Any(), testURI, "").Return(entity.ValidationFact{}, sentinel) + } else { + m.factStore.EXPECT().Get(gomock.Any(), testURI, "").Return(entity.ValidationFact{ + URI: testURI, RequestID: "another-request", Degree: entity.DegreeGreen, + }, nil) + } + case "fact history", "hook publish": + m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + if stage == "fact history" { + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(sentinel) + } else { + m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil) + m.hooks.err = sentinel + } + } + } + err := c.Process(queueContext(), newDLQDelivery(t, ctrl, recordPayload(t, testID), testQueue, failure.Failure{}, false)) + if stage == "different fact owner" { + require.Error(t, err) + } else { + require.ErrorIs(t, err, sentinel) + } + assert.Empty(t, m.hooks.events) + }) + } +} + func TestDLQControllerRetriesDurableStateFailures(t *testing.T) { tests := []struct { name string @@ -153,6 +303,14 @@ func TestDLQControllerAcknowledgesUnresolvableMessages(t *testing.T) { {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: "request queue mismatch", payload: recordPayload(t, testID), tenant: testQueue, + setup: func(m dlqMocks) { + request := requestWithState(entity.RequestStateFailed) + request.Queue = "monorepo/other" + expectDLQRequestLoad(m, request, nil) + }, + }, { name: "unresolvable queue", payload: recordPayload(t, testID), @@ -183,16 +341,40 @@ func newDLQControllerForTest(t *testing.T, ctrl *gomock.Controller) (*DLQControl factory: storagemock.NewMockFactory(ctrl), store: storagemock.NewMockStorage(ctrl), requestStore: storagemock.NewMockRequestStore(ctrl), + factStore: storagemock.NewMockValidationFactStore(ctrl), + hooks: &hookRecorder{}, materializer: requestlogmock.NewMockMaterializer(ctrl), metricsScope: scope, } m.store.EXPECT().GetRequestStore().Return(m.requestStore).AnyTimes() + m.store.EXPECT().GetValidationFactStore().Return(m.factStore).AnyTimes() + publisher := mqmock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + if m.hooks.err != nil { + return m.hooks.err + } + event := &basehook.HookEvent{} + if err := basehook.Unmarshal(msg.Payload, event); err != nil { + return err + } + m.hooks.events = append(m.hooks.events, event) + m.hooks.tenant = msg.Tenant + return nil + }).AnyTimes() + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: basehook.TopicKeyHook, Name: "stovepipe-hook", Queue: queue}, + }) + require.NoError(t, err) return NewDLQController( zap.NewNop().Sugar(), scope, m.factory, m.materializer, + registry, consumer.TopicKey("record_dlq"), "stovepipe-record-dlq", ), m diff --git a/stovepipe/controller/record/facts.go b/stovepipe/controller/record/facts.go new file mode 100644 index 000000000..91db70b75 --- /dev/null +++ b/stovepipe/controller/record/facts.go @@ -0,0 +1,141 @@ +// 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" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "strconv" + "time" + + "github.com/uber-go/tally" + basehook "github.com/uber/submitqueue/api/base/hook" + "github.com/uber/submitqueue/platform/consumer" + platformhook "github.com/uber/submitqueue/platform/hook" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/requestlog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/projectstatus" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +func recordFact( + ctx context.Context, + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + request entity.Request, + results []projectstatus.Result, +) (entity.ValidationFact, bool, error) { + return recordValidationFact(ctx, logger, scope, store.GetValidationFactStore(), entity.ValidationFact{ + URI: request.URI, + Project: wholeRepositoryProject, + Degree: degreeFor(request.State, results), + RequestID: request.ID, + CreatedAt: time.Now().UnixMilli(), + }) +} + +func recordValidationFact( + ctx context.Context, + logger *zap.SugaredLogger, + scope tally.Scope, + factStore storage.ValidationFactStore, + fact entity.ValidationFact, +) (entity.ValidationFact, bool, error) { + err := factStore.Create(ctx, fact) + switch { + case err == nil: + metrics.NamedCounter(scope, _opName, "fact_created", 1, metrics.TagsFromContext(ctx)...) + logger.Infow("recorded validation fact", + "request_id", fact.RequestID, + "uri", fact.URI, + "project", fact.Project, + "degree", fact.Degree, + ) + return fact, true, nil + + case errors.Is(err, storage.ErrAlreadyExists): + stored, getErr := factStore.Get(ctx, fact.URI, fact.Project) + if getErr != nil { + metrics.NamedCounter(scope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return entity.ValidationFact{}, false, fmt.Errorf("failed to load the existing fact for uri %s and project %q: %w", fact.URI, fact.Project, getErr) + } + if stored.RequestID != fact.RequestID { + // Ingest deduplication forbids different requests owning the same immutable fact. + metrics.NamedCounter(scope, _opName, "invariant_errors", 1, metrics.TagsFromContext(ctx)...) + return entity.ValidationFact{}, false, fmt.Errorf( + "fact for uri %s and project %q is owned by request %s, not %s", fact.URI, fact.Project, stored.RequestID, fact.RequestID) + } + metrics.NamedCounter(scope, _opName, "fact_exists", 1, metrics.TagsFromContext(ctx)...) + return stored, false, nil + + default: + metrics.NamedCounter(scope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return entity.ValidationFact{}, false, fmt.Errorf("failed to create the fact for uri %s and project %q: %w", fact.URI, fact.Project, err) + } +} + +func persistValidationFactRecordedLog( + ctx context.Context, + materializer requestlog.Materializer, + store storage.Storage, + request entity.Request, + fact entity.ValidationFact, +) error { + // Hash the composite fact key so an arbitrarily long URI cannot overflow the bounded log ID. + identity := sha256.Sum256([]byte(fact.URI + "\x00" + fact.Project)) + log := requestlog.NewRequestEventLog( + request, + entity.RequestEventValidationFactRecorded, + hex.EncodeToString(identity[:]), + map[string]string{requestlog.MetadataKeyFactDegree: strconv.FormatFloat(fact.Degree, 'g', -1, 64)}, + ) + log.TimestampMs = fact.CreatedAt + if err := materializer.PersistLog(ctx, store, log); err != nil { + return fmt.Errorf("failed to record validation fact for request %s: %w", request.ID, err) + } + return nil +} + +func publishHookEvent( + ctx context.Context, + logger *zap.SugaredLogger, + scope tally.Scope, + registry consumer.TopicRegistry, + request entity.Request, + event *basehook.HookEvent, +) error { + if err := platformhook.Publish(ctx, registry, request.Queue, event, request.ID); err != nil { + metrics.NamedCounter(scope, _opName, "hook_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to announce %s for request %s: %w", event.GetType(), request.ID, err) + } + + metrics.NamedCounter(scope, _opName, "hook_events_published", 1, + metrics.TagsFromContext(ctx, metrics.NewTag("event_type", event.GetType()))..., + ) + logger.Infow("announced validation outcome", + "queue", request.Queue, + "request_id", request.ID, + "uri", request.URI, + "event_type", event.GetType(), + "event_id", event.GetId(), + ) + return nil +} diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 07ab866f9..45624122d 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -30,8 +30,6 @@ package record import ( "context" - "crypto/sha256" - "encoding/hex" "errors" "fmt" "math" @@ -39,10 +37,8 @@ import ( "time" "github.com/uber-go/tally" - basehook "github.com/uber/submitqueue/api/base/hook" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - platformhook "github.com/uber/submitqueue/platform/hook" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/stovepipe/core/hookevent" "github.com/uber/submitqueue/stovepipe/core/loader" @@ -61,11 +57,11 @@ import ( type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope - stores storage.Factory materializer requestlog.Materializer + registry consumer.TopicRegistry + stores storage.Factory projectStatusResolverFactory projectstatus.ResolverFactory sourceControl sourcecontrol.Factory - registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -96,11 +92,11 @@ func NewController( return &Controller{ logger: logger.Named(name), metricsScope: scope.SubScope(name), - stores: stores, materializer: materializer, + registry: registry, + stores: stores, projectStatusResolverFactory: projectStatusResolverFactory, sourceControl: sourceControl, - registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -152,11 +148,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err != nil { return err } - fact, created, err := c.recordFact(ctx, store, request, results) + fact, created, err := recordFact(ctx, c.logger, c.metricsScope, store, request, results) if err != nil { return err } - if err := c.persistValidationFactRecordedLog(ctx, store, request, fact); err != nil { + if err := persistValidationFactRecordedLog(ctx, c.materializer, store, request, fact); err != nil { return err } if err := c.persistProjectFacts(ctx, store, request, results); err != nil { @@ -165,13 +161,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err := c.applyFactToDerivedCaches(ctx, store, request, fact, created); err != nil { return err } - return c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryRecorded(request)) + return publishHookEvent(ctx, c.logger, c.metricsScope, c.registry, request, hookevent.NewValidationRepositoryRecorded(request)) case entity.RequestStateCancelled: // A cancelled build decided nothing about the commit, so it establishes // no fact. The identity stays unclaimed; the next commit re-validates. metrics.NamedCounter(c.metricsScope, _opName, "cancelled", 1, metrics.TagsFromContext(ctx)...) - return c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryCancelled(request)) + return publishHookEvent(ctx, c.logger, c.metricsScope, c.registry, request, hookevent.NewValidationRepositoryCancelled(request)) case entity.RequestStateSuperseded: // Terminal without a build outcome. buildsignal never publishes for a @@ -206,7 +202,7 @@ func (c *Controller) resolveProjectResults(ctx context.Context, request entity.R func (c *Controller) persistProjectFacts(ctx context.Context, store storage.Storage, request entity.Request, results []projectstatus.Result) error { for _, result := range results { - fact, _, err := c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{ + fact, _, err := recordValidationFact(ctx, c.logger, c.metricsScope, store.GetValidationFactStore(), entity.ValidationFact{ URI: request.URI, Project: result.Project, Degree: result.Degree, @@ -264,27 +260,6 @@ func (c *Controller) persistProjectFactsRecordedLog( return nil } -func (c *Controller) persistValidationFactRecordedLog( - ctx context.Context, - store storage.Storage, - request entity.Request, - fact entity.ValidationFact, -) error { - // Hash the composite fact key so an arbitrarily long URI cannot overflow the bounded log ID. - identity := sha256.Sum256([]byte(fact.URI + "\x00" + fact.Project)) - log := requestlog.NewRequestEventLog( - request, - entity.RequestEventValidationFactRecorded, - hex.EncodeToString(identity[:]), - map[string]string{requestlog.MetadataKeyFactDegree: strconv.FormatFloat(fact.Degree, 'g', -1, 64)}, - ) - log.TimestampMs = fact.CreatedAt - if err := c.materializer.PersistLog(ctx, store, log); err != nil { - return fmt.Errorf("failed to record validation fact for request %s: %w", request.ID, err) - } - return nil -} - // applyFactToDerivedCaches follows the persisted fact: a green fact promotes the // commit when this request holds the queue's bookmark, set by build completion. // A broken fact skips promotion, and instead reports how long the break it names @@ -326,61 +301,6 @@ func (c *Controller) applyFactToDerivedCaches( return c.promote(ctx, request) } -// recordFact writes the request's outcome as an immutable whole-repository fact and -// returns the fact that is actually stored, which is not always the one just built: -// facts are first-writer-wins, so an identity already claimed by this same request — -// a redelivery after the write but before promotion — yields the stored -// fact instead. Every decision downstream reads that stored fact rather than the -// 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, results []projectstatus.Result) (entity.ValidationFact, bool, error) { - return c.recordValidationFact(ctx, store.GetValidationFactStore(), entity.ValidationFact{ - URI: request.URI, - Project: wholeRepositoryProject, - Degree: degreeFor(request.State, results), - RequestID: request.ID, - CreatedAt: time.Now().UnixMilli(), - }) -} - -func (c *Controller) recordValidationFact(ctx context.Context, factStore storage.ValidationFactStore, fact entity.ValidationFact) (entity.ValidationFact, bool, error) { - - err := factStore.Create(ctx, fact) - switch { - case err == nil: - metrics.NamedCounter(c.metricsScope, _opName, "fact_created", 1, metrics.TagsFromContext(ctx)...) - c.logger.Infow("recorded validation fact", - "request_id", fact.RequestID, - "uri", fact.URI, - "project", fact.Project, - "degree", fact.Degree, - ) - return fact, true, nil - - case errors.Is(err, storage.ErrAlreadyExists): - stored, getErr := factStore.Get(ctx, fact.URI, fact.Project) - if getErr != nil { - metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) - return entity.ValidationFact{}, false, fmt.Errorf("failed to load the existing fact for uri %s and project %q: %w", fact.URI, fact.Project, getErr) - } - if stored.RequestID != fact.RequestID { - // Two requests validating one URI would break the dedup ingest - // enforces, so this is a broken invariant rather than a race to - // resolve. Non-retryable: the stored fact is immutable. - metrics.NamedCounter(c.metricsScope, _opName, "invariant_errors", 1, metrics.TagsFromContext(ctx)...) - return entity.ValidationFact{}, false, fmt.Errorf( - "fact for uri %s and project %q is owned by request %s, not %s", fact.URI, fact.Project, stored.RequestID, fact.RequestID) - } - metrics.NamedCounter(c.metricsScope, _opName, "fact_exists", 1, metrics.TagsFromContext(ctx)...) - return stored, false, nil - - default: - metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) - return entity.ValidationFact{}, false, fmt.Errorf("failed to create the fact for uri %s and project %q: %w", fact.URI, fact.Project, err) - } -} - // reportFailureDetectionLatency records how long the break this build failed on went // undetected, measured from the commit timestamp of the base it validated against. A // histogram rather than a gauge because the distribution over failures is the point: @@ -558,34 +478,6 @@ func (c *Controller) promote(ctx context.Context, request entity.Request) error return nil } -// 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 -// moved: the payload names the request rather than snapshotting it, so a hook -// reacting to a green commit by reading the bookmark or resolving the promotion -// ref must not find either still pointing at the previous commit. -// -// 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 { - 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) - } - - metrics.NamedCounter(c.metricsScope, _opName, "hook_events_published", 1, - metrics.TagsFromContext(ctx, metrics.NewTag("event_type", event.GetType()))..., - ) - c.logger.Infow("announced validation outcome", - "queue", request.Queue, - "request_id", request.ID, - "uri", request.URI, - "event_type", event.GetType(), - "event_id", event.GetId(), - ) - return nil -} - // loadRequest loads the request by id. func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) { return loader.ByID(ctx, id, store.GetRequestStore().Get, "request")