Skip to content

fix(stovepipe): finalize builds independently of running observations #1008

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mnoah1 wants to merge 1 commit into
base: mnoah1/stovepipe-task-scoped-closure
Choose a base branch
from
Draft

fix(stovepipe): finalize builds independently of running observations #1008

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions stovepipe/controller/issuetrack/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ Issue tracking maintains current issue state and opening/closing history from pe
## Applying one issue

1. Read whether the build reports the issue and load its current state.
2. Reject a source older than the issue's last processed build, using Request order rather than opaque Build IDs.
2. Skip a source older than the issue's last processed build, using Request order rather than opaque Build IDs.
3. Before changing the issue, conditionally save its task/unit key as the current queue cursor **after** reading the issue row. Repeat this conditional write even when resuming the same cursor.
4. Conditionally create or update the issue row. Record the processed build even when its open/closed status stays the same; update the changing build and timestamp only on an actual transition.
5. Retain opening or closing history from the issue row's changing build and timestamp, then materialize its summary and open-issue lookup. On retry, the inclusive cursor revisits that row and repairs all three before proceeding.
Expand All @@ -14,7 +14,7 @@ Closed issue rows are retained, so their versions cannot reset on reopening. A r

## Ordering requirements

During issue application, the caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later is rejected by source ordering or the conditional cursor write.
During issue application, the caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later skips an issue already processed by a newer source or fails the conditional cursor write.

The [issue-log materializer](../../core/issuelog/README.md) also orders summary updates by source request, with closing after opening within the same request. A paused projection writer cannot overwrite a newer summary: its conditional write fails, and reloading the winner repeats the source-order check. Lookup entries are then maintained from the latest summary, not the incoming transition. Summary and lookup failures keep the cursor unfinished, just like history failures.

Expand All @@ -40,7 +40,7 @@ No closure occurs before all accumulated build issues have been processed. A tas

Every delivery processes at most 100 task/unit keys from one ordered range, including a revisited cursor and retained closed rows. Refreshing reads build issues; finalizing reads issue states. The cursor is always inclusive: revisit it to finish the issue and its history, then continue after it. A full batch retains its active phase and requests redelivery. Moving from refreshing to finalizing conditionally saves a reset cursor before any closing work begins.

An unfinished pass completes or is safely abandoned before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized or its terminal pass was abandoned. A terminal pass requires a terminal build chosen by its Request in both active phases.
An unfinished pass completes or is safely abandoned before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and sources at or below `FinalizedRequestID` are ignored. This terminal boundary advances only when a terminal pass completes or is abandoned, and survives every later refresh. It is independent of the source of the current pass: a newer running observation, empty or nonempty, cannot discard an older terminal pass. Each issue instead retains its newest processed source; older passes skip that issue and continue their range. A newer completed terminal pass still supersedes all earlier notifications, including observations for issues with no retained state. A terminal pass requires a terminal build chosen by its Request in both active phases.

Build IDs are opaque; source order comes from Request IDs. All scans are bounded reads under leading primary-key prefixes. There is no transaction spanning progress, issue state, and history.

Expand All @@ -51,7 +51,7 @@ An error returns to normal consumer retry. The next delivery resumes the cursor
1. Conditionally enter `abandoning`, retaining the source, notification, and cursor. Normal deliveries hold for 30 seconds while this phase is present; they cannot start another pass or resume issue application.
2. If there is a cursor, read its issue and conditionally advance progress after that read. Advance the issue's version without changing its status or transition metadata. This invalidates a paused worker's earlier row guard. If the row has not been created yet, finish that one opening instead: the normal worker can only have saved such a cursor after finding a reported build issue. Creating it prevents the delayed create from succeeding. Closed issue rows are never deleted.
3. Repair the cursor's opening or closing history, summary, and open-issue lookup from the retained row. Record an `issue_tracking_abandoned` Request-history event keyed by the failed notification, with its build ID.
4. Conditionally enter `abandoned`. Keep the source-order boundary and notification identity, so the same delivery and older builds cannot restart the abandoned work.
4. Conditionally enter `abandoned`. Keep the terminal boundary and notification identity, so the same delivery cannot restart the abandoned work. A terminal abandonment also advances `FinalizedRequestID`, preventing that Request and older sources from restarting after later refreshes.

Accepted opens and closes remain in place. Recovery does not scan either issue range, close the remaining issues, change Request or Build outcomes, or release build slots. An abandoned ordinary refresh still permits a new refresh or terminal notification for that Request; an abandoned terminal pass permits only newer Requests. A later successful build can reconcile issues left unchanged by the failed pass.

