From ef36b64368dfc492302a54bb71e5904316181d7c Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Fri, 6 Nov 2026 23:01:28 +0000 Subject: [PATCH] feat(stovepipe): Retain repeat issue reports --- .../controller/issuetrack/delivery_test.go | 18 +++--- stovepipe/controller/issuetrack/dlq_test.go | 8 +-- .../controller/issuetrack/fix_handoff_test.go | 2 +- .../controller/issuetrack/fixture_test.go | 12 ++-- stovepipe/controller/issuetrack/history.go | 2 +- .../controller/issuetrack/issuetrack_test.go | 6 +- stovepipe/core/issuelog/materializer.go | 32 +++++++++-- stovepipe/core/issuelog/materializer_test.go | 44 +++++++++++--- stovepipe/entity/issue_log.go | 43 ++++++++++---- stovepipe/entity/issue_log_test.go | 17 ++++-- stovepipe/extension/storage/README.md | 2 + .../extension/storage/issue_log_store.go | 10 ++-- .../storage/mock/issue_log_store_mock.go | 8 +-- .../storage/mysql/issue_log_store.go | 33 ++++++----- .../storage/mysql/issue_log_store_test.go | 26 ++++----- .../storage/mysql/schema/issue_log.sql | 4 +- .../extension/storage/mysql/storage_test.go | 2 +- .../extension/storage/mysql/issue_log_test.go | 57 +++++++++++++------ .../mysql/open_issue_materializer_test.go | 4 +- .../extension/storage/mysql/storage_test.go | 4 ++ 20 files changed, 223 insertions(+), 111 deletions(-) diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index 7a65f6612..e021c82ca 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -89,8 +89,8 @@ func TestFinalizationOpensKeepsAndClosesIssues(t *testing.T) { 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.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueStatusClosed}) - require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueStatusOpen}) + require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueEventClosed}) + require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened}) } func TestEmptyFinalBuildClosesIssuesButRetainsTheirRows(t *testing.T) { @@ -266,7 +266,7 @@ func TestRetryRepairsHistoryBeforeFinalizing(t *testing.T) { require.Equal(t, "task", f.progress.CursorTaskID) require.Equal(t, "unit", f.progress.CursorUnitID) require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) - require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueStatusOpen}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueEventOpened}].TimestampMs) } func TestRetryFinishesTheCursorIssueBeforeScanningTheNextBatch(t *testing.T) { @@ -428,7 +428,7 @@ func TestHistoryRecordsTheActualChangingBuildNotTheLatestProcessedBuild(t *testi require.Empty(t, f.logs) f.controller.nowMs = func() int64 { return 2000 } require.NoError(t, f.deliver(msg, false)) - require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventClosed}].TimestampMs) require.Equal(t, changed, f.states[issueKey{"task", "unit"}]) f.source("build/101", "101", entity.BuildStatusSucceeded) @@ -466,7 +466,7 @@ func TestTaskWideFailurePreservesAllExistingIssues(t *testing.T) { require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "closed"}].Status) require.Len(t, f.logs, 3) for key := range f.logs { - require.Equal(t, entity.IssueStatusOpen, key.status) + require.Equal(t, entity.IssueEventOpened, key.event) } require.Equal(t, 1, f.failures["issue/list"]) @@ -506,7 +506,7 @@ func TestTaskWideFailureInLaterOpeningBatchPreventsAllClosures(t *testing.T) { require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status) require.Len(t, f.logs, 106) for key := range f.logs { - require.Equal(t, entity.IssueStatusOpen, key.status) + require.Equal(t, entity.IssueEventOpened, key.event) } } @@ -610,7 +610,7 @@ func TestTaskWideHistoryRetryFinishesWithoutClosing(t *testing.T) { require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status) - require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueStatusOpen}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueEventOpened}].TimestampMs) require.Equal(t, 1, f.failures["issue/list"]) } @@ -689,7 +689,7 @@ func TestRequestJobWaitsForAnalysisThenPreservesReopening(t *testing.T) { require.NoError(t, f.deliverRequest(older, 0)) require.NoError(t, f.deliverRequest(older, -1)) require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventClosed}) wait := _analysisWaitMs if newerTerminal { wait = 0 @@ -697,7 +697,7 @@ func TestRequestJobWaitsForAnalysisThenPreservesReopening(t *testing.T) { require.NoError(t, f.deliverRequest(newer, wait)) require.Equal(t, "build/101", f.states[issueKey{"task", "unit"}].FirstSeenBuildID) require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/101", entity.IssueStatusOpen}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/101", entity.IssueEventOpened}) require.NoError(t, f.deliverRequest(older, -1)) require.Equal(t, "101", f.progress.RequestID) }) diff --git a/stovepipe/controller/issuetrack/dlq_test.go b/stovepipe/controller/issuetrack/dlq_test.go index e91911ead..68a8be4de 100644 --- a/stovepipe/controller/issuetrack/dlq_test.go +++ b/stovepipe/controller/issuetrack/dlq_test.go @@ -95,7 +95,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) { for key, state := range f.states { untouched[key] = state } - delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, status}) + delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, entity.IssueEvent(status)}) f.failures["issue/list"], f.failures["build-issues/list"], f.failures["build-issues/get"] = 1, 1, 1 require.NoError(t, f.recover(msg, false)) @@ -106,7 +106,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) { } require.Equal(t, expected, f.states[key]) } - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, status}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, entity.IssueEvent(status)}) require.Len(t, f.events, 1) require.Equal(t, 1, f.failures["issue/list"]) require.Equal(t, 1, f.failures["build-issues/list"]) @@ -193,7 +193,7 @@ func TestRecoveryDoesNotReopenAnExistingClosedCursor(t *testing.T) { 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.IssueStatusOpen}) + require.NotContains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}) } func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { @@ -241,7 +241,7 @@ func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) { require.NoError(t, f.recover(msg, false)) issue := f.states[issueKey{"task", "unit"}] - log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}] + log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}] summary := f.summaries[issueKey{"task", "unit"}] require.Equal(t, issue.FirstSeenBuildID, f.openIssues[issueKey{"task", "unit"}].FirstSeenBuildID) require.Equal(t, entity.IssueStatusOpen, summary.Status) diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go index 2a3aecef9..6d05a71fb 100644 --- a/stovepipe/controller/issuetrack/fix_handoff_test.go +++ b/stovepipe/controller/issuetrack/fix_handoff_test.go @@ -46,7 +46,7 @@ func TestOpeningPassPublishesRepairAfterIssueWrites(t *testing.T) { require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) for _, task := range []string{"new", "reopened"} { require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{task, "unit"}].Status) - require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueStatusOpen}) + require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueEventOpened}) } payload := &stovepipemq.FixRequest{} require.NoError(t, stovepipemq.Unmarshal(msg.Payload, payload)) diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index cae7e9a8d..7563c0e81 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -38,8 +38,8 @@ const testQueue = "validation/main" type logKey struct { issueKey - build string - status entity.IssueStatus + build string + event entity.IssueEvent } type fixture struct { @@ -130,7 +130,7 @@ func newFixture(t *testing.T) *fixture { if err := f.fail("history/create"); err != nil { return err } - key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Status} + key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Event} if _, ok := f.initialLogs[key]; ok { return storage.ErrAlreadyExists } @@ -140,11 +140,11 @@ func newFixture(t *testing.T) *fixture { f.logs[key] = log return nil }).AnyTimes() - history.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit, build string, status entity.IssueStatus) (entity.IssueLog, error) { + history.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit, build string, event entity.IssueEvent) (entity.IssueLog, error) { if err := f.fail("history/get"); err != nil { return entity.IssueLog{}, err } - key := logKey{issueKey{task, unit}, build, status} + key := logKey{issueKey{task, unit}, build, event} if log, ok := f.logs[key]; ok { return log, nil } @@ -433,7 +433,7 @@ func (f *fixture) reportIssue(build, task, unit string) { func (f *fixture) seedIssue(task, unit string, status entity.IssueStatus) { f.states[issueKey{task, unit}] = entity.IssueState{TaskID: task, UnitID: unit, Status: status, FirstSeenBuildID: "older", Version: 7, LastProcessedBuildID: "older", LastChangedBuildID: "older", ChangedAtMs: 500} - f.initialLogs[logKey{issueKey{task, unit}, "older", status}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Status: status, TimestampMs: 500} + f.initialLogs[logKey{issueKey{task, unit}, "older", entity.IssueEvent(status)}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Event: entity.IssueEvent(status), TimestampMs: 500} f.summaries[issueKey{task, unit}] = entity.IssueSummary{TaskID: task, UnitID: unit, Status: status, FirstSeenBuildID: "older", RequestID: "1", BuildID: "older", ChangedAtMs: 500, Version: 1} if status == entity.IssueStatusOpen { diff --git a/stovepipe/controller/issuetrack/history.go b/stovepipe/controller/issuetrack/history.go index db6ada0d1..c4431595c 100644 --- a/stovepipe/controller/issuetrack/history.go +++ b/stovepipe/controller/issuetrack/history.go @@ -76,7 +76,7 @@ func persistIssuesReconciledLog(ctx context.Context, materializer requestlog.Mat } func persistIssueHistory(ctx context.Context, materializer issuelog.Materializer, store storage.Storage, issue entity.IssueState) error { - log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Status: issue.Status, TimestampMs: issue.ChangedAtMs} + log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Event: entity.IssueEvent(issue.Status), TimestampMs: issue.ChangedAtMs} if err := materializer.PersistLog(ctx, store, log, issue.FirstSeenBuildID); err != nil { return fmt.Errorf("record issue transition: %w", err) } diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index b28e8ec32..dc5519379 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -99,13 +99,13 @@ func TestProcessIssueOpensAndCloses(t *testing.T) { require.Equal(t, entity.IssueStatusOpen, issue.Status) require.Equal(t, "build/100", issue.FirstSeenBuildID) require.Equal(t, int32(1), issue.Version) - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}) } else { require.NotContains(t, f.openIssues, issueKey{"task", "unit"}) require.Equal(t, entity.IssueStatusClosed, issue.Status) require.Equal(t, "older", issue.FirstSeenBuildID) require.Equal(t, int32(8), issue.Version) - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventClosed}) } }) } @@ -146,7 +146,7 @@ func TestHistoryRetryUsesTheStoredTransition(t *testing.T) { _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false) require.NoError(t, err) require.Equal(t, applied, f.states[issueKey{"task", "unit"}]) - require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}].TimestampMs) } func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { diff --git a/stovepipe/core/issuelog/materializer.go b/stovepipe/core/issuelog/materializer.go index 72e167b74..dede1a6f4 100644 --- a/stovepipe/core/issuelog/materializer.go +++ b/stovepipe/core/issuelog/materializer.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "math" + "time" "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/metrics" @@ -35,11 +36,14 @@ type Materializer interface { PersistLog(ctx context.Context, stores storage.Storage, log entity.IssueLog, firstSeenBuildID string) error } -type materializer struct{ scope tally.Scope } +type materializer struct { + scope tally.Scope + now func() time.Time +} // NewMaterializer creates an issue-log materializer. func NewMaterializer(scope tally.Scope) Materializer { - return &materializer{scope: scope.SubScope("issue_log_materializer")} + return &materializer{scope: scope.SubScope("issue_log_materializer"), now: time.Now} } func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.IssueLog, firstSeenBuildID string) (retErr error) { @@ -49,10 +53,17 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l if err := ctx.Err(); err != nil { return err } + if log.FirstSeenBuildID != "" && log.FirstSeenBuildID != firstSeenBuildID { + return fmt.Errorf("issue event opening disagrees with source state") + } + log.FirstSeenBuildID = firstSeenBuildID + if log.Event == entity.IssueEventReported && log.TimestampMs == 0 { + log.TimestampMs = m.now().UnixMilli() + } if err := log.Validate(); err != nil { return fmt.Errorf("invalid issue transition: %w", err) } - if firstSeenBuildID == "" || (log.Status == entity.IssueStatusOpen && firstSeenBuildID != log.BuildID) { + if firstSeenBuildID == "" || (log.Event == entity.IssueEventOpened && firstSeenBuildID != log.BuildID) { return fmt.Errorf("issue transition has an invalid opening build") } @@ -68,8 +79,14 @@ 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 + } summary := entity.IssueSummary{ - TaskID: retained.TaskID, UnitID: retained.UnitID, Status: retained.Status, + TaskID: retained.TaskID, UnitID: retained.UnitID, Status: entity.IssueStatus(retained.Event), FirstSeenBuildID: firstSeenBuildID, RequestID: build.RequestID, BuildID: retained.BuildID, ChangedAtMs: retained.TimestampMs, Version: 1, } @@ -87,14 +104,17 @@ func retainIssueLog(ctx context.Context, logs storage.IssueLogStore, log entity. if !errors.Is(err, storage.ErrAlreadyExists) { return entity.IssueLog{}, fmt.Errorf("retain issue transition: %w", err) } - retained, err := logs.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + retained, err := logs.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) if err != nil { return entity.IssueLog{}, fmt.Errorf("read retained issue transition: %w", err) } if retained.TaskID != log.TaskID || retained.UnitID != log.UnitID || - retained.BuildID != log.BuildID || retained.Status != log.Status { + retained.BuildID != log.BuildID || retained.Event != log.Event { return entity.IssueLog{}, fmt.Errorf("issue transition identity conflicts with retained history") } + if retained.FirstSeenBuildID != "" && retained.FirstSeenBuildID != log.FirstSeenBuildID { + return entity.IssueLog{}, fmt.Errorf("issue event opening conflicts with retained history") + } if err := retained.Validate(); err != nil { return entity.IssueLog{}, fmt.Errorf("invalid retained issue transition: %w", err) } diff --git a/stovepipe/core/issuelog/materializer_test.go b/stovepipe/core/issuelog/materializer_test.go index 36af850a5..f888a01e9 100644 --- a/stovepipe/core/issuelog/materializer_test.go +++ b/stovepipe/core/issuelog/materializer_test.go @@ -30,7 +30,7 @@ import ( type logKey struct { task, unit, build string - status entity.IssueStatus + event entity.IssueEvent } type materializerFixture struct { @@ -63,18 +63,18 @@ func newMaterializerFixture(t *testing.T) *materializerFixture { if err := f.failures["log/create"]; err != nil { return err } - key := logKey{log.TaskID, log.UnitID, log.BuildID, log.Status} + key := logKey{log.TaskID, log.UnitID, log.BuildID, log.Event} if _, ok := f.logs[key]; ok { return storage.ErrAlreadyExists } f.logs[key] = log return nil }).AnyTimes() - logs.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit, build string, status entity.IssueStatus) (entity.IssueLog, error) { + logs.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit, build string, event entity.IssueEvent) (entity.IssueLog, error) { if err := f.failures["log/get"]; err != nil { return entity.IssueLog{}, err } - log, ok := f.logs[logKey{task, unit, build, status}] + log, ok := f.logs[logKey{task, unit, build, event}] if !ok { return entity.IssueLog{}, storage.ErrNotFound } @@ -111,7 +111,7 @@ func newMaterializerFixture(t *testing.T) *materializerFixture { if f.summary.Version != 0 { return storage.ErrAlreadyExists } - require.Equal(t, summary.ChangedAtMs, f.logs[logKey{summary.TaskID, summary.UnitID, summary.BuildID, summary.Status}].TimestampMs) + require.Equal(t, summary.ChangedAtMs, f.logs[logKey{summary.TaskID, summary.UnitID, summary.BuildID, entity.IssueEvent(summary.Status)}].TimestampMs) f.summary = summary return nil }).AnyTimes() @@ -191,7 +191,7 @@ func newMaterializerFixture(t *testing.T) *materializerFixture { } func testIssueLog(build string, status entity.IssueStatus) entity.IssueLog { - return entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: build, Status: status, TimestampMs: 1000} + return entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: build, Event: entity.IssueEvent(status), TimestampMs: 1000} } func TestMaterializerOpenCloseReopenAndReplay(t *testing.T) { @@ -319,7 +319,7 @@ func TestMaterializerRejectsInvalidTransitions(t *testing.T) { opening := log.BuildID switch scenario { case "invalid status": - log.Status = entity.IssueStatusUnknown + log.Event = "" case "missing opening": opening = "" case "wrong opening": @@ -517,3 +517,33 @@ func TestOpenIssueRetryHonorsCanceledContext(t *testing.T) { require.ErrorIs(t, f.m.PersistLog(ctx, f.stores, log, log.BuildID), context.Canceled) } + +func TestReportedIssueDoesNotChangeLifecycle(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)) + before, opening := f.summary, f.openIssue + report := entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: "build/a", Event: entity.IssueEventReported} + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID)) + require.Equal(t, before, f.summary) + require.Equal(t, opening, f.openIssue) + retained := f.logs[logKey{"validation", "unit", "build/a", entity.IssueEventReported}] + require.Equal(t, opened.BuildID, retained.FirstSeenBuildID) + require.Positive(t, retained.TimestampMs) + report.TimestampMs = retained.TimestampMs + 100 + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID)) + require.Equal(t, retained, f.logs[logKey{"validation", "unit", "build/a", entity.IssueEventReported}]) +} + +func TestReportedIssueCannotReopenClosedIssue(t *testing.T) { + f := newMaterializerFixture(t) + opened := testIssueLog("build/z", entity.IssueStatusOpen) + closed := testIssueLog("build/m", entity.IssueStatusClosed) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, opened, opened.BuildID)) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, closed, opened.BuildID)) + before := f.summary + report := entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: "build/a", Event: entity.IssueEventReported} + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, report, opened.BuildID)) + require.Equal(t, before, f.summary) + require.Empty(t, f.openIssue) +} diff --git a/stovepipe/entity/issue_log.go b/stovepipe/entity/issue_log.go index d56674564..4d7bba3a9 100644 --- a/stovepipe/entity/issue_log.go +++ b/stovepipe/entity/issue_log.go @@ -16,22 +16,32 @@ package entity import "fmt" -// IssueLog is an immutable issue lifecycle transition within a queue. -// Task, unit, build, and status identify one transition. +// IssueEvent identifies an issue occurrence; reporting does not change status. +type IssueEvent string + +const ( + IssueEventOpened IssueEvent = "open" + IssueEventReported IssueEvent = "reported" + IssueEventClosed IssueEvent = "closed" +) + +// IssueLog is an immutable occurrence identified by task, unit, build, and event within a queue. type IssueLog struct { // TaskID identifies the logical validation task. TaskID string `json:"task_id"` // UnitID identifies the validation unit; empty for a task-wide issue. UnitID string `json:"unit_id"` - // Status is the open or closed status established by this transition. - Status IssueStatus `json:"status"` - // BuildID identifies the build where the issue was first seen or confirmed closed. + // Event identifies an opening, repeat report, or closure. + Event IssueEvent `json:"event"` + // BuildID identifies the build that produced this occurrence. BuildID string `json:"build_id"` - // TimestampMs is when the transition was first recorded, in Unix milliseconds. + // TimestampMs is when this occurrence was first recorded, in Unix milliseconds. TimestampMs int64 `json:"timestamp_ms"` + // FirstSeenBuildID identifies this opening; empty only on legacy lifecycle records. + FirstSeenBuildID string `json:"first_seen_build_id"` } -// Validate verifies the invariants required for a newly persisted issue log. +// Validate checks the event identity, timestamp, and opening reference. func (l IssueLog) Validate() error { if l.BuildID == "" { return fmt.Errorf("issue log build ID must not be empty") @@ -39,10 +49,21 @@ func (l IssueLog) Validate() error { if l.TimestampMs <= 0 { return fmt.Errorf("issue log timestamp must be positive") } - switch l.Status { - case IssueStatusOpen, IssueStatusClosed: - return nil + switch l.Event { + case IssueEventOpened: + if l.FirstSeenBuildID != "" && l.FirstSeenBuildID != l.BuildID { + return fmt.Errorf("issue opening must reference its source build") + } + case IssueEventReported: + if l.FirstSeenBuildID == "" { + return fmt.Errorf("reported issue requires an opening build") + } + if l.FirstSeenBuildID == l.BuildID { + return fmt.Errorf("reported issue must reference an earlier opening") + } + case IssueEventClosed: default: - return fmt.Errorf("invalid issue status %q", l.Status) + return fmt.Errorf("invalid issue event %q", l.Event) } + return nil } diff --git a/stovepipe/entity/issue_log_test.go b/stovepipe/entity/issue_log_test.go index 432df1e2c..bf957c982 100644 --- a/stovepipe/entity/issue_log_test.go +++ b/stovepipe/entity/issue_log_test.go @@ -27,17 +27,17 @@ func TestIssueLogValidate(t *testing.T) { invalid bool }{ {name: "open"}, - {name: "closed", change: func(l *IssueLog) { l.Status = "closed" }}, + {name: "closed", change: func(l *IssueLog) { l.Event = "closed" }}, {name: "task-wide issue", change: func(l *IssueLog) { l.UnitID = "" }}, {name: "missing build", change: func(l *IssueLog) { l.BuildID = "" }, invalid: true}, {name: "zero timestamp", change: func(l *IssueLog) { l.TimestampMs = 0 }, invalid: true}, {name: "negative timestamp", change: func(l *IssueLog) { l.TimestampMs = -1 }, invalid: true}, - {name: "unknown status", change: func(l *IssueLog) { l.Status = IssueStatusUnknown }, invalid: true}, - {name: "unsupported status", change: func(l *IssueLog) { l.Status = "other" }, invalid: true}, + {name: "unknown event", change: func(l *IssueLog) { l.Event = "" }, invalid: true}, + {name: "unsupported event", change: func(l *IssueLog) { l.Event = "other" }, invalid: true}, } { t.Run(tt.name, func(t *testing.T) { log := IssueLog{TaskID: "task", UnitID: "unit", - Status: IssueStatusOpen, BuildID: "build/1", TimestampMs: 1000} + Event: IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} if tt.change != nil { tt.change(&log) } @@ -50,3 +50,12 @@ func TestIssueLogValidate(t *testing.T) { }) } } + +func TestReportedIssueRequiresEarlierOpening(t *testing.T) { + log := IssueLog{TaskID: "test", BuildID: "build/2", Event: IssueEventReported, TimestampMs: 1000} + require.Error(t, log.Validate()) + log.FirstSeenBuildID = log.BuildID + require.Error(t, log.Validate()) + log.FirstSeenBuildID = "build/1" + require.NoError(t, log.Validate()) +} diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 31b103e7c..c04f76d14 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -95,3 +95,5 @@ The issue-tracking table must include the notification, cursor, `finalize`, and ## Request issue results The request-issue summary store retains one materialized result per request, task, and unit in its bound queue. New includes reopened issues; existing means this request reported an issue that was already open; closed means this request actually closed it. Issues merely left open because closure was suppressed are not existing results. Reads use a bounded request-prefix scan in task/unit order. These results describe the request's accepted updates, not the queue's full open-issue set. + +Issue history retains openings, repeat reports, and closures. The existing event column and its open/closed values remain unchanged; reported events do not change lifecycle projections. New occurrences retain the opening build reference so request projections can be reconstructed without current issue state. Deploy the additive opening-reference column before updated writers. Existing lifecycle records may have an empty reference and continue using the opening supplied to lifecycle materialization. diff --git a/stovepipe/extension/storage/issue_log_store.go b/stovepipe/extension/storage/issue_log_store.go index 4238323e6..6b9c262f3 100644 --- a/stovepipe/extension/storage/issue_log_store.go +++ b/stovepipe/extension/storage/issue_log_store.go @@ -22,13 +22,13 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) -// IssueLogStore retains immutable issue transitions within the bound queue. -// Task, unit, build, and status identify one transition. +// IssueLogStore retains immutable issue events within the bound queue. +// Task, unit, build, and event identify one occurrence. type IssueLogStore interface { - // Create returns ErrAlreadyExists for a previously recorded transition. + // Create returns ErrAlreadyExists for a previously recorded occurrence. // A duplicate never replaces the original timestamp. Create(ctx context.Context, log entity.IssueLog) error - // Get returns ErrNotFound when the transition has not been recorded. - Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (entity.IssueLog, error) + // Get returns ErrNotFound when the occurrence has not been recorded. + Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (entity.IssueLog, error) } diff --git a/stovepipe/extension/storage/mock/issue_log_store_mock.go b/stovepipe/extension/storage/mock/issue_log_store_mock.go index ed1881f22..54453faba 100644 --- a/stovepipe/extension/storage/mock/issue_log_store_mock.go +++ b/stovepipe/extension/storage/mock/issue_log_store_mock.go @@ -56,16 +56,16 @@ func (mr *MockIssueLogStoreMockRecorder) Create(ctx, log any) *gomock.Call { } // Get mocks base method. -func (m *MockIssueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (entity.IssueLog, error) { +func (m *MockIssueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (entity.IssueLog, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID, buildID, status) + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID, buildID, event) ret0, _ := ret[0].(entity.IssueLog) ret1, _ := ret[1].(error) return ret0, ret1 } // Get indicates an expected call of Get. -func (mr *MockIssueLogStoreMockRecorder) Get(ctx, taskID, unitID, buildID, status any) *gomock.Call { +func (mr *MockIssueLogStoreMockRecorder) Get(ctx, taskID, unitID, buildID, event any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueLogStore)(nil).Get), ctx, taskID, unitID, buildID, status) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueLogStore)(nil).Get), ctx, taskID, unitID, buildID, event) } diff --git a/stovepipe/extension/storage/mysql/issue_log_store.go b/stovepipe/extension/storage/mysql/issue_log_store.go index f3914fcc2..7cd2dbb9d 100644 --- a/stovepipe/extension/storage/mysql/issue_log_store.go +++ b/stovepipe/extension/storage/mysql/issue_log_store.go @@ -44,18 +44,21 @@ func (s *issueLogStore) Create(ctx context.Context, log entity.IssueLog) (retErr if err := log.Validate(); err != nil { return err } - if err := s.validateKey(log.TaskID, log.UnitID, log.BuildID, log.Status); err != nil { + if err := validateBuildIssueKey(s.queue, log.FirstSeenBuildID, log.TaskID, log.UnitID); err != nil { + return err + } + if err := s.validateKey(log.TaskID, log.UnitID, log.BuildID, log.Event); err != nil { return err } _, err := s.db.ExecContext(ctx, - `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms) - VALUES (?, ?, ?, ?, ?, ?)`, - s.queue, log.TaskID, log.UnitID, log.BuildID, log.Status, log.TimestampMs) + `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms, first_seen_build_id) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + s.queue, log.TaskID, log.UnitID, log.BuildID, log.Event, log.TimestampMs, log.FirstSeenBuildID) if err != nil { if isDuplicateEntry(err) { - return fmt.Errorf("issue log task=%q unit=%q build=%q status=%q: %w", - log.TaskID, log.UnitID, log.BuildID, log.Status, storage.ErrAlreadyExists) + return fmt.Errorf("issue log task=%q unit=%q build=%q event=%q: %w", + log.TaskID, log.UnitID, log.BuildID, log.Event, storage.ErrAlreadyExists) } return fmt.Errorf("failed to create issue log: %w", err) @@ -64,20 +67,20 @@ func (s *issueLogStore) Create(ctx context.Context, log entity.IssueLog) (retErr return nil } -func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (ret entity.IssueLog, retErr error) { +func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (ret entity.IssueLog, retErr error) { op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - if err := s.validateKey(taskID, unitID, buildID, status); err != nil { + if err := s.validateKey(taskID, unitID, buildID, event); err != nil { return entity.IssueLog{}, err } var log entity.IssueLog err := s.db.QueryRowContext(ctx, - `SELECT task_id, unit_id, build_id, event, timestamp_ms + `SELECT task_id, unit_id, build_id, event, timestamp_ms, first_seen_build_id FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`, - s.queue, taskID, unitID, buildID, status, - ).Scan(&log.TaskID, &log.UnitID, &log.BuildID, &log.Status, &log.TimestampMs) + s.queue, taskID, unitID, buildID, event, + ).Scan(&log.TaskID, &log.UnitID, &log.BuildID, &log.Event, &log.TimestampMs, &log.FirstSeenBuildID) if errors.Is(err, sql.ErrNoRows) { return entity.IssueLog{}, storage.WrapNotFound(err) } @@ -88,18 +91,18 @@ func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, return log, nil } -func (s *issueLogStore) validateKey(taskID, unitID, buildID string, status entity.IssueStatus) error { +func (s *issueLogStore) validateKey(taskID, unitID, buildID string, event entity.IssueEvent) error { if err := validateBuildIssueKey(s.queue, buildID, taskID, unitID); err != nil { return err } if buildID == "" { return fmt.Errorf("issue log build ID must not be empty") } - switch status { - case entity.IssueStatusOpen, entity.IssueStatusClosed: + switch event { + case entity.IssueEventOpened, entity.IssueEventReported, entity.IssueEventClosed: return nil default: - return fmt.Errorf("invalid issue status %q", status) + return fmt.Errorf("invalid issue event %q", event) } } diff --git a/stovepipe/extension/storage/mysql/issue_log_store_test.go b/stovepipe/extension/storage/mysql/issue_log_store_test.go index 33b4343bc..2b2fdb9dc 100644 --- a/stovepipe/extension/storage/mysql/issue_log_store_test.go +++ b/stovepipe/extension/storage/mysql/issue_log_store_test.go @@ -31,7 +31,7 @@ import ( func issueLogFixture() entity.IssueLog { return entity.IssueLog{TaskID: "task", UnitID: "unit", - Status: entity.IssueStatusClosed, BuildID: "build/2", TimestampMs: 2000} + Event: entity.IssueEventClosed, BuildID: "build/2", TimestampMs: 2000} } func TestIssueLogCreate(t *testing.T) { @@ -52,9 +52,9 @@ func TestIssueLogCreate(t *testing.T) { store := NewIssueLogStore(db, testMetrics(), "queue") log := issueLogFixture() query := mock.ExpectExec(regexp.QuoteMeta( - `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms) - VALUES (?, ?, ?, ?, ?, ?)`)). - WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, "closed", log.TimestampMs) + `INSERT INTO issue_log (queue, task_id, unit_id, build_id, event, timestamp_ms, first_seen_build_id) + VALUES (?, ?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, "closed", log.TimestampMs, log.FirstSeenBuildID) if tt.dbErr != nil { query.WillReturnError(tt.dbErr) } else { @@ -81,7 +81,7 @@ func TestIssueLogGet(t *testing.T) { store := NewIssueLogStore(db, testMetrics(), "queue") log := issueLogFixture() query := mock.ExpectQuery(regexp.QuoteMeta( - `SELECT task_id, unit_id, build_id, event, timestamp_ms + `SELECT task_id, unit_id, build_id, event, timestamp_ms, first_seen_build_id FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`)). WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, "closed") switch kind { @@ -94,11 +94,11 @@ func TestIssueLogGet(t *testing.T) { if kind == "scan error" { timestamp = "invalid" } - query.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "build_id", "event", "timestamp_ms"}). - AddRow(log.TaskID, log.UnitID, log.BuildID, log.Status, timestamp)) + query.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "build_id", "event", "timestamp_ms", "first_seen_build_id"}). + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Event, timestamp, log.FirstSeenBuildID)) } - got, err := store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err := store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) switch kind { case "found": require.NoError(t, err) @@ -128,8 +128,8 @@ func TestIssueLogRejectsInvalidKeys(t *testing.T) { {name: "oversized unit", change: func(l *entity.IssueLog, _ *string) { l.UnitID = strings.Repeat("😀", 257) }}, {name: "oversized build", change: func(l *entity.IssueLog, _ *string) { l.BuildID = strings.Repeat("b", 1021) }}, {name: "empty build", change: func(l *entity.IssueLog, _ *string) { l.BuildID = "" }}, - {name: "unknown status", change: func(l *entity.IssueLog, _ *string) { l.Status = entity.IssueStatusUnknown }}, - {name: "unsupported status", change: func(l *entity.IssueLog, _ *string) { l.Status = "other" }}, + {name: "unknown event", change: func(l *entity.IssueLog, _ *string) { l.Event = "" }}, + {name: "unsupported event", change: func(l *entity.IssueLog, _ *string) { l.Event = "other" }}, } { t.Run(tt.name, func(t *testing.T) { db, mock, err := sqlmock.New() @@ -142,11 +142,11 @@ func TestIssueLogRejectsInvalidKeys(t *testing.T) { mock.MatchExpectationsInOrder(false) mock.ExpectExec("INSERT INTO issue_log").WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows( - []string{"task_id", "unit_id", "build_id", "event", "timestamp_ms"}). - AddRow(log.TaskID, log.UnitID, log.BuildID, log.Status, log.TimestampMs)) + []string{"task_id", "unit_id", "build_id", "event", "timestamp_ms", "first_seen_build_id"}). + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Event, log.TimestampMs, log.FirstSeenBuildID)) require.Error(t, store.Create(context.Background(), log)) - _, err = store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Status) + _, err = store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) require.Error(t, err) }) } diff --git a/stovepipe/extension/storage/mysql/schema/issue_log.sql b/stovepipe/extension/storage/mysql/schema/issue_log.sql index 3c26086d0..ae50b6c57 100644 --- a/stovepipe/extension/storage/mysql/schema/issue_log.sql +++ b/stovepipe/extension/storage/mysql/schema/issue_log.sql @@ -1,4 +1,4 @@ --- Each issue transition is retained once per build and event. +-- Each opening, repeat report, or closure is retained once per build and event. CREATE TABLE IF NOT EXISTS issue_log ( queue VARBINARY(255) NOT NULL, task_id VARBINARY(255) NOT NULL, @@ -8,5 +8,7 @@ CREATE TABLE IF NOT EXISTS issue_log ( timestamp_ms BIGINT NOT NULL, + first_seen_build_id VARBINARY(1020) NOT NULL DEFAULT '', + PRIMARY KEY (queue, task_id, unit_id, build_id, event) ) ENGINE=InnoDB; diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index a96407ea6..00146fe31 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -170,7 +170,7 @@ func TestStorageForQueueMetrics(t *testing.T) { name: "issue_log_store", op: "get", read: func(bound storage.Storage, _ string) error { - _, err := bound.GetIssueLogStore().Get(context.Background(), "task", "unit", "build-id", entity.IssueStatusOpen) + _, err := bound.GetIssueLogStore().Get(context.Background(), "task", "unit", "build-id", entity.IssueEventOpened) return err }, }, diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go index f9c6d56aa..8bc4d05c9 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go @@ -32,7 +32,7 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, err) store := bound.GetIssueLogStore() opened := entity.IssueLog{TaskID: "validation", UnitID: "unit", - Status: entity.IssueStatusOpen, BuildID: "build/1", TimestampMs: 1000} + Event: entity.IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} t.Run("history survives closure and reopening", func(t *testing.T) { issue := entity.IssueState{TaskID: opened.TaskID, UnitID: opened.UnitID, FirstSeenBuildID: opened.BuildID, Status: entity.IssueStatusOpen, Version: 1, LastProcessedBuildID: opened.BuildID, LastChangedBuildID: opened.BuildID, ChangedAtMs: 1000} @@ -41,7 +41,7 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, store.Create(ctx, opened)) closed := opened - closed.Status, closed.BuildID, closed.TimestampMs = entity.IssueStatusClosed, "build/2", 2000 + closed.Event, closed.BuildID, closed.TimestampMs = entity.IssueEventClosed, "build/2", 2000 require.NoError(t, store.Create(ctx, closed)) issue.Status = entity.IssueStatusClosed require.NoError(t, bound.GetIssueStateStore().Update(ctx, issue, 1, 2)) @@ -57,11 +57,11 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, store.Create(ctx, reopened)) for _, want := range []entity.IssueLog{opened, closed, reopened} { - got, err := store.Get(ctx, want.TaskID, want.UnitID, want.BuildID, want.Status) + got, err := store.Get(ctx, want.TaskID, want.UnitID, want.BuildID, want.Event) require.NoError(t, err) require.Equal(t, want, got) } - _, err = store.Get(ctx, reopened.TaskID, reopened.UnitID, reopened.BuildID, entity.IssueStatusClosed) + _, err = store.Get(ctx, reopened.TaskID, reopened.UnitID, reopened.BuildID, entity.IssueEventClosed) require.ErrorIs(t, err, storage.ErrNotFound) _, err = bound.GetBuildIssueStore().Get(ctx, opened.BuildID, opened.TaskID, opened.UnitID) require.NoError(t, err) @@ -71,18 +71,18 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor duplicate := opened duplicate.TimestampMs = 4000 require.ErrorIs(t, store.Create(ctx, duplicate), storage.ErrAlreadyExists) - got, err := store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) + got, err := store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) require.NoError(t, err) require.Equal(t, opened, got) }) t.Run("status distinguishes transitions within one build", func(t *testing.T) { - for _, status := range []entity.IssueStatus{entity.IssueStatusOpen, entity.IssueStatusClosed} { + for _, status := range []entity.IssueEvent{entity.IssueEventOpened, entity.IssueEventClosed} { log := opened - log.UnitID, log.Status = "same-build", status + log.UnitID, log.Event = "same-build", status require.NoError(t, store.Create(ctx, log)) - got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) require.NoError(t, err) require.Equal(t, log, got) } @@ -91,12 +91,12 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor t.Run("queue and byte-exact identity isolation", func(t *testing.T) { other, err := factory.For(storage.Config{QueueName: "Issue-history/main"}) require.NoError(t, err) - _, err = other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) + _, err = other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) require.ErrorIs(t, err, storage.ErrNotFound) otherLog := opened otherLog.TimestampMs++ require.NoError(t, other.GetIssueLogStore().Create(ctx, otherLog)) - got, err := other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) + got, err := other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) require.NoError(t, err) require.Equal(t, otherLog, got) @@ -118,12 +118,12 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor log := opened log.TaskID, log.UnitID, log.BuildID = tt.taskID, tt.unitID, tt.buildID require.NoError(t, store.Create(ctx, log)) - got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) require.NoError(t, err) require.Equal(t, log, got) }) } - got, err = store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) + got, err = store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) require.NoError(t, err) require.Equal(t, opened, got) }) @@ -149,7 +149,7 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor } require.Equal(t, 1, successes) require.Equal(t, 1, duplicates) - got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) require.NoError(t, err) require.Equal(t, log, got) }) @@ -159,26 +159,47 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, err) maxStore := maxBound.GetIssueLogStore() log := entity.IssueLog{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), - Status: entity.IssueStatusClosed, + Event: entity.IssueEventClosed, BuildID: strings.Repeat("😀", 255), TimestampMs: 1000} require.NoError(t, maxStore.Create(ctx, log)) - got, err := maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err := maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) require.NoError(t, err) require.Equal(t, log, got) oversized := log oversized.BuildID += "x" require.Error(t, maxStore.Create(ctx, oversized)) - _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Status) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Event) require.Error(t, err) oversized = log oversized.UnitID += "x" require.Error(t, maxStore.Create(ctx, oversized)) - _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Status) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Event) require.Error(t, err) - got, err = maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) + got, err = maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) require.NoError(t, err) require.Equal(t, log, got) }) } + +func testReportedIssueLog(t *testing.T, ctx context.Context, factory storage.Factory) { + bound, err := factory.For(storage.Config{QueueName: "reported-issue-log"}) + require.NoError(t, err) + logs := bound.GetIssueLogStore() + log := entity.IssueLog{TaskID: "test", Event: entity.IssueEventReported, BuildID: "build/2", FirstSeenBuildID: "build/1", TimestampMs: 1000} + require.NoError(t, logs.Create(ctx, log)) + got, err := logs.Get(ctx, "test", "", "build/2", entity.IssueEventReported) + require.NoError(t, err) + require.Equal(t, log, got) + duplicate := log + duplicate.TimestampMs = 2000 + require.ErrorIs(t, logs.Create(ctx, duplicate), storage.ErrAlreadyExists) + got, err = logs.Get(ctx, "test", "", "build/2", entity.IssueEventReported) + require.NoError(t, err) + require.Equal(t, log, got) + other, err := factory.For(storage.Config{QueueName: "other-reported-issue-log"}) + require.NoError(t, err) + _, err = other.GetIssueLogStore().Get(ctx, "test", "", "build/2", entity.IssueEventReported) + require.ErrorIs(t, err, storage.ErrNotFound) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/open_issue_materializer_test.go b/test/integration/stovepipe/extension/storage/mysql/open_issue_materializer_test.go index 5fc6a2207..c05cbfc14 100644 --- a/test/integration/stovepipe/extension/storage/mysql/open_issue_materializer_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/open_issue_materializer_test.go @@ -38,9 +38,9 @@ func testOpenIssueMaterialization(t *testing.T, ctx context.Context, factory sto } materializer := issuelog.NewMaterializer(tally.NoopScope) openIssues := stores.GetOpenIssueStore() - opened := entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: "build/z", Status: entity.IssueStatusOpen, TimestampMs: 1000} + opened := entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: "build/z", Event: entity.IssueEventOpened, TimestampMs: 1000} closed := opened - closed.BuildID, closed.Status, closed.TimestampMs = "build/a", entity.IssueStatusClosed, 2000 + closed.BuildID, closed.Event, closed.TimestampMs = "build/a", entity.IssueEventClosed, 2000 reopened := opened reopened.BuildID, reopened.TimestampMs = "build/m", 3000 for _, step := range []struct { diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 3b3ab46c9..fa522d62e 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("ReportedIssueLog", func(t *testing.T) { + testReportedIssueLog(t, ctx, factory) + }) + t.Run("IssueLogStore", func(t *testing.T) { resetStorage(t, db) testIssueLogStore(t, ctx, factory)