Skip to content
Merged
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
2 changes: 1 addition & 1 deletion stovepipe/entity/open_issue.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ package entity
// Its task and unit identify the condition; the first-seen build identifies
// this opening and is immutable until removal.
type OpenIssue struct {
// TaskID identifies the logical validation task.
// TaskID identifies the logical validation task and must be nonempty.
TaskID string `json:"task_id"`
// UnitID identifies the validation unit; empty for a task-wide issue.
UnitID string `json:"unit_id"`
Expand Down
8 changes: 6 additions & 2 deletions stovepipe/extension/storage/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -36,13 +36,17 @@ The MySQL primary key contains the queue, build ID, task ID, and unit ID directl

## Currently open issues

`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes immutable per-build observations. There is no open/resolved flag; a resolved issue has no row here.
`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes immutable per-build observations. Task IDs must be nonempty; an empty unit identifies a task-wide issue. There is no open/resolved flag; a resolved issue has no row here.

An open row is immutable. Its first-seen build identifies the current opening, not its latest observation. Repeated observations remain in the per-build store. Reopening after removal uses a different first-seen build ID; no last-observed build, version counter, or update operation is stored here.

Listing scans the bound queue's primary-key prefix in ascending bytewise task/unit order, returning at most the requested positive limit. Continue after the last returned task/unit key; the zero cursor starts the range. A cursor with a unit but no task is invalid. Empty units remain task-wide issues and can be used as continuation boundaries. No secondary index or schema change is required.

Pages are not a snapshot across calls. Concurrent changes can affect which rows appear, and the cursor row need not still exist. Readers must not treat listing as an atomic replacement or resolution decision; guarded deletion still protects a later opening.
Comment thread
mnoah1 marked this conversation as resolved.

Deletion matches the task, unit, and first-seen build within the bound queue. It is safe to repeat: an absent opening is a no-op, and a stale deletion never removes a different opening of the same condition. Database failures are returned for retry.

MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, enumerate issues, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store.
MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store.

## Issue lifecycle history

Expand Down
16 changes: 16 additions & 0 deletions stovepipe/extension/storage/mock/open_issue_store_mock.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions stovepipe/extension/storage/mysql/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ go_test(
"build_issue_store_test.go",
"build_store_test.go",
"issue_log_store_test.go",
"open_issue_list_test.go",
"open_issue_store_test.go",
"project_fact_by_degree_store_test.go",
"queue_store_test.go",
Expand Down
134 changes: 134 additions & 0 deletions stovepipe/extension/storage/mysql/open_issue_list_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
// 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 mysql

import (
"context"
"database/sql/driver"
"regexp"
"strings"
"testing"

"github.com/DATA-DOG/go-sqlmock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/uber/submitqueue/stovepipe/entity"
"github.com/uber/submitqueue/stovepipe/extension/storage"
)

func TestOpenIssueList(t *testing.T) {
for _, tt := range []struct {
name string
after storage.OpenIssueCursor
empty bool
dbErr error
scanError bool
rowError bool
}{
{name: "first page"},
{name: "after task-wide issue", after: storage.OpenIssueCursor{TaskID: "task"}},
{name: "empty page", empty: true},
{name: "query error", dbErr: assert.AnError},
{name: "cancelled", dbErr: context.Canceled},
{name: "scan error", scanError: true},
{name: "iteration error", rowError: true},
} {
t.Run(tt.name, func(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)
defer db.Close()
store := NewOpenIssueStore(db, testMetrics(), "queue")
issue := openIssueFixture()
query := listOpenIssuesQuery
args := []driver.Value{"queue", int64(2)}
if tt.after.TaskID != "" {
query = listOpenIssuesAfterCursorQuery
args = []driver.Value{"queue", tt.after.TaskID, tt.after.TaskID, tt.after.UnitID, int64(2)}
}
expectation := mock.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(args...)
if tt.dbErr != nil {
expectation.WillReturnError(tt.dbErr)
} else {
rows := sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"})
if !tt.empty {
var firstBuild any = issue.FirstSeenBuildID
if tt.scanError {
firstBuild = nil
}
rows.AddRow(issue.TaskID, issue.UnitID, firstBuild)
if tt.rowError {
rows.AddRow("task", "unit-2", "build/2").RowError(1, assert.AnError)
}
}
expectation.WillReturnRows(rows).RowsWillBeClosed()
}

got, err := store.List(context.Background(), storage.OpenIssueRange{After: tt.after, Limit: 2})
switch {
case tt.dbErr != nil:
require.ErrorIs(t, err, tt.dbErr)
require.Nil(t, got)
case tt.scanError:
require.Error(t, err)
require.Nil(t, got)
case tt.rowError:
require.ErrorIs(t, err, assert.AnError)
require.Nil(t, got)
case tt.empty:
require.NoError(t, err)
require.NotNil(t, got)
require.Empty(t, got)
default:
require.NoError(t, err)
require.Equal(t, []entity.OpenIssue{issue}, got)
}
require.NoError(t, mock.ExpectationsWereMet())
})
}
}

func TestOpenIssueListRejectsInvalidRange(t *testing.T) {
for _, tt := range []struct {
name string
queue string
bounds storage.OpenIssueRange
}{
{name: "zero limit", bounds: storage.OpenIssueRange{}},
{name: "negative limit", bounds: storage.OpenIssueRange{Limit: -1}},
{name: "unit without task", bounds: storage.OpenIssueRange{Limit: 1,
After: storage.OpenIssueCursor{UnitID: "unit"}}},
{name: "oversized task", bounds: storage.OpenIssueRange{Limit: 1,
After: storage.OpenIssueCursor{TaskID: strings.Repeat("t", 256)}}},
{name: "oversized unit", bounds: storage.OpenIssueRange{Limit: 1,
After: storage.OpenIssueCursor{TaskID: "task", UnitID: strings.Repeat("😀", 257)}}},
{name: "oversized queue", queue: strings.Repeat("q", 256), bounds: storage.OpenIssueRange{Limit: 1}},
} {
t.Run(tt.name, func(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)
defer db.Close()
queue := tt.queue
if queue == "" {
queue = "queue"
}
store := NewOpenIssueStore(db, testMetrics(), queue)
mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}))

got, err := store.List(context.Background(), tt.bounds)
require.Error(t, err)
require.Nil(t, got)
})
}
}
62 changes: 62 additions & 0 deletions stovepipe/extension/storage/mysql/open_issue_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,17 @@ import (
"github.com/uber/submitqueue/stovepipe/extension/storage"
)