Expand All @@ -61,8 +61,10 @@ A notification can fail while helping an older pass. Recovery abandons that save

Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged.

The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The reference host also consumes repair notifications with a no-op executor; fix dead-letter recovery remains separate.
The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. Final pass admission, completion, and superseded final notifications are logged at info level with their build and request IDs. The reference host also consumes repair notifications with a no-op executor; fix dead-letter recovery remains separate.

## Analysis handoff

Build analysis publishes refresh or final notifications after persisting build issues. Final notification requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host uses a no-op detector; deployments supply their own detector and register the same normal and recovery paths.

The issue-tracking table must include `finalized_request_id` before deploying these controllers. It is a non-null binary request ID with an empty default. Existing completed or abandoned terminal records supply their own Request ID as the initial boundary; existing refresh records have no retained terminal boundary. Upgrade normal and recovery controllers together: older writers do not preserve the terminal boundary, and older readers reject closing passes containing task-wide failures.
125 changes: 125 additions & 0 deletions stovepipe/controller/issuetrack/delivery_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -614,3 +614,128 @@ func TestRetryResetsCursorBeforeClosing(t *testing.T) {
require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
require.Len(t, f.logs, 2)
}

func TestNewerRefreshDoesNotDiscardOlderFinalization(t *testing.T) {
for _, observation := range []string{"empty", "unrelated failure", "same failure"} {
t.Run(observation, func(t *testing.T) {
f := newFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusRunning)
f.seedIssue("old", "unit", entity.IssueStatusOpen)
if observation == "unrelated failure" {
f.reportIssue("build/101", "new", "unit")
} else if observation == "same failure" {
f.reportIssue("build/101", "old", "unit")
}
require.NoError(t, f.deliver(f.message("build/101", false), false))
final := f.message("build/100", true)
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, false))
require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
if observation == "same failure" {
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
require.Equal(t, "build/101", f.states[issueKey{"old", "unit"}].LastProcessedBuildID)
require.Empty(t, f.logs)
} else {
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "unit"}].Status)
require.Contains(t, f.logs, logKey{issueKey{"old", "unit"}, "build/100", entity.IssueStatusClosed})
}
if observation == "unrelated failure" {
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", "unit"}].Status)
require.Equal(t, "build/101", f.states[issueKey{"new", "unit"}].LastProcessedBuildID)
}
})
}
}

func TestFinalizedBoundarySurvivesLaterRefreshes(t *testing.T) {
f := newFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusSucceeded)
f.source("build/102", "102", entity.BuildStatusRunning)
f.seedIssue("resolved", "unit", entity.IssueStatusOpen)
final := f.message("build/101", true)
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, false))
f.reportIssue("build/102", "new", "unit")
require.NoError(t, f.deliver(f.message("build/102", false), false))
before := f.progress
for _, build := range []string{"build/100", "build/101"} {
f.reportIssue(build, "resolved", "unit")
f.reportIssue(build, "stale", "unit")
for _, terminal := range []bool{false, true} {
require.NoError(t, f.deliver(f.message(build, terminal), false))
}
}
require.Equal(t, before, f.progress)
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"resolved", "unit"}].Status)
require.NotContains(t, f.states, issueKey{"stale", "unit"})
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", "unit"}].Status)
require.Len(t, f.logs, 2)
}

func TestOlderFinalizationSkipsNewerIssuesAcrossBatches(t *testing.T) {
f := newFixture(t)
f.source("build/100", "100", entity.BuildStatusFailed)
f.source("build/101", "101", entity.BuildStatusRunning)
f.seedIssue("zzz", "resolved", entity.IssueStatusOpen)
for i := 0; i < 105; i++ {
unit := fmt.Sprintf("unit/%03d", i)
f.reportIssue("build/100", "task", unit)
f.reportIssue("build/101", "task", unit)
}
refresh := f.message("build/101", false)
require.NoError(t, f.deliver(refresh, true))
require.NoError(t, f.deliver(refresh, false))
final := f.message("build/100", true)
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, false))
require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
for key, state := range f.states {
if key.task == "task" {
require.Equal(t, entity.IssueStatusOpen, state.Status)
require.Equal(t, "build/101", state.LastProcessedBuildID)
}
}
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"zzz", "resolved"}].Status)
require.Len(t, f.logs, 106)
}

