Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
36 changes: 36 additions & 0 deletions stovepipe/controller/issuetrack/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
load("@rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "go_default_library",
srcs = ["issuetrack.go"],
importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack",
visibility = ["//visibility:public"],
deps = [
"//platform/consumer:go_default_library",
"//platform/metrics:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/storage:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
)

go_test(
name = "go_default_test",
srcs = [
"issuetrack_test.go",
"planned_change_race_test.go",
],
embed = [":go_default_library"],
deps = [
"//stovepipe/core/messagequeue:go_default_library",
"//stovepipe/entity:go_default_library",
"//stovepipe/extension/storage:go_default_library",
"//stovepipe/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
)
13 changes: 13 additions & 0 deletions stovepipe/controller/issuetrack/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# Issue tracking

## Applying a planned change

The controller applies one `PlannedIssueChange` already saved in the queue's progress record. The cursor identifies the issue; the change holds only its write details. Expected issue version zero means create-if-absent. A positive version means update that exact version. A retry never adopts a newly loaded version to execute old instructions.

After an issue write succeeds, the controller writes the opening or closing history event, then clears the planned change while retaining the cursor using progress CAS. If an issue write already happened, a retry verifies the exact intended issue state and version before repairing history. The saved timestamp makes duplicate history stable. Failure leaves the original planned change available for another worker to finish.

A replacement must finish that same planned change before accepting later work. Closed issue rows retain their versions. Together, these rules prevent a paused old worker from reopening or closing an issue after newer work: its create-if-absent or original version guard fails. A conflicting issue state is not treated as a successful replay.

All writes use complete primary keys. There are no cross-record transactions, leases, snapshot indexes, or additional operation counters. Records and versions must not be deleted or reset while an old delivery could resume. All issue-state writers must follow this protocol.

This layer does not scan build issues, receive queue messages, publish repairs, or register a runtime consumer. Planning, delivery orchestration, and terminal recovery are separate additions.
117 changes: 117 additions & 0 deletions stovepipe/controller/issuetrack/issuetrack.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,117 @@
// Copyright (c) 2026 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package issuetrack maintains issue state and lifecycle history from build issues.
package issuetrack

import (
"context"
"errors"
"fmt"
"math"

"github.com/uber-go/tally"
"github.com/uber/submitqueue/platform/consumer"
"github.com/uber/submitqueue/platform/metrics"
"github.com/uber/submitqueue/stovepipe/entity"
"github.com/uber/submitqueue/stovepipe/extension/storage"
"go.uber.org/zap"
)

// Controller retains issue transitions and their queue-scoped progress.
type Controller struct {
logger *zap.SugaredLogger
metricsScope tally.Scope
stores storage.Factory
topicKey consumer.TopicKey
consumerGroup string
}

func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string) *Controller {
return &Controller{
logger: logger.Named("issuetrack_controller"),
metricsScope: scope.SubScope("issuetrack_controller"),
stores: stores,
topicKey: topicKey,
consumerGroup: consumerGroup,
}
}

func (c *Controller) applyPlannedChange(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) {
op := metrics.Begin(c.metricsScope, "apply_change", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
defer func() { op.Complete(retErr) }()

if err := ctx.Err(); err != nil {
return entity.IssueTrackingState{}, err
}
if err := progress.Validate(); err != nil {
return entity.IssueTrackingState{}, err
}
if !progress.HasPlannedChange() {
return entity.IssueTrackingState{}, fmt.Errorf("issue tracking has no saved change")
}
if progress.Version == math.MaxInt32 {
return entity.IssueTrackingState{}, fmt.Errorf("issue tracking version exhausted")
}

change := progress.PlannedChange
target := entity.IssueState{
TaskID: progress.CursorTaskID,
UnitID: progress.CursorUnitID,
Status: change.Status,
FirstSeenBuildID: change.FirstSeenBuildID,
Version: change.ExpectedVersion + 1,
}
issues := store.GetIssueStateStore()
var err error
if change.ExpectedVersion == 0 {
err = issues.Create(ctx, target)
} else {
err = issues.Update(ctx, target, change.ExpectedVersion, target.Version)
}
if errors.Is(err, storage.ErrAlreadyExists) || errors.Is(err, storage.ErrVersionMismatch) {
retained, readErr := issues.Get(ctx, progress.CursorTaskID, progress.CursorUnitID)
if readErr != nil {
return entity.IssueTrackingState{}, fmt.Errorf("read planned issue change result: %w", readErr)
}
// A retry may finish history, but must never rebase the planned change onto a newer version.
if retained != target {
return entity.IssueTrackingState{}, fmt.Errorf("planned issue change has been superseded: %w", storage.ErrVersionMismatch)
}
} else if err != nil {
return entity.IssueTrackingState{}, fmt.Errorf("apply planned issue change: %w", err)
}

event := entity.IssueEventOpened
if change.Status == entity.IssueStatusClosed {
event = entity.IssueEventResolved
}
log := entity.IssueLog{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID, BuildID: progress.BuildID, Event: event, TimestampMs: change.TimestampMs}
if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
return entity.IssueTrackingState{}, fmt.Errorf("record issue transition: %w", err)
}

next := progress
next.PlannedChange = entity.PlannedIssueChange{}
newVersion := progress.Version + 1
if err := store.GetIssueTrackingStore().Update(ctx, next, progress.Version, newVersion); err != nil {
return entity.IssueTrackingState{}, fmt.Errorf("complete planned issue change: %w", err)
}
next.Version = newVersion
return next, nil
}

func (c *Controller) Name() string { return "issuetrack" }
func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
func (c *Controller) ConsumerGroup() string { return c.consumerGroup }
Loading