feat(stovepipe): Maintain open issue lookups

stovepipe/controller/issuetrack/README.md

@@ -8,15 +8,15 @@ 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. 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.
+5. Retain opening or closing history from the issue row's changing build and timestamp, then materialize its summary and open-issue lookup. On retry, the inclusive cursor revisits that row and repairs all three 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 [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. Lookup entries are then maintained from the latest summary, not the incoming transition. Summary and lookup 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.
 
@@ -50,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 and summary 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, summary, and open-issue lookup 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.

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", "history/get", "summary/get", "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", "open-issue/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", "summary/create", "recovery/history"} {
+	for _, step := range []string{"issue/create", "history/create", "summary/create", "open-issue/create", "recovery/history"} {
 		t.Run(step, func(t *testing.T) {
 			f := newRecoveryFixture(t)
 			f.source("build/100", "100", entity.BuildStatusRunning)
@@ -237,6 +237,7 @@ func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) {
 			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, issue.FirstSeenBuildID, f.openIssues[issueKey{"task", "unit"}].FirstSeenBuildID)
 			require.Equal(t, entity.IssueStatusOpen, summary.Status)
 			require.Equal(t, issue.FirstSeenBuildID, summary.FirstSeenBuildID)
 			require.Equal(t, issue.ChangedAtMs, summary.ChangedAtMs)
@@ -251,6 +252,29 @@ func TestRecoveryRetainsAnOpeningAcrossRetries(t *testing.T) {
 	}
 }
 
+func TestRecoveryRepairsClosedIssueLookupBeforeAbandonment(t *testing.T) {
+	f := newRecoveryFixture(t)
+	f.source("build/100", "100", entity.BuildStatusFailed)
+	f.seedIssue("task", "unit", entity.IssueStatusClosed)
+	key := issueKey{"task", "unit"}
+	f.openIssues[key] = entity.OpenIssue{TaskID: key.task, UnitID: key.unit, FirstSeenBuildID: "older"}
+	progress := currentCursor()
+	msg := f.message("build/100", true)
+	progress.Finalize, progress.NotificationID = true, msg.ID
+	f.retain(progress)
+	f.failures["open-issue/delete"] = 1
+
+	require.ErrorIs(t, f.recover(msg, false), assert.AnError)
+	require.Equal(t, entity.IssueTrackingPhaseAbandoning, f.progress.Phase)
+	require.Contains(t, f.openIssues, key)
+
+	require.NoError(t, f.recover(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase)
+	require.NotContains(t, f.openIssues, key)
+	require.Equal(t, entity.IssueStatusClosed, f.states[key].Status)
+	require.Equal(t, entity.IssueStatusClosed, f.summaries[key].Status)
+}
+
 func TestPrimaryWaitsForRecoveryThenAdmitsANewerBuild(t *testing.T) {
 	f := newRecoveryFixture(t)
 	f.source("build/100", "100", entity.BuildStatusFailed)

stovepipe/controller/issuetrack/fixture_test.go

@@ -50,6 +50,7 @@ type fixture struct {
 	store       storage.Storage
 	states      map[issueKey]entity.IssueState
 	summaries   map[issueKey]entity.IssueSummary
+	openIssues  map[issueKey]entity.OpenIssue
 	logs        map[logKey]entity.IssueLog
 	progress    entity.IssueTrackingState
 	hasProgress bool
@@ -72,17 +73,20 @@ func newFixture(t *testing.T) *fixture {
 	f.requests = map[string]entity.Request{}
 	f.buildIssues = map[string]map[issueKey]entity.BuildIssue{}
 	f.summaries = map[issueKey]entity.IssueSummary{}
+	f.openIssues = map[issueKey]entity.OpenIssue{}
 	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)
+	openIssues := storagemock.NewMockOpenIssueStore(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().GetOpenIssueStore().Return(openIssues).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) {
@@ -180,6 +184,49 @@ func newFixture(t *testing.T) *fixture {
 		f.summaries[key] = summary
 		return nil
 	}).AnyTimes()
+	openIssues.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit string) (entity.OpenIssue, error) {
+		if err := f.fail("open-issue/get"); err != nil {
+			return entity.OpenIssue{}, err
+		}
+		issue, ok := f.openIssues[issueKey{task, unit}]
+		if !ok {
+			return entity.OpenIssue{}, storage.ErrNotFound
+		}
+		return issue, nil
+	}).AnyTimes()
+	openIssues.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue) error {
+		if err := f.fail("open-issue/create"); err != nil {
+			return err
+		}
+		key := issueKey{issue.TaskID, issue.UnitID}
+		if _, ok := f.openIssues[key]; ok {
+			return storage.ErrAlreadyExists
+		}
+		f.openIssues[key] = issue
+		return nil
+	}).AnyTimes()
+	openIssues.EXPECT().Replace(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue, expected string) (bool, error) {
+		if err := f.fail("open-issue/replace"); err != nil {
+			return false, err
+		}
+		key := issueKey{issue.TaskID, issue.UnitID}
+		if current, ok := f.openIssues[key]; !ok || current.FirstSeenBuildID != expected {
+			return false, nil
+		}
+		f.openIssues[key] = issue
+		return true, nil
+	}).AnyTimes()
+	openIssues.EXPECT().Delete(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.OpenIssue) (bool, error) {
+		if err := f.fail("open-issue/delete"); err != nil {
+			return false, err
+		}
+		key := issueKey{issue.TaskID, issue.UnitID}
+		if current, ok := f.openIssues[key]; !ok || current.FirstSeenBuildID != issue.FirstSeenBuildID {
+			return false, nil
+		}
+		delete(f.openIssues, key)
+		return true, 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
@@ -358,4 +405,7 @@ func (f *fixture) seedIssue(task, unit string, status entity.IssueStatus) {
 	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}
+	if status == entity.IssueStatusOpen {
+		f.openIssues[issueKey{task, unit}] = entity.OpenIssue{TaskID: task, UnitID: unit, FirstSeenBuildID: "older"}
+	}
 }