func TestFailedCompletionCheckpointDoesNotLoseFinalization(t *testing.T) {
f := newFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusRunning)
f.seedIssue("old", "unit", entity.IssueStatusOpen)
final := f.message("build/100", true)
require.NoError(t, f.deliver(final, true))
f.failOnPhase = entity.IssueTrackingPhaseFinalized
require.ErrorIs(t, f.deliver(final, false), assert.AnError)
refresh := f.message("build/101", false)
require.NoError(t, f.deliver(refresh, true))
require.NoError(t, f.deliver(refresh, false))
f.reportIssue("build/100", "stale", "unit")
require.NoError(t, f.deliver(f.message("build/100", false), false))
require.NotContains(t, f.states, issueKey{"stale", "unit"})
require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "unit"}].Status)
require.Len(t, f.logs, 1)
}

func TestLegacyTerminalBoundarySurvivesANewerRefresh(t *testing.T) {
f := newFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusRunning)
legacy := currentCursor()
legacy.Finalize, legacy.Phase = true, entity.IssueTrackingPhaseFinalized
f.retain(legacy)
f.reportIssue("build/101", "new", "unit")
require.NoError(t, f.deliver(f.message("build/101", false), false))
before := f.progress
f.reportIssue("build/100", "stale", "unit")
require.NoError(t, f.deliver(f.message("build/100", false), false))
require.NoError(t, f.deliver(f.message("build/100", true), false))
require.Equal(t, before, f.progress)
require.NotContains(t, f.states, issueKey{"stale", "unit"})
require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", "unit"}].Status)
}
12 changes: 8 additions & 4 deletions stovepipe/controller/issuetrack/dlq.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,20 +138,24 @@ func (c *DLQController) abandonPass(ctx context.Context, store storage.Storage,
return progress, c.recordAbandonment(ctx, store, request.Queue, progress)
}
} else {
order, err := entity.CompareRequestID(request.ID, progress.RequestID)
superseded, err := notificationSuperseded(progress, request.ID)
if err != nil {
return entity.IssueTrackingState{}, err
}
if order < 0 || (order == 0 && progress.Finalize) {
if superseded {
return entity.IssueTrackingState{}, nil
}
order, err := entity.CompareRequestID(request.ID, progress.RequestID)
if err != nil {
return entity.IssueTrackingState{}, err
}
switch progress.Phase {
case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing, entity.IssueTrackingPhaseAbandoning:
if order == 0 {
if order <= 0 {
return entity.IssueTrackingState{}, nil
}
default:
progress = entity.IssueTrackingState{Version: progress.Version}
progress = entity.IssueTrackingState{Version: progress.Version, FinalizedRequestID: finalizedRequestID(progress)}
}
}
}
Expand Down
34 changes: 34 additions & 0 deletions stovepipe/controller/issuetrack/dlq_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,3 +436,37 @@ func TestRecoveryAcknowledgesUnusableMessages(t *testing.T) {
})
}
}

func TestRecoveryRetainsTerminalBoundaryAcrossANewerAbandonedRefresh(t *testing.T) {
f := newRecoveryFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusRunning)
final := f.message("build/100", true)
require.NoError(t, f.deliver(final, true))
require.NoError(t, f.deliver(final, false))
require.NoError(t, f.recover(f.message("build/101", false), false))
before := f.progress
f.reportIssue("build/100", "stale", "unit")
require.NoError(t, f.deliver(f.message("build/100", false), false))
require.NoError(t, f.recover(f.message("build/100", true), false))
require.Equal(t, before, f.progress)
require.Empty(t, f.states)
}

func TestRecoveryOfOlderTerminalAfterNewerRefreshRetainsNewerEvidence(t *testing.T) {
f := newRecoveryFixture(t)
f.source("build/100", "100", entity.BuildStatusSucceeded)
f.source("build/101", "101", entity.BuildStatusRunning)
f.reportIssue("build/101", "new", "unit")
require.NoError(t, f.deliver(f.message("build/101", false), false))
newer := f.states[issueKey{"new", "unit"}]
require.NoError(t, f.recover(f.message("build/100", true), false))
require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase)
require.Equal(t, "build/100", f.progress.BuildID)
require.Equal(t, newer, f.states[issueKey{"new", "unit"}])
require.Len(t, f.events, 1)
require.NoError(t, f.deliver(f.message("build/101", false), false))
before := f.progress
require.NoError(t, f.recover(f.message("build/100", true), false))
require.Equal(t, before, f.progress)
}
Loading