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
1 change: 1 addition & 0 deletions stovepipe/extension/storage/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
"build_issue_store.go",
"build_store.go",
"queue_log_store.go",
"queue_store.go",
Expand Down
10 changes: 9 additions & 1 deletion stovepipe/extension/storage/README.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
# Storage

Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`.
Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`). Implementations live under `extension/storage/<impl>/`.

The aggregate is resolved per queue through a factory keyed by queue name, mirroring the extension contract and the [submitqueue storage seam](../../../submitqueue/extension/storage/README.md#queue-scoped-resolution): a resolved instance is bound to its queue, entity arguments whose queue disagrees with the binding are rejected, and reads never surface another queue's records. Stovepipe has no cross-queue read paths, so every store — including `QueueStore`, whose row key is the queue name itself — lives inside the queue-scoped aggregate; there is no global remainder.

Expand All @@ -23,3 +23,11 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
`RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate.

`QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md).

## Build issue observations

`BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Empty unit IDs represent task-wide observations.

Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or resolved. No update, deletion, or absence-based recovery is implied by this store.

The MySQL primary key contains the queue, build ID, task ID, and unit ID directly as binary strings, preserving byte-exact identity. Their limits are 255, 1020, 255, and 1024 bytes respectively, for a combined maximum of 2554 bytes. Creates and reads reject identifiers exceeding these limits before querying the database; identifiers are never truncated.
34 changes: 34 additions & 0 deletions stovepipe/extension/storage/build_issue_store.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
// 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 storage

//go:generate mockgen -source=build_issue_store.go -destination=mock/build_issue_store_mock.go -package=mock

import (
"context"

"github.com/uber/submitqueue/stovepipe/entity"
)

// BuildIssueStore retains immutable observations keyed by build, task, and unit
// within the bound queue. Presence records an observation, not a currently open issue.
type BuildIssueStore interface {
// Create returns ErrAlreadyExists for a previously recorded build/task/unit.
// Callers may treat that duplicate as success when accumulating observations.
Create(ctx context.Context, buildID string, issue entity.BuildIssue) error

// Get returns ErrNotFound when the build/task/unit has not been recorded.
Get(ctx context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error)
}
1 change: 1 addition & 0 deletions stovepipe/extension/storage/mock/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = [
"build_issue_store_mock.go",
"build_store_mock.go",
"queue_log_store_mock.go",
"queue_store_mock.go",
Expand Down
71 changes: 71 additions & 0 deletions stovepipe/extension/storage/mock/build_issue_store_mock.go

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

14 changes: 14 additions & 0 deletions stovepipe/extension/storage/mock/storage_mock.go

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

2 changes: 2 additions & 0 deletions stovepipe/extension/storage/mysql/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
"build_issue_store.go",
"build_store.go",
"queue_log_store.go",
"queue_store.go",
Expand All @@ -29,6 +30,7 @@ go_library(
go_test(
name = "go_default_test",
srcs = [
"build_issue_store_test.go",
"build_store_test.go",
"queue_store_test.go",
"queue_summary_store_test.go",
Expand Down
100 changes: 100 additions & 0 deletions stovepipe/extension/storage/mysql/build_issue_store.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
// 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"
"errors"
"fmt"

"github.com/uber-go/tally"

"github.com/uber/submitqueue/platform/metrics"
"github.com/uber/submitqueue/stovepipe/entity"
"github.com/uber/submitqueue/stovepipe/extension/storage"
)

type buildIssueStore struct {
db *sql.DB
scope tally.Scope
queue string
}

// NewBuildIssueStore binds observations to queue over the shared database.
func NewBuildIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.BuildIssueStore {
return &buildIssueStore{db: db, scope: scope, queue: queue}
}

func (s *buildIssueStore) Create(ctx context.Context, buildID string, issue entity.BuildIssue) (retErr error) {
op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

if err := validateBuildIssueKey(s.queue, buildID, issue.TaskID, issue.UnitID); err != nil {
return err
}
_, err := s.db.ExecContext(ctx,
`INSERT INTO build_issue (queue, build_id, task_id, unit_id)
VALUES (?, ?, ?, ?)`,
s.queue, buildID, issue.TaskID, issue.UnitID,
)
if err != nil {
if isDuplicateEntry(err) {
return fmt.Errorf("build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, storage.ErrAlreadyExists)
}
return fmt.Errorf("failed to create build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, err)
}
return nil
}

func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID string) (ret entity.BuildIssue, retErr error) {
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

if err := validateBuildIssueKey(s.queue, buildID, taskID, unitID); err != nil {
return entity.BuildIssue{}, err
}
var issue entity.BuildIssue
err := s.db.QueryRowContext(ctx,
`SELECT task_id, unit_id FROM build_issue
WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ?`,
s.queue, buildID, taskID, unitID,
).Scan(&issue.TaskID, &issue.UnitID)
if errors.Is(err, sql.ErrNoRows) {
return entity.BuildIssue{}, storage.WrapNotFound(err)
}
if err != nil {
return entity.BuildIssue{}, fmt.Errorf("failed to get build issue build=%q task=%q unit=%q: %w", buildID, taskID, unitID, err)
}
return issue, nil
}

func validateBuildIssueKey(queue, buildID, taskID, unitID string) error {
for _, field := range []struct {
name string
value string
maxBytes int
}{
{"queue", queue, 255},
{"build ID", buildID, 1020},
{"task ID", taskID, 255},
{"unit ID", unitID, 1024},
} {
if len(field.value) > field.maxBytes {
return fmt.Errorf("build issue %s exceeds %d-byte limit", field.name, field.maxBytes)
}
}
return nil
}
Loading