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..9447f6bc6 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 { + taskWide, err := hasTaskWideIssue(ctx, store.GetBuildIssueStore(), progress.BuildID, key.task) + if err != nil { + return entity.IssueTrackingState{}, err + } + if taskWide { + desired = current.Status + } + } if !missing && current.LastProcessedBuildID != progress.BuildID { previous, err := loader.ByID(ctx, current.LastProcessedBuildID, store.GetBuildStore().Get, "last processed build") @@ -398,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 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.