From 081c9e15b073438fcf7b4723b5f2d9e51788c608 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 29 Oct 2026 03:45:17 +0000 Subject: [PATCH] feat(stovepipe): Materialize issue summaries --- stovepipe/controller/issuetrack/BUILD.bazel | 1 + stovepipe/controller/issuetrack/README.md | 6 +- stovepipe/controller/issuetrack/dlq_test.go | 8 +- .../controller/issuetrack/fixture_test.go | 53 +++ stovepipe/controller/issuetrack/issuetrack.go | 5 +- .../controller/issuetrack/issuetrack_test.go | 32 ++ stovepipe/core/issuelog/BUILD.bazel | 30 ++ stovepipe/core/issuelog/README.md | 9 + stovepipe/core/issuelog/materializer.go | 156 +++++++++ stovepipe/core/issuelog/materializer_test.go | 308 ++++++++++++++++++ stovepipe/extension/storage/README.md | 4 +- 11 files changed, 605 insertions(+), 7 deletions(-) create mode 100644 stovepipe/core/issuelog/BUILD.bazel create mode 100644 stovepipe/core/issuelog/README.md create mode 100644 stovepipe/core/issuelog/materializer.go create mode 100644 stovepipe/core/issuelog/materializer_test.go diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index cea99a725..f0a302ea4 100644 --- a/stovepipe/controller/issuetrack/BUILD.bazel +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -14,6 +14,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/fixrequest:go_default_library", + "//stovepipe/core/issuelog:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 68cbbf2cd..866349edc 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -8,7 +8,7 @@ Issue tracking maintains current issue state and opening/closing history from pe 2. Reject a source older than the issue's last processed build, using Request order rather than opaque Build IDs. 3. Before changing the issue, conditionally save its task/unit key as the current queue cursor **after** reading the issue row. Repeat this conditional write even when resuming the same cursor. 4. Conditionally create or update the issue row. Record the processed build even when its open/closed status stays the same; update the changing build and timestamp only on an actual transition. -5. Record opening or closing history from the issue row's changing build and timestamp. On retry, the inclusive cursor revisits that row and repairs its history before proceeding. +5. Retain opening or closing history from the issue row's changing build and timestamp, then materialize its summary. On retry, the inclusive cursor revisits that row and repairs both before proceeding. Closed issue rows are retained, so their versions cannot reset on reopening. A repeated result from the same build needs no new issue write. Its history is repaired when revisiting the current cursor. @@ -16,6 +16,8 @@ Closed issue rows are retained, so their versions cannot reset on reopening. A r During issue application, the caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later is rejected by source ordering or the conditional cursor write. +The [issue-log materializer](../../core/issuelog/README.md) also orders summary updates by source request, with closing after opening within the same request. A paused projection writer cannot overwrite a newer summary: its conditional write fails, and reloading the winner repeats the source-order check. Summary failures keep the cursor unfinished, just like history failures. + The post-read cursor write and processed-build updates, including unchanged statuses, are required for this ordering. A progress version alone is not a lock around issue writes. These operations use individual K-V reads, conditional writes, and immutable history creates; no cross-record transaction or stored future-change payload is needed. ## Refreshing and finalizing @@ -48,7 +50,7 @@ An error returns to normal consumer retry. The next delivery resumes the cursor 1. Conditionally enter `abandoning`, retaining the source, notification, and cursor. Normal deliveries hold for 30 seconds while this phase is present; they cannot start another pass or resume issue application. 2. If there is a cursor, read its issue and conditionally advance progress after that read. Advance the issue's version without changing its status or transition metadata. This invalidates a paused worker's earlier row guard. If the row has not been created yet, finish that one opening instead: the normal worker can only have saved such a cursor after finding a reported build issue. Creating it prevents the delayed create from succeeding. Closed issue rows are never deleted. -3. Repair the cursor's opening or closing history from the retained row. Record an `issue_tracking_abandoned` Request-history event keyed by the failed notification, with its build ID. +3. Repair the cursor's opening or closing history and summary from the retained row. Record an `issue_tracking_abandoned` Request-history event keyed by the failed notification, with its build ID. 4. Conditionally enter `abandoned`. Keep the source-order boundary and notification identity, so the same delivery and older builds cannot restart the abandoned work. Accepted opens and closes remain in place. Recovery does not scan either issue range, close the remaining issues, change Request or Build outcomes, or release build slots. An abandoned ordinary refresh still permits a new refresh or terminal notification for that Request; an abandoned terminal pass permits only newer Requests. A later successful build can reconcile issues left unchanged by the failed pass. diff --git a/stovepipe/controller/issuetrack/dlq_test.go b/stovepipe/controller/issuetrack/dlq_test.go index c71dc4e37..78705e5a4 100644 --- a/stovepipe/controller/issuetrack/dlq_test.go +++ b/stovepipe/controller/issuetrack/dlq_test.go @@ -193,7 +193,7 @@ func TestRecoveryDoesNotReopenAnExistingClosedCursor(t *testing.T) { } func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { - for _, step := range []string{"build/get", "request/get", "progress/get", "progress/update", "issue/get", "issue/update", "history/create", "recovery/history", "finish abandonment"} { + for _, step := range []string{"build/get", "request/get", "progress/get", "progress/update", "issue/get", "issue/update", "history/create", "history/get", "summary/get", "recovery/history", "finish abandonment"} { t.Run(step, func(t *testing.T) { f := newRecoveryFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) @@ -221,7 +221,7 @@ func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { } func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) { - for _, step := range []string{"issue/create", "history/create", "recovery/history"} { + for _, step := range []string{"issue/create", "history/create", "summary/create", "recovery/history"} { t.Run(step, func(t *testing.T) { f := newRecoveryFixture(t) f.source("build/100", "100", entity.BuildStatusRunning) @@ -236,6 +236,10 @@ 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}] + summary := f.summaries[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusOpen, summary.Status) + require.Equal(t, issue.FirstSeenBuildID, summary.FirstSeenBuildID) + require.Equal(t, issue.ChangedAtMs, summary.ChangedAtMs) require.Equal(t, "build/100", issue.FirstSeenBuildID) require.Equal(t, issue.ChangedAtMs, log.TimestampMs) if step == "issue/create" { diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index 96b87eb40..4ab7d03db 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -49,6 +49,7 @@ type fixture struct { factory *storagemock.MockFactory store storage.Storage states map[issueKey]entity.IssueState + summaries map[issueKey]entity.IssueSummary logs map[logKey]entity.IssueLog progress entity.IssueTrackingState hasProgress bool @@ -70,15 +71,18 @@ func newFixture(t *testing.T) *fixture { f.builds = map[string]entity.Build{} f.requests = map[string]entity.Request{} f.buildIssues = map[string]map[issueKey]entity.BuildIssue{} + f.summaries = map[issueKey]entity.IssueSummary{} f.source("older", "1", entity.BuildStatusSucceeded) store := storagemock.NewMockStorage(f.mocks) f.store = store states := storagemock.NewMockIssueStateStore(f.mocks) history := storagemock.NewMockIssueLogStore(f.mocks) + summaries := storagemock.NewMockIssueSummaryStore(f.mocks) tracking := storagemock.NewMockIssueTrackingStore(f.mocks) f.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(store, nil).AnyTimes() store.EXPECT().GetIssueStateStore().Return(states).AnyTimes() store.EXPECT().GetIssueLogStore().Return(history).AnyTimes() + store.EXPECT().GetIssueSummaryStore().Return(summaries).AnyTimes() store.EXPECT().GetIssueTrackingStore().Return(tracking).AnyTimes() states.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit string) (entity.IssueState, error) { @@ -129,6 +133,53 @@ 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) { + if err := f.fail("history/get"); err != nil { + return entity.IssueLog{}, err + } + key := logKey{issueKey{task, unit}, build, status} + if log, ok := f.logs[key]; ok { + return log, nil + } + if log, ok := f.initialLogs[key]; ok { + return log, nil + } + return entity.IssueLog{}, storage.ErrNotFound + }).AnyTimes() + summaries.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit string) (entity.IssueSummary, error) { + if err := f.fail("summary/get"); err != nil { + return entity.IssueSummary{}, err + } + summary, ok := f.summaries[issueKey{task, unit}] + if !ok { + return entity.IssueSummary{}, storage.ErrNotFound + } + return summary, nil + }).AnyTimes() + summaries.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.IssueSummary) error { + if err := f.fail("summary/create"); err != nil { + return err + } + key := issueKey{summary.TaskID, summary.UnitID} + if _, ok := f.summaries[key]; ok { + return storage.ErrAlreadyExists + } + f.summaries[key] = summary + return nil + }).AnyTimes() + summaries.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.IssueSummary, oldVersion, newVersion int32) error { + if err := f.fail("summary/update"); err != nil { + return err + } + key := issueKey{summary.TaskID, summary.UnitID} + current, ok := f.summaries[key] + if !ok || current.Version != oldVersion { + return storage.ErrVersionMismatch + } + summary.Version = newVersion + f.summaries[key] = summary + return nil + }).AnyTimes() tracking.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.IssueTrackingState, error) { if err := f.fail("progress/get"); err != nil { return entity.IssueTrackingState{}, err @@ -305,4 +356,6 @@ 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.summaries[issueKey{task, unit}] = entity.IssueSummary{TaskID: task, UnitID: unit, Status: status, + FirstSeenBuildID: "older", RequestID: "1", BuildID: "older", ChangedAtMs: 500, Version: 1} } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 44d14748e..f6beedf3b 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/stovepipe/core/fixrequest" + "github.com/uber/submitqueue/stovepipe/core/issuelog" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -45,6 +46,7 @@ type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory + materializer issuelog.Materializer registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string @@ -56,6 +58,7 @@ func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage. logger: logger.Named("issuetrack_controller"), metricsScope: scope.SubScope("issuetrack_controller"), stores: stores, + materializer: issuelog.NewMaterializer(scope), registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, @@ -389,7 +392,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 { 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) { + if err := c.materializer.PersistLog(ctx, store, log, issue.FirstSeenBuildID); err != nil { return fmt.Errorf("record issue transition: %w", err) } return nil diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index 42f6b9f46..7c2af9ed6 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -85,6 +85,12 @@ func TestProcessIssueOpensAndCloses(t *testing.T) { require.Equal(t, "unit", next.CursorUnitID) require.Equal(t, int32(2), next.Version) issue := f.states[issueKey{"task", "unit"}] + summary := f.summaries[issueKey{"task", "unit"}] + require.Equal(t, issue.Status, summary.Status) + require.Equal(t, issue.FirstSeenBuildID, summary.FirstSeenBuildID) + require.Equal(t, "100", summary.RequestID) + require.Equal(t, issue.LastChangedBuildID, summary.BuildID) + require.Equal(t, issue.ChangedAtMs, summary.ChangedAtMs) require.Equal(t, "build/100", issue.LastProcessedBuildID) require.Equal(t, "build/100", issue.LastChangedBuildID) require.Equal(t, int64(1000), issue.ChangedAtMs) @@ -154,6 +160,32 @@ func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { require.Equal(t, progress, f.progress) } +func TestSummaryRetryRepairsTheRetainedTransition(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + f.retain(progress) + f.failures["summary/create"] = 1 + + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.ErrorIs(t, err, assert.AnError) + applied := f.states[issueKey{"task", "unit"}] + require.Len(t, f.logs, 1) + require.Empty(t, f.summaries) + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + f.controller.nowMs = func() int64 { return 2000 } + + _, 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.Len(t, f.logs, 1) + summary := f.summaries[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusOpen, summary.Status) + require.Equal(t, applied.ChangedAtMs, summary.ChangedAtMs) + require.Equal(t, applied.FirstSeenBuildID, summary.FirstSeenBuildID) +} + func TestOlderSourceCannotChangeTheIssue(t *testing.T) { f := newFixture(t) f.source("newer", "101", entity.BuildStatusSucceeded) diff --git a/stovepipe/core/issuelog/BUILD.bazel b/stovepipe/core/issuelog/BUILD.bazel new file mode 100644 index 000000000..b891157b0 --- /dev/null +++ b/stovepipe/core/issuelog/BUILD.bazel @@ -0,0 +1,30 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["materializer.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/issuelog", + visibility = ["//visibility:public"], + deps = [ + "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["materializer_test.go"], + embed = [":go_default_library"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/core/issuelog/README.md b/stovepipe/core/issuelog/README.md new file mode 100644 index 000000000..d2f609ca0 --- /dev/null +++ b/stovepipe/core/issuelog/README.md @@ -0,0 +1,9 @@ +# Issue history and summaries + +The materializer retains an issue transition before updating its summary. Issue tracking calls it inline during normal processing and recovery; there is no separate consumer. A failed summary write returns an error, so the current issue remains at the cursor for retry. Replaying an existing transition reads its original timestamp and still repairs the summary. + +The source issue state supplies the opening build ID because closing log entries identify the closing build, not the opening. That value comes from the same retained issue snapshot as the transition; the materializer does not look up a potentially newer issue state to reconstruct an older opening. + +Summary ordering uses the source build's request ID in ingest order. Within one request, closing supersedes opening. Build IDs and timestamps do not establish order. Older transitions remain in history but cannot replace newer summaries. Conditional-create and CAS conflicts reload the winning summary and repeat the ordering check. Closed summaries remain present across reopening, with increasing projection versions. + +This materializer does not backfill existing history, maintain an open-only mapping, or expose a list API. Source builds must remain readable while their transitions can be replayed. diff --git a/stovepipe/core/issuelog/materializer.go b/stovepipe/core/issuelog/materializer.go new file mode 100644 index 000000000..a84cce234 --- /dev/null +++ b/stovepipe/core/issuelog/materializer.go @@ -0,0 +1,156 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package issuelog retains issue transitions and projects their latest status. +package issuelog + +import ( + "context" + "errors" + "fmt" + "math" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/loader" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +// Materializer retains immutable issue history and updates its summary. +type Materializer interface { + // PersistLog repairs the summary even when the transition already exists. + // firstSeenBuildID identifies the opening retained by the source issue state. + PersistLog(ctx context.Context, stores storage.Storage, log entity.IssueLog, firstSeenBuildID string) error +} + +type materializer struct{ scope tally.Scope } + +// NewMaterializer creates an issue-log materializer. +func NewMaterializer(scope tally.Scope) Materializer { + return &materializer{scope: scope.SubScope("issue_log_materializer")} +} + +func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.IssueLog, firstSeenBuildID string) (retErr error) { + op := metrics.Begin(m.scope, "persist_log", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if err := ctx.Err(); err != nil { + return err + } + if err := log.Validate(); err != nil { + return fmt.Errorf("invalid issue transition: %w", err) + } + if firstSeenBuildID == "" || (log.Status == entity.IssueStatusOpen && firstSeenBuildID != log.BuildID) { + return fmt.Errorf("issue transition has an invalid opening build") + } + + retained, err := retainIssueLog(ctx, stores.GetIssueLogStore(), log) + if err != nil { + return err + } + build, err := loader.ByID(ctx, retained.BuildID, stores.GetBuildStore().Get, "issue transition build") + if err != nil { + return err + } + if build.ID != retained.BuildID { + return fmt.Errorf("issue transition build identity does not match retained history") + } + + summary := entity.IssueSummary{ + TaskID: retained.TaskID, UnitID: retained.UnitID, Status: retained.Status, + FirstSeenBuildID: firstSeenBuildID, RequestID: build.RequestID, BuildID: retained.BuildID, + ChangedAtMs: retained.TimestampMs, Version: 1, + } + if err := summary.Validate(); err != nil { + return fmt.Errorf("invalid issue summary: %w", err) + } + return materializeIssueSummary(ctx, stores.GetIssueSummaryStore(), summary) +} + +func retainIssueLog(ctx context.Context, logs storage.IssueLogStore, log entity.IssueLog) (entity.IssueLog, error) { + if err := logs.Create(ctx, log); err != nil { + 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) + 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 { + return entity.IssueLog{}, fmt.Errorf("issue transition identity conflicts with retained history") + } + if err := retained.Validate(); err != nil { + return entity.IssueLog{}, fmt.Errorf("invalid retained issue transition: %w", err) + } + return retained, nil + } + return log, nil +} + +func materializeIssueSummary(ctx context.Context, summaries storage.IssueSummaryStore, incoming entity.IssueSummary) error { + for { + if err := ctx.Err(); err != nil { + return err + } + current, err := summaries.Get(ctx, incoming.TaskID, incoming.UnitID) + if errors.Is(err, storage.ErrNotFound) { + if err := summaries.Create(ctx, incoming); errors.Is(err, storage.ErrAlreadyExists) { + continue + } else if err != nil { + return fmt.Errorf("create issue summary: %w", err) + } + return nil + } + if err != nil { + return fmt.Errorf("read issue summary: %w", err) + } + if err := current.Validate(); err != nil { + return fmt.Errorf("invalid stored issue summary: %w", err) + } + if current.TaskID != incoming.TaskID || current.UnitID != incoming.UnitID { + return fmt.Errorf("stored issue summary identity does not match transition") + } + + order, err := entity.CompareRequestID(incoming.RequestID, current.RequestID) + if err != nil { + return err + } + // A request's closing transition wins over its opening, regardless of delivery order. + if order < 0 || (order == 0 && current.Status == entity.IssueStatusClosed && incoming.Status == entity.IssueStatusOpen) { + return nil + } + if order == 0 && current.Status == incoming.Status { + if current.BuildID != incoming.BuildID || current.FirstSeenBuildID != incoming.FirstSeenBuildID || current.ChangedAtMs != incoming.ChangedAtMs { + return fmt.Errorf("issue summary conflicts with retained transition") + } + return nil + } + if current.Version == math.MaxInt32 { + return fmt.Errorf("issue summary version exhausted") + } + + replacement := incoming + replacement.Version = current.Version + if err := summaries.Update(ctx, replacement, current.Version, current.Version+1); errors.Is(err, storage.ErrVersionMismatch) { + continue + } else if err != nil { + return fmt.Errorf("replace issue summary: %w", err) + } + return nil + } +} + +var _ Materializer = (*materializer)(nil) diff --git a/stovepipe/core/issuelog/materializer_test.go b/stovepipe/core/issuelog/materializer_test.go new file mode 100644 index 000000000..48e425b44 --- /dev/null +++ b/stovepipe/core/issuelog/materializer_test.go @@ -0,0 +1,308 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package issuelog + +import ( + "context" + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +type logKey struct { + task, unit, build string + status entity.IssueStatus +} + +type materializerFixture struct { + m Materializer + stores *storagemock.MockStorage + logs map[logKey]entity.IssueLog + summary entity.IssueSummary + failures map[string]error + beforeCreate, beforeUpdate func() +} + +func newMaterializerFixture(t *testing.T) *materializerFixture { + t.Helper() + ctrl := gomock.NewController(t) + f := &materializerFixture{m: NewMaterializer(tally.NoopScope), stores: storagemock.NewMockStorage(ctrl), + logs: make(map[logKey]entity.IssueLog), failures: make(map[string]error)} + logs := storagemock.NewMockIssueLogStore(ctrl) + summaries := storagemock.NewMockIssueSummaryStore(ctrl) + builds := storagemock.NewMockBuildStore(ctrl) + f.stores.EXPECT().GetIssueLogStore().Return(logs).AnyTimes() + f.stores.EXPECT().GetIssueSummaryStore().Return(summaries).AnyTimes() + f.stores.EXPECT().GetBuildStore().Return(builds).AnyTimes() + logs.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.IssueLog) error { + if err := f.failures["log/create"]; err != nil { + return err + } + key := logKey{log.TaskID, log.UnitID, log.BuildID, log.Status} + 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) { + if err := f.failures["log/get"]; err != nil { + return entity.IssueLog{}, err + } + log, ok := f.logs[logKey{task, unit, build, status}] + if !ok { + return entity.IssueLog{}, storage.ErrNotFound + } + return log, nil + }).AnyTimes() + builds.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, buildID string) (entity.Build, error) { + if err := f.failures["build/get"]; err != nil { + return entity.Build{}, err + } + requestID, ok := map[string]string{"build/z": "9", "build/a": "10", "build/m": "11"}[buildID] + if !ok { + return entity.Build{}, storage.ErrNotFound + } + return entity.Build{ID: buildID, RequestID: requestID, Version: 1}, nil + }).AnyTimes() + summaries.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, string, string) (entity.IssueSummary, error) { + if err := f.failures["summary/get"]; err != nil { + return entity.IssueSummary{}, err + } + if f.summary.Version == 0 { + return entity.IssueSummary{}, storage.ErrNotFound + } + return f.summary, nil + }).AnyTimes() + summaries.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.IssueSummary) error { + if f.beforeCreate != nil { + hook := f.beforeCreate + f.beforeCreate = nil + hook() + } + if err := f.failures["summary/create"]; err != nil { + return err + } + 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) + f.summary = summary + return nil + }).AnyTimes() + summaries.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.IssueSummary, oldVersion, newVersion int32) error { + if f.beforeUpdate != nil { + hook := f.beforeUpdate + f.beforeUpdate = nil + hook() + } + if err := f.failures["summary/update"]; err != nil { + return err + } + if f.summary.Version != oldVersion { + return storage.ErrVersionMismatch + } + require.Equal(t, oldVersion, summary.Version) + summary.Version = newVersion + f.summary = summary + return nil + }).AnyTimes() + return f +} + +func testIssueLog(build string, status entity.IssueStatus) entity.IssueLog { + return entity.IssueLog{TaskID: "validation", UnitID: "unit", BuildID: build, Status: status, TimestampMs: 1000} +} + +func TestMaterializerOpenCloseReopenAndReplay(t *testing.T) { + f := newMaterializerFixture(t) + opened := testIssueLog("build/z", entity.IssueStatusOpen) + closed := testIssueLog("build/a", entity.IssueStatusClosed) + reopened := testIssueLog("build/m", entity.IssueStatusOpen) + for _, step := range []struct { + log entity.IssueLog + opening string + status entity.IssueStatus + version int32 + }{ + {opened, opened.BuildID, entity.IssueStatusOpen, 1}, + {closed, opened.BuildID, entity.IssueStatusClosed, 2}, + {opened, opened.BuildID, entity.IssueStatusClosed, 2}, + {reopened, reopened.BuildID, entity.IssueStatusOpen, 3}, + {closed, opened.BuildID, entity.IssueStatusOpen, 3}, + {reopened, reopened.BuildID, entity.IssueStatusOpen, 3}, + } { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, step.log, step.opening)) + require.Equal(t, step.status, f.summary.Status) + require.Equal(t, step.version, f.summary.Version) + } + require.Equal(t, "11", f.summary.RequestID) + require.Equal(t, reopened.BuildID, f.summary.FirstSeenBuildID) + require.Len(t, f.logs, 3) +} + +func TestMaterializerCloseWinsWithinRequest(t *testing.T) { + for _, closeFirst := range []bool{false, true} { + name := "opening first" + if closeFirst { + name = "closing first" + } + t.Run(name, func(t *testing.T) { + f := newMaterializerFixture(t) + opened := testIssueLog("build/z", entity.IssueStatusOpen) + closed := testIssueLog("build/z", entity.IssueStatusClosed) + logs := []entity.IssueLog{opened, closed} + if closeFirst { + logs = []entity.IssueLog{closed, opened} + } + for _, log := range logs { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log, opened.BuildID)) + } + require.Equal(t, entity.IssueStatusClosed, f.summary.Status) + require.Equal(t, "9", f.summary.RequestID) + require.Equal(t, opened.BuildID, f.summary.FirstSeenBuildID) + }) + } +} + +func TestMaterializerRepairsProjectionAfterFailure(t *testing.T) { + for _, step := range []string{"summary/create", "summary/update"} { + t.Run(step, func(t *testing.T) { + f := newMaterializerFixture(t) + log := testIssueLog("build/z", entity.IssueStatusOpen) + opening := log.BuildID + if step == "summary/update" { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log, opening)) + log = testIssueLog("build/a", entity.IssueStatusClosed) + } + f.failures[step] = assert.AnError + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, log, opening), assert.AnError) + delete(f.failures, step) + retry := log + retry.TimestampMs = 9999 + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, retry, opening)) + require.Equal(t, log.TimestampMs, f.summary.ChangedAtMs) + require.Equal(t, log.BuildID, f.summary.BuildID) + }) + } +} + +func TestMaterializerReloadsConcurrentWinner(t *testing.T) { + for _, step := range []string{"create", "update"} { + t.Run(step, func(t *testing.T) { + f := newMaterializerFixture(t) + opened := testIssueLog("build/z", entity.IssueStatusOpen) + incoming := testIssueLog("build/a", entity.IssueStatusClosed) + winner := testIssueLog("build/m", entity.IssueStatusOpen) + if step == "create" { + f.beforeCreate = func() { require.NoError(t, f.m.PersistLog(context.Background(), f.stores, winner, winner.BuildID)) } + } else { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, opened, opened.BuildID)) + f.beforeUpdate = func() { require.NoError(t, f.m.PersistLog(context.Background(), f.stores, winner, winner.BuildID)) } + } + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, incoming, opened.BuildID)) + require.Equal(t, winner.BuildID, f.summary.BuildID) + require.Equal(t, winner.BuildID, f.summary.FirstSeenBuildID) + require.Equal(t, entity.IssueStatusOpen, f.summary.Status) + }) + } +} + +func TestMaterializerReturnsStorageErrors(t *testing.T) { + for _, step := range []string{"log/create", "log/get", "build/get", "summary/get", "summary/create", "summary/update"} { + t.Run(step, func(t *testing.T) { + f := newMaterializerFixture(t) + log := testIssueLog("build/z", entity.IssueStatusOpen) + opening := log.BuildID + if step == "log/get" || step == "summary/update" { + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log, opening)) + } + if step == "summary/update" { + log = testIssueLog("build/a", entity.IssueStatusClosed) + } + f.failures[step] = assert.AnError + require.ErrorIs(t, f.m.PersistLog(context.Background(), f.stores, log, opening), assert.AnError) + }) + } +} + +func TestMaterializerRejectsInvalidTransitions(t *testing.T) { + for _, scenario := range []string{"invalid status", "missing opening", "wrong opening", "invalid timestamp", "missing build"} { + t.Run(scenario, func(t *testing.T) { + f := newMaterializerFixture(t) + log := testIssueLog("build/z", entity.IssueStatusOpen) + opening := log.BuildID + switch scenario { + case "invalid status": + log.Status = entity.IssueStatusUnknown + case "missing opening": + opening = "" + case "wrong opening": + opening = "build/a" + case "invalid timestamp": + log.TimestampMs = 0 + case "missing build": + log.BuildID, opening = "unknown", "unknown" + } + require.Error(t, f.m.PersistLog(context.Background(), f.stores, log, opening)) + require.Empty(t, f.summary) + }) + } +} + +func TestMaterializerRejectsConflictingSummary(t *testing.T) { + for _, scenario := range []string{"build", "opening", "timestamp", "identity", "invalid summary", "exhausted version"} { + t.Run(scenario, func(t *testing.T) { + f := newMaterializerFixture(t) + log := testIssueLog("build/z", entity.IssueStatusOpen) + require.NoError(t, f.m.PersistLog(context.Background(), f.stores, log, log.BuildID)) + switch scenario { + case "build": + f.summary.BuildID = "different" + case "opening": + f.summary.FirstSeenBuildID = "different" + case "timestamp": + f.summary.ChangedAtMs++ + case "identity": + f.summary.TaskID = "different" + case "invalid summary": + f.summary.Status = entity.IssueStatusUnknown + case "exhausted version": + f.summary.Version = math.MaxInt32 + log = testIssueLog("build/m", entity.IssueStatusOpen) + } + current := f.summary + require.Error(t, f.m.PersistLog(context.Background(), f.stores, log, log.BuildID)) + require.Equal(t, current, f.summary) + }) + } +} + +func TestMaterializerHonorsCanceledContext(t *testing.T) { + f := newMaterializerFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + log := testIssueLog("build/z", entity.IssueStatusOpen) + require.ErrorIs(t, f.m.PersistLog(ctx, f.stores, log, log.BuildID), context.Canceled) + require.Empty(t, f.logs) + require.Empty(t, f.summary) +} diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 37bc68699..646a7adea 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -56,7 +56,7 @@ The MySQL backend uses the new `issue_state` table. Create it before deploying t `RequestID` and `BuildID` identify the latest represented transition, while `FirstSeenBuildID` identifies its current or most recent opening. Source order follows request ingest order, not opaque build IDs or timestamps. Within the same request, a closing transition supersedes an opening. `Version` is only the summary's CAS guard, not the source order. The materializer must compare incoming transitions before replacing a summary; the store does not enforce lifecycle ordering. -The MySQL primary key is `(queue, task_id, unit_id)` with byte-exact identity and the same identifier limits as the issue-state store. Source request IDs follow the existing canonical positive decimal ID contract. The new `issue_summary` table must exist before deploying readers or writers. This storage layer does not yet populate summaries or expose a public open-issues list. +The MySQL primary key is `(queue, task_id, unit_id)` with byte-exact identity and the same identifier limits as the issue-state store. Source request IDs follow the existing canonical positive decimal ID contract. The new `issue_summary` table must exist before deploying readers or writers. The [issue-log materializer](../../core/issuelog/README.md) updates summaries during normal issue tracking and recovery. No public open-issues list is exposed yet. ## Issue lifecycle history @@ -66,7 +66,7 @@ The build ID identifies the build where the issue was first seen or confirmed cl 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. The issue-tracking controller owns opening/closing decisions and retry-safe history writes. 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 invokes the issue-log materializer to retain history and update summaries. History enumeration is outside this storage contract. ## Issue reconciliation progress