const listOpenIssuesQuery = `
SELECT task_id, unit_id, first_seen_build_id
FROM open_issue WHERE queue = ?
ORDER BY task_id ASC, unit_id ASC LIMIT ?`

const listOpenIssuesAfterCursorQuery = `
SELECT task_id, unit_id, first_seen_build_id
FROM open_issue WHERE queue = ?
AND (task_id > ? OR (task_id = ? AND unit_id > ?))
ORDER BY task_id ASC, unit_id ASC LIMIT ?`

type openIssueStore struct {
db *sql.DB
scope tally.Scope
Expand Down Expand Up @@ -64,6 +75,9 @@ func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret en
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

if taskID == "" {
return entity.OpenIssue{}, fmt.Errorf("open issue task ID must not be empty")
}
if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil {
return entity.OpenIssue{}, err
}
Expand All @@ -84,6 +98,51 @@ func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret en
return issue, nil
}

func (s *openIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) (ret []entity.OpenIssue, retErr error) {
op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

cursor := bounds.After
if bounds.Limit <= 0 {
return nil, fmt.Errorf("open issue range requires a positive limit")
}
if cursor.TaskID == "" && cursor.UnitID != "" {
return nil, fmt.Errorf("open issue cursor requires a task ID when a unit ID is supplied")
}
if err := validateBuildIssueKey(s.queue, "", cursor.TaskID, cursor.UnitID); err != nil {
return nil, err
}

var (
rows *sql.Rows
err error
)
if cursor.TaskID != "" {
rows, err = s.db.QueryContext(ctx, listOpenIssuesAfterCursorQuery,
s.queue, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit)
} else {
rows, err = s.db.QueryContext(ctx, listOpenIssuesQuery, s.queue, bounds.Limit)
}
if err != nil {
return nil, fmt.Errorf("failed to list open issues: %w", err)
}
defer rows.Close()

issues := make([]entity.OpenIssue, 0)
for rows.Next() {
var issue entity.OpenIssue
if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID); err != nil {
return nil, fmt.Errorf("failed to scan open issue: %w", err)
}
issues = append(issues, issue)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("failed to iterate open issues: %w", err)
}

