diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel new file mode 100644 index 000000000..76d235b97 --- /dev/null +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -0,0 +1,40 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["issuetrack.go"], + importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack", + visibility = ["//visibility:public"], + deps = [ + "//platform/base/id:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["issuetrack_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer/mock:go_default_library", + "//platform/publish:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md new file mode 100644 index 000000000..e2e880f01 --- /dev/null +++ b/stovepipe/controller/issuetrack/README.md @@ -0,0 +1,21 @@ +# Issue tracking + +`issuetrack.Controller` consumes `IssueTrack` notifications in a single queue partition. It alone maintains operational `OpenIssue` membership and records `IssueLog` transitions; build detection and immutable per-build observations remain in `buildanalysis`. The controller rejects messages whose partition key differs from the payload queue. Producers and consumers must preserve per-queue serialization; the progress row is not a distributed lock or a substitute for that delivery contract. + +## Reconciliation + +A refresh notification traverses its build's observations in bounded task/unit batches. A missing issue is opened using this build as its first-seen build. Repeated observations preserve existing openings and do not create another transition. Final reconciliation also traverses open issues and resolves those absent from the source build. Cancelled builds may add confirmed observations but never establish resolution. Final notifications must name the request's winning terminal build; duplicate-build losers are ignored. + +The final-notification contract is an input requirement, not something terminal status alone proves: detection must have succeeded, its persisted observations must be complete and authoritative for the queue's relevant validation scope, and in-flight analysis must not add unaccounted results after the final handoff. Accumulated observations are not automatically a current failure snapshot. A producer that cannot satisfy those requirements must not request absence-based resolution. No producer or consumer is wired by this implementation; the production handoff and recovery policy remain follow-up work. + +## Source ordering and interrupted passes + +Notification decisions are separate from storage writes. New and resumed work use the same execution path, which reloads the build named by saved progress. After resuming, the incoming notification is reconsidered only if the completed work did not already satisfy it. + +`IssueTrackingStore` retains one progress row per queue, including the source request, build, phase, and optimistic version. Request IDs are compared through `CompareRequestID`; provider build IDs and timestamps are not order keys. An older request cannot modify current issues. Later refresh notifications and duplicate finals for an already-finalized request are ignored. Same-request refresh notifications remain eligible until finalization. The progress row remains even when there are no open issues. + +Before applying any issue writes, the controller conditionally records a refreshing or finalizing phase. It records the completed phase only after issue and history writes succeed. A subsequent delivery first finishes pending work, including when the original delivery was retried or dead-lettered; only then may it admit a newer source. A permanent pending failure therefore needs operational repair rather than silently skipping a partially applied transition. Controller code owns version arithmetic; storage performs guarded replacement only. + +Opening history is repaired on replay if an open row was created before its log write failed. Resolution is retained before deleting the matching opening, so failed history persistence leaves the issue available for retry. Duplicate transition identities preserve their original timestamps. Replaying a whole pass avoids a multi-row transaction or a durable cursor over mutable membership. The operational progress and accepted writes are serialized, not an atomic database snapshot. + +This controller does not publish fix requests, expose a user-facing issue read model, or materialize `IssueSummary`. Those are separate responsibilities. Lifecycle history is retained here; a log-derived summary/materializer can be added without changing the observation handoff or the source-order guard. diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go new file mode 100644 index 000000000..7d5ba45ff --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -0,0 +1,391 @@ +// 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 issuetrack applies build observations to operational open issues and lifecycle history. +// All deliveries for one queue must share its partition and consumer group. +package issuetrack + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/id" + 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/loader" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// Each read is bounded; every pass still traverses the entire relevant key range. +const _issueBatchSize = 100 + +// Controller is the queue-serialized writer of open issues and issue transitions. +// Accepted pending work is replayed before a newer source can change issue state. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string + nowMs func() int64 +} + +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + topicKey consumer.TopicKey, + consumerGroup string, +) *Controller { + return &Controller{ + logger: logger.Named("issuetrack_controller"), + metricsScope: scope.SubScope("issuetrack_controller"), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + nowMs: func() int64 { return time.Now().UnixMilli() }, + } +} + +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { + op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + msg := delivery.Message() + notification := &stovepipemq.IssueTrack{} + if err := stovepipemq.Unmarshal(msg.Payload, notification); err != nil { + return fmt.Errorf("deserialize issue tracking notification: %w", err) + } + + queue := notification.GetQueueName() + if err := entityqueue.ValidatePayloadQueue(msg, queue); err != nil { + return fmt.Errorf("invalid message identity: %w", err) + } + if queue == "" || notification.GetBuildId() == "" { + return fmt.Errorf("issue tracking requires queue and build identifiers") + } + if msg.PartitionKey != queue { + return fmt.Errorf("issue tracking must be partitioned by queue %q", queue) + } + + store, err := c.stores.For(storage.Config{QueueName: queue}) + if err != nil { + return fmt.Errorf("resolve issue tracking storage: %w", err) + } + + build, request, err := loadBuildAndRequest(ctx, store, queue, notification.GetBuildId()) + if err != nil { + return err + } + if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID { + return nil + } + + return c.reconcileBuild(ctx, store, build, request, notification.GetFinalize()) +} + +func (c *Controller) reconcileBuild(ctx context.Context, store storage.Storage, build entity.Build, request entity.Request, finalize bool) error { + trackingStore := store.GetIssueTrackingStore() + + for { + if err := ctx.Err(); err != nil { + return err + } + + progress, err := trackingStore.Get(ctx) + if errors.Is(err, storage.ErrNotFound) { + progress = entity.IssueTrackingState{} + } else if err != nil { + return fmt.Errorf("load issue tracking progress: %w", err) + } + + action, err := decideReconciliation(progress, build, request, finalize) + if err != nil { + return err + } + if action == _ignoreNotification { + c.logger.Debugw("ignoring superseded issue analysis", "queue", request.Queue, + "build_id", build.ID, "request_id", request.ID, "current_request_id", progress.RequestID) + return nil + } + + pending := progress + if action == _startRefresh || action == _startFinalization { + pending = entity.IssueTrackingState{ + RequestID: request.ID, + BuildID: build.ID, + Phase: entity.IssueTrackingPhaseRefreshing, + Version: progress.Version + 1, + } + if action == _startFinalization { + pending.Phase = entity.IssueTrackingPhaseFinalizing + } + + if progress.Version == 0 { + err = trackingStore.Create(ctx, pending) + if errors.Is(err, storage.ErrAlreadyExists) { + continue + } + } else { + err = trackingStore.Update(ctx, pending, progress.Version, pending.Version) + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } + } + if err != nil { + return fmt.Errorf("start issue reconciliation: %w", err) + } + } + + if err := c.applyReconciliation(ctx, store, request.Queue, pending); err != nil { + return err + } + if action != _resumeAndRecheckNotification { + return nil + } + } +} + +type reconciliationAction int + +const ( + _ignoreNotification reconciliationAction = iota + _resumeAndFinishNotification + _resumeAndRecheckNotification + _startRefresh + _startFinalization +) + +// A zero progress value represents a queue with no accepted reconciliation. +func decideReconciliation(progress entity.IssueTrackingState, build entity.Build, request entity.Request, finalize bool) (reconciliationAction, error) { + if progress.IsPending() { + if progress.RequestID != request.ID || progress.BuildID != build.ID { + return _resumeAndRecheckNotification, nil + } + if finalize && progress.Phase == entity.IssueTrackingPhaseRefreshing { + return _resumeAndRecheckNotification, nil + } + return _resumeAndFinishNotification, nil + } + + if progress.Version != 0 { + order, err := entity.CompareRequestID(request.ID, progress.RequestID) + if err != nil { + return _ignoreNotification, err + } + if order < 0 || (order == 0 && progress.Phase == entity.IssueTrackingPhaseFinalized) { + return _ignoreNotification, nil + } + } + + if finalize { + if err := validateFinalBuild(build, request); err != nil { + return _ignoreNotification, err + } + return _startFinalization, nil + } + return _startRefresh, nil +} + +func (c *Controller) applyReconciliation(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) error { + build, request, err := loadBuildAndRequest(ctx, store, queue, progress.BuildID) + if err != nil { + return err + } + if request.ID != progress.RequestID { + return fmt.Errorf("pending issue reconciliation belongs to another request") + } + finalizing := progress.Phase == entity.IssueTrackingPhaseFinalizing + if finalizing { + if err := validateFinalBuild(build, request); err != nil { + return err + } + } + + if err := c.openBuildIssues(ctx, store, build.ID); err != nil { + return err + } + + if finalizing && build.Status != entity.BuildStatusCancelled { + if err := c.resolveMissingIssues(ctx, store, build.ID); err != nil { + return err + } + } + + completed := progress + completed.Phase = entity.IssueTrackingPhaseRefreshed + if finalizing { + completed.Phase = entity.IssueTrackingPhaseFinalized + } + + newVersion := progress.Version + 1 + if err := store.GetIssueTrackingStore().Update(ctx, completed, progress.Version, newVersion); err != nil { + return fmt.Errorf("complete issue reconciliation: %w", err) + } + return nil +} + +func (c *Controller) openBuildIssues(ctx context.Context, store storage.Storage, buildID string) error { + bounds := storage.BuildIssueRange{Limit: _issueBatchSize} + for { + if err := ctx.Err(); err != nil { + return err + } + + issues, err := store.GetBuildIssueStore().List(ctx, buildID, bounds) + if err != nil { + return fmt.Errorf("list build observations: %w", err) + } + if len(issues) == 0 { + return nil + } + + for _, issue := range issues { + if err := c.openObservedIssue(ctx, store, buildID, issue); err != nil { + return err + } + } + + last := issues[len(issues)-1] + bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + } +} + +func (c *Controller) openObservedIssue(ctx context.Context, store storage.Storage, buildID string, issue entity.BuildIssue) error { + openStore := store.GetOpenIssueStore() + openIssue, err := openStore.Get(ctx, issue.TaskID, issue.UnitID) + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return fmt.Errorf("read open issue: %w", err) + } + + if errors.Is(err, storage.ErrNotFound) { + openIssue = entity.OpenIssue{ + TaskID: issue.TaskID, + UnitID: issue.UnitID, + FirstSeenBuildID: buildID, + } + err = openStore.Create(ctx, openIssue) + if errors.Is(err, storage.ErrAlreadyExists) { + openIssue, err = openStore.Get(ctx, issue.TaskID, issue.UnitID) + } + if err != nil { + return fmt.Errorf("open observed issue: %w", err) + } + } + + if openIssue.FirstSeenBuildID == buildID { + return c.recordIssueEvent(ctx, store, buildID, entity.IssueEventOpened, issue.TaskID, issue.UnitID) + } + return nil +} + +func (c *Controller) resolveMissingIssues(ctx context.Context, store storage.Storage, buildID string) error { + bounds := storage.OpenIssueRange{Limit: _issueBatchSize} + for { + if err := ctx.Err(); err != nil { + return err + } + + issues, err := store.GetOpenIssueStore().List(ctx, bounds) + if err != nil { + return fmt.Errorf("list open issues: %w", err) + } + if len(issues) == 0 { + return nil + } + + for _, issue := range issues { + if err := c.resolveIssueIfAbsent(ctx, store, buildID, issue); err != nil { + return err + } + } + + last := issues[len(issues)-1] + bounds.After = storage.OpenIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + } +} + +func (c *Controller) resolveIssueIfAbsent(ctx context.Context, store storage.Storage, buildID string, issue entity.OpenIssue) error { + _, err := store.GetBuildIssueStore().Get(ctx, buildID, issue.TaskID, issue.UnitID) + if err == nil { + return nil + } + if !errors.Is(err, storage.ErrNotFound) { + return fmt.Errorf("check retained build observation: %w", err) + } + + // Record resolution before deletion so a failed history write leaves the issue retryable. + if err := c.recordIssueEvent(ctx, store, buildID, entity.IssueEventResolved, issue.TaskID, issue.UnitID); err != nil { + return err + } + if err := store.GetOpenIssueStore().Delete(ctx, issue); err != nil { + return fmt.Errorf("resolve open issue: %w", err) + } + return nil +} + +func (c *Controller) recordIssueEvent(ctx context.Context, store storage.Storage, buildID string, event entity.IssueEvent, taskID, unitID string) error { + log := entity.IssueLog{ + TaskID: taskID, + UnitID: unitID, + BuildID: buildID, + Event: event, + TimestampMs: c.nowMs(), + } + if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("record issue transition: %w", err) + } + return nil +} + +func loadBuildAndRequest(ctx context.Context, store storage.Storage, queue, buildID string) (entity.Build, entity.Request, error) { + build, err := loader.ByID(ctx, buildID, store.GetBuildStore().Get, "build") + if err != nil { + return entity.Build{}, entity.Request{}, err + } + + request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request") + if err != nil { + return entity.Build{}, entity.Request{}, err + } + + if request.Queue != queue { + return entity.Build{}, entity.Request{}, fmt.Errorf("issue tracking request belongs to another queue") + } + if err := id.Validate(request.ID); err != nil { + return entity.Build{}, entity.Request{}, fmt.Errorf("invalid issue tracking request ID: %w", err) + } + + return build, request, nil +} + +func validateFinalBuild(build entity.Build, request entity.Request) error { + if !build.Status.IsTerminal() || request.TerminalBuildID != build.ID { + return fmt.Errorf("final issue reconciliation requires the request's terminal build") + } + return nil +} + +func (c *Controller) Name() string { return "issuetrack" } +func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } +func (c *Controller) ConsumerGroup() string { return c.consumerGroup } + +var _ consumer.Controller = (*Controller)(nil) diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go new file mode 100644 index 000000000..e7b9e212f --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -0,0 +1,833 @@ +// 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 issuetrack + +import ( + "context" + "fmt" + "sort" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "github.com/uber/submitqueue/platform/publish" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "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" +) + +const testQueue = "validation/main" + +type issueKey struct{ task, unit string } +type logKey struct { + issueKey + build string + event entity.IssueEvent +} + +type fixture struct { + t *testing.T + ctrl *gomock.Controller + controller *Controller + builds map[string]entity.Build + requests map[string]entity.Request + observations map[string]map[issueKey]entity.BuildIssue + opens map[issueKey]entity.OpenIssue + logs map[logKey]entity.IssueLog + progress entity.IssueTrackingState + hasProgress bool + failures map[string]int + createRace bool + updateRace bool +} + +func newFixture(t *testing.T) *fixture { + t.Helper() + + f := &fixture{ + t: t, + ctrl: gomock.NewController(t), + builds: map[string]entity.Build{}, + requests: map[string]entity.Request{}, + observations: map[string]map[issueKey]entity.BuildIssue{}, + opens: map[issueKey]entity.OpenIssue{}, + logs: map[logKey]entity.IssueLog{}, + failures: map[string]int{}, + } + + stores := storagemock.NewMockFactory(f.ctrl) + store := storagemock.NewMockStorage(f.ctrl) + builds := storagemock.NewMockBuildStore(f.ctrl) + requests := storagemock.NewMockRequestStore(f.ctrl) + observations := storagemock.NewMockBuildIssueStore(f.ctrl) + opens := storagemock.NewMockOpenIssueStore(f.ctrl) + logs := storagemock.NewMockIssueLogStore(f.ctrl) + progress := storagemock.NewMockIssueTrackingStore(f.ctrl) + + stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(store, nil).AnyTimes() + store.EXPECT().GetBuildStore().Return(builds).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetBuildIssueStore().Return(observations).AnyTimes() + store.EXPECT().GetOpenIssueStore().Return(opens).AnyTimes() + store.EXPECT().GetIssueLogStore().Return(logs).AnyTimes() + store.EXPECT().GetIssueTrackingStore().Return(progress).AnyTimes() + + builds.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, key string) (entity.Build, error) { + value, ok := f.builds[key] + if !ok { + return entity.Build{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + + requests.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, key string) (entity.Request, error) { + value, ok := f.requests[key] + if !ok { + return entity.Request{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + + observations.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, buildID string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) { + if err := f.fail("observations/list"); err != nil { + return nil, err + } + keys := make([]issueKey, 0) + for key := range f.observations[buildID] { + if bounds.After.TaskID == "" || key.task > bounds.After.TaskID || (key.task == bounds.After.TaskID && key.unit > bounds.After.UnitID) { + keys = append(keys, key) + } + } + sortKeys(keys) + if len(keys) > bounds.Limit { + keys = keys[:bounds.Limit] + } + result := make([]entity.BuildIssue, 0, len(keys)) + for _, key := range keys { + result = append(result, f.observations[buildID][key]) + } + return result, nil + }).AnyTimes() + + observations.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error) { + if err := f.fail("observations/get"); err != nil { + return entity.BuildIssue{}, err + } + value, ok := f.observations[buildID][issueKey{taskID, unitID}] + if !ok { + return entity.BuildIssue{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + + opens.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, taskID, unitID string) (entity.OpenIssue, error) { + value, ok := f.opens[issueKey{taskID, unitID}] + if !ok { + return entity.OpenIssue{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + + opens.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue) error { + if err := f.fail("opens/create"); err != nil { + return err + } + key := issueKey{issue.TaskID, issue.UnitID} + if _, ok := f.opens[key]; ok { + return storage.ErrAlreadyExists + } + f.opens[key] = issue + return nil + }).AnyTimes() + + opens.EXPECT().Delete(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue) error { + if err := f.fail("opens/delete"); err != nil { + return err + } + key := issueKey{issue.TaskID, issue.UnitID} + if f.opens[key].FirstSeenBuildID == issue.FirstSeenBuildID { + delete(f.opens, key) + } + return nil + }).AnyTimes() + + opens.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, bounds storage.OpenIssueRange) ([]entity.OpenIssue, error) { + if err := f.fail("opens/list"); err != nil { + return nil, err + } + keys := make([]issueKey, 0) + for key := range f.opens { + if bounds.After.TaskID == "" || key.task > bounds.After.TaskID || (key.task == bounds.After.TaskID && key.unit > bounds.After.UnitID) { + keys = append(keys, key) + } + } + sortKeys(keys) + if len(keys) > bounds.Limit { + keys = keys[:bounds.Limit] + } + result := make([]entity.OpenIssue, 0, len(keys)) + for _, key := range keys { + result = append(result, f.opens[key]) + } + return result, nil + }).AnyTimes() + + logs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.IssueLog) error { + if err := f.fail("logs/" + string(log.Event)); err != nil { + return err + } + key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Event} + if _, ok := f.logs[key]; ok { + return storage.ErrAlreadyExists + } + f.logs[key] = log + return nil + }).AnyTimes() + + progress.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.IssueTrackingState, error) { + if err := f.fail("progress/get"); err != nil { + return entity.IssueTrackingState{}, err + } + if !f.hasProgress { + return entity.IssueTrackingState{}, storage.ErrNotFound + } + return f.progress, nil + }).AnyTimes() + + progress.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, state entity.IssueTrackingState) error { + if err := f.fail("progress/create"); err != nil { + return err + } + if f.createRace { + f.createRace = false + f.hasProgress = true + f.progress = entity.IssueTrackingState{RequestID: "11", BuildID: "build/11", Phase: entity.IssueTrackingPhaseRefreshed, Version: 1} + return storage.ErrAlreadyExists + } + if f.hasProgress { + return storage.ErrAlreadyExists + } + f.progress = state + f.hasProgress = true + return nil + }).AnyTimes() + + progress.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, state entity.IssueTrackingState, oldVersion, newVersion int32) error { + if state.IsPending() { + if err := f.fail("progress/start"); err != nil { + return err + } + } + if f.updateRace && state.IsPending() { + f.updateRace = false + f.progress = entity.IssueTrackingState{RequestID: "11", BuildID: "build/11", Phase: entity.IssueTrackingPhaseRefreshed, Version: oldVersion + 1} + return storage.ErrVersionMismatch + } + if !f.hasProgress || f.progress.Version != oldVersion { + return storage.ErrVersionMismatch + } + if !state.IsPending() { + if err := f.fail("progress/complete"); err != nil { + return err + } + } + state.Version = newVersion + f.progress = state + return nil + }).AnyTimes() + + f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, stores, stovepipemq.TopicKeyIssueTrack, "test-issuetrack") + f.controller.nowMs = func() int64 { return 1000 } + return f +} + +func sortKeys(keys []issueKey) { + sort.Slice(keys, func(i, j int) bool { + if keys[i].task != keys[j].task { + return keys[i].task < keys[j].task + } + return keys[i].unit < keys[j].unit + }) +} + +func (f *fixture) fail(operation string) error { + if f.failures[operation] > 0 { + f.failures[operation]-- + return assert.AnError + } + return nil +} + +func (f *fixture) source(buildID, requestID string, status entity.BuildStatus) { + f.builds[buildID] = entity.Build{ID: buildID, RequestID: requestID, Status: status, Version: 1} + request := entity.Request{ID: requestID, Queue: testQueue, State: entity.RequestStateProcessing} + if status.IsTerminal() { + request.TerminalBuildID = buildID + switch status { + case entity.BuildStatusSucceeded: + request.State = entity.RequestStateSucceeded + case entity.BuildStatusCancelled: + request.State = entity.RequestStateCancelled + default: + request.State = entity.RequestStateFailed + } + } + f.requests[requestID] = request +} + +func (f *fixture) observe(buildID, task, unit string) { + if f.observations[buildID] == nil { + f.observations[buildID] = map[issueKey]entity.BuildIssue{} + } + f.observations[buildID][issueKey{task, unit}] = entity.BuildIssue{TaskID: task, UnitID: unit} +} + +func (f *fixture) seed(task, unit, buildID string) { + f.opens[issueKey{task, unit}] = entity.OpenIssue{TaskID: task, UnitID: unit, FirstSeenBuildID: buildID} + f.logs[logKey{issueKey{task, unit}, buildID, entity.IssueEventOpened}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: buildID, Event: entity.IssueEventOpened, TimestampMs: 1} +} + +func (f *fixture) message(buildID string, finalize bool) entityqueue.Message { + f.t.Helper() + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{BuildId: buildID, QueueName: testQueue, Finalize: finalize}) + require.NoError(f.t, err) + msg := entityqueue.NewMessage(publish.IntentID(buildID, "issuetrack"), payload, testQueue, nil) + msg.Tenant = testQueue + return msg +} + +func (f *fixture) process(buildID string, finalize bool) error { + f.t.Helper() + return f.deliver(context.Background(), f.message(buildID, finalize)) +} + +func (f *fixture) deliver(ctx context.Context, msg entityqueue.Message) error { + f.t.Helper() + delivery := consumermock.NewMockDelivery(f.ctrl) + delivery.EXPECT().Message().Return(msg) + return f.controller.Process(ctx, delivery) +} + +func TestRefreshOpensIssuesWithoutResolving(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + f.seed("old", "unit", "build/1") + f.observe("build/10", "task", "") + + require.NoError(t, f.process("build/10", false)) + require.Len(t, f.opens, 2) + require.Equal(t, "build/10", f.opens[issueKey{"task", ""}].FirstSeenBuildID) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Equal(t, int32(2), f.progress.Version) + require.Len(t, f.logs, 2) + + f.controller.nowMs = func() int64 { return 2000 } + f.observe("build/10", "task", "next") + + require.NoError(t, f.process("build/10", false)) + require.Len(t, f.opens, 3) + require.Len(t, f.logs, 3) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build/10", entity.IssueEventOpened}].TimestampMs) +} + +func TestFinalAnalysisKeepsOpensAndResolves(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusFailed) + f.seed("a", "unit", "build/1") + f.seed("b", "unit", "build/1") + f.observe("build/10", "b", "unit") + f.observe("build/10", "c", "unit") + + require.NoError(t, f.process("build/10", true)) + require.Len(t, f.opens, 2) + require.NotContains(t, f.opens, issueKey{"a", "unit"}) + require.Equal(t, "build/1", f.opens[issueKey{"b", "unit"}].FirstSeenBuildID) + require.Equal(t, "build/10", f.opens[issueKey{"c", "unit"}].FirstSeenBuildID) + require.Contains(t, f.logs, logKey{issueKey{"a", "unit"}, "build/10", entity.IssueEventResolved}) + require.Len(t, f.logs, 4) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + + f.observe("build/10", "late", "unit") + + require.NoError(t, f.process("build/10", false)) + + require.NoError(t, f.process("build/10", true)) + require.Len(t, f.opens, 2) + require.Len(t, f.logs, 4) +} + +func TestEmptyFinalResultResolvesAcrossBatches(t *testing.T) { + f := newFixture(t) + f.source("new", "10", entity.BuildStatusSucceeded) + for i := 0; i < _issueBatchSize+3; i++ { + f.seed("task", fmt.Sprintf("unit/%03d", i), "old") + } + + require.NoError(t, f.process("new", true)) + require.Empty(t, f.opens) + require.Len(t, f.logs, 2*(_issueBatchSize+3)) + require.True(t, f.hasProgress) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + f.source("older", "9", entity.BuildStatusRunning) + f.observe("older", "task", "late") + + require.NoError(t, f.process("older", false)) + require.Empty(t, f.opens) +} + +func TestRefreshTraversesAllBatches(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + for i := 0; i < _issueBatchSize+3; i++ { + f.observe("build/10", "task", fmt.Sprintf("unit/%03d", i)) + } + + require.NoError(t, f.process("build/10", false)) + require.Len(t, f.opens, _issueBatchSize+3) + require.Len(t, f.logs, _issueBatchSize+3) +} + +func TestCancelledFinalDoesNotResolve(t *testing.T) { + f := newFixture(t) + f.source("cancelled", "10", entity.BuildStatusCancelled) + f.seed("old", "unit", "build/1") + f.observe("cancelled", "confirmed", "unit") + + require.NoError(t, f.process("cancelled", true)) + require.Len(t, f.opens, 2) + require.Len(t, f.logs, 2) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) +} + +func TestRequestOrderNotBuildIDOrder(t *testing.T) { + f := newFixture(t) + f.source("aaa-new", "10", entity.BuildStatusRunning) + f.source("zzz-old", "9", entity.BuildStatusSucceeded) + f.observe("aaa-new", "new", "unit") + f.observe("zzz-old", "old", "unit") + + require.NoError(t, f.process("aaa-new", false)) + before := f.progress + + require.NoError(t, f.process("zzz-old", false)) + + require.NoError(t, f.process("zzz-old", true)) + require.Equal(t, before, f.progress) + require.Len(t, f.opens, 1) + require.Len(t, f.logs, 1) + require.Contains(t, f.opens, issueKey{"new", "unit"}) +} + +func TestNewerBuildCanReopenResolvedIssue(t *testing.T) { + f := newFixture(t) + f.seed("task", "unit", "old") + f.source("build/10", "10", entity.BuildStatusSucceeded) + + require.NoError(t, f.process("build/10", true)) + f.source("build/11", "11", entity.BuildStatusRunning) + f.observe("build/11", "task", "unit") + + require.NoError(t, f.process("build/11", false)) + require.Equal(t, "build/11", f.opens[issueKey{"task", "unit"}].FirstSeenBuildID) + require.Len(t, f.logs, 3) +} + +func TestRetriesFinishAcceptedWork(t *testing.T) { + for _, operation := range []string{"logs/opened", "progress/complete"} { + t.Run("opening/"+operation, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + f.observe("build/10", "task", "unit") + f.failures[operation] = 1 + require.ErrorIs(t, f.process("build/10", false), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + + require.NoError(t, f.process("build/10", false)) + require.Len(t, f.opens, 1) + require.Len(t, f.logs, 1) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + }) + } + for _, operation := range []string{"logs/resolved", "opens/delete", "progress/complete"} { + t.Run("resolution/"+operation, func(t *testing.T) { + f := newFixture(t) + f.seed("task", "unit", "old") + f.source("build/10", "10", entity.BuildStatusSucceeded) + f.failures[operation] = 1 + require.ErrorIs(t, f.process("build/10", true), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) + if operation == "logs/resolved" { + require.Len(t, f.opens, 1) + } + f.controller.nowMs = func() int64 { return 2000 } + + require.NoError(t, f.process("build/10", true)) + require.Empty(t, f.opens) + require.Len(t, f.logs, 2) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + }) + } +} + +func TestNewDeliveryResumesPendingResolutionBeforeReopening(t *testing.T) { + f := newFixture(t) + f.seed("task", "unit", "old") + f.source("build/10", "10", entity.BuildStatusSucceeded) + f.failures["opens/delete"] = 1 + require.ErrorIs(t, f.process("build/10", true), assert.AnError) + f.source("build/11", "11", entity.BuildStatusRunning) + f.observe("build/11", "task", "unit") + + require.NoError(t, f.process("build/11", false)) + require.Equal(t, "build/11", f.opens[issueKey{"task", "unit"}].FirstSeenBuildID) + require.Len(t, f.logs, 3) + require.Equal(t, "11", f.progress.RequestID) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) +} + +func TestNewDeliveryResumesPendingRefresh(t *testing.T) { + f := newFixture(t) + f.source("old", "9", entity.BuildStatusRunning) + f.observe("old", "old", "unit") + f.failures["observations/list"] = 1 + require.ErrorIs(t, f.process("old", false), assert.AnError) + f.source("new", "10", entity.BuildStatusRunning) + f.observe("new", "new", "unit") + + require.NoError(t, f.process("new", false)) + require.Len(t, f.opens, 2) + require.Len(t, f.logs, 2) + require.Equal(t, "10", f.progress.RequestID) +} + +func TestClaimConflictsRecheckSourceOrder(t *testing.T) { + for _, kind := range []string{"create", "update"} { + t.Run(kind, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + f.observe("build/10", "task", "unit") + if kind == "create" { + f.createRace = true + } else { + f.hasProgress = true + f.progress = entity.IssueTrackingState{RequestID: "9", BuildID: "build/9", Phase: entity.IssueTrackingPhaseRefreshed, Version: 2} + f.updateRace = true + } + + require.NoError(t, f.process("build/10", false)) + require.Empty(t, f.opens) + require.Empty(t, f.logs) + require.Equal(t, "11", f.progress.RequestID) + }) + } +} + +func TestFinalRequiresTerminalWinningBuild(t *testing.T) { + for _, kind := range []string{"nonterminal", "missing outcome", "different winning build"} { + t.Run(kind, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + if kind != "nonterminal" { + f.source("build/10", "10", entity.BuildStatusSucceeded) + request := f.requests["10"] + if kind == "missing outcome" { + request.TerminalBuildID = "" + } else { + request.TerminalBuildID = "other" + } + f.requests["10"] = request + } + err := f.process("build/10", true) + if kind == "different winning build" { + require.NoError(t, err) + } else { + require.Error(t, err) + } + require.False(t, f.hasProgress) + require.Empty(t, f.opens) + require.Empty(t, f.logs) + }) + } +} + +func TestReadErrorsNeverImplyResolution(t *testing.T) { + for _, operation := range []string{"progress/get", "observations/list", "observations/get", "opens/list"} { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusSucceeded) + f.seed("task", "unit", "old") + f.failures[operation] = 1 + require.ErrorIs(t, f.process("build/10", true), assert.AnError) + require.Len(t, f.opens, 1) + require.Len(t, f.logs, 1) + }) + } +} + +func TestRejectsInvalidNotifications(t *testing.T) { + for _, kind := range []string{"malformed", "missing build", "missing queue", "wrong tenant", "wrong partition", "missing source", "wrong request queue", "invalid request ID"} { + t.Run(kind, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + msg := f.message("build/10", false) + switch kind { + case "malformed": + msg.Payload = []byte("{") + case "missing build": + msg = f.message("", false) + case "missing queue": + msg.Payload = []byte(`{"build_id":"build/10"}`) + case "wrong tenant": + msg.Tenant = "other" + case "wrong partition": + msg.PartitionKey = "build/10" + case "missing source": + delete(f.builds, "build/10") + case "wrong request queue": + request := f.requests["10"] + request.Queue = "other" + f.requests["10"] = request + case "invalid request ID": + build := f.builds["build/10"] + build.RequestID = "request/10" + f.builds["build/10"] = build + f.requests["request/10"] = entity.Request{ID: "request/10", Queue: testQueue} + } + require.Error(t, f.deliver(context.Background(), msg)) + require.False(t, f.hasProgress) + require.Empty(t, f.opens) + }) + } +} + +func TestCancelledContextDoesNotAcceptWork(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.ErrorIs(t, f.deliver(ctx, f.message("build/10", false)), context.Canceled) + require.False(t, f.hasProgress) +} + +func TestFinalNotificationFinishesPendingRefresh(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + f.seed("old", "unit", "build/1") + f.observe("build/10", "new", "unit") + f.failures["logs/opened"] = 1 + require.ErrorIs(t, f.process("build/10", false), assert.AnError) + + f.source("build/10", "10", entity.BuildStatusFailed) + require.NoError(t, f.process("build/10", true)) + + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Len(t, f.opens, 1) + require.Contains(t, f.opens, issueKey{"new", "unit"}) + require.Contains(t, f.logs, logKey{issueKey{"old", "unit"}, "build/10", entity.IssueEventResolved}) +} + +func TestNewFinalFinishesPendingOlderRefresh(t *testing.T) { + f := newFixture(t) + f.source("old", "9", entity.BuildStatusRunning) + f.observe("old", "task", "unit") + f.failures["opens/create"] = 1 + require.ErrorIs(t, f.process("old", false), assert.AnError) + + f.source("new", "10", entity.BuildStatusSucceeded) + require.NoError(t, f.process("new", true)) + + require.Empty(t, f.opens) + require.Len(t, f.logs, 2) + require.Equal(t, "10", f.progress.RequestID) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) +} + +func TestPendingFinalizationValidatesItsOwnBuildAndRequest(t *testing.T) { + for _, invalidSource := range []string{"missing build", "missing request", "wrong request", "nonterminal build", "different winning build"} { + t.Run(invalidSource, func(t *testing.T) { + f := newFixture(t) + f.source("pending", "10", entity.BuildStatusSucceeded) + f.seed("old", "unit", "build/1") + f.failures["logs/resolved"] = 1 + require.ErrorIs(t, f.process("pending", true), assert.AnError) + pending := f.progress + + build := f.builds["pending"] + request := f.requests["10"] + switch invalidSource { + case "missing build": + delete(f.builds, "pending") + case "missing request": + delete(f.requests, "10") + case "wrong request": + request.ID = "9" + f.requests["10"] = request + case "nonterminal build": + build.Status = entity.BuildStatusRunning + f.builds["pending"] = build + case "different winning build": + request.TerminalBuildID = "other" + f.requests["10"] = request + } + + f.source("new", "11", entity.BuildStatusRunning) + f.observe("new", "new", "unit") + + require.Error(t, f.process("new", false)) + require.Equal(t, pending, f.progress) + require.Len(t, f.opens, 1) + require.Contains(t, f.opens, issueKey{"old", "unit"}) + require.Len(t, f.logs, 1) + }) + } +} + +func TestRefreshNotificationCompletesPendingFinalization(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusSucceeded) + f.seed("old", "unit", "build/1") + f.failures["opens/delete"] = 1 + require.ErrorIs(t, f.process("build/10", true), assert.AnError) + + require.NoError(t, f.process("build/10", false)) + + require.Empty(t, f.opens) + require.Len(t, f.logs, 2) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) +} + +func TestDecideReconciliationForSameBuild(t *testing.T) { + tests := []struct { + phase entity.IssueTrackingPhase + finalize bool + want reconciliationAction + }{ + {entity.IssueTrackingPhaseUnknown, false, _startRefresh}, + {entity.IssueTrackingPhaseUnknown, true, _startFinalization}, + {entity.IssueTrackingPhaseRefreshing, false, _resumeAndFinishNotification}, + {entity.IssueTrackingPhaseRefreshing, true, _resumeAndRecheckNotification}, + {entity.IssueTrackingPhaseRefreshed, false, _startRefresh}, + {entity.IssueTrackingPhaseRefreshed, true, _startFinalization}, + {entity.IssueTrackingPhaseFinalizing, false, _resumeAndFinishNotification}, + {entity.IssueTrackingPhaseFinalizing, true, _resumeAndFinishNotification}, + {entity.IssueTrackingPhaseFinalized, false, _ignoreNotification}, + {entity.IssueTrackingPhaseFinalized, true, _ignoreNotification}, + } + + for _, tt := range tests { + t.Run(fmt.Sprintf("%s/finalize=%t", tt.phase, tt.finalize), func(t *testing.T) { + build := entity.Build{ID: "build/10", RequestID: "10", Status: entity.BuildStatusFailed} + request := entity.Request{ID: "10", Queue: testQueue, TerminalBuildID: build.ID} + var progress entity.IssueTrackingState + if tt.phase != entity.IssueTrackingPhaseUnknown { + progress = entity.IssueTrackingState{RequestID: request.ID, BuildID: build.ID, Phase: tt.phase, Version: 1} + } + + action, err := decideReconciliation(progress, build, request, tt.finalize) + + require.NoError(t, err) + require.Equal(t, tt.want, action) + }) + } +} + +func TestDecideReconciliationOrdersPendingAndIncomingWork(t *testing.T) { + tests := []struct { + name string + phase entity.IssueTrackingPhase + requestID string + buildID string + nonterminal bool + noOutcome bool + want reconciliationAction + wantErr bool + }{ + {name: "resume older refresh", phase: entity.IssueTrackingPhaseRefreshing, requestID: "9", want: _resumeAndRecheckNotification}, + {name: "resume newer finalization", phase: entity.IssueTrackingPhaseFinalizing, requestID: "11", want: _resumeAndRecheckNotification}, + {name: "resume another build of same request", phase: entity.IssueTrackingPhaseRefreshing, requestID: "10", buildID: "other", want: _resumeAndRecheckNotification}, + {name: "ignore older final", phase: entity.IssueTrackingPhaseRefreshed, requestID: "11", nonterminal: true, want: _ignoreNotification}, + {name: "ignore already finalized", phase: entity.IssueTrackingPhaseFinalized, requestID: "10", nonterminal: true, want: _ignoreNotification}, + {name: "accept after older refresh", phase: entity.IssueTrackingPhaseRefreshed, requestID: "9", want: _startFinalization}, + {name: "accept after older final", phase: entity.IssueTrackingPhaseFinalized, requestID: "9", want: _startFinalization}, + {name: "finish pending before validating final", phase: entity.IssueTrackingPhaseRefreshing, requestID: "10", nonterminal: true, want: _resumeAndRecheckNotification}, + {name: "resume final based on saved work", phase: entity.IssueTrackingPhaseFinalizing, requestID: "10", nonterminal: true, want: _resumeAndFinishNotification}, + {name: "reject nonterminal final", nonterminal: true, wantErr: true}, + {name: "reject final without outcome", noOutcome: true, wantErr: true}, + {name: "reject invalid ordering ID", phase: entity.IssueTrackingPhaseRefreshed, requestID: "invalid", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + build := entity.Build{ID: "build/10", RequestID: "10", Status: entity.BuildStatusFailed} + request := entity.Request{ID: "10", Queue: testQueue, TerminalBuildID: build.ID} + if tt.nonterminal { + build.Status = entity.BuildStatusRunning + } + if tt.noOutcome { + request.TerminalBuildID = "" + } + var progress entity.IssueTrackingState + if tt.phase != entity.IssueTrackingPhaseUnknown { + buildID := tt.buildID + if buildID == "" { + buildID = build.ID + } + progress = entity.IssueTrackingState{RequestID: tt.requestID, BuildID: buildID, Phase: tt.phase, Version: 1} + } + + action, err := decideReconciliation(progress, build, request, true) + + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Equal(t, tt.want, action) + } + }) + } +} + +func TestFailedAcceptanceDoesNotApplyIssues(t *testing.T) { + for _, operation := range []string{"progress/create", "progress/start"} { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + f.source("build/10", "10", entity.BuildStatusRunning) + f.seed("old", "unit", "build/1") + f.observe("build/10", "new", "unit") + if operation == "progress/start" { + f.hasProgress = true + f.progress = entity.IssueTrackingState{RequestID: "9", BuildID: "build/9", Phase: entity.IssueTrackingPhaseRefreshed, Version: 2} + } + before := f.progress + f.failures[operation] = 1 + + require.ErrorIs(t, f.process("build/10", false), assert.AnError) + require.Equal(t, before, f.progress) + require.Len(t, f.opens, 1) + require.Len(t, f.logs, 1) + + require.NoError(t, f.process("build/10", false)) + require.Len(t, f.opens, 2) + require.Len(t, f.logs, 2) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + }) + } +} diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index a4f05b33d..473b0c01a 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -14,10 +14,10 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali ## Issue tracking handoff -`IssueTrack` binds to `TopicKeyIssueTrack` and identifies a persisted build and its queue. Notifications are partitioned by queue so both running and final issue-state updates share one serialized path, independently of build-partitioned analysis polling. The message carries identifiers rather than a failure collection; per-build observations are read through `BuildIssueStore.List`. +`IssueTrack` binds to `TopicKeyIssueTrack` and identifies a persisted build and its queue. Notifications are partitioned by queue so both refresh and final issue-state updates share one serialized path, independently of build-partitioned analysis polling. The message carries identifiers rather than a failure collection; per-build observations are read through `BuildIssueStore.List`. Publish only after detection and observation persistence succeed. With `finalize` unset, the notification permits opening newly detected issues, not resolution by absence. With `finalize` set, it requests final reconciliation after a successful terminal analysis. Publish that notification even when no failures were detected; terminal Build status alone does not replace the handoff. Publication retries must also follow duplicate observation writes so a crash between persistence and publication does not strand work. -Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before resolving issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID. +Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. A final notification attests that its persisted observations are complete and authoritative for the queue's relevant validation scope; a producer unable to establish that must not request absence-based resolution. The reconciler rejects older requests and ignores refresh notifications after finalization. Notification identities must not suppress later refreshes or finalization. Consumers must tolerate duplicate publications. Business behavior follows the payload, not a parsed message ID. -Only the handoff contract is defined. No publisher, consumer, topic registration, reconciliation progress, or writes to open issues and lifecycle history are enabled. +The [issue-tracking controller](../../controller/issuetrack/README.md) implements reconciliation and operational progress. No publisher or consumer registration is enabled in the reference host; runtime handoff, recovery wiring, and user-facing summary materialization remain follow-up work. diff --git a/stovepipe/core/messagequeue/proto/issuetrack.proto b/stovepipe/core/messagequeue/proto/issuetrack.proto index 52acce9f5..d9ac6e415 100644 --- a/stovepipe/core/messagequeue/proto/issuetrack.proto +++ b/stovepipe/core/messagequeue/proto/issuetrack.proto @@ -34,6 +34,8 @@ message IssueTrack { string queue_name = 2; // finalize requests final reconciliation after successful terminal detection // and persistence, including when no failures were detected. + // Persisted observations must be the complete authoritative failure set for + // the queue's relevant validation scope before absence may imply resolution. // Otherwise the notification permits opening issues, not resolution by absence. bool finalize = 3; } diff --git a/stovepipe/core/messagequeue/protopb/issuetrack.pb.go b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go index 38a282c26..4491da397 100644 --- a/stovepipe/core/messagequeue/protopb/issuetrack.pb.go +++ b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go @@ -47,6 +47,8 @@ type IssueTrack struct { QueueName string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` // finalize requests final reconciliation after successful terminal detection // and persistence, including when no failures were detected. + // Persisted observations must be the complete authoritative failure set for + // the queue's relevant validation scope before absence may imply resolution. // Otherwise the notification permits opening issues, not resolution by absence. Finalize bool `protobuf:"varint,3,opt,name=finalize,proto3" json:"finalize,omitempty"` unknownFields protoimpl.UnknownFields diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index a36d1ad77..b8511a224 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -58,10 +58,10 @@ The build ID identifies the build where the issue was first seen or confirmed re The MySQL key uses the same binary identifier limits as the issue stores, plus a 32-byte event field, for a maximum of 2586 bytes. Reads and creates use primary-key operations; there is no update or deletion API, metadata collection, or repair state. -History persistence does not alter open-issue membership or validate transitions against it. No runtime writer is enabled here. A later reconciliation controller must own opening/resolution decisions, retry-safe writes to both stores, and protection against stale build analysis. History enumeration is outside this storage contract. +History persistence does not alter open-issue membership or validate transitions against it. The issue-tracking controller owns opening/resolution decisions, retry-safe issue/history writes, and protection against stale build analysis; the reference host does not enable it yet. History enumeration is outside this storage contract. ## Issue reconciliation progress -`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, whether observation or final writes are pending or complete, and its optimistic version. Keeping it after the last open row is removed preserves the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It identifies the accepted source request and build, whether observation/final writes are pending or complete, and its optimistic version. Keeping it after the last open row is removed preserves the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. -The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of pending writes, and completion; no runtime writer is enabled here. +The controller computes replacement versions and decides source ordering. A pending pass must be resumed before another source is accepted, and the completed phase must follow all issue/history writes. Store updates neither infer request ordering nor advance versions. See [issue tracking](../../controller/issuetrack/README.md) for the replay protocol and final-result input requirements. The reference host does not enable these writes yet.