diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index b14946102..77a81ea96 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -221,7 +221,7 @@ The `project_facts_recorded` event also projects its retained timestamp into `Re | 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`; 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. | -| Issue tracking DLQ | Protect the current cursor from delayed writes, retain its issue history and `issue_tracking_abandoned`, then mark the pass abandoned. | A stable hash of the notification ID deduplicates the event. Recovery retains accepted issue changes and never processes the remaining range; consumer registration is separate. | +| Issue tracking DLQ | Protect the current cursor from delayed writes, retain its issue history and `issue_tracking_abandoned`, then mark the pass abandoned. | A stable hash of the notification ID deduplicates the event. Recovery retains accepted issue changes and never processes the remaining range. | | 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/buildanalysis.md b/doc/rfc/stovepipe/steps/buildanalysis.md index 7c088d7bb..474dd84d4 100644 --- a/doc/rfc/stovepipe/steps/buildanalysis.md +++ b/doc/rfc/stovepipe/steps/buildanalysis.md @@ -1,6 +1,6 @@ # Build issue detection -Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host registers an issue-tracking publication route but does not yet consume those notifications. +Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host consumes issue-tracking notifications and registers their DLQ recovery alongside analysis. ## Start and completion notices diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 2bd7b0b44..7e67d287d 100644 --- a/service/stovepipe/README.md +++ b/service/stovepipe/README.md @@ -9,9 +9,11 @@ Runnable wiring for the **Stovepipe** domain — a single-service domain (the do - **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`). - **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`. - **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`. +- **buildanalysis consumer** (`TopicKeyBuildAnalysis`) — periodically detects and persists build issues, then publishes queue-partitioned issue-tracking notifications. +- **issuetrack consumer** (`TopicKeyIssueTrack`) — applies reported build issues to current issue state and history, closing absent issues only when final reconciliation permits it. - **record consumer** (`TopicKeyRecord`) — writes the whole-repository validation fact and, for a green fact, advances the queue's last-green bookmark and promotion ref. - **hook consumer** (`TopicKeyHook`) — receives `validation.repository.started` from process and `validation.repository.recorded` or `validation.repository.cancelled` from record. The current resolver invokes only `noop`, so these events have no external side effect. Its topic name is domain-qualified (`stovepipe-hook`) because the key is shared across domains. -- **DLQ consumers** — registered for process, build, buildsignal, record, and hook, with stage-specific reconciliation behavior. +- **DLQ consumers** — registered for process, build, buildsignal, buildanalysis, issuetrack, record, and hook, with stage-specific recovery behavior. Issue-tracking recovery abandons incomplete passes while preserving accepted changes; it does not rerun their remaining batches. The ingest → process → build → buildsignal → record flow stays inside one service and one store, so messages carry thin identifiers: request IDs on process, build, and record; a build ID on buildsignal. Consumers reload the full entity from storage, which keeps messages small and redelivery idempotent. The topic keys and internal wire contract are owned by the domain under `stovepipe/core/messagequeue/`. diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 2989150ee..73bfdd181 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -29,6 +29,7 @@ go_library( "//stovepipe/controller/buildanalysis:go_default_library", "//stovepipe/controller/buildsignal:go_default_library", "//stovepipe/controller/dlq:go_default_library", + "//stovepipe/controller/issuetrack:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/controller/record:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 509122734..0cb10c206 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -51,6 +51,7 @@ import ( "github.com/uber/submitqueue/stovepipe/controller/buildanalysis" "github.com/uber/submitqueue/stovepipe/controller/buildsignal" "github.com/uber/submitqueue/stovepipe/controller/dlq" + "github.com/uber/submitqueue/stovepipe/controller/issuetrack" "github.com/uber/submitqueue/stovepipe/controller/process" "github.com/uber/submitqueue/stovepipe/controller/record" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -415,6 +416,12 @@ func registerPrimaryControllers( } count++ + issueTrackController := issuetrack.NewController(logger, scope, store, registry, stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack") + if err := c.Register(issueTrackController); err != nil { + return count, fmt.Errorf("failed to register issuetrack controller: %w", err) + } + count++ + recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) @@ -466,6 +473,12 @@ func registerDLQControllers( } count++ + issueTrackDLQController := issuetrack.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyIssueTrack), "stovepipe-issuetrack-dlq") + if err := c.Register(issueTrackDLQController); err != nil { + return count, fmt.Errorf("failed to register issuetrack dlq controller: %w", err) + } + count++ + 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) @@ -563,9 +576,16 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-hook-dlq"), }, { - Key: stovepipemq.TopicKeyIssueTrack, - Name: "issuetrack", - Queue: q, + Key: stovepipemq.TopicKeyIssueTrack, + Name: "issuetrack", + Queue: q, + Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-issuetrack"), + }, + { + Key: dlq.TopicKey(stovepipemq.TopicKeyIssueTrack), + Name: "issuetrack_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-issuetrack-dlq"), }, { Key: stovepipemq.TopicKeyFix, diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 6dd54825e..aa1c10267 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -92,13 +92,32 @@ func TestEveryRegisteredControllerResolvesInTheTopicRegistry(t *testing.T) { } } -func TestIssueTrackingHasPublicationRouteWithoutConsumption(t *testing.T) { +func TestIssueTrackingRegistration(t *testing.T) { registry, primary, deadLetter := registeredControllers(t) name, ok := registry.TopicName(stovepipemq.TopicKeyIssueTrack) require.True(t, ok) assert.Equal(t, "issuetrack", name) - assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyIssueTrack) - assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyIssueTrack)) + assert.Contains(t, topicKeys(primary), stovepipemq.TopicKeyIssueTrack) + assert.Contains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyIssueTrack)) + + dlqName, ok := registry.TopicName(dlq.TopicKey(stovepipemq.TopicKeyIssueTrack)) + require.True(t, ok) + assert.Equal(t, "issuetrack_dlq", dlqName) + + cfg, ok := registry.SubscriptionConfig(stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack") + require.True(t, ok) + assert.True(t, cfg.DLQ.Enabled) + assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix) +} + +func TestEveryPrimaryStageHasRecovery(t *testing.T) { + _, primary, deadLetter := registeredControllers(t) + recoveryKeys := topicKeys(deadLetter) + for _, controller := range primary { + t.Run(controller.Name(), func(t *testing.T) { + assert.Contains(t, recoveryKeys, dlq.TopicKey(controller.TopicKey())) + }) + } } func TestFixHasPublicationRouteWithoutConsumption(t *testing.T) { @@ -148,6 +167,7 @@ func TestPipelineSubscriptionRetryPolicies(t *testing.T) { {name: "build", topicKey: stovepipemq.TopicKeyBuild, consumerGroup: "stovepipe-build", wantRetry: defaultRetry}, {name: "buildsignal", topicKey: stovepipemq.TopicKeyBuildSignal, consumerGroup: "stovepipe-buildsignal", wantRetry: extendedRetry}, {name: "buildanalysis", topicKey: stovepipemq.TopicKeyBuildAnalysis, consumerGroup: "stovepipe-buildanalysis", wantRetry: extendedRetry}, + {name: "issuetrack", topicKey: stovepipemq.TopicKeyIssueTrack, consumerGroup: "stovepipe-issuetrack", wantRetry: extendedRetry}, {name: "record", topicKey: stovepipemq.TopicKeyRecord, consumerGroup: "stovepipe-record", wantRetry: extendedRetry}, {name: "hook", topicKey: basehook.TopicKeyHook, consumerGroup: "stovepipe-hook", wantRetry: defaultRetry}, } diff --git a/stovepipe/controller/buildanalysis/README.md b/stovepipe/controller/buildanalysis/README.md index fb5f71027..65be8d984 100644 --- a/stovepipe/controller/buildanalysis/README.md +++ b/stovepipe/controller/buildanalysis/README.md @@ -36,4 +36,4 @@ State-read and history-write errors are returned for retry. Hosts must run recov ## Host integration -The reference service supplies a no-op detector and publishes analysis notices at build start and completion. It also registers a publish-only issue-tracking topic; issue-tracking consumption and recovery are not enabled yet. It registers the normal analysis controller with bounded retries and the recovery controller with the existing DLQ consumer. See [build issue detection](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration. +The reference service supplies a no-op detector and publishes analysis notices at build start and completion. It registers both analysis and issue tracking with bounded normal retries and the existing DLQ consumer for recovery. See [issue tracking](../issuetrack/README.md) for reconciliation and recovery behavior and [build issue detection](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration. diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index b3bfbd376..591a39099 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -1,6 +1,6 @@ # Issue tracking -Issue tracking maintains current issue state and opening/closing history from persisted build issues. Each delivery applies a bounded batch of issues and requests immediate redelivery when more work remains. No consumer is registered yet. +Issue tracking maintains current issue state and opening/closing history from persisted build issues. Each delivery applies a bounded batch of issues and requests immediate redelivery when more work remains. The reference service registers both the normal and recovery consumers. ## Applying one issue @@ -57,8 +57,10 @@ Recovery must finish fencing the current issue and retaining its history before A notification can fail while helping an older pass. Recovery abandons that saved pass first, then requests immediate redelivery to abandon the failed notification's own work if it has not already been superseded. A delayed DLQ delivery never abandons a newer build or a different notification from the same Request. -Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged. No normal or DLQ consumer is registered yet. Completed opening passes publish repair notifications; recovery does not. The host must provide a fix publication route and a repair executor when registering the fix consumer. +Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged. + +The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The fix publication route is registered, but fix consumption and executor configuration remain separate. ## Analysis handoff -Build analysis publishes refresh or final notifications after persisting build issues. Final notification requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host registers the publication route but does not enable either issue-tracking consumer. +Build analysis publishes refresh or final notifications after persisting build issues. Final notification requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host uses a no-op detector; deployments supply their own detector and register the same normal and recovery paths. diff --git a/test/e2e/stovepipe/BUILD.bazel b/test/e2e/stovepipe/BUILD.bazel index 9cd77929e..ccb790b2b 100644 --- a/test/e2e/stovepipe/BUILD.bazel +++ b/test/e2e/stovepipe/BUILD.bazel @@ -5,6 +5,7 @@ go_test( srcs = [ "harness_test.go", "ingest_test.go", + "issuetrack_test.go", "list_test.go", "queue_status_test.go", "suite_test.go", @@ -23,8 +24,14 @@ go_test( ], deps = [ "//api/stovepipe/protopb:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mysql:go_default_library", + "//platform/publish:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/queuelog:go_default_library", "//stovepipe/core/queuestate:go_default_library", + "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/sourcecontrol/fake:go_default_library", @@ -40,5 +47,6 @@ go_test( "@org_golang_google_grpc//codes:go_default_library", "@org_golang_google_grpc//status:go_default_library", "@org_golang_google_protobuf//proto:go_default_library", + "@org_uber_go_zap//:go_default_library", ], ) diff --git a/test/e2e/stovepipe/issuetrack_test.go b/test/e2e/stovepipe/issuetrack_test.go new file mode 100644 index 000000000..41ac01f85 --- /dev/null +++ b/test/e2e/stovepipe/issuetrack_test.go @@ -0,0 +1,129 @@ +// 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 e2e_test + +import ( + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemysql "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/platform/publish" + 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 ( + issueTrackingQueue = "e2e-stovepipe/issue-tracking" + issueTrackingDLQQueue = "e2e-stovepipe/issue-tracking-dlq" +) + +func (s *StovepipeE2ESuite) awaitIssueTracking(queue string, phase entity.IssueTrackingPhase) entity.IssueTrackingState { + s.T().Helper() + store, err := s.appStorage.For(queue) + require.NoError(s.T(), err) + var progress entity.IssueTrackingState + pollUntil(convergencePollInterval, func() bool { + progress, err = store.GetIssueTrackingStore().Get(s.ctx) + if storage.IsNotFound(err) { + return false + } + require.NoError(s.T(), err) + s.log.Logf("issue tracking queue=%s build=%s phase=%s (want %s)", queue, progress.BuildID, progress.Phase, phase) + return progress.Phase == phase + }) + return progress +} + +func (s *StovepipeE2ESuite) TestIssueTracking_FinalizesAfterAnalysis() { + t := s.T() + requestID := s.ingest(issueTrackingQueue) + progress := s.awaitIssueTracking(issueTrackingQueue, entity.IssueTrackingPhaseFinalized) + require.Equal(t, requestID, progress.RequestID) + require.True(t, progress.Finalize) + + store, err := s.appStorage.For(issueTrackingQueue) + require.NoError(t, err) + request, err := store.GetRequestStore().Get(s.ctx, requestID) + require.NoError(t, err) + require.Equal(t, entity.RequestStateSucceeded, request.State) + require.Equal(t, request.TerminalBuildID, progress.BuildID) + issues, err := store.GetIssueStateStore().List(s.ctx, storage.IssueStateRange{Limit: 1}) + require.NoError(t, err) + require.Empty(t, issues) +} + +func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotification() { + t := s.T() + const queue = issueTrackingDLQQueue + store, err := s.appStorage.For(queue) + require.NoError(t, err) + request := entity.Request{ + ID: "1", Queue: queue, URI: "git://" + queue + "/HEAD", + State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyFull, Version: 1, + } + build := entity.Build{ID: "running-build", RequestID: request.ID, Status: entity.BuildStatusRunning, Version: 1} + require.NoError(t, store.GetRequestStore().Create(s.ctx, request)) + require.NoError(t, store.GetBuildStore().Create(s.ctx, build)) + queueState, err := store.GetQueueStore().Get(s.ctx, queue) + require.NoError(t, err) + queueState.InFlightCount = 1 + require.NoError(t, store.GetQueueStore().Update(s.ctx, queueState, queueState.Version, queueState.Version+1)) + + mq, err := queuemysql.NewQueue(queuemysql.Params{ + DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{queue}, + }) + require.NoError(t, err) + defer func() { require.NoError(t, mq.Close()) }() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: stovepipemq.TopicKeyIssueTrack, Name: "issuetrack", Queue: mq}, + }) + require.NoError(t, err) + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{QueueName: queue, BuildId: build.ID, Finalize: true}) + require.NoError(t, err) + notificationID := publish.IntentID(build.ID, "premature-final") + require.NoError(t, publish.Message(entityqueue.WithQueueName(s.ctx, queue), registry, stovepipemq.TopicKeyIssueTrack, publish.MessageParams{ + Tenant: queue, ID: notificationID, Payload: payload, PartitionKey: queue, + })) + + progress := s.awaitIssueTracking(queue, entity.IssueTrackingPhaseAbandoned) + require.Equal(t, notificationID, progress.NotificationID) + require.Equal(t, build.ID, progress.BuildID) + logs, err := store.GetRequestLogStore().List(s.ctx, request.ID) + require.NoError(t, err) + require.Len(t, logs, 1) + require.Equal(t, entity.RequestEventIssueTrackingAbandoned, logs[0].Event) + require.Equal(t, build.ID, logs[0].Metadata[requestlog.MetadataKeyBuildID]) + retainedRequest, err := store.GetRequestStore().Get(s.ctx, request.ID) + require.NoError(t, err) + require.Equal(t, request, retainedRequest) + retainedBuild, err := store.GetBuildStore().Get(s.ctx, build.ID) + require.NoError(t, err) + require.Equal(t, build, retainedBuild) + require.Equal(t, int32(1), s.inFlightCount(queue)) + + // The DLQ watermark survives message garbage collection and proves recovery acknowledged. + pollUntil(convergencePollInterval, func() bool { + var offset int64 + err := s.queueDB.QueryRowContext(s.ctx, ` + SELECT offset_acked FROM queue_offsets + WHERE tenant = ? AND topic = ? AND consumer_group = ? AND partition_key = ?`, + queue, "issuetrack_dlq", "stovepipe-issuetrack-dlq", queue).Scan(&offset) + return err == nil && offset > 0 + }) +} diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 0c1fdd7b2..37be8583f 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -65,6 +65,7 @@ func (s *StovepipeE2ESuite) SetupSuite() { t.Setenv("MQ_TENANTS", strings.Join([]string{ "monorepo/main", "monorepo/release", "monorepo/slow?buildrunner-fake=build-slow", listPaginationQueue, listTimeBoundsQueue, listPrimaryQueue, listOtherQueue, listValidationQueue, queueStateTestQueue, + issueTrackingQueue, issueTrackingDLQQueue, }, ",")) s.log.Logf("Starting Stovepipe e2e test suite using docker-compose")