Skip to content

Commit 4db5cf1

Browse files
prathshenoyprathshenoy
authored
refactor(stovepipe): Reuse issue status in history (#985)
## Summary **What**: - Reuse the same open/closed statuses for current issue records and retained history. **Why**: - Remove two vocabularies for the same issue lifecycle. Current records use open/closed, while history uses opened/resolved and requires a translation. Sharing the status type makes the records consistent without changing when issues open or close. ## Test Plan - [x] Add unit tests. - [x] Verify issue-history persistence against a real MySQL database. ## Revert Plan - Revert this PR before enabling issue-history writes. The table layout remains unchanged, with no existing history to convert. ## Issue Links - [CODEM-607](https://linear.app/uber/issue/CODEM-607/build-analysis)
1 parent 31b95a4 commit 4db5cf1

17 files changed

Lines changed: 91 additions & 116 deletions

File tree

‎stovepipe/controller/fix/fix_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ func TestProcessBuildChecksLaterPagesForRecordedFix(t *testing.T) {
162162
Return([]entity.BuildIssue{issueIdentity(issue)}, nil),
163163
m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil),
164164
m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).
165-
Return([]entity.BuildIssue{{TaskID: "resolved", FixID: "fix/1"}}, nil),
165+
Return([]entity.BuildIssue{{TaskID: "closed", FixID: "fix/1"}}, nil),
166166
)
167167
require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID))
168168
}

‎stovepipe/controller/issuetrack/delivery_test.go‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -88,8 +88,8 @@ func TestFinalizationOpensKeepsAndClosesIssues(t *testing.T) {
8888
require.Equal(t, "build", f.states[issueKey{"c", "unit"}].FirstSeenBuildID)
8989
require.Equal(t, int32(8), f.states[issueKey{"closed", ""}].Version)
9090
require.Len(t, f.logs, 2)
91-
require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueEventResolved})
92-
require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened})
91+
require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueStatusClosed})
92+
require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueStatusOpen})
9393
}
9494

9595
func TestEmptyFinalBuildClosesIssuesButRetainsTheirRows(t *testing.T) {
@@ -259,7 +259,7 @@ func TestRetryRepairsHistoryBeforeFinalizing(t *testing.T) {
259259
require.Equal(t, "task", f.progress.CursorTaskID)
260260
require.Equal(t, "unit", f.progress.CursorUnitID)
261261
require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
262-
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueEventOpened}].TimestampMs)
262+
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueStatusOpen}].TimestampMs)
263263
}
264264

265265
func TestRetryFinishesTheCursorIssueBeforeScanningTheNextBatch(t *testing.T) {
@@ -411,7 +411,7 @@ func TestHistoryRecordsTheActualChangingBuildNotTheLatestProcessedBuild(t *testi
411411
require.Empty(t, f.logs)
412412
f.controller.nowMs = func() int64 { return 2000 }
413413
require.NoError(t, f.deliver(msg, false))
414-
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventResolved}].TimestampMs)
414+
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}].TimestampMs)
415415
require.Equal(t, changed, f.states[issueKey{"task", "unit"}])
416416

417417
f.source("build/101", "101", entity.BuildStatusSucceeded)
@@ -449,7 +449,7 @@ func TestTaskWideFailurePreservesAllExistingIssues(t *testing.T) {
449449
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "closed"}].Status)
450450
require.Len(t, f.logs, 3)
451451
for key := range f.logs {
452-
require.Equal(t, entity.IssueEventOpened, key.event)
452+
require.Equal(t, entity.IssueStatusOpen, key.status)
453453
}
454454
require.Equal(t, 1, f.failures["issue/list"])
455455

@@ -489,7 +489,7 @@ func TestTaskWideFailureInLaterOpeningBatchPreventsAllClosures(t *testing.T) {
489489
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
490490
require.Len(t, f.logs, 106)
491491
for key := range f.logs {
492-
require.Equal(t, entity.IssueEventOpened, key.event)
492+
require.Equal(t, entity.IssueStatusOpen, key.status)
493493
}
494494
}
495495

@@ -591,7 +591,7 @@ func TestTaskWideHistoryRetryFinishesWithoutClosing(t *testing.T) {
591591
require.NoError(t, f.deliver(msg, false))
592592
require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
593593
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
594-
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueEventOpened}].TimestampMs)
594+
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueStatusOpen}].TimestampMs)
595595
require.Equal(t, 1, f.failures["issue/list"])
596596
}
597597

‎stovepipe/controller/issuetrack/dlq_test.go‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -91,11 +91,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) {
9191
for key, state := range f.states {
9292
untouched[key] = state
9393
}
94-
event := entity.IssueEventOpened
95-
if status == entity.IssueStatusClosed {
96-
event = entity.IssueEventResolved
97-
}
98-
delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event})
94+
delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, status})
9995
f.failures["issue/list"], f.failures["build-issues/list"], f.failures["build-issues/get"] = 1, 1, 1
10096