stovepipe/controller/issuetrack/issuetrack_test.go

@@ -95,11 +95,13 @@ func TestProcessIssueOpensAndCloses(t *testing.T) {
 			require.Equal(t, "build/100", issue.LastChangedBuildID)
 			require.Equal(t, int64(1000), issue.ChangedAtMs)
 			if reported {
+				require.Equal(t, issue.FirstSeenBuildID, f.openIssues[issueKey{"task", "unit"}].FirstSeenBuildID)
 				require.Equal(t, entity.IssueStatusOpen, issue.Status)
 				require.Equal(t, "build/100", issue.FirstSeenBuildID)
 				require.Equal(t, int32(1), issue.Version)
 				require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusOpen})
 			} else {
+				require.NotContains(t, f.openIssues, issueKey{"task", "unit"})
 				require.Equal(t, entity.IssueStatusClosed, issue.Status)
 				require.Equal(t, "older", issue.FirstSeenBuildID)
 				require.Equal(t, int32(8), issue.Version)
@@ -202,3 +204,27 @@ func TestOlderSourceCannotChangeTheIssue(t *testing.T) {
 	require.Equal(t, progress, f.progress)
 	require.Empty(t, f.logs)
 }
+
+func TestOpenIssueFailureKeepsTheCursorForRetry(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["open-issue/create"] = 1
+
+	_, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true)
+	require.ErrorIs(t, err, assert.AnError)
+	accepted := f.states[issueKey{"task", "unit"}]
+	require.Contains(t, f.summaries, issueKey{"task", "unit"})
+	require.Empty(t, f.openIssues)
+	require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+	require.Equal(t, "task", f.progress.CursorTaskID)
+	require.Equal(t, "unit", f.progress.CursorUnitID)
+
+	_, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false)
+	require.NoError(t, err)
+	require.Equal(t, accepted, f.states[issueKey{"task", "unit"}])
+	require.Equal(t, accepted.FirstSeenBuildID, f.openIssues[issueKey{"task", "unit"}].FirstSeenBuildID)
+	require.Len(t, f.logs, 1)
+}

stovepipe/core/issuelog/README.md

