Skip to content

Commit 8e7fea2

Browse files
prathshenoyprathshenoy
committed
feat(stovepipe): Persist build issue observations
1 parent 3c505ea commit 8e7fea2

16 files changed

Lines changed: 490 additions & 1 deletion

File tree

‎stovepipe/extension/storage/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"build_issue_store.go",
67
"build_store.go",
78
"queue_log_store.go",
89
"queue_store.go",

‎stovepipe/extension/storage/README.md‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Storage
22

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

55
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.
66

@@ -23,3 +23,11 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
2323
`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.
2424

2525
`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).
26+
27+
## Build issue observations
28+
29+
`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.
30+
31+
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.
32+
33+
The MySQL backend retains the original task and unit identifiers in binary payload columns and uses their SHA-256 digests in the primary key to avoid index-length and collation constraints. It assumes collision resistance for identity. Queue and build identifiers are byte-exact and each limited to 1020 bytes; task and unit identifiers are not truncated to fit the index.

‎stovepipe/extension/storage/build_issue_store.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package storage
16+
17+
//go:generate mockgen -source=build_issue_store.go -destination=mock/build_issue_store_mock.go -package=mock
18+
19+
import (
20+
"context"
21+
22+
"github.com/uber/submitqueue/stovepipe/entity"
23+
)
24+
25+
// BuildIssueStore retains immutable observations keyed by build, task, and unit
26+
// within the bound queue. Presence records an observation, not a currently open issue.
27+
type BuildIssueStore interface {
28+
// Create returns ErrAlreadyExists for a previously recorded build/task/unit.
29+
// Callers may treat that duplicate as success when accumulating observations.
30+
Create(ctx context.Context, buildID string, issue entity.BuildIssue) error
31+
32+
// Get returns ErrNotFound when the build/task/unit has not been recorded.
33+
Get(ctx context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error)
34+
}

‎stovepipe/extension/storage/mock/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"build_issue_store_mock.go",
67
"build_store_mock.go",
78
"queue_log_store_mock.go",
89
"queue_store_mock.go",

‎stovepipe/extension/storage/mock/build_issue_store_mock.go‎

Lines changed: 71 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎stovepipe/extension/storage/mock/storage_mock.go‎

Lines changed: 14 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎stovepipe/extension/storage/mysql/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"build_issue_store.go",
67
"build_store.go",
78
"queue_log_store.go",
89
"queue_store.go",
@@ -29,6 +30,7 @@ go_library(
2930
go_test(
3031
name = "go_default_test",
3132
srcs = [
33+
"build_issue_store_test.go",
3234
"build_store_test.go",
3335
"queue_store_test.go",
3436
"queue_summary_store_test.go",

‎stovepipe/extension/storage/mysql/build_issue_store.go‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package mysql
16+
17+
import (
18+
"context"
19+
"crypto/sha256"
20+
"database/sql"
21+
"errors"
22+
"fmt"
23+
24+
"github.com/uber-go/tally"
25+
26+
"github.com/uber/submitqueue/platform/metrics"
27+
"github.com/uber/submitqueue/stovepipe/entity"
28+
"github.com/uber/submitqueue/stovepipe/extension/storage"
29+
)
30+
31+
type buildIssueStore struct {
32+
db *sql.DB
33+
scope tally.Scope
34+
queue string
35+
}
36+
37+
// NewBuildIssueStore binds observations to queue over the shared database.
38+
func NewBuildIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.BuildIssueStore {
39+
return &buildIssueStore{db: db, scope: scope, queue: queue}
40+
}
41+
42+
func (s *buildIssueStore) Create(ctx context.Context, buildID string, issue entity.BuildIssue) (retErr error) {
43+
op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
44+
defer func() { op.Complete(retErr) }()
45+
46+
taskHash := sha256.Sum256([]byte(issue.TaskID))
47+
unitHash := sha256.Sum256([]byte(issue.UnitID))
48+
_, err := s.db.ExecContext(ctx,
49+
`INSERT INTO build_issue (queue, build_id, task_hash, unit_hash, task_id, unit_id)
50+
VALUES (?, ?, ?, ?, ?, ?)`,
51+
s.queue, buildID, taskHash[:], unitHash[:], issue.TaskID, issue.UnitID,
52+
)
53+
if err != nil {
54+
if isDuplicateEntry(err) {
55+
return fmt.Errorf("build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, storage.ErrAlreadyExists)
56+
}
57+
return fmt.Errorf("failed to create build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, err)
58+
}
59+
return nil
60+
}
61+
62+
func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID string) (ret entity.BuildIssue, retErr error) {
63+
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
64+
defer func() { op.Complete(retErr) }()
65+
66+
taskHash := sha256.Sum256([]byte(taskID))
67+
unitHash := sha256.Sum256([]byte(unitID))
68+
var issue entity.BuildIssue
69+
err := s.db.QueryRowContext(ctx,
70+
`SELECT task_id, unit_id FROM build_issue
71+
WHERE queue = ? AND build_id = ? AND task_hash = ? AND unit_hash = ?`,
72+
s.queue, buildID, taskHash[:], unitHash[:],
73+
).Scan(&issue.TaskID, &issue.UnitID)
74+
if errors.Is(err, sql.ErrNoRows) {
75+
return entity.BuildIssue{}, storage.WrapNotFound(err)
76+
}
77+
if err != nil {
78+
return entity.BuildIssue{}, fmt.Errorf("failed to get build issue build=%q task=%q unit=%q: %w", buildID, taskID, unitID, err)
79+
}
80+
return issue, nil
81+
}

‎stovepipe/extension/storage/mysql/build_issue_store_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package mysql
16+
17+
import (
18+
"context"
19+
"crypto/sha256"
20+
"database/sql"
21+
"errors"
22+
"regexp"
23+
"strings"
24+
"testing"
25+
26+
"github.com/DATA-DOG/go-sqlmock"
27+
"github.com/go-sql-driver/mysql"
28+
"github.com/stretchr/testify/require"
29+
30+
"github.com/uber/submitqueue/stovepipe/entity"
31+
"github.com/uber/submitqueue/stovepipe/extension/storage"
32+
)
33+
34+
func TestBuildIssueStoreCreate(t *testing.T) {
35+
dbErr := errors.New("database unavailable")
36+
for _, tt := range []struct {
37+
name string
38+
unitID string
39+
dbErr error
40+
wantErr error
41+
}{
42+
{name: "unit observation", unitID: "unit"},
43+
{name: "task-wide observation"},
44+
{name: "long unit identifier", unitID: strings.Repeat("unit/", 1000)},
45+
{name: "duplicate observation", unitID: "unit", dbErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists},
46+
{name: "dependency error", unitID: "unit", dbErr: dbErr, wantErr: dbErr},
47+
} {
48+
t.Run(tt.name, func(t *testing.T) {
49+
db, mock, err := sqlmock.New()
50+
require.NoError(t, err)
51+
defer db.Close()
52+
store := NewBuildIssueStore(db, testMetrics(), "queue")
53+
issue := entity.BuildIssue{TaskID: "task", UnitID: tt.unitID}
54+
taskHash := sha256.Sum256([]byte(issue.TaskID))
55+
unitHash := sha256.Sum256([]byte(issue.UnitID))
56+
expectation := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO build_issue (queue, build_id, task_hash, unit_hash, task_id, unit_id) VALUES (?, ?, ?, ?, ?, ?)`)).
57+
WithArgs("queue", "build", taskHash[:], unitHash[:], issue.TaskID, issue.UnitID)
58+
if tt.dbErr != nil {
59+
expectation.WillReturnError(tt.dbErr)
60+
} else {
61+
expectation.WillReturnResult(sqlmock.NewResult(0, 1))
62+
}
63+
64+
err = store.Create(context.Background(), "build", issue)
65+
if tt.wantErr != nil {
66+
require.ErrorIs(t, err, tt.wantErr)
67+
} else {
68+
require.NoError(t, err)
69+
}
70+
require.NoError(t, mock.ExpectationsWereMet())
71+
})
72+
}
73+
}
74+
75+
func TestBuildIssueStoreGet(t *testing.T) {
76+
dbErr := errors.New("database unavailable")
77+
for _, tt := range []struct {
78+
name string
79+
unitID string
80+
queryErr error
81+
scanErr bool
82+
wantErr error
83+
}{
84+
{name: "unit observation", unitID: "unit"},
85+
{name: "task-wide observation"},
86+
{name: "missing observation", unitID: "unit", queryErr: sql.ErrNoRows, wantErr: storage.ErrNotFound},
87+
{name: "dependency error", unitID: "unit", queryErr: dbErr, wantErr: dbErr},
88+
{name: "scan error", unitID: "unit", scanErr: true},
89+
} {
90+
t.Run(tt.name, func(t *testing.T) {
91+
db, mock, err := sqlmock.New()
92+
require.NoError(t, err)
93+
defer db.Close()
94+
store := NewBuildIssueStore(db, testMetrics(), "queue")
95+
issue := entity.BuildIssue{TaskID: "task", UnitID: tt.unitID}
96+
taskHash := sha256.Sum256([]byte(issue.TaskID))
97+
unitHash := sha256.Sum256([]byte(issue.UnitID))
98+
expectation := mock.ExpectQuery(regexp.QuoteMeta(`SELECT task_id, unit_id FROM build_issue WHERE queue = ? AND build_id = ? AND task_hash = ? AND unit_hash = ?`)).
99+
WithArgs("queue", "build", taskHash[:], unitHash[:])
100+
if tt.queryErr != nil {
101+
expectation.WillReturnError(tt.queryErr)
102+
} else if tt.scanErr {
103+
expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(nil, issue.UnitID))
104+
} else {
105+
expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(issue.TaskID, issue.UnitID))
106+
}
107+
108+
got, err := store.Get(context.Background(), "build", issue.TaskID, issue.UnitID)
109+
if tt.wantErr != nil {
110+
require.ErrorIs(t, err, tt.wantErr)
111+
} else if tt.scanErr {
112+
require.Error(t, err)
113+
} else {
114+
require.NoError(t, err)
115+
require.Equal(t, issue, got)
116+
}
117+
if tt.wantErr != nil || tt.scanErr {
118+
require.Equal(t, entity.BuildIssue{}, got)
119+
}
120+
require.NoError(t, mock.ExpectationsWereMet())
121+
})
122+
}
123+
}

‎stovepipe/extension/storage/mysql/schema/build_issue.sql‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
-- One immutable observation per (queue, build, task, unit), not an open-issue projection.
2+
-- Hashes keep arbitrary-length task/unit identifiers out of the InnoDB primary key;
3+
-- original identifiers are retained separately without collation or truncation.
4+
CREATE TABLE IF NOT EXISTS build_issue (
5+
queue VARBINARY(1020) NOT NULL,
6+
build_id VARBINARY(1020) NOT NULL,
7+
task_hash BINARY(32) NOT NULL,
8+
unit_hash BINARY(32) NOT NULL,
9+
task_id LONGBLOB NOT NULL,
10+
unit_id LONGBLOB NOT NULL,
11+
PRIMARY KEY (queue, build_id, task_hash, unit_hash)
12+
) ENGINE=InnoDB;

0 commit comments

Comments
 (0)