prathshenoy · Oct 30, 2026 · Oct 29, 2026
@@ -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",

@@ -8,14 +8,16 @@ 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.
 
 ## Ordering requirements
 
 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.

@@ -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" {

@@ -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}
 }
@@ -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

@@ -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)

@@ -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",
+    ],
+)
@@ -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.