From 120be5c1219d4a7d8244ff5d2f40e6ca2fdc8997 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 3 Nov 2026 03:46:49 +0000 Subject: [PATCH 1/3] fix(stovepipe): scope task-wide closure suppression to its task --- doc/rfc/stovepipe/steps/buildanalysis.md | 2 +- stovepipe/controller/buildanalysis/README.md | 2 +- stovepipe/controller/issuetrack/README.md | 8 +-- .../controller/issuetrack/delivery_test.go | 69 +++++++++---------- .../controller/issuetrack/fix_handoff_test.go | 6 +- stovepipe/controller/issuetrack/issuetrack.go | 11 ++- .../controller/issuetrack/issuetrack_test.go | 4 +- stovepipe/entity/issue_tracking_state.go | 3 - stovepipe/entity/issue_tracking_state_test.go | 5 -- stovepipe/extension/storage/README.md | 2 +- 10 files changed, 55 insertions(+), 57 deletions(-) diff --git a/doc/rfc/stovepipe/steps/buildanalysis.md b/doc/rfc/stovepipe/steps/buildanalysis.md index 474dd84d4..77a96f853 100644 --- a/doc/rfc/stovepipe/steps/buildanalysis.md +++ b/doc/rfc/stovepipe/steps/buildanalysis.md @@ -12,7 +12,7 @@ Publication errors are returned through the producing stage's retry/DLQ policy. ## Build issues and polling -The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues and publishes queue-partitioned issue-tracking notifications after persistence; it does not maintain open issues itself. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls, and analysis persists that set before publishing final reconciliation. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules. +The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues and publishes queue-partitioned issue-tracking notifications after persistence; it does not maintain open issues itself. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls, and analysis persists that set before publishing final reconciliation. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves missing target issues of the same task while allowing unrelated issues to close. See the controller documentation for handoff and deduplication rules. Status-polling DLQ recovery does not publish a build-completion notice. The analysis polling loop observes the ended Request through storage instead. diff --git a/stovepipe/controller/buildanalysis/README.md b/stovepipe/controller/buildanalysis/README.md index 65be8d984..5974e9f2e 100644 --- a/stovepipe/controller/buildanalysis/README.md +++ b/stovepipe/controller/buildanalysis/README.md @@ -8,7 +8,7 @@ After persistence, the controller publishes an issue-tracking notification parti ## Issue-tracking handoff -Refresh notifications let issue tracking open issues reported by the build without closing other open issues. Final notifications let it also close open issues absent from the build's accumulated issues, unless any task-wide failure was recorded or the build was canceled. A successful terminal detection must return the complete cumulative issue set for the build, including issues returned by earlier calls. Persisting that complete result before publication makes finalization independent of whether an earlier worker finished saving its results. An empty terminal result is valid only if the build has no reportable issues; inability to determine the complete set must return an error instead. See the [detector contract](../../extension/buildissue/README.md). +Refresh notifications let issue tracking open issues reported by the build without closing other open issues. Final notifications let it also close open issues absent from the build's accumulated issues, unless the same task has a task-wide failure or the build was canceled. A successful terminal detection must return the complete cumulative issue set for the build, including issues returned by earlier calls. Persisting that complete result before publication makes finalization independent of whether an earlier worker finished saving its results. An empty terminal result is valid only if the build has no reportable issues; inability to determine the complete set must return an error instead. See the [detector contract](../../extension/buildissue/README.md). All analysis deliveries for a build must share its build-ID partition and consumer group. The controller validates that partition key. Finalization requires a successful pass that began with a terminal Build and the Request's recorded winning build ID. If status becomes terminal before that outcome is recorded, analysis publishes a refresh and holds until the outcome is available. Known losing builds do not publish. A terminal Request without a recorded winning build permits a refresh but not closing issues based on absence. diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 701af184f..1d7606dc0 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -29,12 +29,12 @@ Once the entire build-issue range is exhausted, an opening pass with a nonempty Once the entire build-issue range is exhausted: - An ordinary refresh becomes `refreshed`. -- A terminal pass with a task-wide failure, or a canceled build, becomes `finalized` without closing anything. Reported closed issues may still have reopened during refreshing; unreported closed issues remain closed. -- Any other terminal pass becomes `finalizing`, resetting its cursor to the start of the issue-state range. +- A canceled build becomes `finalized` without closing anything. Reported closed issues may still have reopened during refreshing; unreported closed issues remain closed. +- Every other terminal pass becomes `finalizing`, resetting its cursor to the start of the issue-state range. -The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. +The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes unless the same task has a task-wide failure. Such a failure preserves missing target issues in their current status, including closed issues; it does not reopen unreported targets. Issues from other tasks can close, including task-wide issues. The processed build advances for preserved issues as well. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. -No closure occurs before all accumulated build issues have been processed. A task-wide failure in a later batch therefore protects existing issues across every task, not just keys encountered afterward. A later build makes its own decision; the restriction does not carry over. Both walks consume one ordered range at a time; no merge is required. +No closure occurs before all accumulated build issues have been processed. A task-wide failure in a later batch therefore protects missing targets of that task before any closing work begins. A later build makes its own decision; the restriction does not carry over. Both walks consume one ordered range at a time; no merge is required. ## Bounded passes diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index f3d66990a..1545efe4c 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -426,49 +426,50 @@ func TestHistoryRecordsTheActualChangingBuildNotTheLatestProcessedBuild(t *testi require.Len(t, f.logs, 1) } -func TestTaskWideFailurePreservesAllExistingIssues(t *testing.T) { +func TestTaskWideFailureSuppressesClosuresOnlyForItsTask(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) - f.seedIssue("old", "unit", entity.IssueStatusOpen) - f.seedIssue("old", "", entity.IssueStatusOpen) - f.seedIssue("old", "closed", entity.IssueStatusClosed) - f.seedIssue("reopen", "unit", entity.IssueStatusClosed) - f.reportIssue("build/100", "new", "unit") - f.reportIssue("build/100", "reopen", "unit") - f.reportIssue("build/100", "unrelated", "") - f.failures["issue/list"] = 1 + for _, task := range []string{"affected", "unrelated"} { + f.seedIssue(task, "", entity.IssueStatusOpen) + f.seedIssue(task, "unit", entity.IssueStatusOpen) + f.seedIssue(task, "closed", entity.IssueStatusClosed) + } + f.seedIssue("affected", "reopen", entity.IssueStatusClosed) + f.reportIssue("build/100", "affected", "") + f.reportIssue("build/100", "affected", "reopen") msg := f.message("build/100", true) + require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, false)) - require.True(t, f.progress.Finalize) - require.True(t, f.progress.HasTaskWideIssues) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - for _, key := range []issueKey{{"old", "unit"}, {"old", ""}, {"new", "unit"}, {"reopen", "unit"}, {"unrelated", ""}} { + for _, key := range []issueKey{{"affected", ""}, {"affected", "unit"}, {"affected", "reopen"}} { require.Equal(t, entity.IssueStatusOpen, f.states[key].Status) + require.Contains(t, f.openIssues, key) } - 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.IssueStatusOpen, key.status) + for _, key := range []issueKey{{"affected", "closed"}, {"unrelated", ""}, {"unrelated", "unit"}, {"unrelated", "closed"}} { + require.Equal(t, entity.IssueStatusClosed, f.states[key].Status) + require.NotContains(t, f.openIssues, key) } - require.Equal(t, 1, f.failures["issue/list"]) + require.Len(t, f.logs, 3) + require.Contains(t, f.logs, logKey{issueKey{"affected", "reopen"}, "build/100", entity.IssueStatusOpen}) + require.Contains(t, f.logs, logKey{issueKey{"unrelated", ""}, "build/100", entity.IssueStatusClosed}) + require.Contains(t, f.logs, logKey{issueKey{"unrelated", "unit"}, "build/100", entity.IssueStatusClosed}) - f.failures["issue/list"] = 0 f.source("build/101", "101", entity.BuildStatusSucceeded) next := f.message("build/101", true) require.NoError(t, f.deliver(next, true)) - require.False(t, f.progress.HasTaskWideIssues) - require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) require.NoError(t, f.deliver(next, false)) for _, issue := range f.states { require.Equal(t, entity.IssueStatusClosed, issue.Status) } + require.Empty(t, f.openIssues) } -func TestTaskWideFailureInLaterOpeningBatchPreventsAllClosures(t *testing.T) { +func TestTaskWideFailureInLaterBatchProtectsOnlyItsTask(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusFailed) f.seedIssue("old", "unit", entity.IssueStatusOpen) + f.seedIssue("zzz", "unit", entity.IssueStatusOpen) for i := 0; i < 105; i++ { f.reportIssue("build", fmt.Sprintf("task/%03d", i), "unit") } @@ -477,20 +478,15 @@ func TestTaskWideFailureInLaterOpeningBatchPreventsAllClosures(t *testing.T) { require.NoError(t, f.deliver(msg, true)) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) - require.True(t, f.progress.Finalize) - require.False(t, f.progress.HasTaskWideIssues) - require.Equal(t, "task/099", f.progress.CursorTaskID) - require.Len(t, f.logs, 100) require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status) - + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) + require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.True(t, f.progress.HasTaskWideIssues) - 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.IssueStatusOpen, key.status) - } + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "unit"}].Status) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"zzz", "unit"}].Status) + require.Len(t, f.logs, 107) } func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { @@ -522,6 +518,8 @@ func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { f.failures["build-issues/list"] = 1 require.ErrorIs(t, f.deliver(msg, false), assert.AnError) require.True(t, f.progress.HasTaskWideIssues) + require.NoError(t, f.deliver(msg, true)) + require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) require.True(t, f.progress.HasTaskWideIssues) @@ -574,13 +572,12 @@ func TestOpeningCursorCannotCloseAnUnreportedIssue(t *testing.T) { require.Empty(t, f.logs) } -func TestTaskWideHistoryRetryFinishesWithoutClosing(t *testing.T) { +func TestTaskWideHistoryRetryThenClosesUnrelatedIssues(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusFailed) f.seedIssue("old", "unit", entity.IssueStatusOpen) f.reportIssue("build", "task", "") f.failures["history/create"] = 1 - f.failures["issue/list"] = 1 msg := f.message("build", true) require.ErrorIs(t, f.deliver(msg, false), assert.AnError) @@ -588,11 +585,11 @@ func TestTaskWideHistoryRetryFinishesWithoutClosing(t *testing.T) { require.True(t, f.progress.Finalize) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) f.controller.nowMs = func() int64 { return 2000 } + require.NoError(t, f.deliver(msg, true)) 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, entity.IssueStatusClosed, f.states[issueKey{"old", "unit"}].Status) require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueStatusOpen}].TimestampMs) - require.Equal(t, 1, f.failures["issue/list"]) } func TestRetryResetsCursorBeforeClosing(t *testing.T) { diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go index 2239e0eae..dfc40ab04 100644 --- a/stovepipe/controller/issuetrack/fix_handoff_test.go +++ b/stovepipe/controller/issuetrack/fix_handoff_test.go @@ -190,8 +190,10 @@ func TestTaskWideFailureStillPublishesRepair(t *testing.T) { f.source("build", "10", entity.BuildStatusFailed) f.seedIssue("existing", "unit", entity.IssueStatusOpen) f.reportIssue("build", "setup", "") - require.NoError(t, f.deliver(f.message("build", true), false)) + msg := f.message("build", true) + require.NoError(t, f.deliver(msg, true)) + require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"existing", "unit"}].Status) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"existing", "unit"}].Status) require.Len(t, f.fixMessages, 1) } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index f6beedf3b..61840440b 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -268,7 +268,7 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, next.Phase = entity.IssueTrackingPhaseFinalized case !progress.Finalize: next.Phase = entity.IssueTrackingPhaseRefreshed - case progress.HasTaskWideIssues || build.Status == entity.BuildStatusCancelled: + case build.Status == entity.BuildStatusCancelled: next.Phase = entity.IssueTrackingPhaseFinalized default: next.Phase = entity.IssueTrackingPhaseFinalizing @@ -317,6 +317,15 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr if missing && !reported { return entity.IssueTrackingState{}, fmt.Errorf("closing cursor requires an existing issue") } + if !reported && key.unit != "" && progress.HasTaskWideIssues { + _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, "") + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.IssueTrackingState{}, fmt.Errorf("check task-wide issue: %w", err) + } + if err == nil { + desired = current.Status + } + } if !missing && current.LastProcessedBuildID != progress.BuildID { previous, err := loader.ByID(ctx, current.LastProcessedBuildID, store.GetBuildStore().Get, "last processed build") diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index b28e8ec32..b8f167fa7 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -25,7 +25,7 @@ import ( ) func TestBatchRejectsInvalidClosingPass(t *testing.T) { - for _, scenario := range []string{"missing terminal intent", "task-wide failure", "canceled build", "running build", "not terminal build"} { + for _, scenario := range []string{"missing terminal intent", "canceled build", "running build", "not terminal build"} { t.Run(scenario, func(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusSucceeded) @@ -38,8 +38,6 @@ func TestBatchRejectsInvalidClosingPass(t *testing.T) { switch scenario { case "missing terminal intent": progress.Finalize = false - case "task-wide failure": - progress.HasTaskWideIssues = true case "canceled build": f.source("build/100", "100", entity.BuildStatusCancelled) case "running build": diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 453e49e25..05c0f3a92 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/stovepipe/entity/issue_tracking_state.go @@ -90,9 +90,6 @@ func (s IssueTrackingState) Validate() error { if !s.Finalize { return fmt.Errorf("finalizing requires terminal intent") } - if s.HasTaskWideIssues { - return fmt.Errorf("task-wide failures prevent closing missing issues") - } case IssueTrackingPhaseFinalized: if !s.Finalize { return fmt.Errorf("finalized requires terminal intent") diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go index 245cc2512..c7f2d581b 100644 --- a/stovepipe/entity/issue_tracking_state_test.go +++ b/stovepipe/entity/issue_tracking_state_test.go @@ -39,11 +39,6 @@ func TestIssueTrackingState(t *testing.T) { {"closing without terminal intent", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseFinalizing }}, {"finalized without terminal intent", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseFinalized }}, {"terminal pass marked refreshed", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseRefreshed; s.Finalize = true }}, - {"closing with task-wide failure", func(s *IssueTrackingState) { - s.Phase = IssueTrackingPhaseFinalizing - s.Finalize = true - s.HasTaskWideIssues = true - }}, {"missing build", func(s *IssueTrackingState) { s.BuildID = "" }}, {"unknown phase", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseUnknown }}, {"unsupported phase", func(s *IssueTrackingState) { s.Phase = "other" }}, diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 4a335c196..47645b848 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -80,7 +80,7 @@ History persistence does not alter issue state or validate transitions against i ## Issue reconciliation progress -`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a terminal pass with task-wide failures finishes without closing; otherwise it resets the cursor and finalizes against issue states. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a non-canceled terminal pass resets the cursor and finalizes against issue states. A task-wide failure suppresses closures only for missing targets of the same task. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of cursor work, and completion; no runtime writer is enabled here. From cee21bec158add70eb861c9367835399235dd541 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 4 Nov 2026 15:06:54 +0000 Subject: [PATCH 2/3] refactor(stovepipe): derive task-wide suppression from build issues --- stovepipe/controller/issuetrack/README.md | 4 +- .../controller/issuetrack/delivery_test.go | 89 +++++++++++++++++-- .../controller/issuetrack/fixture_test.go | 10 ++- stovepipe/controller/issuetrack/issuetrack.go | 29 +++--- .../controller/issuetrack/issuetrack_test.go | 20 ++--- stovepipe/entity/issue_tracking_state.go | 2 - stovepipe/extension/storage/README.md | 4 +- .../storage/mysql/issue_tracking_store.go | 12 +-- .../mysql/issue_tracking_store_test.go | 12 +-- .../storage/mysql/schema/issue_tracking.sql | 1 - .../storage/mysql/issue_tracking_test.go | 8 +- 11 files changed, 135 insertions(+), 56 deletions(-) diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 1d7606dc0..85c592cf8 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -22,7 +22,7 @@ The post-read cursor write and processed-build updates, including unchanged stat ## Refreshing and finalizing -Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. +Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity combines the build and persisted pass notification IDs: retries share an identity, while later passes can trigger another evaluation. The fix controller selects an eligible issue and submits at most one repair per build. @@ -32,7 +32,7 @@ Once the entire build-issue range is exhausted: - A canceled build becomes `finalized` without closing anything. Reported closed issues may still have reopened during refreshing; unreported closed issues remain closed. - Every other terminal pass becomes `finalizing`, resetting its cursor to the start of the issue-state range. -The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes unless the same task has a task-wide failure. Such a failure preserves missing target issues in their current status, including closed issues; it does not reopen unreported targets. Issues from other tasks can close, including task-wide issues. The processed build advances for preserved issues as well. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. +The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes unless the same task has a task-wide failure. Such a failure preserves missing target issues in their current status, including closed issues; it does not reopen unreported targets. Issues from other tasks can close, including task-wide issues. Task-wide observations use an empty unit ID. Their presence or absence is cached per task within each batch; subsequent deliveries rebuild the cache from stored build issues. The processed build advances for preserved issues as well. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. No closure occurs before all accumulated build issues have been processed. A task-wide failure in a later batch therefore protects missing targets of that task before any closing work begins. A later build makes its own decision; the restriction does not carry over. Both walks consume one ordered range at a time; no merge is required. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index 1545efe4c..b0a9c3d74 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -465,6 +465,76 @@ func TestTaskWideFailureSuppressesClosuresOnlyForItsTask(t *testing.T) { require.Empty(t, f.openIssues) } +func TestTaskWideLookupIsCachedPerTaskWithinEachBatch(t *testing.T) { + for _, taskWide := range []bool{false, true} { + t.Run(fmt.Sprintf("task-wide=%t", taskWide), func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + if taskWide { + f.source("build/100", "100", entity.BuildStatusFailed) + f.reportIssue("build/100", "affected", "") + } + for i := 0; i < _issueBatchSize+1; i++ { + f.seedIssue("affected", fmt.Sprintf("unit/%03d", i), entity.IssueStatusOpen) + } + f.seedIssue("unrelated", "a", entity.IssueStatusOpen) + f.seedIssue("unrelated", "b", entity.IssueStatusOpen) + progress := currentCursor() + progress.Phase = entity.IssueTrackingPhaseFinalizing + progress.Finalize = true + progress.CursorTaskID, progress.CursorUnitID = "", "" + f.retain(progress) + reads := make(map[string]int) + f.onBuildIssueGet = func(_, task, unit string) error { + if unit == "" { + reads[task]++ + } + return nil + } + msg := f.message("build/100", true) + msg.ID = progress.NotificationID + + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) + require.Equal(t, map[string]int{"affected": 1}, reads) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Equal(t, map[string]int{"affected": 2, "unrelated": 1}, reads) + for key, issue := range f.states { + desired := entity.IssueStatusClosed + if taskWide && key.task == "affected" { + desired = entity.IssueStatusOpen + } + require.Equal(t, desired, issue.Status) + require.Equal(t, "build/100", issue.LastProcessedBuildID) + } + }) + } +} + +func TestTaskWideLookupFailureLeavesTheIssueForRetry(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + msg := f.message("build/100", true) + require.NoError(t, f.deliver(msg, true)) + f.onBuildIssueGet = func(_, _, unit string) error { + if unit == "" { + return assert.AnError + } + return nil + } + + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) + require.Equal(t, "older", f.states[issueKey{"task", "unit"}].LastProcessedBuildID) + require.Empty(t, f.logs) + f.onBuildIssueGet = nil + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) +} + func TestTaskWideFailureInLaterBatchProtectsOnlyItsTask(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusFailed) @@ -489,7 +559,7 @@ func TestTaskWideFailureInLaterBatchProtectsOnlyItsTask(t *testing.T) { require.Len(t, f.logs, 107) } -func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { +func TestTaskWideSuppressionSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) for i := 0; i < 104; i++ { @@ -504,6 +574,7 @@ func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { issue.LastProcessedBuildID = "build/100" f.states[key] = issue } + f.seedIssue("task", "unreported", entity.IssueStatusOpen) prior := currentCursor() prior.Phase = entity.IssueTrackingPhaseRefreshed f.retain(prior) @@ -512,19 +583,22 @@ func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { require.NoError(t, f.deliver(msg, true)) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) require.True(t, f.progress.Finalize) - require.True(t, f.progress.HasTaskWideIssues) require.Equal(t, "unit/099", f.progress.CursorUnitID) require.Empty(t, f.logs) f.failures["build-issues/list"] = 1 require.ErrorIs(t, f.deliver(msg, false), assert.AnError) - require.True(t, f.progress.HasTaskWideIssues) require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.True(t, f.progress.HasTaskWideIssues) - for _, issue := range f.states { - require.Equal(t, int32(7), issue.Version) + for key, issue := range f.states { + require.Equal(t, entity.IssueStatusOpen, issue.Status) + require.Equal(t, "build/100", issue.LastProcessedBuildID) + version := int32(7) + if key.unit == "unreported" { + version++ + } + require.Equal(t, version, issue.Version) } } @@ -550,7 +624,6 @@ func TestAnotherNotificationRetainsTheSavedTerminalIntent(t *testing.T) { require.NoError(t, f.deliver(msg, false)) require.Equal(t, "build/101", f.progress.BuildID) require.False(t, f.progress.Finalize) - require.False(t, f.progress.HasTaskWideIssues) require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) } @@ -560,7 +633,6 @@ func TestOpeningCursorCannotCloseAnUnreportedIssue(t *testing.T) { f.seedIssue("task", "unit", entity.IssueStatusOpen) progress := currentCursor() progress.Finalize = true - progress.HasTaskWideIssues = true f.retain(progress) msg := f.message("build/100", true) msg.ID = progress.NotificationID @@ -581,7 +653,6 @@ func TestTaskWideHistoryRetryThenClosesUnrelatedIssues(t *testing.T) { msg := f.message("build", true) require.ErrorIs(t, f.deliver(msg, false), assert.AnError) - require.True(t, f.progress.HasTaskWideIssues) require.True(t, f.progress.Finalize) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) f.controller.nowMs = func() int64 { return 2000 } diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index 232abd91b..f663d7427 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -61,8 +61,9 @@ type fixture struct { buildIssues map[string]map[issueKey]entity.BuildIssue sequence int - fixMessages []entityqueue.Message - onFixPublish func(entityqueue.Message) + fixMessages []entityqueue.Message + onFixPublish func(entityqueue.Message) + onBuildIssueGet func(build, task, unit string) error } func newFixture(t *testing.T) *fixture { @@ -290,6 +291,11 @@ func newFixture(t *testing.T) *fixture { return value, nil }).AnyTimes() buildIssues.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, build, task, unit string) (entity.BuildIssue, error) { + if f.onBuildIssueGet != nil { + if err := f.onBuildIssueGet(build, task, unit); err != nil { + return entity.BuildIssue{}, err + } + } if err := f.fail("build-issues/get"); err != nil { return entity.BuildIssue{}, err } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 61840440b..b0c227925 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -209,8 +209,9 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return entity.IssueTrackingState{}, fmt.Errorf("canceled builds cannot close missing issues") } + taskWideIssues := make(map[string]bool) if progress.CursorTaskID != "" { - next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false) + next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false, taskWideIssues) if err != nil { return entity.IssueTrackingState{}, err } @@ -249,7 +250,7 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return entity.IssueTrackingState{}, err } previousVersion := progress.Version - progress, err = c.processIssue(ctx, store, progress, key, !finalizing) + progress, err = c.processIssue(ctx, store, progress, key, !finalizing, taskWideIssues) if err != nil { return entity.IssueTrackingState{}, err } @@ -283,7 +284,7 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return progress, nil } -func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool) (entity.IssueTrackingState, error) { +func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool, taskWideIssues map[string]bool) (entity.IssueTrackingState, error) { if err := ctx.Err(); err != nil { return entity.IssueTrackingState{}, err } @@ -294,14 +295,13 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr } reported = err == nil } + if key.unit == "" { + taskWideIssues[key.task] = reported + } if !reported && progress.Phase != entity.IssueTrackingPhaseFinalizing { return entity.IssueTrackingState{}, fmt.Errorf("only finalizing passes can close missing issues") } - if reported && key.unit == "" && progress.Phase == entity.IssueTrackingPhaseRefreshing { - progress.HasTaskWideIssues = true - } - desired := entity.IssueStatusClosed if reported { desired = entity.IssueStatusOpen @@ -317,12 +317,17 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr if missing && !reported { return entity.IssueTrackingState{}, fmt.Errorf("closing cursor requires an existing issue") } - if !reported && key.unit != "" && progress.HasTaskWideIssues { - _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, "") - if err != nil && !errors.Is(err, storage.ErrNotFound) { - return entity.IssueTrackingState{}, fmt.Errorf("check task-wide issue: %w", err) + if !reported && key.unit != "" { + taskWide, cached := taskWideIssues[key.task] + if !cached { + _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, "") + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.IssueTrackingState{}, fmt.Errorf("check task-wide issue: %w", err) + } + taskWide = err == nil + taskWideIssues[key.task] = taskWide } - if err == nil { + if taskWide { desired = current.Status } } diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index b8f167fa7..55506357b 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -77,7 +77,7 @@ func TestProcessIssueOpensAndCloses(t *testing.T) { f.seedIssue("task", "unit", entity.IssueStatusOpen) } - next, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, reported) + next, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, reported, make(map[string]bool)) require.NoError(t, err) require.Equal(t, "task", next.CursorTaskID) require.Equal(t, "unit", next.CursorUnitID) @@ -116,7 +116,7 @@ func TestUnchangedIssueRecordsTheProcessedBuild(t *testing.T) { progress := currentCursor() f.retain(progress) - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.NoError(t, err) issue := f.states[issueKey{"task", "unit"}] require.Equal(t, "build/100", issue.LastProcessedBuildID) @@ -135,13 +135,13 @@ func TestHistoryRetryUsesTheStoredTransition(t *testing.T) { f.retain(progress) f.failures["history/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.ErrorIs(t, err, assert.AnError) applied := f.states[issueKey{"task", "unit"}] require.Empty(t, f.logs) f.controller.nowMs = func() int64 { return 2000 } - _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false) + _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false, make(map[string]bool)) 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.IssueStatusOpen}].TimestampMs) @@ -153,7 +153,7 @@ func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { f.retain(progress) f.failures["progress/update"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.ErrorIs(t, err, assert.AnError) require.Empty(t, f.states) require.Empty(t, f.logs) @@ -168,7 +168,7 @@ func TestSummaryRetryRepairsTheRetainedTransition(t *testing.T) { f.retain(progress) f.failures["summary/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.ErrorIs(t, err, assert.AnError) applied := f.states[issueKey{"task", "unit"}] require.Len(t, f.logs, 1) @@ -176,7 +176,7 @@ func TestSummaryRetryRepairsTheRetainedTransition(t *testing.T) { 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) + _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false, make(map[string]bool)) require.NoError(t, err) require.Equal(t, applied, f.states[issueKey{"task", "unit"}]) require.Len(t, f.logs, 1) @@ -196,7 +196,7 @@ func TestOlderSourceCannotChangeTheIssue(t *testing.T) { progress := currentCursor() f.retain(progress) - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.ErrorIs(t, err, storage.ErrVersionMismatch) require.Equal(t, current, f.states[issueKey{"task", "unit"}]) require.Equal(t, progress, f.progress) @@ -211,7 +211,7 @@ func TestOpenIssueFailureKeepsTheCursorForRetry(t *testing.T) { f.retain(progress) f.failures["open-issue/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) require.ErrorIs(t, err, assert.AnError) accepted := f.states[issueKey{"task", "unit"}] require.Contains(t, f.summaries, issueKey{"task", "unit"}) @@ -220,7 +220,7 @@ func TestOpenIssueFailureKeepsTheCursorForRetry(t *testing.T) { 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) + _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false, make(map[string]bool)) require.NoError(t, err) require.Equal(t, accepted, f.states[issueKey{"task", "unit"}]) require.Equal(t, accepted.FirstSeenBuildID, f.openIssues[issueKey{"task", "unit"}].FirstSeenBuildID) diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 05c0f3a92..7763975ab 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/stovepipe/entity/issue_tracking_state.go @@ -55,8 +55,6 @@ type IssueTrackingState struct { CursorUnitID string `json:"cursor_unit_id"` // Finalize identifies a terminal reconciliation pass rather than an ordinary refresh. Finalize bool `json:"finalize"` - // HasTaskWideIssues records whether this pass has encountered a task-wide issue. - HasTaskWideIssues bool `json:"has_task_wide_issues"` // Phase distinguishes active, completed, and abandoned work. Phase IssueTrackingPhase `json:"phase"` // Version is the positive optimistic-lock version of this state. diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 47645b848..42035bedb 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -80,13 +80,13 @@ History persistence does not alter issue state or validate transitions against i ## Issue reconciliation progress -`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a non-canceled terminal pass resets the cursor and finalizes against issue states. A task-wide failure suppresses closures only for missing targets of the same task. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing. Once all build issues have been applied, a non-canceled terminal pass resets the cursor and finalizes against issue states. A task-wide failure suppresses closures only for missing targets of the same task. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of cursor work, and completion; no runtime writer is enabled here. Recovery uses the same record's `abandoning` and `abandoned` phases; no additional schema fields or storage operations are required. The source, notification, cursor, and terminal intent remain present after abandonment. See [issue tracking recovery](../../controller/issuetrack/README.md#recovery-and-activation) for the conditional-write ordering that protects the current cursor before newer work is admitted. -The issue-tracking table must include the notification, cursor, `finalize`, and `has_task_wide_issues` columns before updated readers or writers are deployed. Both boolean columns default to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. +The issue-tracking table must include the notification, cursor, and `finalize` columns before updated readers or writers are deployed. The `finalize` column defaults to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. The obsolete `has_task_wide_issues` column can remain in existing tables; updated readers and writers ignore it. Keep it until all older readers and writers have been retired. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. ### Applied transition facts diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go index 9e81deeb9..d052989f5 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store.go +++ b/stovepipe/extension/storage/mysql/issue_tracking_store.go @@ -43,13 +43,13 @@ func (s *issueTrackingStore) Get(ctx context.Context) (state entity.IssueTrackin err := s.db.QueryRowContext(ctx, `SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, - finalize, has_task_wide_issues, phase, version + finalize, phase, version FROM issue_tracking WHERE queue = ?`, s.queue). Scan( &state.RequestID, &state.BuildID, &state.NotificationID, &state.CursorTaskID, &state.CursorUnitID, - &state.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version, + &state.Finalize, &state.Phase, &state.Version, ) if errors.Is(err, sql.ErrNoRows) { return entity.IssueTrackingState{}, storage.WrapNotFound(err) @@ -69,10 +69,10 @@ func (s *issueTrackingStore) Create(ctx context.Context, state entity.IssueTrack _, err := s.db.ExecContext(ctx, `INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, - finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + finalize, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, s.queue, state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, - state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, + state.Finalize, state.Phase, state.Version, ) if isDuplicateEntry(err) { return storage.ErrAlreadyExists @@ -97,11 +97,11 @@ func (s *issueTrackingStore) Update(ctx context.Context, state entity.IssueTrack result, err := s.db.ExecContext(ctx, `UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, - finalize = ?, has_task_wide_issues = ?, phase = ?, version = ? + finalize = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`, state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, - state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion, + state.Finalize, state.Phase, newVersion, s.queue, oldVersion, ) if err != nil { diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go index 935ea6682..25e7091f2 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go +++ b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go @@ -43,7 +43,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") + query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") switch kind { case "missing": query.WillReturnError(sql.ErrNoRows) @@ -54,7 +54,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { if kind == "scan error" { request = nil } - query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "has_task_wide_issues", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version)) + query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, state.Version)) } got, err := store.Get(context.Background()) @@ -94,8 +94,8 @@ func TestIssueTrackingStoreCreate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)). - WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version) + query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, state.Version) if tt.dbErr != nil { query.WillReturnError(tt.dbErr) } else { @@ -133,8 +133,8 @@ func TestIssueTrackingStoreUpdate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, finalize = ?, has_task_wide_issues = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). - WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, int32(7), "queue", int32(1)) + query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, finalize = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). + WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, int32(7), "queue", int32(1)) switch { case tt.dbErr != nil: query.WillReturnError(tt.dbErr) diff --git a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql index 9d367d84c..9ca367153 100644 --- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql +++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql @@ -9,7 +9,6 @@ CREATE TABLE IF NOT EXISTS issue_tracking ( cursor_unit_id VARBINARY(1024) NOT NULL, finalize BOOLEAN NOT NULL DEFAULT FALSE, - has_task_wide_issues BOOLEAN NOT NULL DEFAULT FALSE, phase VARBINARY(32) NOT NULL, version INT NOT NULL, PRIMARY KEY (queue) diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go index e30f5aaa5..7304fa592 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go @@ -62,11 +62,11 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F require.NoError(t, err) require.Empty(t, opens) - t.Run("terminal opening checkpoint and task-wide flag", func(t *testing.T) { + t.Run("terminal opening checkpoint", func(t *testing.T) { bound, err := factory.For(storage.Config{QueueName: "tracking/terminal"}) require.NoError(t, err) store := bound.GetIssueTrackingStore() - opening := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "terminal", Phase: entity.IssueTrackingPhaseRefreshing, Finalize: true, HasTaskWideIssues: true, CursorTaskID: "task", CursorUnitID: "unit", Version: 1} + opening := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "terminal", Phase: entity.IssueTrackingPhaseRefreshing, Finalize: true, CursorTaskID: "task", CursorUnitID: "unit", Version: 1} require.NoError(t, store.Create(ctx, opening)) got, err := store.Get(ctx) require.NoError(t, err) @@ -99,8 +99,8 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F state := entity.IssueTrackingState{ RequestID: "10", BuildID: "build/10", NotificationID: "failed-notification", CursorTaskID: "task", CursorUnitID: "unit", - Finalize: true, HasTaskWideIssues: true, - Phase: entity.IssueTrackingPhaseAbandoning, Version: 1, + Finalize: true, + Phase: entity.IssueTrackingPhaseAbandoning, Version: 1, } require.NoError(t, store.Create(ctx, state)) got, err := store.Get(ctx) From cac06cabc31176b58e8400f1bb19e47c695163f1 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Wed, 4 Nov 2026 15:25:17 +0000 Subject: [PATCH 3/3] refactor(stovepipe): retain task-wide flag and clarify issue lookup --- stovepipe/controller/issuetrack/README.md | 4 +- .../controller/issuetrack/delivery_test.go | 89 ++----------------- .../controller/issuetrack/fixture_test.go | 10 +-- stovepipe/controller/issuetrack/issuetrack.go | 39 ++++---- .../controller/issuetrack/issuetrack_test.go | 20 ++--- stovepipe/entity/issue_tracking_state.go | 2 + stovepipe/extension/storage/README.md | 4 +- .../storage/mysql/issue_tracking_store.go | 12 +-- .../mysql/issue_tracking_store_test.go | 12 +-- .../storage/mysql/schema/issue_tracking.sql | 1 + .../storage/mysql/issue_tracking_test.go | 8 +- 11 files changed, 67 insertions(+), 134 deletions(-) diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 85c592cf8..1d7606dc0 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -22,7 +22,7 @@ The post-read cursor write and processed-build updates, including unchanged stat ## Refreshing and finalizing -Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. +Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity combines the build and persisted pass notification IDs: retries share an identity, while later passes can trigger another evaluation. The fix controller selects an eligible issue and submits at most one repair per build. @@ -32,7 +32,7 @@ Once the entire build-issue range is exhausted: - A canceled build becomes `finalized` without closing anything. Reported closed issues may still have reopened during refreshing; unreported closed issues remain closed. - Every other terminal pass becomes `finalizing`, resetting its cursor to the start of the issue-state range. -The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes unless the same task has a task-wide failure. Such a failure preserves missing target issues in their current status, including closed issues; it does not reopen unreported targets. Issues from other tasks can close, including task-wide issues. Task-wide observations use an empty unit ID. Their presence or absence is cached per task within each batch; subsequent deliveries rebuild the cache from stored build issues. The processed build advances for preserved issues as well. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. +The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes unless the same task has a task-wide failure. Such a failure preserves missing target issues in their current status, including closed issues; it does not reopen unreported targets. Issues from other tasks can close, including task-wide issues. The processed build advances for preserved issues as well. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored. No closure occurs before all accumulated build issues have been processed. A task-wide failure in a later batch therefore protects missing targets of that task before any closing work begins. A later build makes its own decision; the restriction does not carry over. Both walks consume one ordered range at a time; no merge is required. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index b0a9c3d74..1545efe4c 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -465,76 +465,6 @@ func TestTaskWideFailureSuppressesClosuresOnlyForItsTask(t *testing.T) { require.Empty(t, f.openIssues) } -func TestTaskWideLookupIsCachedPerTaskWithinEachBatch(t *testing.T) { - for _, taskWide := range []bool{false, true} { - t.Run(fmt.Sprintf("task-wide=%t", taskWide), func(t *testing.T) { - f := newFixture(t) - f.source("build/100", "100", entity.BuildStatusSucceeded) - if taskWide { - f.source("build/100", "100", entity.BuildStatusFailed) - f.reportIssue("build/100", "affected", "") - } - for i := 0; i < _issueBatchSize+1; i++ { - f.seedIssue("affected", fmt.Sprintf("unit/%03d", i), entity.IssueStatusOpen) - } - f.seedIssue("unrelated", "a", entity.IssueStatusOpen) - f.seedIssue("unrelated", "b", entity.IssueStatusOpen) - progress := currentCursor() - progress.Phase = entity.IssueTrackingPhaseFinalizing - progress.Finalize = true - progress.CursorTaskID, progress.CursorUnitID = "", "" - f.retain(progress) - reads := make(map[string]int) - f.onBuildIssueGet = func(_, task, unit string) error { - if unit == "" { - reads[task]++ - } - return nil - } - msg := f.message("build/100", true) - msg.ID = progress.NotificationID - - require.NoError(t, f.deliver(msg, true)) - require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) - require.Equal(t, map[string]int{"affected": 1}, reads) - require.NoError(t, f.deliver(msg, false)) - require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.Equal(t, map[string]int{"affected": 2, "unrelated": 1}, reads) - for key, issue := range f.states { - desired := entity.IssueStatusClosed - if taskWide && key.task == "affected" { - desired = entity.IssueStatusOpen - } - require.Equal(t, desired, issue.Status) - require.Equal(t, "build/100", issue.LastProcessedBuildID) - } - }) - } -} - -func TestTaskWideLookupFailureLeavesTheIssueForRetry(t *testing.T) { - f := newFixture(t) - f.source("build/100", "100", entity.BuildStatusSucceeded) - f.seedIssue("task", "unit", entity.IssueStatusOpen) - msg := f.message("build/100", true) - require.NoError(t, f.deliver(msg, true)) - f.onBuildIssueGet = func(_, _, unit string) error { - if unit == "" { - return assert.AnError - } - return nil - } - - require.ErrorIs(t, f.deliver(msg, false), assert.AnError) - require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) - require.Equal(t, "older", f.states[issueKey{"task", "unit"}].LastProcessedBuildID) - require.Empty(t, f.logs) - f.onBuildIssueGet = nil - require.NoError(t, f.deliver(msg, false)) - require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) -} - func TestTaskWideFailureInLaterBatchProtectsOnlyItsTask(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusFailed) @@ -559,7 +489,7 @@ func TestTaskWideFailureInLaterBatchProtectsOnlyItsTask(t *testing.T) { require.Len(t, f.logs, 107) } -func TestTaskWideSuppressionSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { +func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) for i := 0; i < 104; i++ { @@ -574,7 +504,6 @@ func TestTaskWideSuppressionSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { issue.LastProcessedBuildID = "build/100" f.states[key] = issue } - f.seedIssue("task", "unreported", entity.IssueStatusOpen) prior := currentCursor() prior.Phase = entity.IssueTrackingPhaseRefreshed f.retain(prior) @@ -583,22 +512,19 @@ func TestTaskWideSuppressionSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { require.NoError(t, f.deliver(msg, true)) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) require.True(t, f.progress.Finalize) + require.True(t, f.progress.HasTaskWideIssues) require.Equal(t, "unit/099", f.progress.CursorUnitID) require.Empty(t, f.logs) f.failures["build-issues/list"] = 1 require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.True(t, f.progress.HasTaskWideIssues) require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, true)) require.NoError(t, f.deliver(msg, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - for key, issue := range f.states { - require.Equal(t, entity.IssueStatusOpen, issue.Status) - require.Equal(t, "build/100", issue.LastProcessedBuildID) - version := int32(7) - if key.unit == "unreported" { - version++ - } - require.Equal(t, version, issue.Version) + require.True(t, f.progress.HasTaskWideIssues) + for _, issue := range f.states { + require.Equal(t, int32(7), issue.Version) } } @@ -624,6 +550,7 @@ func TestAnotherNotificationRetainsTheSavedTerminalIntent(t *testing.T) { require.NoError(t, f.deliver(msg, false)) require.Equal(t, "build/101", f.progress.BuildID) require.False(t, f.progress.Finalize) + require.False(t, f.progress.HasTaskWideIssues) require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) } @@ -633,6 +560,7 @@ func TestOpeningCursorCannotCloseAnUnreportedIssue(t *testing.T) { f.seedIssue("task", "unit", entity.IssueStatusOpen) progress := currentCursor() progress.Finalize = true + progress.HasTaskWideIssues = true f.retain(progress) msg := f.message("build/100", true) msg.ID = progress.NotificationID @@ -653,6 +581,7 @@ func TestTaskWideHistoryRetryThenClosesUnrelatedIssues(t *testing.T) { msg := f.message("build", true) require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.True(t, f.progress.HasTaskWideIssues) require.True(t, f.progress.Finalize) require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) f.controller.nowMs = func() int64 { return 2000 } diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index f663d7427..232abd91b 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -61,9 +61,8 @@ type fixture struct { buildIssues map[string]map[issueKey]entity.BuildIssue sequence int - fixMessages []entityqueue.Message - onFixPublish func(entityqueue.Message) - onBuildIssueGet func(build, task, unit string) error + fixMessages []entityqueue.Message + onFixPublish func(entityqueue.Message) } func newFixture(t *testing.T) *fixture { @@ -291,11 +290,6 @@ func newFixture(t *testing.T) *fixture { return value, nil }).AnyTimes() buildIssues.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, build, task, unit string) (entity.BuildIssue, error) { - if f.onBuildIssueGet != nil { - if err := f.onBuildIssueGet(build, task, unit); err != nil { - return entity.BuildIssue{}, err - } - } if err := f.fail("build-issues/get"); err != nil { return entity.BuildIssue{}, err } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index b0c227925..9447f6bc6 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -209,9 +209,8 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return entity.IssueTrackingState{}, fmt.Errorf("canceled builds cannot close missing issues") } - taskWideIssues := make(map[string]bool) if progress.CursorTaskID != "" { - next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false, taskWideIssues) + next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false) if err != nil { return entity.IssueTrackingState{}, err } @@ -250,7 +249,7 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return entity.IssueTrackingState{}, err } previousVersion := progress.Version - progress, err = c.processIssue(ctx, store, progress, key, !finalizing, taskWideIssues) + progress, err = c.processIssue(ctx, store, progress, key, !finalizing) if err != nil { return entity.IssueTrackingState{}, err } @@ -284,7 +283,7 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, return progress, nil } -func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool, taskWideIssues map[string]bool) (entity.IssueTrackingState, error) { +func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool) (entity.IssueTrackingState, error) { if err := ctx.Err(); err != nil { return entity.IssueTrackingState{}, err } @@ -295,13 +294,14 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr } reported = err == nil } - if key.unit == "" { - taskWideIssues[key.task] = reported - } if !reported && progress.Phase != entity.IssueTrackingPhaseFinalizing { return entity.IssueTrackingState{}, fmt.Errorf("only finalizing passes can close missing issues") } + if reported && key.unit == "" && progress.Phase == entity.IssueTrackingPhaseRefreshing { + progress.HasTaskWideIssues = true + } + desired := entity.IssueStatusClosed if reported { desired = entity.IssueStatusOpen @@ -317,15 +317,10 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr if missing && !reported { return entity.IssueTrackingState{}, fmt.Errorf("closing cursor requires an existing issue") } - if !reported && key.unit != "" { - taskWide, cached := taskWideIssues[key.task] - if !cached { - _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, "") - if err != nil && !errors.Is(err, storage.ErrNotFound) { - return entity.IssueTrackingState{}, fmt.Errorf("check task-wide issue: %w", err) - } - taskWide = err == nil - taskWideIssues[key.task] = taskWide + if !reported && key.unit != "" && progress.HasTaskWideIssues { + taskWide, err := hasTaskWideIssue(ctx, store.GetBuildIssueStore(), progress.BuildID, key.task) + if err != nil { + return entity.IssueTrackingState{}, err } if taskWide { desired = current.Status @@ -412,6 +407,18 @@ func (c *Controller) recordHistory(ctx context.Context, store storage.Storage, i return nil } +func hasTaskWideIssue(ctx context.Context, store storage.BuildIssueStore, buildID, taskID string) (bool, error) { + // An empty unit ID identifies a task-wide observation. + _, err := store.Get(ctx, buildID, taskID, "") + if errors.Is(err, storage.ErrNotFound) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("check task-wide issue: %w", err) + } + return true, nil +} + func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next entity.IssueTrackingState) (entity.IssueTrackingState, error) { if next.Version == math.MaxInt32 { return entity.IssueTrackingState{}, fmt.Errorf("issue tracking version exhausted") diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index 55506357b..b8f167fa7 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -77,7 +77,7 @@ func TestProcessIssueOpensAndCloses(t *testing.T) { f.seedIssue("task", "unit", entity.IssueStatusOpen) } - next, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, reported, make(map[string]bool)) + next, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, reported) require.NoError(t, err) require.Equal(t, "task", next.CursorTaskID) require.Equal(t, "unit", next.CursorUnitID) @@ -116,7 +116,7 @@ func TestUnchangedIssueRecordsTheProcessedBuild(t *testing.T) { progress := currentCursor() f.retain(progress) - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) require.NoError(t, err) issue := f.states[issueKey{"task", "unit"}] require.Equal(t, "build/100", issue.LastProcessedBuildID) @@ -135,13 +135,13 @@ func TestHistoryRetryUsesTheStoredTransition(t *testing.T) { f.retain(progress) f.failures["history/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, 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.Empty(t, f.logs) f.controller.nowMs = func() int64 { return 2000 } - _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false, make(map[string]bool)) + _, 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.IssueStatusOpen}].TimestampMs) @@ -153,7 +153,7 @@ func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { f.retain(progress) f.failures["progress/update"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) require.ErrorIs(t, err, assert.AnError) require.Empty(t, f.states) require.Empty(t, f.logs) @@ -168,7 +168,7 @@ func TestSummaryRetryRepairsTheRetainedTransition(t *testing.T) { f.retain(progress) f.failures["summary/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, 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) @@ -176,7 +176,7 @@ func TestSummaryRetryRepairsTheRetainedTransition(t *testing.T) { 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, make(map[string]bool)) + _, 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) @@ -196,7 +196,7 @@ func TestOlderSourceCannotChangeTheIssue(t *testing.T) { progress := currentCursor() f.retain(progress) - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) require.ErrorIs(t, err, storage.ErrVersionMismatch) require.Equal(t, current, f.states[issueKey{"task", "unit"}]) require.Equal(t, progress, f.progress) @@ -211,7 +211,7 @@ func TestOpenIssueFailureKeepsTheCursorForRetry(t *testing.T) { f.retain(progress) f.failures["open-issue/create"] = 1 - _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true, make(map[string]bool)) + _, 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"}) @@ -220,7 +220,7 @@ func TestOpenIssueFailureKeepsTheCursorForRetry(t *testing.T) { 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, make(map[string]bool)) + _, 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) diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 7763975ab..05c0f3a92 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/stovepipe/entity/issue_tracking_state.go @@ -55,6 +55,8 @@ type IssueTrackingState struct { CursorUnitID string `json:"cursor_unit_id"` // Finalize identifies a terminal reconciliation pass rather than an ordinary refresh. Finalize bool `json:"finalize"` + // HasTaskWideIssues records whether this pass has encountered a task-wide issue. + HasTaskWideIssues bool `json:"has_task_wide_issues"` // Phase distinguishes active, completed, and abandoned work. Phase IssueTrackingPhase `json:"phase"` // Version is the positive optimistic-lock version of this state. diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 42035bedb..47645b848 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -80,13 +80,13 @@ History persistence does not alter issue state or validate transitions against i ## Issue reconciliation progress -`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing. Once all build issues have been applied, a non-canceled terminal pass resets the cursor and finalizes against issue states. A task-wide failure suppresses closures only for missing targets of the same task. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a non-canceled terminal pass resets the cursor and finalizes against issue states. A task-wide failure suppresses closures only for missing targets of the same task. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of cursor work, and completion; no runtime writer is enabled here. Recovery uses the same record's `abandoning` and `abandoned` phases; no additional schema fields or storage operations are required. The source, notification, cursor, and terminal intent remain present after abandonment. See [issue tracking recovery](../../controller/issuetrack/README.md#recovery-and-activation) for the conditional-write ordering that protects the current cursor before newer work is admitted. -The issue-tracking table must include the notification, cursor, and `finalize` columns before updated readers or writers are deployed. The `finalize` column defaults to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. The obsolete `has_task_wide_issues` column can remain in existing tables; updated readers and writers ignore it. Keep it until all older readers and writers have been retired. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. +The issue-tracking table must include the notification, cursor, `finalize`, and `has_task_wide_issues` columns before updated readers or writers are deployed. Both boolean columns default to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores. ### Applied transition facts diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go index d052989f5..9e81deeb9 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store.go +++ b/stovepipe/extension/storage/mysql/issue_tracking_store.go @@ -43,13 +43,13 @@ func (s *issueTrackingStore) Get(ctx context.Context) (state entity.IssueTrackin err := s.db.QueryRowContext(ctx, `SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, - finalize, phase, version + finalize, has_task_wide_issues, phase, version FROM issue_tracking WHERE queue = ?`, s.queue). Scan( &state.RequestID, &state.BuildID, &state.NotificationID, &state.CursorTaskID, &state.CursorUnitID, - &state.Finalize, &state.Phase, &state.Version, + &state.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version, ) if errors.Is(err, sql.ErrNoRows) { return entity.IssueTrackingState{}, storage.WrapNotFound(err) @@ -69,10 +69,10 @@ func (s *issueTrackingStore) Create(ctx context.Context, state entity.IssueTrack _, err := s.db.ExecContext(ctx, `INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, - finalize, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, s.queue, state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, - state.Finalize, state.Phase, state.Version, + state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, ) if isDuplicateEntry(err) { return storage.ErrAlreadyExists @@ -97,11 +97,11 @@ func (s *issueTrackingStore) Update(ctx context.Context, state entity.IssueTrack result, err := s.db.ExecContext(ctx, `UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, - finalize = ?, phase = ?, version = ? + finalize = ?, has_task_wide_issues = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`, state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, - state.Finalize, state.Phase, newVersion, + state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion, s.queue, oldVersion, ) if err != nil { diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go index 25e7091f2..935ea6682 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go +++ b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go @@ -43,7 +43,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") + query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") switch kind { case "missing": query.WillReturnError(sql.ErrNoRows) @@ -54,7 +54,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { if kind == "scan error" { request = nil } - query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, state.Version)) + query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "has_task_wide_issues", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version)) } got, err := store.Get(context.Background()) @@ -94,8 +94,8 @@ func TestIssueTrackingStoreCreate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`)). - WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, state.Version) + query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version) if tt.dbErr != nil { query.WillReturnError(tt.dbErr) } else { @@ -133,8 +133,8 @@ func TestIssueTrackingStoreUpdate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, finalize = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). - WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.Phase, int32(7), "queue", int32(1)) + query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, finalize = ?, has_task_wide_issues = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). + WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, int32(7), "queue", int32(1)) switch { case tt.dbErr != nil: query.WillReturnError(tt.dbErr) diff --git a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql index 9ca367153..9d367d84c 100644 --- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql +++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql @@ -9,6 +9,7 @@ CREATE TABLE IF NOT EXISTS issue_tracking ( cursor_unit_id VARBINARY(1024) NOT NULL, finalize BOOLEAN NOT NULL DEFAULT FALSE, + has_task_wide_issues BOOLEAN NOT NULL DEFAULT FALSE, phase VARBINARY(32) NOT NULL, version INT NOT NULL, PRIMARY KEY (queue) diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go index 7304fa592..e30f5aaa5 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go @@ -62,11 +62,11 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F require.NoError(t, err) require.Empty(t, opens) - t.Run("terminal opening checkpoint", func(t *testing.T) { + t.Run("terminal opening checkpoint and task-wide flag", func(t *testing.T) { bound, err := factory.For(storage.Config{QueueName: "tracking/terminal"}) require.NoError(t, err) store := bound.GetIssueTrackingStore() - opening := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "terminal", Phase: entity.IssueTrackingPhaseRefreshing, Finalize: true, CursorTaskID: "task", CursorUnitID: "unit", Version: 1} + opening := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "terminal", Phase: entity.IssueTrackingPhaseRefreshing, Finalize: true, HasTaskWideIssues: true, CursorTaskID: "task", CursorUnitID: "unit", Version: 1} require.NoError(t, store.Create(ctx, opening)) got, err := store.Get(ctx) require.NoError(t, err) @@ -99,8 +99,8 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F state := entity.IssueTrackingState{ RequestID: "10", BuildID: "build/10", NotificationID: "failed-notification", CursorTaskID: "task", CursorUnitID: "unit", - Finalize: true, - Phase: entity.IssueTrackingPhaseAbandoning, Version: 1, + Finalize: true, HasTaskWideIssues: true, + Phase: entity.IssueTrackingPhaseAbandoning, Version: 1, } require.NoError(t, store.Create(ctx, state)) got, err := store.Get(ctx)