@@ -1,9 +1,17 @@
 # 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 materializer retains an issue transition before updating its summary and open-issue lookup. Issue tracking calls it inline during normal processing and recovery; there is no separate consumer. A failed projection write returns an error, so the current issue remains at the cursor for retry. Replaying an existing transition reads its original timestamp and repairs both projections.
 
 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.
+## Open-issue lookup
+
+After summary materialization, the lookup is repaired from the latest summary, including when an incoming transition was a duplicate or older than the summary. An open summary requires a reference to its opening; a closed summary requires no open-issue row.
+
+Each iteration reads the open-issue row before the summary. Otherwise, a worker could read an old closed summary, then read a newer opening and use that opening's ID to delete it. Reading the open-issue row first means a concurrent reopening changes the expected ID and prevents the stale delete or replacement. Creates require absence; replacement and deletion compare the opening that was read. Both successful writes and guard mismatches repeat the reads. A delayed create can briefly leave a stale reference after a close; the next iteration removes it. Storage implementations must satisfy the causal read-after-write contract.
+
+The two projections are not updated atomically. A future list reader must still verify that each referenced summary is open and its opening matches. No status filtering, cross-record transaction, or queue-wide scan is required to maintain one open-issue row.
+
+This materializer does not backfill existing history or expose a list API. Source builds must remain readable while their transitions can be replayed.

stovepipe/core/issuelog/materializer.go

@@ -28,9 +28,9 @@ import (
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
-// Materializer retains immutable issue history and updates its summary.
+// Materializer retains issue history and updates its summary and open-issue lookup.
 type Materializer interface {
-	// PersistLog repairs the summary even when the transition already exists.
+	// PersistLog repairs the summary and lookup 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
 }
@@ -76,7 +76,10 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
 	if err := summary.Validate(); err != nil {
 		return fmt.Errorf("invalid issue summary: %w", err)
 	}
-	return materializeIssueSummary(ctx, stores.GetIssueSummaryStore(), summary)
+	if err := materializeIssueSummary(ctx, stores.GetIssueSummaryStore(), summary); err != nil {
+		return err
+	}
+	return materializeOpenIssue(ctx, stores, summary.TaskID, summary.UnitID)
 }
 
 func retainIssueLog(ctx context.Context, logs storage.IssueLogStore, log entity.IssueLog) (entity.IssueLog, error) {
@@ -153,4 +156,52 @@ func materializeIssueSummary(ctx context.Context, summaries storage.IssueSummary
 	}
 }
 
+func materializeOpenIssue(ctx context.Context, stores storage.Storage, taskID, unitID string) error {
+	openIssues := stores.GetOpenIssueStore()
+	summaries := stores.GetIssueSummaryStore()
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		// Read the open-issue row before the summary so a stale closed
+		// summary cannot authorize deleting a newer opening.
+		current, err := openIssues.Get(ctx, taskID, unitID)
+		missing := errors.Is(err, storage.ErrNotFound)
+		if err != nil && !missing {
+			return fmt.Errorf("read open issue: %w", err)
+		}
+		summary, err := summaries.Get(ctx, taskID, unitID)
+		if err != nil {
+			return fmt.Errorf("read issue summary for open issue: %w", err)
+		}
+
+		switch summary.Status {
+		case entity.IssueStatusOpen:
+			if !missing && current.FirstSeenBuildID == summary.FirstSeenBuildID {
+				return nil
+			}
+			desired := entity.OpenIssue{TaskID: taskID, UnitID: unitID, FirstSeenBuildID: summary.FirstSeenBuildID}
+			if missing {
+				err = openIssues.Create(ctx, desired)
+				if errors.Is(err, storage.ErrAlreadyExists) {
+					continue
+				}
+			} else {
+				_, err = openIssues.Replace(ctx, desired, current.FirstSeenBuildID)
+			}
+		case entity.IssueStatusClosed:
+			if missing {
+				return nil
+			}
+			_, err = openIssues.Delete(ctx, current)
+		default:
+			return fmt.Errorf("invalid issue summary status %q", summary.Status)
+		}
+		if err != nil {
+			return fmt.Errorf("materialize open issue: %w", err)
+		}
+	}
+}
+
 var _ Materializer = (*materializer)(nil)