From a87126766a46b4d4758c8369c57830147e2d3ad2 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 29 Oct 2026 04:16:04 +0000 Subject: [PATCH] refactor(stovepipe): Reuse issue status in history --- stovepipe/controller/fix/fix_test.go | 2 +- .../controller/issuetrack/delivery_test.go | 14 +++--- stovepipe/controller/issuetrack/dlq_test.go | 12 ++--- .../controller/issuetrack/fix_handoff_test.go | 2 +- .../controller/issuetrack/fixture_test.go | 12 ++--- stovepipe/controller/issuetrack/issuetrack.go | 7 +-- .../controller/issuetrack/issuetrack_test.go | 6 +-- stovepipe/core/messagequeue/README.md | 4 +- stovepipe/entity/issue_log.go | 26 +++-------- stovepipe/entity/issue_log_test.go | 10 ++-- stovepipe/extension/storage/README.md | 10 ++-- .../extension/storage/issue_log_store.go | 4 +- .../storage/mock/issue_log_store_mock.go | 8 ++-- .../storage/mysql/issue_log_store.go | 24 +++++----- .../storage/mysql/issue_log_store_test.go | 18 ++++---- .../extension/storage/mysql/storage_test.go | 2 +- .../extension/storage/mysql/issue_log_test.go | 46 +++++++++---------- 17 files changed, 91 insertions(+), 116 deletions(-) diff --git a/stovepipe/controller/fix/fix_test.go b/stovepipe/controller/fix/fix_test.go index da21a4856..6d4cdea4a 100644 --- a/stovepipe/controller/fix/fix_test.go +++ b/stovepipe/controller/fix/fix_test.go @@ -162,7 +162,7 @@ func TestProcessBuildChecksLaterPagesForRecordedFix(t *testing.T) { Return([]entity.BuildIssue{issueIdentity(issue)}, nil), m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil), m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))). - Return([]entity.BuildIssue{{TaskID: "resolved", FixID: "fix/1"}}, nil), + Return([]entity.BuildIssue{{TaskID: "closed", FixID: "fix/1"}}, nil), ) require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID)) } diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index 337c362d5..f3d66990a 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -88,8 +88,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.IssueEventResolved}) - require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened}) + require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueStatusClosed}) + require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueStatusOpen}) } func TestEmptyFinalBuildClosesIssuesButRetainsTheirRows(t *testing.T) { @@ -259,7 +259,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.IssueEventOpened}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueStatusOpen}].TimestampMs) } func TestRetryFinishesTheCursorIssueBeforeScanningTheNextBatch(t *testing.T) { @@ -411,7 +411,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.IssueEventResolved}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}].TimestampMs) require.Equal(t, changed, f.states[issueKey{"task", "unit"}]) f.source("build/101", "101", entity.BuildStatusSucceeded) @@ -449,7 +449,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.IssueEventOpened, key.event) + require.Equal(t, entity.IssueStatusOpen, key.status) } require.Equal(t, 1, f.failures["issue/list"]) @@ -489,7 +489,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.IssueEventOpened, key.event) + require.Equal(t, entity.IssueStatusOpen, key.status) } } @@ -591,7 +591,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.IssueEventOpened}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueStatusOpen}].TimestampMs) require.Equal(t, 1, f.failures["issue/list"]) } diff --git a/stovepipe/controller/issuetrack/dlq_test.go b/stovepipe/controller/issuetrack/dlq_test.go index 5fff7947e..c71dc4e37 100644 --- a/stovepipe/controller/issuetrack/dlq_test.go +++ b/stovepipe/controller/issuetrack/dlq_test.go @@ -91,11 +91,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) { for key, state := range f.states { untouched[key] = state } - event := entity.IssueEventOpened - if status == entity.IssueStatusClosed { - event = entity.IssueEventResolved - } - delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event}) + delete(f.initialLogs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, 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 +102,7 @@ func TestRecoveryKeepsAcceptedIssueChanges(t *testing.T) { } require.Equal(t, expected, f.states[key]) } - require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, event}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, original.LastChangedBuildID, 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 +189,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.IssueEventOpened}) + require.NotContains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}) } func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { @@ -239,7 +235,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.IssueEventOpened}] + log := f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}] require.Equal(t, "build/100", issue.FirstSeenBuildID) require.Equal(t, issue.ChangedAtMs, log.TimestampMs) if step == "issue/create" { diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go index 1b82ea3bc..2239e0eae 100644 --- a/stovepipe/controller/issuetrack/fix_handoff_test.go +++ b/stovepipe/controller/issuetrack/fix_handoff_test.go @@ -46,7 +46,7 @@ func TestOpeningPassPublishesOneBuildRepairRequest(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.IssueEventOpened}) + require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueStatusOpen}) } 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 929067ed0..96b87eb40 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -37,8 +37,8 @@ const testQueue = "validation/main" type logKey struct { issueKey - build string - event entity.IssueEvent + build string + status entity.IssueStatus } type fixture struct { @@ -119,7 +119,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.Event} + key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Status} if _, ok := f.initialLogs[key]; ok { return storage.ErrAlreadyExists } @@ -304,9 +304,5 @@ 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} - event := entity.IssueEventOpened - if status == entity.IssueStatusClosed { - event = entity.IssueEventResolved - } - f.initialLogs[logKey{issueKey{task, unit}, "older", event}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Event: event, TimestampMs: 500} + f.initialLogs[logKey{issueKey{task, unit}, "older", status}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Status: status, TimestampMs: 500} } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 0796733c1..44d14748e 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -388,12 +388,7 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr } func (c *Controller) recordHistory(ctx context.Context, store storage.Storage, issue entity.IssueState) error { - event := entity.IssueEventOpened - if issue.Status == entity.IssueStatusClosed { - event = entity.IssueEventResolved - } - - log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Event: event, TimestampMs: issue.ChangedAtMs} + log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Status: issue.Status, TimestampMs: issue.ChangedAtMs} if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { 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 9e104cb00..42f6b9f46 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -92,12 +92,12 @@ 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.IssueEventOpened}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}) } else { 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.IssueEventResolved}) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}) } }) } @@ -138,7 +138,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.IssueEventOpened}].TimestampMs) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen}].TimestampMs) } func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index a4f05b33d..815f9f6f3 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -16,8 +16,8 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali `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`. -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. +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. -Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before resolving issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID. +Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. 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. Only the handoff contract is defined. No publisher, consumer, topic registration, reconciliation progress, or writes to open issues and lifecycle history are enabled. diff --git a/stovepipe/entity/issue_log.go b/stovepipe/entity/issue_log.go index 99619eac9..d56674564 100644 --- a/stovepipe/entity/issue_log.go +++ b/stovepipe/entity/issue_log.go @@ -16,28 +16,16 @@ package entity import "fmt" -// IssueEvent identifies an issue lifecycle transition. -type IssueEvent string - -const ( - // IssueEventUnknown is the unset event value. - IssueEventUnknown IssueEvent = "" - // IssueEventOpened records that an issue became open. - IssueEventOpened IssueEvent = "opened" - // IssueEventResolved records that an issue was resolved. - IssueEventResolved IssueEvent = "resolved" -) - // IssueLog is an immutable issue lifecycle transition within a queue. -// Task, unit, build, and event identify one transition. +// Task, unit, build, and status identify one transition. 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"` - // Event identifies the opening or resolution. - Event IssueEvent `json:"event"` - // BuildID identifies the build where the issue was first seen or confirmed resolved. + // 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. BuildID string `json:"build_id"` // TimestampMs is when the transition was first recorded, in Unix milliseconds. TimestampMs int64 `json:"timestamp_ms"` @@ -51,10 +39,10 @@ func (l IssueLog) Validate() error { if l.TimestampMs <= 0 { return fmt.Errorf("issue log timestamp must be positive") } - switch l.Event { - case IssueEventOpened, IssueEventResolved: + switch l.Status { + case IssueStatusOpen, IssueStatusClosed: return nil default: - return fmt.Errorf("invalid issue event %q", l.Event) + return fmt.Errorf("invalid issue status %q", l.Status) } } diff --git a/stovepipe/entity/issue_log_test.go b/stovepipe/entity/issue_log_test.go index aa6387149..432df1e2c 100644 --- a/stovepipe/entity/issue_log_test.go +++ b/stovepipe/entity/issue_log_test.go @@ -26,18 +26,18 @@ func TestIssueLogValidate(t *testing.T) { change func(*IssueLog) invalid bool }{ - {name: "opened"}, - {name: "resolved", change: func(l *IssueLog) { l.Event = IssueEventResolved }}, + {name: "open"}, + {name: "closed", change: func(l *IssueLog) { l.Status = "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 event", change: func(l *IssueLog) { l.Event = IssueEventUnknown }, invalid: true}, - {name: "unsupported event", change: func(l *IssueLog) { l.Event = "other" }, 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}, } { t.Run(tt.name, func(t *testing.T) { log := IssueLog{TaskID: "task", UnitID: "unit", - Event: IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} + Status: IssueStatusOpen, BuildID: "build/1", TimestampMs: 1000} if tt.change != nil { tt.change(&log) } diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index b456ee8b5..7eb97d1be 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -32,7 +32,7 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte Listing fixes the queue/build primary-key prefix and returns bounded pages in ascending bytewise task/unit order. A task/unit cursor continues after the last returned key; the zero cursor starts the range, and a unit without a task is invalid. Pages are not a snapshot across calls: concurrent observations can affect membership. Empty pages represent no observations in that range, not build success or completed analysis. -Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or resolved. Repair metadata is mutable; observation identities remain durable, with no deletion or absence-based recovery. +Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or closed. Repair metadata is mutable; observation identities remain durable, with no deletion or absence-based recovery. The MySQL primary key contains the queue, build ID, task ID, and unit ID directly as binary strings, preserving byte-exact identity. Their limits are 255, 1020, 255, and 1024 bytes respectively, for a combined maximum of 2554 bytes. Creates and reads reject identifiers exceeding these limits before querying the database; identifiers are never truncated. @@ -52,13 +52,13 @@ The MySQL backend uses the new `issue_state` table. Create it before deploying t ## Issue lifecycle history -`IssueLogStore` retains immutable opened and resolved transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and event within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to. +`IssueLogStore` retains immutable open/closed status transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and status within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to. -The build ID identifies the build where the issue was first seen or confirmed resolved. The application supplies a positive Unix-millisecond timestamp for when the transition was first recorded, not when the build finished. A duplicate identity returns `ErrAlreadyExists` without replacing the original timestamp. Callers can treat replay as success or read the original event. Build IDs are opaque and do not establish transition ordering. +The build ID identifies the build where the issue was first seen or confirmed closed. The application supplies a positive Unix-millisecond timestamp for when the transition was first recorded, not when the build finished. A duplicate identity returns `ErrAlreadyExists` without replacing the original timestamp. Callers can treat replay as success or read the original transition. Build IDs are opaque and do not establish transition ordering. -The MySQL key uses the same binary identifier limits as the issue stores, plus a 32-byte event field, for a maximum of 2586 bytes. Reads and creates use primary-key operations; there is no update or deletion API, metadata collection, or repair state. +The MySQL key uses the same binary identifier limits as the issue stores, plus the existing 32-byte `event` column containing the status, for a maximum of 2586 bytes. Its column name and primary key remain unchanged. Reads and creates use primary-key operations; there is no update or deletion API, metadata collection, or repair state. -History persistence does not alter issue state or validate transitions against it. No runtime writer is enabled here. A later reconciliation controller must own opening/resolution decisions, retry-safe writes to both stores, and protection against stale build analysis. History enumeration is outside this storage contract. +History persistence does not alter issue state or validate transitions against it. The issue-tracking controller owns opening/closing decisions and retry-safe history writes. History enumeration is outside this storage contract. ## Issue reconciliation progress diff --git a/stovepipe/extension/storage/issue_log_store.go b/stovepipe/extension/storage/issue_log_store.go index 9ea03eafe..4238323e6 100644 --- a/stovepipe/extension/storage/issue_log_store.go +++ b/stovepipe/extension/storage/issue_log_store.go @@ -23,12 +23,12 @@ import ( ) // IssueLogStore retains immutable issue transitions within the bound queue. -// Task, unit, build, and event identify one transition. +// Task, unit, build, and status identify one transition. type IssueLogStore interface { // Create returns ErrAlreadyExists for a previously recorded transition. // 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, event entity.IssueEvent) (entity.IssueLog, error) + Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (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 54453faba..ed1881f22 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, event entity.IssueEvent) (entity.IssueLog, error) { +func (m *MockIssueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (entity.IssueLog, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID, buildID, event) + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID, buildID, status) 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, event any) *gomock.Call { +func (mr *MockIssueLogStoreMockRecorder) Get(ctx, taskID, unitID, buildID, status 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, event) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueLogStore)(nil).Get), ctx, taskID, unitID, buildID, status) } diff --git a/stovepipe/extension/storage/mysql/issue_log_store.go b/stovepipe/extension/storage/mysql/issue_log_store.go index 8ef1d1089..f3914fcc2 100644 --- a/stovepipe/extension/storage/mysql/issue_log_store.go +++ b/stovepipe/extension/storage/mysql/issue_log_store.go @@ -44,18 +44,18 @@ 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.Event); err != nil { + if err := s.validateKey(log.TaskID, log.UnitID, log.BuildID, log.Status); 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.Event, log.TimestampMs) + s.queue, log.TaskID, log.UnitID, log.BuildID, log.Status, log.TimestampMs) if err != nil { if isDuplicateEntry(err) { - 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("issue log task=%q unit=%q build=%q status=%q: %w", + log.TaskID, log.UnitID, log.BuildID, log.Status, storage.ErrAlreadyExists) } return fmt.Errorf("failed to create issue log: %w", err) @@ -64,11 +64,11 @@ func (s *issueLogStore) Create(ctx context.Context, log entity.IssueLog) (retErr return nil } -func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, event entity.IssueEvent) (ret entity.IssueLog, retErr error) { +func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, status entity.IssueStatus) (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, event); err != nil { + if err := s.validateKey(taskID, unitID, buildID, status); err != nil { return entity.IssueLog{}, err } @@ -76,8 +76,8 @@ func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, err := s.db.QueryRowContext(ctx, `SELECT task_id, unit_id, build_id, event, timestamp_ms FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`, - s.queue, taskID, unitID, buildID, event, - ).Scan(&log.TaskID, &log.UnitID, &log.BuildID, &log.Event, &log.TimestampMs) + s.queue, taskID, unitID, buildID, status, + ).Scan(&log.TaskID, &log.UnitID, &log.BuildID, &log.Status, &log.TimestampMs) if errors.Is(err, sql.ErrNoRows) { return entity.IssueLog{}, storage.WrapNotFound(err) } @@ -88,18 +88,18 @@ func (s *issueLogStore) Get(ctx context.Context, taskID, unitID, buildID string, return log, nil } -func (s *issueLogStore) validateKey(taskID, unitID, buildID string, event entity.IssueEvent) error { +func (s *issueLogStore) validateKey(taskID, unitID, buildID string, status entity.IssueStatus) 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 event { - case entity.IssueEventOpened, entity.IssueEventResolved: + switch status { + case entity.IssueStatusOpen, entity.IssueStatusClosed: return nil default: - return fmt.Errorf("invalid issue event %q", event) + return fmt.Errorf("invalid issue status %q", status) } } diff --git a/stovepipe/extension/storage/mysql/issue_log_store_test.go b/stovepipe/extension/storage/mysql/issue_log_store_test.go index ebeec0375..33b4343bc 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", - Event: entity.IssueEventResolved, BuildID: "build/2", TimestampMs: 2000} + Status: entity.IssueStatusClosed, BuildID: "build/2", TimestampMs: 2000} } func TestIssueLogCreate(t *testing.T) { @@ -54,7 +54,7 @@ func TestIssueLogCreate(t *testing.T) { 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, log.Event, log.TimestampMs) + WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, "closed", log.TimestampMs) if tt.dbErr != nil { query.WillReturnError(tt.dbErr) } else { @@ -83,7 +83,7 @@ func TestIssueLogGet(t *testing.T) { query := mock.ExpectQuery(regexp.QuoteMeta( `SELECT task_id, unit_id, build_id, event, timestamp_ms FROM issue_log WHERE queue = ? AND task_id = ? AND unit_id = ? AND build_id = ? AND event = ?`)). - WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, log.Event) + WithArgs("queue", log.TaskID, log.UnitID, log.BuildID, "closed") switch kind { case "missing": query.WillReturnError(sql.ErrNoRows) @@ -95,10 +95,10 @@ func TestIssueLogGet(t *testing.T) { timestamp = "invalid" } query.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "build_id", "event", "timestamp_ms"}). - AddRow(log.TaskID, log.UnitID, log.BuildID, log.Event, timestamp)) + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Status, timestamp)) } - got, err := store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) + got, err := store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Status) 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 event", change: func(l *entity.IssueLog, _ *string) { l.Event = entity.IssueEventUnknown }}, - {name: "unsupported event", change: func(l *entity.IssueLog, _ *string) { l.Event = "other" }}, + {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" }}, } { t.Run(tt.name, func(t *testing.T) { db, mock, err := sqlmock.New() @@ -143,10 +143,10 @@ func TestIssueLogRejectsInvalidKeys(t *testing.T) { 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.Event, log.TimestampMs)) + AddRow(log.TaskID, log.UnitID, log.BuildID, log.Status, log.TimestampMs)) require.Error(t, store.Create(context.Background(), log)) - _, err = store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Event) + _, err = store.Get(context.Background(), log.TaskID, log.UnitID, log.BuildID, log.Status) require.Error(t, err) }) } diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index a8e7a4f4b..2d2fda26b 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -152,7 +152,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.IssueEventOpened) + _, err := bound.GetIssueLogStore().Get(context.Background(), "task", "unit", "build-id", entity.IssueStatusOpen) 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 0cd9dcf3d..f9c6d56aa 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", - Event: entity.IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} + Status: entity.IssueStatusOpen, 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} @@ -40,9 +40,9 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, bound.GetBuildIssueStore().Create(ctx, opened.BuildID, entity.BuildIssue{TaskID: opened.TaskID, UnitID: opened.UnitID})) require.NoError(t, store.Create(ctx, opened)) - resolved := opened - resolved.Event, resolved.BuildID, resolved.TimestampMs = entity.IssueEventResolved, "build/2", 2000 - require.NoError(t, store.Create(ctx, resolved)) + closed := opened + closed.Status, closed.BuildID, closed.TimestampMs = entity.IssueStatusClosed, "build/2", 2000 + require.NoError(t, store.Create(ctx, closed)) issue.Status = entity.IssueStatusClosed require.NoError(t, bound.GetIssueStateStore().Update(ctx, issue, 1, 2)) retained, err := bound.GetIssueStateStore().Get(ctx, issue.TaskID, issue.UnitID) @@ -56,12 +56,12 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor require.NoError(t, bound.GetIssueStateStore().Update(ctx, issue, 2, 3)) require.NoError(t, store.Create(ctx, reopened)) - for _, want := range []entity.IssueLog{opened, resolved, reopened} { - got, err := store.Get(ctx, want.TaskID, want.UnitID, want.BuildID, want.Event) + for _, want := range []entity.IssueLog{opened, closed, reopened} { + got, err := store.Get(ctx, want.TaskID, want.UnitID, want.BuildID, want.Status) require.NoError(t, err) require.Equal(t, want, got) } - _, err = store.Get(ctx, reopened.TaskID, reopened.UnitID, reopened.BuildID, entity.IssueEventResolved) + _, err = store.Get(ctx, reopened.TaskID, reopened.UnitID, reopened.BuildID, entity.IssueStatusClosed) 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.Event) + got, err := store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) require.NoError(t, err) require.Equal(t, opened, got) }) - t.Run("event distinguishes transitions within one build", func(t *testing.T) { - for _, event := range []entity.IssueEvent{entity.IssueEventOpened, entity.IssueEventResolved} { + t.Run("status distinguishes transitions within one build", func(t *testing.T) { + for _, status := range []entity.IssueStatus{entity.IssueStatusOpen, entity.IssueStatusClosed} { log := opened - log.UnitID, log.Event = "same-build", event + log.UnitID, log.Status = "same-build", status require.NoError(t, store.Create(ctx, log)) - got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) 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.Event) + _, err = other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) 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.Event) + got, err := other.GetIssueLogStore().Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) require.NoError(t, err) require.Equal(t, otherLog, got) @@ -118,17 +118,17 @@ 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.Event) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) require.NoError(t, err) require.Equal(t, log, got) }) } - got, err = store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Event) + got, err = store.Get(ctx, opened.TaskID, opened.UnitID, opened.BuildID, opened.Status) require.NoError(t, err) require.Equal(t, opened, got) }) - t.Run("concurrent replay retains one event", func(t *testing.T) { + t.Run("concurrent replay retains one transition", func(t *testing.T) { log := opened log.UnitID = "concurrent" results := make(chan error, 2) @@ -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.Event) + got, err := store.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) require.NoError(t, err) require.Equal(t, log, got) }) @@ -159,25 +159,25 @@ 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), - Event: entity.IssueEventResolved, + Status: entity.IssueStatusClosed, 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.Event) + got, err := maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) 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.Event) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Status) 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.Event) + _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID, oversized.BuildID, oversized.Status) require.Error(t, err) - got, err = maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Event) + got, err = maxStore.Get(ctx, log.TaskID, log.UnitID, log.BuildID, log.Status) require.NoError(t, err) require.Equal(t, log, got) })