diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index a4b3dc2ea..b14946102 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -9,7 +9,8 @@ Stovepipe retains an append-only request log for each validation request. Its in - `validation_fact_recorded`; - `project_facts_recorded`; - `record_abandoned`; -- `build_analysis_abandoned`. +- `build_analysis_abandoned`; +- `issue_tracking_abandoned`. The model deliberately follows SubmitQueue's distinction between statuses describing where a request is and events describing important activity that does not move it. It remains a bounded request-lifecycle log rather than a generic event bus or an audit of every correlated operation. @@ -128,6 +129,7 @@ Immutable Request context such as URI, build strategy, base URI, and terminal bu | `project_facts_recorded` | All project validation facts returned for a request became durable. | Project-fact count metadata and completion time | | `record_abandoned` | Record-stage work stopped after exhausting primary retries. | Event retention time | | `build_analysis_abandoned` | Build analysis could not complete; issue observations may be incomplete. This is not a validation verdict. | Build ID metadata and event retention time | +| `issue_tracking_abandoned` | Issue tracking stopped before completing a pass; existing issue changes remain, but reconciliation may be incomplete. This is not a validation verdict. | Build ID metadata and event retention time | Build running and unchanged polls are not retained. Trigger and terminal result explain the request outcome without turning polling into an unbounded log. Individual project facts remain outside the vocabulary; one batch-completion event summarizes their durable recording. @@ -219,6 +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. | | 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/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index f8d6359fe..cea99a725 100644 --- a/stovepipe/controller/issuetrack/BUILD.bazel +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["issuetrack.go"], + srcs = [ + "dlq.go", + "issuetrack.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack", visibility = ["//visibility:public"], deps = [ @@ -13,6 +16,7 @@ go_library( "//stovepipe/core/fixrequest:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", @@ -24,17 +28,21 @@ go_test( name = "go_default_test", srcs = [ "delivery_test.go", + "dlq_test.go", "fix_handoff_test.go", "fixture_test.go", "issuetrack_test.go", ], embed = [":go_default_library"], deps = [ + "//platform/base/failure:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/core/requestlog:go_default_library", + "//stovepipe/core/requestlog/mock:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index abd4fb740..b3bfbd376 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -38,14 +38,27 @@ No closure occurs before all accumulated build issues have been processed. A tas Every delivery processes at most 100 task/unit keys from one ordered range, including a revisited cursor and retained closed rows. Refreshing reads build issues; finalizing reads issue states. The cursor is always inclusive: revisit it to finish the issue and its history, then continue after it. A full batch retains its active phase and requests redelivery. Moving from refreshing to finalizing conditionally saves a reset cursor before any closing work begins. -An unfinished pass completes before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized. A terminal pass requires a terminal build chosen by its Request in both active phases. +An unfinished pass completes or is safely abandoned before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized or its terminal pass was abandoned. A terminal pass requires a terminal build chosen by its Request in both active phases. Build IDs are opaque; source order comes from Request IDs. All scans are bounded reads under leading primary-key prefixes. There is no transaction spanning progress, issue state, and history. ## Recovery and activation -An error returns to normal consumer retry. The next delivery resumes the cursor before advancing. Permanently failing cursor work blocks issue tracking for that queue; terminal recovery and consumer registration remain follow-up work. This PR does not enable consumption. The host must provide a fix publication route and a repair executor when registering the fix consumer. +An error returns to normal consumer retry. The next delivery resumes the cursor before advancing. Once that retry policy dead-letters a notification, the recovery controller abandons the failed pass rather than rerunning its remaining batches: + +1. Conditionally enter `abandoning`, retaining the source, notification, and cursor. Normal deliveries hold for 30 seconds while this phase is present; they cannot start another pass or resume issue application. +2. If there is a cursor, read its issue and conditionally advance progress after that read. Advance the issue's version without changing its status or transition metadata. This invalidates a paused worker's earlier row guard. If the row has not been created yet, finish that one opening instead: the normal worker can only have saved such a cursor after finding a reported build issue. Creating it prevents the delayed create from succeeding. Closed issue rows are never deleted. +3. Repair the cursor's opening or closing history from the retained row. Record an `issue_tracking_abandoned` Request-history event keyed by the failed notification, with its build ID. +4. Conditionally enter `abandoned`. Keep the source-order boundary and notification identity, so the same delivery and older builds cannot restart the abandoned work. + +Accepted opens and closes remain in place. Recovery does not scan either issue range, close the remaining issues, change Request or Build outcomes, or release build slots. An abandoned ordinary refresh still permits a new refresh or terminal notification for that Request; an abandoned terminal pass permits only newer Requests. A later successful build can reconcile issues left unchanged by the failed pass. + +Recovery must finish fencing the current issue and retaining its history before admitting newer work. If a normal write wins the issue CAS first, recovery retries against that accepted row. If a replacement recovery worker advances first, its issue-version change rejects the old write. Merely marking the pass abandoned without touching the cursor row would not provide this protection. + +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. ## 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 issue-tracking consumption or recovery. +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. diff --git a/stovepipe/controller/issuetrack/dlq.go b/stovepipe/controller/issuetrack/dlq.go new file mode 100644 index 000000000..14440e55c --- /dev/null +++ b/stovepipe/controller/issuetrack/dlq.go @@ -0,0 +1,248 @@ +// 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" + "crypto/sha256" + "errors" + "fmt" + "math" + + "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/core/requestlog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// Avoid spinning the primary consumer while recovery retries storage or history writes. +const _recoveryDelayMs int64 = 30000 + +// DLQController abandons incomplete issue tracking while retaining accepted issue changes. +type DLQController struct { + controller *Controller + materializer requestlog.Materializer +} + +func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, topicKey consumer.TopicKey, consumerGroup string) *DLQController { + return &DLQController{ + controller: NewController(logger, scope, stores, consumer.TopicRegistry{}, topicKey, consumerGroup), + materializer: materializer, + } +} + +// Process stops a failed pass without traversing its remaining issues. +// Recovery errors require an always-retryable DLQ consumer; unusable messages are logged and acknowledged. +func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + notification := &stovepipemq.IssueTrack{} + if err := stovepipemq.Unmarshal(msg.Payload, notification); err != nil { + c.controller.logger.Errorw("discarding malformed issue tracking dlq message", "message_id", msg.ID, "error", err) + return nil + } + queue := notification.GetQueueName() + if err := entityqueue.ValidatePayloadQueue(msg, queue); err != nil { + c.controller.logger.Errorw("discarding issue tracking dlq message with invalid queue identity", "message_id", msg.ID, "error", err) + return nil + } + if msg.ID == "" || queue == "" || notification.GetBuildId() == "" || msg.PartitionKey != queue { + c.controller.logger.Errorw("discarding issue tracking dlq message with invalid identity", "message_id", msg.ID, "queue", queue) + return nil + } + + store, err := c.controller.stores.For(storage.Config{QueueName: queue}) + if err != nil { + c.controller.logger.Errorw("discarding issue tracking dlq message for unresolvable queue", "queue", queue, "message_id", msg.ID, "error", err) + return nil + } + build, err := loader.ByID(ctx, notification.GetBuildId(), store.GetBuildStore().Get, "build") + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + c.controller.logger.Errorw("discarding issue tracking dlq message for missing build", "message_id", msg.ID, "error", err) + return nil + } + return err + } + request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request") + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + c.controller.logger.Errorw("discarding issue tracking dlq message for missing request", "message_id", msg.ID, "error", err) + return nil + } + return err + } + if request.Queue != queue { + c.controller.logger.Errorw("discarding issue tracking dlq message for request in another queue", "message_id", msg.ID, "request_id", request.ID) + return nil + } + if err := id.Validate(request.ID); err != nil { + c.controller.logger.Errorw("discarding issue tracking dlq message with invalid request id", "message_id", msg.ID, "error", err) + return nil + } + + progress, err := c.abandonPass(ctx, store, request, notification, msg.ID) + if err != nil { + return err + } + if progress.Phase != entity.IssueTrackingPhaseAbandoned { + return nil + } + + originalFailure, hasFailure := delivery.Failure() + fields := []any{"queue", queue, "message_id", msg.ID, "build_id", progress.BuildID, "request_id", progress.RequestID, "notification_id", progress.NotificationID} + if hasFailure { + fields = append(fields, "failure", originalFailure.Message, "failure_subjects", originalFailure.Subjects, "failure_detail", originalFailure.Detail) + } + c.controller.logger.Errorw("issue tracking abandoned; issue state may be incomplete", fields...) + metrics.NamedCounter(c.controller.metricsScope, "dlq", "tracking_abandoned", 1, metrics.TagsFromContext(ctx)...) + if progress.NotificationID != msg.ID { + delivery.Hold(0) + } + return nil +} + +func (c *DLQController) abandonPass(ctx context.Context, store storage.Storage, request entity.Request, notification *stovepipemq.IssueTrack, notificationID string) (entity.IssueTrackingState, error) { + tracking := store.GetIssueTrackingStore() + progress, err := tracking.Get(ctx) + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.IssueTrackingState{}, fmt.Errorf("load issue tracking recovery: %w", err) + } + if err == nil { + if err := progress.Validate(); err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("invalid issue tracking recovery: %w", err) + } + if progress.NotificationID == notificationID { + switch progress.Phase { + case entity.IssueTrackingPhaseRefreshed, entity.IssueTrackingPhaseFinalized: + return progress, nil + case entity.IssueTrackingPhaseAbandoned: + return progress, c.recordAbandonment(ctx, store, request.Queue, progress) + } + } else { + order, err := entity.CompareRequestID(request.ID, progress.RequestID) + if err != nil { + return entity.IssueTrackingState{}, err + } + if order < 0 || (order == 0 && progress.Finalize) { + return entity.IssueTrackingState{}, nil + } + switch progress.Phase { + case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing, entity.IssueTrackingPhaseAbandoning: + if order == 0 { + return entity.IssueTrackingState{}, nil + } + default: + progress = entity.IssueTrackingState{Version: progress.Version} + } + } + } + + if progress.NotificationID == "" { + if request.TerminalBuildID != "" && request.TerminalBuildID != notification.GetBuildId() { + return entity.IssueTrackingState{}, nil + } + progress.RequestID, progress.BuildID, progress.NotificationID = request.ID, notification.GetBuildId(), notificationID + progress.Finalize = notification.GetFinalize() + } + progress.Phase = entity.IssueTrackingPhaseAbandoning + if progress.Version == 0 { + progress.Version = 1 + if err := tracking.Create(ctx, progress); err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("start issue tracking recovery: %w", err) + } + } else { + progress, err = saveProgress(ctx, tracking, progress) + if err != nil { + return entity.IssueTrackingState{}, err + } + } + + return c.finishAbandonment(ctx, store, request.Queue, progress) +} + +func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (entity.IssueTrackingState, error) { + if progress.CursorTaskID != "" { + issues := store.GetIssueStateStore() + current, err := issues.Get(ctx, progress.CursorTaskID, progress.CursorUnitID) + missing := errors.Is(err, storage.ErrNotFound) + if err != nil && !missing { + return entity.IssueTrackingState{}, fmt.Errorf("read recovery cursor: %w", err) + } + if current.Version == math.MaxInt32 { + return entity.IssueTrackingState{}, fmt.Errorf("issue state version exhausted") + } + + // As in normal application, the post-read progress CAS must precede the issue CAS. + progress, err = saveProgress(ctx, store.GetIssueTrackingStore(), progress) + if err != nil { + return entity.IssueTrackingState{}, err + } + if missing { + current = entity.IssueState{ + TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID, + Status: entity.IssueStatusOpen, FirstSeenBuildID: progress.BuildID, + LastProcessedBuildID: progress.BuildID, LastChangedBuildID: progress.BuildID, + ChangedAtMs: c.controller.nowMs(), Version: 1, + } + err = issues.Create(ctx, current) + } else { + newVersion := current.Version + 1 + err = issues.Update(ctx, current, current.Version, newVersion) + if err == nil { + current.Version = newVersion + } + } + if errors.Is(err, storage.ErrAlreadyExists) { + return entity.IssueTrackingState{}, fmt.Errorf("recovery cursor was concurrently created: %w", storage.ErrVersionMismatch) + } + if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("fence recovery cursor: %w", err) + } + if err := c.controller.recordHistory(ctx, store, current); err != nil { + return entity.IssueTrackingState{}, err + } + } + + if err := c.recordAbandonment(ctx, store, queue, progress); err != nil { + return entity.IssueTrackingState{}, err + } + progress.Phase = entity.IssueTrackingPhaseAbandoned + return saveProgress(ctx, store.GetIssueTrackingStore(), progress) +} + +func (c *DLQController) recordAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) error { + request := entity.Request{ID: progress.RequestID, Queue: queue} + // Notification IDs can exceed the request log's 255-character key. + identity := sha256.Sum256([]byte(progress.NotificationID)) + log := requestlog.NewRequestEventLog(request, entity.RequestEventIssueTrackingAbandoned, fmt.Sprintf("%x", identity), + map[string]string{requestlog.MetadataKeyBuildID: progress.BuildID}) + if err := c.materializer.PersistLog(ctx, store, log); err != nil { + return fmt.Errorf("record incomplete issue tracking: %w", err) + } + return nil +} + +func (c *DLQController) Name() string { return string(c.controller.topicKey) } +func (c *DLQController) TopicKey() consumer.TopicKey { return c.controller.topicKey } +func (c *DLQController) ConsumerGroup() string { return c.controller.consumerGroup } + +var _ consumer.Controller = (*DLQController)(nil) diff --git a/stovepipe/controller/issuetrack/dlq_test.go b/stovepipe/controller/issuetrack/dlq_test.go new file mode 100644 index 000000000..5fff7947e --- /dev/null +++ b/stovepipe/controller/issuetrack/dlq_test.go @@ -0,0 +1,414 @@ +// 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" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "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" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +type recoveryFixture struct { + *fixture + dlq *DLQController + events map[string]entity.RequestLog +} + +func newRecoveryFixture(t *testing.T) *recoveryFixture { + t.Helper() + f := &recoveryFixture{fixture: newFixture(t), events: map[string]entity.RequestLog{}} + materializer := requestlogmock.NewMockMaterializer(f.mocks) + materializer.EXPECT().PersistLog(gomock.Any(), f.store, gomock.Any()).DoAndReturn( + func(_ context.Context, _ storage.Storage, log entity.RequestLog) error { + if err := f.fail("recovery/history"); err != nil { + return err + } + log.TimestampMs = 1000 + require.NoError(t, log.Validate()) + require.Equal(t, entity.RequestEventIssueTrackingAbandoned, log.Event) + f.events[log.RequestID+"/"+log.ID] = log + return nil + }).AnyTimes() + f.dlq = NewDLQController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, materializer, consumer.TopicKey("issuetrack_dlq"), "test-dlq") + f.dlq.controller.nowMs = func() int64 { return 1000 } + return f +} + +func (f *recoveryFixture) recover(msg entityqueue.Message, hold bool) error { + f.t.Helper() + delivery := consumermock.NewMockDelivery(f.mocks) + delivery.EXPECT().Message().Return(msg) + delivery.EXPECT().Failure().Return(failure.Failure{Message: "storage unavailable"}, true).AnyTimes() + if hold { + delivery.EXPECT().Hold(int64(0)) + } + return f.dlq.Process(context.Background(), delivery) +} + +func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) { + for _, status := range []entity.IssueStatus{entity.IssueStatusOpen, entity.IssueStatusClosed} { + t.Run(string(status), func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.seedIssue("already-opened", "unit", entity.IssueStatusOpen) + f.seedIssue("already-closed", "unit", entity.IssueStatusClosed) + f.seedIssue("unprocessed", "unit", entity.IssueStatusOpen) + f.seedIssue("task", "unit", status) + progress := currentCursor() + progress.Phase, progress.Finalize = entity.IssueTrackingPhaseFinalizing, true + msg := f.message("build/100", true) + progress.NotificationID = msg.ID + f.retain(progress) + original := f.states[issueKey{"task", "unit"}] + untouched := map[issueKey]entity.IssueState{} + for key, state := range f.states { + untouched[key] = state + } + event := entity.IssueEventOpened + if status == entity.IssueStatusClosed { + event = entity.IssueEventResolved + } + delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event}) + f.failures["issue/list"], f.failures["build-issues/list"], f.failures["build-issues/get"] = 1, 1, 1 + + require.NoError(t, f.recover(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + for key, expected := range untouched { + if key == (issueKey{"task", "unit"}) { + expected.Version++ + } + require.Equal(t, expected, f.states[key]) + } + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event}) + require.Len(t, f.events, 1) + require.Equal(t, 1, f.failures["issue/list"]) + require.Equal(t, 1, f.failures["build-issues/list"]) + require.Equal(t, 1, f.failures["build-issues/get"]) + for _, log := range f.events { + require.Equal(t, "100", log.RequestID) + require.Equal(t, testQueue, log.Queue) + require.Equal(t, map[string]string{requestlog.MetadataKeyBuildID: "build/100"}, log.Metadata) + } + }) + } +} + +func TestRecoveryFinishesOnlyTheUncreatedCursor(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.reportIssue("build/100", "task", "unit") + f.reportIssue("build/100", "unprocessed", "unit") + progress := currentCursor() + msg := f.message("build/100", false) + progress.NotificationID = msg.ID + f.retain(progress) + + require.NoError(t, f.recover(msg, false)) + require.Len(t, f.states, 1) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusOpen, issue.Status) + require.Equal(t, "build/100", issue.FirstSeenBuildID) + require.Equal(t, int64(1000), issue.ChangedAtMs) + require.Len(t, f.logs, 1) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + request, build := f.requests["100"], f.builds["build/100"] + require.NoError(t, f.recover(msg, false)) + require.Equal(t, issue, f.states[issueKey{"task", "unit"}]) + require.Len(t, f.events, 1) + require.Equal(t, request, f.requests["100"]) + require.Equal(t, build, f.builds["build/100"]) +} + +func TestRecoveryKeepsCompletedClosingBatches(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + for i := 0; i < 205; i++ { + f.seedIssue(fmt.Sprintf("task-%03d", i), "unit", entity.IssueStatusOpen) + } + f.reportIssue("build/100", "reported", "unit") + msg := f.message("build/100", true) + require.NoError(t, f.deliver(msg, true)) + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) + f.failures["issue/list"] = 1 + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + + accepted := make(map[issueKey]entity.IssueState, len(f.states)) + closed := 0 + for key, state := range f.states { + accepted[key] = state + if state.Status == entity.IssueStatusClosed { + closed++ + } + } + require.Equal(t, 99, closed) + cursor := issueKey{f.progress.CursorTaskID, f.progress.CursorUnitID} + require.NoError(t, f.recover(msg, false)) + for key, expected := range accepted { + if key == cursor { + expected.Version++ + } + require.Equal(t, expected, f.states[key]) + } + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) +} + +func TestRecoveryDoesNotReopenAnExistingClosedCursor(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.seedIssue("task", "unit", entity.IssueStatusClosed) + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + msg := f.message("build/100", false) + progress.NotificationID = msg.ID + f.retain(progress) + + require.NoError(t, f.recover(msg, false)) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + require.Equal(t, "older", f.states[issueKey{"task", "unit"}].FirstSeenBuildID) + require.NotContains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}) +} + +func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { + for _, step := range []string{"build/get", "request/get", "progress/get", "progress/update", "issue/get", "issue/update", "history/create", "recovery/history", "finish abandonment"} { + t.Run(step, func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.seedIssue("task", "unit", entity.IssueStatusClosed) + progress := currentCursor() + msg := f.message("build/100", true) + progress.Finalize, progress.NotificationID = true, msg.ID + f.retain(progress) + if step == "finish abandonment" { + f.failOnPhase = entity.IssueTrackingPhaseAbandoned + } else { + f.failures[step] = 1 + } + require.ErrorIs(t, f.recover(msg, false), assert.AnError) + require.NotEqual(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + + require.NoError(t, f.recover(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + require.Len(t, f.events, 1) + require.NoError(t, f.recover(msg, false)) + require.Len(t, f.events, 1) + }) + } +} + +func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) { + for _, step := range []string{"issue/create", "history/create", "recovery/history"} { + t.Run(step, func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + progress := currentCursor() + msg := f.message("build/100", false) + progress.NotificationID = msg.ID + f.retain(progress) + f.failures[step] = 1 + require.ErrorIs(t, f.recover(msg, false), assert.AnError) + f.dlq.controller.nowMs = func() int64 { return 2000 } + + require.NoError(t, f.recover(msg, false)) + issue := f.states[issueKey{"task", "unit"}] + log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}] + require.Equal(t, "build/100", issue.FirstSeenBuildID) + require.Equal(t, issue.ChangedAtMs, log.TimestampMs) + if step == "issue/create" { + require.Equal(t, int64(2000), issue.ChangedAtMs) + } else { + require.Equal(t, int64(1000), issue.ChangedAtMs) + } + }) + } +} + +func TestPrimaryWaitsForRecoveryThenAdmitsANewerBuild(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.source("build/101", "101", entity.BuildStatusRunning) + f.reportIssue("build/101", "new", "unit") + progress := currentCursor() + failed := f.message("build/100", true) + progress.Finalize, progress.NotificationID = true, failed.ID + f.retain(progress) + f.failures["issue/get"] = 1 + require.ErrorIs(t, f.recover(failed, false), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseAbandoning, f.progress.Phase) + + newer := f.message("build/101", false) + delivery := consumermock.NewMockDelivery(f.mocks) + delivery.EXPECT().Message().Return(newer) + delivery.EXPECT().Hold(_recoveryDelayMs) + require.NoError(t, f.controller.Process(context.Background(), delivery)) + require.Empty(t, f.states) + + require.NoError(t, f.recover(failed, false)) + require.NoError(t, f.deliver(newer, false)) + require.Equal(t, "build/101", f.progress.BuildID) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", "unit"}].Status) + require.NoError(t, f.recover(failed, false)) + require.Equal(t, "build/101", f.progress.BuildID) +} + +func TestRecoveryOfAHelperAbandonsTheSavedPassFirst(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.source("build/101", "101", entity.BuildStatusFailed) + f.retain(currentCursor()) + msg := f.message("build/101", true) + + require.NoError(t, f.recover(msg, true)) + require.Equal(t, "build/100", f.progress.BuildID) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.NoError(t, f.recover(msg, false)) + require.Equal(t, "build/101", f.progress.BuildID) + require.True(t, f.progress.Finalize) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Len(t, f.events, 2) +} + +func TestTerminalNotificationCanFollowAnAbandonedRefresh(t *testing.T) { + for _, terminal := range []bool{false, true} { + t.Run(map[bool]string{false: "abandoned refresh", true: "abandoned terminal pass"}[terminal], func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + f.seedIssue("previous", "unit", entity.IssueStatusOpen) + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + failed := f.message("build/100", terminal) + progress.Finalize, progress.NotificationID = terminal, failed.ID + f.retain(progress) + require.NoError(t, f.recover(failed, false)) + + final := f.message("build/100", true) + if terminal { + require.NoError(t, f.deliver(final, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"previous", "unit"}].Status) + } else { + require.NoError(t, f.deliver(final, true)) + require.NoError(t, f.deliver(final, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"previous", "unit"}].Status) + } + }) + } +} + +func TestRecoveryBeforeAStateWasCreated(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + msg := f.message("build/100", false) + require.NoError(t, f.recover(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Empty(t, f.states) + old := currentCursor() + require.ErrorIs(t, f.store.GetIssueTrackingStore().Create(context.Background(), old), storage.ErrAlreadyExists) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) +} + +func TestRecoveryHistorySupportsLongNotificationIDs(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + msg := f.message("build/100", false) + msg.ID = strings.Repeat("n", 1020) + require.NoError(t, f.recover(msg, false)) + require.NoError(t, f.recover(msg, false)) + require.Len(t, f.events, 1) + for _, log := range f.events { + require.LessOrEqual(t, len(log.ID), 255) + } +} + +func TestRecoveryIgnoresCompletedAndUnrelatedPasses(t *testing.T) { + for _, scenario := range []string{"completed refresh", "completed terminal pass", "newer build", "different refresh"} { + t.Run(scenario, func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + msg := f.message("build/100", false) + progress := currentCursor() + progress.NotificationID = msg.ID + switch scenario { + case "completed refresh": + progress.Phase = entity.IssueTrackingPhaseRefreshed + case "completed terminal pass": + progress.Phase, progress.Finalize = entity.IssueTrackingPhaseFinalized, true + case "newer build": + progress.RequestID, progress.BuildID, progress.NotificationID = "101", "build/101", "newer" + case "different refresh": + progress.NotificationID = "another-notification" + } + f.retain(progress) + require.NoError(t, f.recover(msg, false)) + require.Equal(t, progress, f.progress) + require.Empty(t, f.events) + require.Empty(t, f.states) + }) + } +} + +func TestRecoveryAcknowledgesUnusableMessages(t *testing.T) { + for _, scenario := range []string{"payload", "queue", "partition", "message ID", "missing build", "missing request", "request queue", "request ID"} { + t.Run(scenario, func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + msg := f.message("build/100", true) + switch scenario { + case "payload": + msg.Payload = []byte("invalid") + case "queue": + msg.Tenant = "another-queue" + case "partition": + msg.PartitionKey = "another-queue" + case "message ID": + msg.ID = "" + case "missing build": + delete(f.builds, "build/100") + case "missing request": + delete(f.requests, "100") + case "request queue": + request := f.requests["100"] + request.Queue = "another-queue" + f.requests["100"] = request + case "request ID": + request := f.requests["100"] + request.ID = "invalid" + f.requests["100"] = request + } + require.NoError(t, f.recover(msg, false)) + require.False(t, f.hasProgress) + require.Empty(t, f.states) + }) + } +} diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index a39576d01..0796733c1 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -105,6 +105,11 @@ func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, return fmt.Errorf("load issue tracking progress: %w", err) } + if progress.Phase == entity.IssueTrackingPhaseAbandoning { + delivery.Hold(_recoveryDelayMs) + return nil + } + if progress.Phase != entity.IssueTrackingPhaseRefreshing && progress.Phase != entity.IssueTrackingPhaseFinalizing { if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID { return nil @@ -117,7 +122,7 @@ func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, if err != nil { return err } - if order < 0 || (order == 0 && progress.Phase == entity.IssueTrackingPhaseFinalized) { + if order < 0 || (order == 0 && progress.Finalize) { c.logger.Debugw("ignoring superseded issue tracking notification", "queue", queue, "build_id", build.ID, "request_id", request.ID, "current_request_id", progress.RequestID) return nil diff --git a/stovepipe/core/requestlog/materializer_test.go b/stovepipe/core/requestlog/materializer_test.go index 037fd344a..46131122e 100644 --- a/stovepipe/core/requestlog/materializer_test.go +++ b/stovepipe/core/requestlog/materializer_test.go @@ -391,6 +391,7 @@ func TestMaterializerRetainsOtherEventsWithoutProjectingSummary(t *testing.T) { entity.RequestEventValidationFactRecorded, entity.RequestEventRecordAbandoned, entity.RequestEventBuildAnalysisAbandoned, + entity.RequestEventIssueTrackingAbandoned, } { t.Run(string(event), func(t *testing.T) { materializer, stores, logs, _, _ := newTestMaterializer(t) diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 664f83264..453e49e25 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/stovepipe/entity/issue_tracking_state.go @@ -34,6 +34,10 @@ const ( IssueTrackingPhaseFinalizing IssueTrackingPhase = "finalizing" // IssueTrackingPhaseFinalized means final reconciliation completed; later refresh notifications are ignored. IssueTrackingPhaseFinalized IssueTrackingPhase = "finalized" + // IssueTrackingPhaseAbandoning means an incomplete pass is being stopped without processing its remaining range. + IssueTrackingPhaseAbandoning IssueTrackingPhase = "abandoning" + // IssueTrackingPhaseAbandoned means a pass stopped before completing its range. + IssueTrackingPhaseAbandoned IssueTrackingPhase = "abandoned" ) // IssueTrackingState is the queue-scoped source-order boundary and resumable reconciliation state. @@ -53,7 +57,7 @@ type IssueTrackingState struct { Finalize bool `json:"finalize"` // HasTaskWideIssues records whether this pass has encountered a task-wide issue. HasTaskWideIssues bool `json:"has_task_wide_issues"` - // Phase distinguishes running passes from completed refresh or final passes. + // Phase distinguishes active, completed, and abandoned work. Phase IssueTrackingPhase `json:"phase"` // Version is the positive optimistic-lock version of this state. Version int32 `json:"version"` @@ -77,7 +81,7 @@ func (s IssueTrackingState) Validate() error { return fmt.Errorf("issue tracking version must be positive") } switch s.Phase { - case IssueTrackingPhaseRefreshing: + case IssueTrackingPhaseRefreshing, IssueTrackingPhaseAbandoning, IssueTrackingPhaseAbandoned: case IssueTrackingPhaseRefreshed: if s.Finalize { return fmt.Errorf("a terminal pass cannot finish as an ordinary refresh") diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go index 58eb40f24..245cc2512 100644 --- a/stovepipe/entity/issue_tracking_state_test.go +++ b/stovepipe/entity/issue_tracking_state_test.go @@ -21,7 +21,7 @@ import ( ) func TestIssueTrackingState(t *testing.T) { - for _, phase := range []IssueTrackingPhase{IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized} { + for _, phase := range []IssueTrackingPhase{IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized, IssueTrackingPhaseAbandoning, IssueTrackingPhaseAbandoned} { t.Run(string(phase), func(t *testing.T) { state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: phase, Finalize: phase == IssueTrackingPhaseFinalizing || phase == IssueTrackingPhaseFinalized, Version: 1} require.NoError(t, state.Validate()) diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index 7c53434c4..1c15d67a9 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -34,6 +34,8 @@ const ( RequestEventRecordAbandoned RequestEvent = "record_abandoned" // RequestEventBuildAnalysisAbandoned records incomplete build analysis, not a validation outcome. RequestEventBuildAnalysisAbandoned RequestEvent = "build_analysis_abandoned" + // RequestEventIssueTrackingAbandoned records incomplete issue tracking, not a validation outcome. + RequestEventIssueTrackingAbandoned RequestEvent = "issue_tracking_abandoned" ) // RequestOutcomeReason identifies the durable domain reason for a terminal request state. @@ -139,7 +141,7 @@ func (e RequestLog) validateEvent() error { return fmt.Errorf("event log must not contain request-state context") } switch e.Event { - case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned: + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned, RequestEventIssueTrackingAbandoned: default: return fmt.Errorf("unknown request event %q", e.Event) } diff --git a/stovepipe/entity/request_log_test.go b/stovepipe/entity/request_log_test.go index dc6a82975..7dcb41263 100644 --- a/stovepipe/entity/request_log_test.go +++ b/stovepipe/entity/request_log_test.go @@ -155,6 +155,16 @@ func TestRequestLogValidate(t *testing.T) { return entry }, }, + { + name: "incomplete issue tracking event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventIssueTrackingAbandoned + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + }, {name: "missing ID", mutate: func(entry RequestLog) RequestLog { entry.ID = ""; return entry }, wantErr: true}, {name: "missing queue", mutate: func(entry RequestLog) RequestLog { entry.Queue = ""; return entry }, wantErr: true}, {name: "missing request ID", mutate: func(entry RequestLog) RequestLog { entry.RequestID = ""; return entry }, wantErr: true}, diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 4e0de28ea..b456ee8b5 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -66,6 +66,8 @@ History persistence does not alter issue state or validate transitions against i 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 cursor work, and completion; no runtime writer is enabled here. +Recovery uses the same record's `abandoning` and `abandoned` phases; no additional schema fields or storage operations are required. The source, notification, cursor, and terminal intent remain present after abandonment. See [issue tracking recovery](../../controller/issuetrack/README.md#recovery-and-activation) for the conditional-write ordering that protects the current cursor before newer work is admitted. + The issue-tracking table must include the notification, cursor, `finalize`, and `has_task_wide_issues` columns before updated readers or writers are deployed. Both boolean columns default to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. ### Applied transition facts diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go index b8b63bd4f..e30f5aaa5 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go @@ -92,6 +92,30 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F require.Equal(t, state, got) }) + t.Run("recovery retains cursor and source boundary", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: "tracking/recovery"}) + require.NoError(t, err) + store := bound.GetIssueTrackingStore() + state := entity.IssueTrackingState{ + RequestID: "10", BuildID: "build/10", NotificationID: "failed-notification", + CursorTaskID: "task", CursorUnitID: "unit", + Finalize: true, HasTaskWideIssues: true, + Phase: entity.IssueTrackingPhaseAbandoning, Version: 1, + } + require.NoError(t, store.Create(ctx, state)) + got, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + + state.Phase = entity.IssueTrackingPhaseAbandoned + require.NoError(t, store.Update(ctx, state, 1, 2)) + state.Version = 2 + got, err = store.Get(ctx) + require.NoError(t, err) + require.Equal(t, state, got) + require.ErrorIs(t, store.Update(ctx, state, 1, 2), storage.ErrVersionMismatch) + }) + t.Run("one conditional update wins", func(t *testing.T) { results := make(chan error, 2) for _, requestID := range []string{"12", "13"} {