10197
require.NoError(t, f.recover(msg, false))
@@ -106,7 +102,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) {
106102
}
107103
require.Equal(t, expected, f.states[key])
108104
}
109-
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event})
105+
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, status})
110106
require.Len(t, f.events, 1)
111107
require.Equal(t, 1, f.failures["issue/list"])
112108
require.Equal(t, 1, f.failures["build-issues/list"])
@@ -193,7 +189,7 @@ func TestRecoveryDoesNotReopenAnExistingClosedCursor(t *testing.T) {
193189
require.NoError(t, f.recover(msg, false))
194190
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status)
195191
require.Equal(t, "older", f.states[issueKey{"task", "unit"}].FirstSeenBuildID)
196-
require.NotContains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened})
192+
require.NotContains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen})
197193
}
198194

199195
func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) {
@@ -239,7 +235,7 @@ func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) {
239235

240236
require.NoError(t, f.recover(msg, false))
241237
issue := f.states[issueKey{"task", "unit"}]
242-
log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}]
238+
log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}]
243239
require.Equal(t, "build/100", issue.FirstSeenBuildID)
244240
require.Equal(t, issue.ChangedAtMs, log.TimestampMs)
245241
if step == "issue/create" {

‎stovepipe/controller/issuetrack/fix_handoff_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) {
4646
require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
4747
for _, task := range []string{"new", "reopened"} {
4848
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{task, "unit"}].Status)
49-
require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueEventOpened})
49+
require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueStatusOpen})
5050
}
5151
payload := &stovepipemq.FixRequest{}
5252
require.NoError(t, stovepipemq.Unmarshal(msg.Payload, payload))

‎stovepipe/controller/issuetrack/fixture_test.go‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,8 @@ const testQueue = "validation/main"
3737

3838
type logKey struct {
3939
issueKey
40-
build string
41-
event entity.IssueEvent
40+
build string
41+
status entity.IssueStatus
4242
}
4343

4444
type fixture struct {
@@ -119,7 +119,7 @@ func newFixture(t *testing.T) *fixture {
119119
if err := f.fail("history/create"); err != nil {
120120
return err
121121
}
122-
key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Event}
122+
key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Status}
123123
if _, ok := f.initialLogs[key]; ok {
124124
return storage.ErrAlreadyExists
125125
}
@@ -304,9 +304,5 @@ func (f *fixture) reportIssue(build, task, unit string) {
304304

305305
func (f *fixture) seedIssue(task, unit string, status entity.IssueStatus) {
306306
f.states[issueKey{task, unit}] = entity.IssueState{TaskID: task, UnitID: unit, Status: status, FirstSeenBuildID: "older", Version: 7, LastProcessedBuildID: "older", LastChangedBuildID: "older", ChangedAtMs: 500}
307-
event := entity.IssueEventOpened
308-
if status == entity.IssueStatusClosed {
309-
event = entity.IssueEventResolved
310-
}
311-
f.initialLogs[logKey{issueKey{task, unit}, "older", event}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Event: event, TimestampMs: 500}
307+
f.initialLogs[logKey{issueKey{task, unit}, "older", status}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Status: status, TimestampMs: 500}
312308
}

‎stovepipe/controller/issuetrack/issuetrack.go‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -388,12 +388,7 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr
388388
}
389389

390390
func (c *Controller) recordHistory(ctx context.Context, store storage.Storage, issue entity.IssueState) error {
391-
event := entity.IssueEventOpened
392-
if issue.Status == entity.IssueStatusClosed {
393-
event = entity.IssueEventResolved
394-
}
395-
396-
log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Event: event, TimestampMs: issue.ChangedAtMs}
391+
log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Status: issue.Status, TimestampMs: issue.ChangedAtMs}
397392
if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
398393
return fmt.Errorf("record issue transition: %w", err)
399394
}

‎stovepipe/controller/issuetrack/issuetrack_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -92,12 +92,12 @@ func TestProcessIssueOpensAndCloses(t *testing.T) {
9292
require.Equal(t, entity.IssueStatusOpen, issue.Status)
9393
require.Equal(t, "build/100", issue.FirstSeenBuildID)
9494
require.Equal(t, int32(1), issue.Version)
95-
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened})
95+
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen})
9696
} else {
9797
require.Equal(t, entity.IssueStatusClosed, issue.Status)
9898
require.Equal(t, "older", issue.FirstSeenBuildID)
9999
require.Equal(t, int32(8), issue.Version)
100-
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventResolved})
100+
require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed})
101101
}
102102
})
103103
}
@@ -138,7 +138,7 @@ func TestHistoryRetryUsesTheStoredTransition(t *testing.T) {
138138
_, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false)
139139
require.NoError(t, err)
140140
require.Equal(t, applied, f.states[issueKey{"task", "unit"}])
141-
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}].TimestampMs)
141+
require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}].TimestampMs)
142142
}
143143