return issues, nil
}

func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (retErr error) {
op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()
Expand All @@ -104,6 +163,9 @@ func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (re
}

func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error {
if issue.TaskID == "" {
return fmt.Errorf("open issue task ID must not be empty")
}
if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil {
return err
}
Expand Down
6 changes: 4 additions & 2 deletions stovepipe/extension/storage/mysql/open_issue_store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ func TestOpenIssueDelete(t *testing.T) {
}

func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
for _, kind := range []string{"queue", "task", "unit", "first build", "empty first build"} {
for _, kind := range []string{"queue", "task", "empty task", "unit", "first build", "empty first build"} {
t.Run(kind, func(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)
Expand All @@ -179,6 +179,8 @@ func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
queue = strings.Repeat("q", 256)
case "task":
issue.TaskID = strings.Repeat("t", 256)
case "empty task":
issue.TaskID = ""
case "unit":
issue.UnitID = strings.Repeat("😀", 257)
case "first build":
Expand All @@ -200,7 +202,7 @@ func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
require.Error(t, store.Delete(context.Background(), issue))

switch kind {
case "queue", "task", "unit":
case "queue", "task", "empty task", "unit":
_, err = store.Get(context.Background(), issue.TaskID, issue.UnitID)
require.Error(t, err)
}
Expand Down
22 changes: 22 additions & 0 deletions stovepipe/extension/storage/open_issue_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,25 @@ import (
"github.com/uber/submitqueue/stovepipe/entity"
)

// OpenIssueCursor is an exclusive position in ascending bytewise task/unit order.
type OpenIssueCursor struct {
// TaskID is the task component; empty with an empty UnitID starts the range.
TaskID string
// UnitID is the unit component of the last returned key; empty is a task-wide issue.
UnitID string
}

// OpenIssueRange bounds a primary-key scan within the store's queue.
type OpenIssueRange struct {
// After excludes this key and earlier keys; the zero cursor starts the range.
After OpenIssueCursor
// Limit is the positive maximum number of issues to return.
Limit int
}

// OpenIssueStore contains only currently open issues, keyed by task and unit
// within the bound queue. A reopened issue must use a new first-seen build ID.
// Issue keys require a nonempty task ID; an empty unit is a task-wide issue.
type OpenIssueStore interface {
// Create requires a nonempty first-seen build ID.
// An existing task/unit returns ErrAlreadyExists.
Expand All @@ -32,6 +49,11 @@ type OpenIssueStore interface {
// Get returns ErrNotFound when the task/unit is not currently open.
Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error)

// List returns at most Limit issues in ascending bytewise (task, unit) order.
// An empty slice ends the range; invalid bounds return an error.
// Pages are not a snapshot: concurrent creates/deletes may change membership.
List(ctx context.Context, bounds OpenIssueRange) ([]entity.OpenIssue, error)

// Delete removes only the matching opening. It succeeds if that opening is
// already absent and never removes a row with a different first-seen build.
Delete(ctx context.Context, issue entity.OpenIssue) error
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ go_test(
srcs = [
"build_issue_test.go",
"issue_log_test.go",
"open_issue_list_test.go",
"open_issue_test.go",
"project_fact_by_degree_test.go",
"queue_state_transitions_test.go",
Expand Down
Loading
Loading