diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index 1717d25b9..005970f22 100644 --- a/stovepipe/controller/issuetrack/BUILD.bazel +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -35,6 +35,7 @@ go_test( "fixture_test.go", "history_test.go", "issuetrack_test.go", + "request_issue_summary_test.go", ], embed = [":go_default_library"], deps = [ diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 77e414a38..cf18b5ddb 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -86,3 +86,11 @@ Each validation uses the single job registered at admission; analysis does not p Provision the non-null `request.build_id` column with an empty default before deploying. This is an additive association; existing terminal requests can still use their winning build ID. No new tables, secondary indexes, or cross-key transactions are required. Pause admission and drain pre-upgrade validations, completing or explicitly abandoning the existing tracking checkpoint before enabling request-scoped registration. Upgrade normal and recovery consumers together and use the single-head tracking subscription before resuming admission. Old per-analysis notifications are not supported. Do not mix old publishers or consumers with the request-scoped flow. To revert, pause admission and finish or safely abandon request-scoped jobs before restoring the old consumers; the added column can remain. + +## Per-request issue summaries + +The existing issue log is the source for both queue-level lifecycle projections and request-level summaries. Opening and closing events produce New and Closed request results. A reported event records an already-open issue encountered by a later build and produces Existing; it does not change issue status, transition time, or the open-issue lookup. Each event retains its opening build reference, and the materializer loads the source build to identify the request. Request-wide completion and abandonment events remain in request history. + +The cursor cannot advance until the accepted event and all its projections are durable. Retry repairs missing projections from the retained event, preserving the original timestamp and opening. A closure wins over an opening or report from the same request, with conditional writes preventing an older replay from changing it back. Recovery repairs lifecycle history and any accepted report before abandonment; merely fencing an older row does not attribute a report to the abandoned request. + +Existing lifecycle events can lack the additive opening reference. Their lifecycle projections continue using the supplied opening, but those records do not create request summaries from information absent from retained history. Requests begun before recording was deployed have no coverage marker and are not backfilled. Deploy the opening-reference column before updated writers; recording coverage assumes every worker uses the updated contract. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index e021c82ca..a912d3e6e 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -88,7 +88,8 @@ func TestFinalizationOpensKeepsAndClosesIssues(t *testing.T) { require.Equal(t, int32(8), f.states[issueKey{"b", "unit"}].Version) require.Equal(t, "build", f.states[issueKey{"c", "unit"}].FirstSeenBuildID) require.Equal(t, int32(8), f.states[issueKey{"closed", ""}].Version) - require.Len(t, f.logs, 2) + require.Len(t, f.logs, 3) + require.Contains(t, f.logs, logKey{issueKey{"b", "unit"}, "build", entity.IssueEventReported}) require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueEventClosed}) require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened}) } @@ -535,7 +536,10 @@ func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { require.True(t, f.progress.Finalize) require.True(t, f.progress.HasTaskWideIssues) require.Equal(t, "unit/099", f.progress.CursorUnitID) - require.Empty(t, f.logs) + require.Len(t, f.logs, 100) + for _, log := range f.logs { + require.Equal(t, entity.IssueEventReported, log.Event) + } f.failures["build-issues/list"] = 1 require.ErrorIs(t, f.deliver(msg, false), assert.AnError) require.True(t, f.progress.HasTaskWideIssues) diff --git a/stovepipe/controller/issuetrack/dlq.go b/stovepipe/controller/issuetrack/dlq.go index 9a5a36b3c..1176f65ec 100644 --- a/stovepipe/controller/issuetrack/dlq.go +++ b/stovepipe/controller/issuetrack/dlq.go @@ -229,6 +229,9 @@ func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Sto if err := persistIssueHistory(ctx, c.materializer, store, current); err != nil { return entity.IssueTrackingState{}, err } + if err := persistReportedIssue(ctx, c.materializer, store, progress, current); err != nil { + return entity.IssueTrackingState{}, err + } } if err := c.recordAbandonment(ctx, store, queue, progress); err != nil { diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index 7563c0e81..5d098ee0f 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -44,6 +44,7 @@ type logKey struct { type fixture struct { requestHistory map[string]entity.RequestLog + requestIssues map[string]map[issueKey]entity.RequestIssueSummary materializer requestlog.Materializer failOnPhase entity.IssueTrackingPhase t *testing.T @@ -72,6 +73,7 @@ func newFixture(t *testing.T) *fixture { t.Helper() f := &fixture{t: t, mocks: gomock.NewController(t), states: map[issueKey]entity.IssueState{}, logs: map[logKey]entity.IssueLog{}, failures: map[string]int{}, initialLogs: map[logKey]entity.IssueLog{}} f.factory = storagemock.NewMockFactory(f.mocks) + f.requestIssues = map[string]map[issueKey]entity.RequestIssueSummary{} f.builds = map[string]entity.Build{} f.requests = map[string]entity.Request{} f.buildIssues = map[string]map[issueKey]entity.BuildIssue{} @@ -84,6 +86,45 @@ func newFixture(t *testing.T) *fixture { history := storagemock.NewMockIssueLogStore(f.mocks) summaries := storagemock.NewMockIssueSummaryStore(f.mocks) openIssues := storagemock.NewMockOpenIssueStore(f.mocks) + requestIssues := storagemock.NewMockRequestIssueSummaryStore(f.mocks) + store.EXPECT().GetRequestIssueSummaryStore().Return(requestIssues).AnyTimes() + requestIssues.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.RequestIssueSummary) error { + if err := f.fail("request-issue/create"); err != nil { + return err + } + if f.requestIssues[issue.RequestID] == nil { + f.requestIssues[issue.RequestID] = map[issueKey]entity.RequestIssueSummary{} + } + key := issueKey{issue.TaskID, issue.UnitID} + if _, ok := f.requestIssues[issue.RequestID][key]; ok { + return storage.ErrAlreadyExists + } + f.requestIssues[issue.RequestID][key] = issue + return nil + }).AnyTimes() + requestIssues.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, request, task, unit string) (entity.RequestIssueSummary, error) { + if err := f.fail("request-issue/get"); err != nil { + return entity.RequestIssueSummary{}, err + } + issue, ok := f.requestIssues[request][issueKey{task, unit}] + if !ok { + return entity.RequestIssueSummary{}, storage.ErrNotFound + } + return issue, nil + }).AnyTimes() + requestIssues.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.RequestIssueSummary, oldVersion, newVersion int32) error { + if err := f.fail("request-issue/update"); err != nil { + return err + } + key := issueKey{issue.TaskID, issue.UnitID} + current, ok := f.requestIssues[issue.RequestID][key] + if !ok || current.Version != oldVersion { + return storage.ErrVersionMismatch + } + issue.Version = newVersion + f.requestIssues[issue.RequestID][key] = issue + return nil + }).AnyTimes() tracking := storagemock.NewMockIssueTrackingStore(f.mocks) f.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(store, nil).AnyTimes() store.EXPECT().GetIssueStateStore().Return(states).AnyTimes() diff --git a/stovepipe/controller/issuetrack/history.go b/stovepipe/controller/issuetrack/history.go index c4431595c..8b7646f8b 100644 --- a/stovepipe/controller/issuetrack/history.go +++ b/stovepipe/controller/issuetrack/history.go @@ -82,3 +82,22 @@ func persistIssueHistory(ctx context.Context, materializer issuelog.Materializer } return nil } + +// persistReportedIssue records only reports accepted for this pass, not older rows fenced by recovery. +func persistReportedIssue(ctx context.Context, materializer issuelog.Materializer, store storage.Storage, progress entity.IssueTrackingState, issue entity.IssueState) error { + if issue.LastProcessedBuildID != progress.BuildID { + return nil + } + if issue.Status != entity.IssueStatusOpen { + return nil + } + if issue.FirstSeenBuildID == progress.BuildID { + return nil + } + log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, Event: entity.IssueEventReported, + BuildID: progress.BuildID, FirstSeenBuildID: issue.FirstSeenBuildID} + if err := materializer.PersistLog(ctx, store, log, issue.FirstSeenBuildID); err != nil { + return fmt.Errorf("record reported issue: %w", err) + } + return nil +} diff --git a/stovepipe/controller/issuetrack/history_test.go b/stovepipe/controller/issuetrack/history_test.go index 13a3c6bcc..6c94725c2 100644 --- a/stovepipe/controller/issuetrack/history_test.go +++ b/stovepipe/controller/issuetrack/history_test.go @@ -57,7 +57,7 @@ func TestReconciliationHistoryClosureModes(t *testing.T) { if mode == "reconciled" { require.NoError(t, f.deliver(msg, false)) } - require.Len(t, f.requestHistory, 2) + require.Len(t, f.requestHistory, 3) log := f.requestHistory["10:event/issues_reconciled/build"] require.Equal(t, entity.RequestEventIssuesReconciled, log.Event) require.Equal(t, mode, log.Metadata[requestlog.MetadataKeyClosureMode]) @@ -91,7 +91,7 @@ func TestCompletedReconciliationRepairsHistoryBeforeReplacingCheckpoint(t *testi require.NoError(t, f.deliver(newer, false)) require.Equal(t, "new", f.progress.BuildID) require.Contains(t, f.requestHistory, "10:event/issues_reconciled/old") - require.Len(t, f.requestHistory, 2) // Only the older validation has completion and reconciliation events. + require.Len(t, f.requestHistory, 4) } func TestIssueTrackingDLQRepairsCompletedPassInsteadOfAbandoning(t *testing.T) { @@ -105,6 +105,6 @@ func TestIssueTrackingDLQRepairsCompletedPassInsteadOfAbandoning(t *testing.T) { checkpoint := f.progress require.NoError(t, f.recoverHistory(msg, failure.Failure{})) require.Equal(t, checkpoint, f.progress) - require.Len(t, f.requestHistory, 2) + require.Len(t, f.requestHistory, 3) require.Contains(t, f.requestHistory, "10:event/issues_reconciled/build") } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 383bfe48c..429440c19 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -208,6 +208,12 @@ func (c *Controller) trackRequest(ctx context.Context, delivery consumer.Deliver Finalize: finalize, Version: progress.Version + 1, } + if progress.RequestID != request.ID { + log := requestlog.NewRequestEventLog(request, entity.RequestEventIssueTrackingStarted, "", nil) + if err := c.requestMaterializer.PersistLog(ctx, store, log); err != nil { + return fmt.Errorf("record start of request issue results: %w", err) + } + } if progress.Version == 0 { err = tracking.Create(ctx, next) } else { @@ -411,9 +417,11 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr next.CursorTaskID, next.CursorUnitID = key.task, key.unit if !missing && current.LastProcessedBuildID == progress.BuildID && current.Status == desired { if key == (issueKey{progress.CursorTaskID, progress.CursorUnitID}) { - return next, persistIssueHistory(ctx, c.materializer, store, current) + if err := persistIssueHistory(ctx, c.materializer, store, current); err != nil { + return entity.IssueTrackingState{}, err + } } - return next, nil + return next, persistReportedIssue(ctx, c.materializer, store, progress, current) } // Read the issue before this CAS, including when resuming a cursor. A @@ -460,9 +468,11 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr } if missing || current.Status != desired { - return next, persistIssueHistory(ctx, c.materializer, store, target) + if err := persistIssueHistory(ctx, c.materializer, store, target); err != nil { + return entity.IssueTrackingState{}, err + } } - return next, nil + return next, persistReportedIssue(ctx, c.materializer, store, progress, target) } func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next entity.IssueTrackingState) (entity.IssueTrackingState, error) { diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index dc5519379..aceed38b9 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -126,7 +126,9 @@ func TestUnchangedIssueRecordsTheProcessedBuild(t *testing.T) { require.Equal(t, "older", issue.FirstSeenBuildID) require.Equal(t, int64(500), issue.ChangedAtMs) require.Equal(t, int32(8), issue.Version) - require.Empty(t, f.logs) + require.Len(t, f.logs, 1) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventReported}) + require.Equal(t, "older", f.summaries[issueKey{"task", "unit"}].BuildID) } func TestHistoryRetryUsesTheStoredTransition(t *testing.T) { diff --git a/stovepipe/controller/issuetrack/request_issue_summary_test.go b/stovepipe/controller/issuetrack/request_issue_summary_test.go new file mode 100644 index 000000000..91206c0da --- /dev/null +++ b/stovepipe/controller/issuetrack/request_issue_summary_test.go @@ -0,0 +1,200 @@ +// 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" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" +) + +func TestRequestIssuesAcrossBatches(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.seedIssue("closed", "unit", entity.IssueStatusClosed) + f.seedIssue("close-me", "unit", entity.IssueStatusOpen) + f.seedIssue("existing", "unit", entity.IssueStatusOpen) + f.seedIssue("reopen", "unit", entity.IssueStatusClosed) + f.reportIssue("build/100", "existing", "unit") + f.reportIssue("build/100", "reopen", "unit") + for i := 0; i < 230; i++ { + f.reportIssue("build/100", "test", fmt.Sprintf("unit-%03d", i)) + } + progress := currentCursor() + progress.CursorTaskID, progress.CursorUnitID = "", "" + progress.Finalize = true + f.retain(progress) + for f.progress.Phase != entity.IssueTrackingPhaseFinalized { + next, err := f.controller.reconcileBatch(context.Background(), f.store, testQueue, f.progress) + require.NoError(t, err) + require.Equal(t, next, f.progress) + } + results := f.requestIssues["100"] + require.Len(t, results, 233) + require.Equal(t, entity.RequestIssueKindExisting, results[issueKey{"existing", "unit"}].Kind) + require.Equal(t, "older", results[issueKey{"existing", "unit"}].FirstSeenBuildID) + require.Equal(t, entity.RequestIssueKindNew, results[issueKey{"reopen", "unit"}].Kind) + require.Equal(t, entity.RequestIssueKindClosed, results[issueKey{"close-me", "unit"}].Kind) + require.NotContains(t, results, issueKey{"closed", "unit"}) + for i := 0; i < 230; i++ { + require.Equal(t, entity.RequestIssueKindNew, results[issueKey{"test", fmt.Sprintf("unit-%03d", i)}].Kind) + } + + f.source("build/101", "101", entity.BuildStatusSucceeded) + next := currentCursor() + next.RequestID, next.BuildID = "101", "build/101" + next.Version = f.progress.Version + next.Finalize = true + next.CursorTaskID, next.CursorUnitID = "", "" + f.retain(next) + for f.progress.Phase != entity.IssueTrackingPhaseFinalized { + _, err := f.controller.reconcileBatch(context.Background(), f.store, testQueue, f.progress) + require.NoError(t, err) + } + require.Equal(t, results, f.requestIssues["100"]) + require.Equal(t, entity.RequestIssueKindNew, f.requestIssues["100"][issueKey{"reopen", "unit"}].Kind) + require.Equal(t, entity.RequestIssueKindClosed, f.requestIssues["101"][issueKey{"reopen", "unit"}].Kind) +} + +func TestRequestIssueFailureKeepsCursor(t *testing.T) { + for _, existing := range []bool{false, true} { + t.Run(fmt.Sprint(existing), func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + if existing { + f.seedIssue("task", "unit", entity.IssueStatusOpen) + } + f.reportIssue("build/100", "task", "unit") + f.reportIssue("build/100", "z", "unit") + progress := currentCursor() + progress.CursorTaskID, progress.CursorUnitID = "", "" + f.retain(progress) + f.failures["request-issue/create"] = 1 + _, err := f.controller.reconcileBatch(context.Background(), f.store, testQueue, progress) + require.ErrorIs(t, err, assert.AnError) + require.Equal(t, "task", f.progress.CursorTaskID) + require.NotContains(t, f.states, issueKey{"z", "unit"}) + require.Empty(t, f.requestIssues) + event := entity.IssueEventOpened + if existing { + event = entity.IssueEventReported + } + retained := f.logs[logKey{issueKey{"task", "unit"}, "build/100", event}] + require.NotZero(t, retained.TimestampMs) + _, err = f.controller.reconcileBatch(context.Background(), f.store, testQueue, f.progress) + require.NoError(t, err) + kind := entity.RequestIssueKindNew + if existing { + kind = entity.RequestIssueKindExisting + } + require.Equal(t, kind, f.requestIssues["100"][issueKey{"task", "unit"}].Kind) + require.Len(t, f.requestIssues["100"], 2) + require.Equal(t, retained, f.logs[logKey{issueKey{"task", "unit"}, "build/100", event}]) + }) + } +} + +func TestRecoveryRepairsRequestIssueBeforeAbandoning(t *testing.T) { + for _, kind := range []entity.RequestIssueKind{entity.RequestIssueKindNew, entity.RequestIssueKindExisting, entity.RequestIssueKindClosed} { + t.Run(string(kind), func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + progress := currentCursor() + if kind != entity.RequestIssueKindNew { + f.seedIssue("task", "unit", entity.IssueStatusOpen) + } + if kind == entity.RequestIssueKindClosed { + progress.Phase, progress.Finalize = entity.IssueTrackingPhaseFinalizing, true + } else { + f.reportIssue("build/100", "task", "unit") + } + msg := f.message("build/100", true) + progress.NotificationID = msg.ID + f.retain(progress) + f.failures["request-issue/create"] = 1 + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, kind != entity.RequestIssueKindClosed) + require.ErrorIs(t, err, assert.AnError) + f.failures["request-issue/create"] = 1 + require.ErrorIs(t, f.recover(msg, false), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseAbandoning, f.progress.Phase) + require.NoError(t, f.recover(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, kind, f.requestIssues["100"][issueKey{"task", "unit"}].Kind) + }) + } +} + +func TestRecoveryDoesNotAttributeUnappliedCursor(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + progress := currentCursor() + msg := f.message("build/100", true) + progress.NotificationID = msg.ID + f.retain(progress) + require.NoError(t, f.recover(msg, false)) + require.Empty(t, f.requestIssues) +} + +func TestTaskWideSuppressionDoesNotReportUnseenIssues(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusFailed) + f.seedIssue("old", "unit", entity.IssueStatusOpen) + f.reportIssue("build/100", "entry", "") + msg := f.message("build/100", true) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Len(t, f.requestIssues["100"], 1) + require.Equal(t, entity.RequestIssueKindNew, f.requestIssues["100"][issueKey{"entry", ""}].Kind) + require.NotContains(t, f.requestIssues["100"], issueKey{"old", "unit"}) +} + +func TestRequestIssueReplayRejectsConflict(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + progress := currentCursor() + f.retain(progress) + f.reportIssue("build/100", "task", "unit") + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.NoError(t, err) + retained := f.requestIssues["100"][issueKey{"task", "unit"}] + retained.Kind = entity.RequestIssueKindExisting + f.requestIssues["100"][issueKey{"task", "unit"}] = retained + _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, true) + require.ErrorContains(t, err, "conflicts") +} + +func TestReportedIssueFailureKeepsCursor(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + progress.CursorTaskID, progress.CursorUnitID = "", "" + f.retain(progress) + f.failures["history/create"] = 1 + _, err := f.controller.reconcileBatch(context.Background(), f.store, testQueue, progress) + require.ErrorIs(t, err, assert.AnError) + require.Equal(t, "task", f.progress.CursorTaskID) + require.Empty(t, f.requestIssues["100"]) + _, err = f.controller.reconcileBatch(context.Background(), f.store, testQueue, f.progress) + require.NoError(t, err) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventReported}) + require.Equal(t, entity.RequestIssueKindExisting, f.requestIssues["100"][issueKey{"task", "unit"}].Kind) +} diff --git a/stovepipe/core/issuelog/materializer.go b/stovepipe/core/issuelog/materializer.go index dede1a6f4..75852de19 100644 --- a/stovepipe/core/issuelog/materializer.go +++ b/stovepipe/core/issuelog/materializer.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package issuelog retains issue transitions and projects their latest status. +// Package issuelog retains issue events and materializes lifecycle and request summaries. package issuelog import ( @@ -29,9 +29,10 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) -// Materializer retains issue history and updates its summary and open-issue lookup. +// Materializer retains issue events and repairs lifecycle and request-level projections. type Materializer interface { - // PersistLog repairs the summary and lookup even when the transition already exists. + // PersistLog repairs projections even when the event already exists. + // Reported events leave lifecycle projections unchanged and assign a zero timestamp on retention. // firstSeenBuildID identifies the opening retained by the source issue state. PersistLog(ctx context.Context, stores storage.Storage, log entity.IssueLog, firstSeenBuildID string) error } @@ -79,24 +80,40 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l return fmt.Errorf("issue transition build identity does not match retained history") } - if retained.Event == entity.IssueEventReported { - return nil - } - if retained.FirstSeenBuildID != "" { - firstSeenBuildID = retained.FirstSeenBuildID + if retained.Event != entity.IssueEventReported { + if retained.FirstSeenBuildID != "" { + firstSeenBuildID = retained.FirstSeenBuildID + } + summary := entity.IssueSummary{TaskID: retained.TaskID, UnitID: retained.UnitID, + Status: entity.IssueStatus(retained.Event), FirstSeenBuildID: firstSeenBuildID, + RequestID: build.RequestID, BuildID: retained.BuildID, ChangedAtMs: retained.TimestampMs, Version: 1} + if err := summary.Validate(); err != nil { + return fmt.Errorf("invalid issue summary: %w", err) + } + if err := materializeIssueSummary(ctx, stores.GetIssueSummaryStore(), summary); err != nil { + return err + } + if err := materializeOpenIssue(ctx, stores, summary.TaskID, summary.UnitID); err != nil { + return err + } } - summary := entity.IssueSummary{ - TaskID: retained.TaskID, UnitID: retained.UnitID, Status: entity.IssueStatus(retained.Event), - FirstSeenBuildID: firstSeenBuildID, RequestID: build.RequestID, BuildID: retained.BuildID, - ChangedAtMs: retained.TimestampMs, Version: 1, + // Legacy events lack the retained opening needed to reconstruct request summaries. + if retained.FirstSeenBuildID == "" { + return nil } - if err := summary.Validate(); err != nil { - return fmt.Errorf("invalid issue summary: %w", err) + kind := entity.RequestIssueKindClosed + switch retained.Event { + case entity.IssueEventOpened: + kind = entity.RequestIssueKindNew + case entity.IssueEventReported: + kind = entity.RequestIssueKindExisting } - if err := materializeIssueSummary(ctx, stores.GetIssueSummaryStore(), summary); err != nil { - return err + result := entity.RequestIssueSummary{RequestID: build.RequestID, TaskID: retained.TaskID, UnitID: retained.UnitID, + Kind: kind, BuildID: retained.BuildID, FirstSeenBuildID: retained.FirstSeenBuildID} + if err := result.Validate(); err != nil { + return fmt.Errorf("invalid request issue summary: %w", err) } - return materializeOpenIssue(ctx, stores, summary.TaskID, summary.UnitID) + return materializeRequestIssueSummary(ctx, stores.GetRequestIssueSummaryStore(), result) } func retainIssueLog(ctx context.Context, logs storage.IssueLogStore, log entity.IssueLog) (entity.IssueLog, error) { @@ -225,3 +242,49 @@ func materializeOpenIssue(ctx context.Context, stores storage.Storage, taskID, u } var _ Materializer = (*materializer)(nil) + +func materializeRequestIssueSummary(ctx context.Context, store storage.RequestIssueSummaryStore, incoming entity.RequestIssueSummary) error { + for { + if err := ctx.Err(); err != nil { + return err + } + current, err := store.Get(ctx, incoming.RequestID, incoming.TaskID, incoming.UnitID) + if errors.Is(err, storage.ErrNotFound) { + err := store.Create(ctx, incoming) + if errors.Is(err, storage.ErrAlreadyExists) { + continue + } + if err != nil { + return fmt.Errorf("create request issue summary: %w", err) + } + return nil + } + if err != nil { + return fmt.Errorf("read request issue summary: %w", err) + } + if err := current.Validate(); err != nil { + return fmt.Errorf("invalid retained request issue summary: %w", err) + } + expected := incoming + expected.Kind = current.Kind + expected.Version = current.Version + if current != expected { + return fmt.Errorf("request issue summary conflicts with retained history") + } + if current.Kind == entity.RequestIssueKindClosed || current.Kind == incoming.Kind { + return nil + } + if incoming.Kind != entity.RequestIssueKindClosed { + return fmt.Errorf("request issue summary conflicts with retained history") + } + if current.Version == math.MaxInt32 { + return fmt.Errorf("request issue summary version exhausted") + } + if err := store.Update(ctx, incoming, current.Version, current.Version+1); errors.Is(err, storage.ErrVersionMismatch) { + continue + } else if err != nil { + return fmt.Errorf("update request issue summary: %w", err) + } + return nil + } +} diff --git a/stovepipe/core/issuelog/materializer_test.go b/stovepipe/core/issuelog/materializer_test.go index f888a01e9..8cecc296d 100644 --- a/stovepipe/core/issuelog/materializer_test.go +++ b/stovepipe/core/issuelog/materializer_test.go @@ -34,6 +34,7 @@ type logKey struct { } type materializerFixture struct { + requestSummaries map[string]entity.RequestIssueSummary m Materializer stores *storagemock.MockStorage logs map[logKey]entity.IssueLog @@ -49,8 +50,44 @@ type materializerFixture struct { func newMaterializerFixture(t *testing.T) *materializerFixture { t.Helper() ctrl := gomock.NewController(t) - f := &materializerFixture{m: NewMaterializer(tally.NoopScope), stores: storagemock.NewMockStorage(ctrl), + f := &materializerFixture{requestSummaries: map[string]entity.RequestIssueSummary{}, m: NewMaterializer(tally.NoopScope), stores: storagemock.NewMockStorage(ctrl), logs: make(map[logKey]entity.IssueLog), failures: make(map[string]error)} + requestSummaries := storagemock.NewMockRequestIssueSummaryStore(ctrl) + f.stores.EXPECT().GetRequestIssueSummaryStore().Return(requestSummaries).AnyTimes() + requestSummaries.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, req, task, unit string) (entity.RequestIssueSummary, error) { + if err := f.failures["request-summary/get"]; err != nil { + return entity.RequestIssueSummary{}, err + } + result, ok := f.requestSummaries[req+":"+task+":"+unit] + if !ok { + return entity.RequestIssueSummary{}, storage.ErrNotFound + } + return result, nil + }).AnyTimes() + requestSummaries.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, result entity.RequestIssueSummary) error { + if err := f.failures["request-summary/create"]; err != nil { + return err + } + key := result.RequestID + ":" + result.TaskID + ":" + result.UnitID + if _, ok := f.requestSummaries[key]; ok { + return storage.ErrAlreadyExists + } + f.requestSummaries[key] = result + return nil + }).AnyTimes() + requestSummaries.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, result entity.RequestIssueSummary, oldVersion, newVersion int32) error { + if err := f.failures["request-summary/update"]; err != nil { + return err + } + key := result.RequestID + ":" + result.TaskID + ":" + result.UnitID + current, ok := f.requestSummaries[key] + if !ok || current.Version != oldVersion { + return storage.ErrVersionMismatch + } + result.Version = newVersion + f.requestSummaries[key] = result + return nil + }).AnyTimes() logs := storagemock.NewMockIssueLogStore(ctrl) summaries := storagemock.NewMockIssueSummaryStore(ctrl) openIssues := storagemock.NewMockOpenIssueStore(ctrl) @@ -547,3 +584,37 @@ func TestReportedIssueCannotReopenClosedIssue(t *testing.T) { require.Equal(t, before, f.summary) require.Empty(t, f.openIssue) } + +func TestRequestSummaryRepairAndClosureWins(t *testing.T) { + f := newMaterializerFixture(t) + opened := testIssueLog("build/z", entity.IssueStatusOpen) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, opened, opened.BuildID)) + require.Equal(t, entity.RequestIssueKindNew, f.requestSummaries["9:validation:unit"].Kind) + report := entity.IssueLog{TaskID: "validation", UnitID: "unit", Event: entity.IssueEventReported, BuildID: "build/a"} + f.failures["request-summary/create"] = assert.AnError + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID), assert.AnError) + retained := f.logs[logKey{"validation", "unit", "build/a", entity.IssueEventReported}] + delete(f.failures, "request-summary/create") + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID)) + require.Equal(t, entity.RequestIssueKindExisting, f.requestSummaries["10:validation:unit"].Kind) + require.Equal(t, retained, f.logs[logKey{"validation", "unit", "build/a", entity.IssueEventReported}]) + closed := testIssueLog("build/a", entity.IssueStatusClosed) + f.failures["request-summary/update"] = assert.AnError + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, closed, opened.BuildID), assert.AnError) + delete(f.failures, "request-summary/update") + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, closed, opened.BuildID)) + require.Equal(t, entity.RequestIssueKindClosed, f.requestSummaries["10:validation:unit"].Kind) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID)) + require.Equal(t, entity.RequestIssueKindClosed, f.requestSummaries["10:validation:unit"].Kind) + require.Equal(t, entity.IssueStatusClosed, f.summary.Status) + require.Empty(t, f.openIssue) +} + +func TestLegacyHistoryDoesNotInventRequestSummary(t *testing.T) { + f := newMaterializerFixture(t) + log := testIssueLog("build/z", entity.IssueStatusOpen) + f.logs[logKey{log.TaskID, log.UnitID, log.BuildID, log.Event}] = log + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log, log.BuildID)) + require.Equal(t, entity.IssueStatusOpen, f.summary.Status) + require.Empty(t, f.requestSummaries) +} diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go index 33aa791e3..c81426b1a 100644 --- a/stovepipe/entity/request_log.go +++ b/stovepipe/entity/request_log.go @@ -36,6 +36,8 @@ const ( RequestEventBuildAnalysisAbandoned RequestEvent = "build_analysis_abandoned" // RequestEventBuildAnalysisCompleted records completed issue analysis for the winning terminal build. RequestEventBuildAnalysisCompleted RequestEvent = "build_analysis_completed" + // RequestEventIssueTrackingStarted records the start of per-request issue result recording. + RequestEventIssueTrackingStarted RequestEvent = "issue_tracking_started" // RequestEventIssuesReconciled records completed final issue reconciliation for a build. RequestEventIssuesReconciled RequestEvent = "issues_reconciled" // RequestEventIssueTrackingAbandoned records unfinished issue reconciliation after primary processing stopped. @@ -153,7 +155,7 @@ func (e RequestLog) validateEvent() error { switch e.Event { case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded, RequestEventProjectFactsRecorded, RequestEventRecordAbandoned, RequestEventBuildAnalysisAbandoned, RequestEventBuildAnalysisCompleted, - RequestEventIssuesReconciled, RequestEventIssueTrackingAbandoned, + RequestEventIssueTrackingStarted, RequestEventIssuesReconciled, RequestEventIssueTrackingAbandoned, RequestEventFixSubmitted, RequestEventFixSubmissionSkipped, RequestEventFixSubmissionAbandoned: default: return fmt.Errorf("unknown request event %q", e.Event) diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 80833ee2f..8a6df3ead 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -17,6 +17,7 @@ go_test( "queue_state_transitions_test.go", "queue_summary_test.go", "request_acceptance_test.go", + "request_issue_summary_materializer_test.go", "request_issue_summary_test.go", "storage_test.go", ], diff --git a/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_materializer_test.go b/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_materializer_test.go new file mode 100644 index 000000000..d0e726584 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/request_issue_summary_materializer_test.go @@ -0,0 +1,69 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/core/issuelog" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testRequestIssueSummaryMaterializer(t *testing.T, ctx context.Context, db *sql.DB, factory storage.Factory) { + const queue = "request-issue-summary-history" + bound, err := factory.For(storage.Config{QueueName: queue}) + require.NoError(t, err) + for _, build := range []entity.Build{ + {ID: "b/12", RequestID: "12", Status: entity.BuildStatusFailed, Version: 1}, + {ID: "b/13", RequestID: "13", Status: entity.BuildStatusFailed, Version: 1}, + {ID: "b/14", RequestID: "14", Status: entity.BuildStatusSucceeded, Version: 1}, + } { + require.NoError(t, bound.GetBuildStore().Create(ctx, build)) + } + m := issuelog.NewMaterializer(tally.NoopScope) + for _, log := range []entity.IssueLog{ + {TaskID: "test", BuildID: "b/12", Event: entity.IssueEventOpened, TimestampMs: 1000}, + {TaskID: "test", BuildID: "b/13", Event: entity.IssueEventReported}, + {TaskID: "test", BuildID: "b/14", Event: entity.IssueEventClosed, TimestampMs: 3000}, + } { + require.NoError(t, m.PersistLog(ctx, bound, log, "b/12")) + } + results := bound.GetRequestIssueSummaryStore() + for request, kind := range map[string]entity.RequestIssueKind{"12": entity.RequestIssueKindNew, "13": entity.RequestIssueKindExisting, "14": entity.RequestIssueKindClosed} { + got, err := results.Get(ctx, request, "test", "") + require.NoError(t, err) + require.Equal(t, kind, got.Kind) + require.Equal(t, "b/12", got.FirstSeenBuildID) + } + retained, err := bound.GetIssueLogStore().Get(ctx, "test", "", "b/13", entity.IssueEventReported) + require.NoError(t, err) + _, err = db.ExecContext(ctx, `DELETE FROM request_issue_summary WHERE queue = ? AND request_id = ? AND task_id = ? AND unit_id = ?`, queue, "13", "test", "") + require.NoError(t, err) + require.NoError(t, m.PersistLog(ctx, bound, retained, retained.FirstSeenBuildID)) + rebuilt, err := results.Get(ctx, "13", "test", "") + require.NoError(t, err) + require.Equal(t, entity.RequestIssueKindExisting, rebuilt.Kind) + global, err := bound.GetIssueSummaryStore().Get(ctx, "test", "") + require.NoError(t, err) + require.Equal(t, entity.IssueStatusClosed, global.Status) + require.Equal(t, int64(3000), global.ChangedAtMs) + _, err = bound.GetOpenIssueStore().Get(ctx, "test", "") + require.ErrorIs(t, err, storage.ErrNotFound) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index fa522d62e..458d4843b 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -139,6 +139,10 @@ func TestMySQLStorage(t *testing.T) { testRequestIssueSummaryStore(t, ctx, factory) }) + t.Run("RequestIssueSummaryMaterializer", func(t *testing.T) { + testRequestIssueSummaryMaterializer(t, ctx, db, factory) + }) + t.Run("ReportedIssueLog", func(t *testing.T) { testReportedIssueLog(t, ctx, factory) })