144144
func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) {

‎stovepipe/core/messagequeue/README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,8 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali
1616

1717
`IssueTrack` binds to `TopicKeyIssueTrack` and identifies a persisted build and its queue. Notifications are partitioned by queue so both running and final issue-state updates share one serialized path, independently of build-partitioned analysis polling. The message carries identifiers rather than a failure collection; per-build observations are read through `BuildIssueStore.List`.
1818

19-
Publish only after detection and observation persistence succeed. With `finalize` unset, the notification permits opening newly detected issues, not resolution by absence. With `finalize` set, it requests final reconciliation after a successful terminal analysis. Publish that notification even when no failures were detected; terminal Build status alone does not replace the handoff. Publication retries must also follow duplicate observation writes so a crash between persistence and publication does not strand work.
19+
Publish only after detection and observation persistence succeed. With `finalize` unset, the notification permits opening newly detected issues, not closing by absence. With `finalize` set, it requests final reconciliation after a successful terminal analysis. Publish that notification even when no failures were detected; terminal Build status alone does not replace the handoff. Publication retries must also follow duplicate observation writes so a crash between persistence and publication does not strand work.
2020

21-
Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before resolving issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID.
21+
Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before closing issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID.
2222

2323
Only the handoff contract is defined. No publisher, consumer, topic registration, reconciliation progress, or writes to open issues and lifecycle history are enabled.

‎stovepipe/entity/issue_log.go‎

Lines changed: 7 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,28 +16,16 @@ package entity
1616

1717
import "fmt"
1818

19-
// IssueEvent identifies an issue lifecycle transition.
20-
type IssueEvent string
21-
22-
const (
23-
// IssueEventUnknown is the unset event value.
24-
IssueEventUnknown IssueEvent = ""
25-
// IssueEventOpened records that an issue became open.
26-
IssueEventOpened IssueEvent = "opened"
27-
// IssueEventResolved records that an issue was resolved.
28-
IssueEventResolved IssueEvent = "resolved"
29-
)
30-
3119
// IssueLog is an immutable issue lifecycle transition within a queue.
32-
// Task, unit, build, and event identify one transition.
20+
// Task, unit, build, and status identify one transition.
3321
type IssueLog struct {
3422
// TaskID identifies the logical validation task.
3523
TaskID string `json:"task_id"`
3624
// UnitID identifies the validation unit; empty for a task-wide issue.
3725
UnitID string `json:"unit_id"`
38-
// Event identifies the opening or resolution.
39-
Event IssueEvent `json:"event"`
40-
// BuildID identifies the build where the issue was first seen or confirmed resolved.
26+
// Status is the open or closed status established by this transition.
27+
Status IssueStatus `json:"status"`
28+
// BuildID identifies the build where the issue was first seen or confirmed closed.
4129
BuildID string `json:"build_id"`
4230
// TimestampMs is when the transition was first recorded, in Unix milliseconds.
4331
TimestampMs int64 `json:"timestamp_ms"`
@@ -51,10 +39,10 @@ func (l IssueLog) Validate() error {
5139
if l.TimestampMs <= 0 {
5240
return fmt.Errorf("issue log timestamp must be positive")
5341
}
54-
switch l.Event {
55-
case IssueEventOpened, IssueEventResolved:
42+
switch l.Status {
43+
case IssueStatusOpen, IssueStatusClosed:
5644
return nil
5745
default:
58-
return fmt.Errorf("invalid issue event %q", l.Event)
46+
return fmt.Errorf("invalid issue status %q", l.Status)
5947
}
6048
}

‎stovepipe/entity/issue_log_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,18 +26,18 @@ func TestIssueLogValidate(t *testing.T) {
2626
change func(*IssueLog)
2727
invalid bool
2828
}{
29-
{name: "opened"},
30-
{name: "resolved", change: func(l *IssueLog) { l.Event = IssueEventResolved }},
29+
{name: "open"},
30+
{name: "closed", change: func(l *IssueLog) { l.Status = "closed" }},
3131
{name: "task-wide issue", change: func(l *IssueLog) { l.UnitID = "" }},
3232
{name: "missing build", change: func(l *IssueLog) { l.BuildID = "" }, invalid: true},
3333
{name: "zero timestamp", change: func(l *IssueLog) { l.TimestampMs = 0 }, invalid: true},
3434
{name: "negative timestamp", change: func(l *IssueLog) { l.TimestampMs = -1 }, invalid: true},
35-
{name: "unknown event", change: func(l *IssueLog) { l.Event = IssueEventUnknown }, invalid: true},
36-
{name: "unsupported event", change: func(l *IssueLog) { l.Event = "other" }, invalid: true},
35+
{name: "unknown status", change: func(l *IssueLog) { l.Status = IssueStatusUnknown }, invalid: true},
36+
{name: "unsupported status", change: func(l *IssueLog) { l.Status = "other" }, invalid: true},
3737
} {
3838
t.Run(tt.name, func(t *testing.T) {
3939
log := IssueLog{TaskID: "task", UnitID: "unit",
40-
Event: IssueEventOpened, BuildID: "build/1", TimestampMs: 1000}
40+
Status: IssueStatusOpen, BuildID: "build/1", TimestampMs: 1000}
4141
if tt.change != nil {
4242
tt.change(&log)
4343
}

0 commit comments

Comments
 (0)