Skip to content

Commit 6feb948

Browse files
mnoah1mnoah1
committed
feat(stovepipe): persist first-failure repair submissions
1 parent 7f695f7 commit 6feb948

11 files changed

Lines changed: 514 additions & 0 deletions

File tree

‎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
@@ -5,6 +5,7 @@ go_library(
55
srcs = [
66
"build_issue_store.go",
77
"build_store.go",
8+
"fix_store.go",
89
"open_issue_store.go",
910
"queue_log_store.go",
1011
"queue_store.go",
@@ -33,6 +34,7 @@ go_test(
3334
srcs = [
3435
"build_issue_store_test.go",
3536
"build_store_test.go",
37+
"fix_store_test.go",
3638
"open_issue_store_test.go",
3739
"queue_store_test.go",
3840
"queue_summary_store_test.go",

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

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,172 @@
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+
"database/sql"
20+
"encoding/json"
21+
"errors"
22+
"fmt"
23+
24+
"github.com/uber-go/tally"
25+
"github.com/uber/submitqueue/platform/metrics"
26+
"github.com/uber/submitqueue/stovepipe/entity"
27+
"github.com/uber/submitqueue/stovepipe/extension/storage"
28+
)
29+
30+
type fixStore struct {
31+
db *sql.DB
32+
scope tally.Scope
33+
queue string
34+
}
35+
36+
// NewFixStore creates a MySQL-backed, queue-scoped FixStore.
37+
func NewFixStore(db *sql.DB, scope tally.Scope, queue string) storage.FixStore {
38+
return &fixStore{db: db, scope: scope, queue: queue}
39+
}
40+
41+
func (s *fixStore) Create(ctx context.Context, fix entity.Fix) (retErr error) {
42+
op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
43+
defer func() { op.Complete(retErr) }()
44+
45+
if fix.Queue != s.queue || fix.ID == "" {
46+
return fmt.Errorf("repair requires a nonempty ID and queue matching %q", s.queue)
47+
}
48+
issuesJSON, metadataJSON, err := marshalFixJSON(fix)
49+
if err != nil {
50+
return fmt.Errorf("failed to marshal repair data: %w", err)
51+
}
52+
53+
_, err = s.db.ExecContext(ctx,
54+
`INSERT INTO fix (queue, id, build_id, state, issues, execution_ref, metadata, created_at, updated_at, version)
55+
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
56+
s.queue,
57+
fix.ID,
58+
fix.BuildID,
59+
fix.State,
60+
issuesJSON,
61+
fix.ExecutionRef,
62+
metadataJSON,
63+
fix.CreatedAt,
64+
fix.UpdatedAt,
65+
fix.Version,
66+
)
67+
if err != nil {
68+
if isDuplicateEntry(err) {
69+
return fmt.Errorf("fix queue=%q: %w", s.queue, storage.ErrAlreadyExists)
70+
}
71+
return fmt.Errorf("failed to insert fix queue=%q: %w", s.queue, err)
72+
}
73+
return nil
74+
}
75+
76+
func (s *fixStore) Get(ctx context.Context, id string) (ret entity.Fix, retErr error) {
77+
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
78+
defer func() { op.Complete(retErr) }()
79+
80+
var fix entity.Fix
81+
fix.Queue = s.queue
82+
var issuesJSON []byte
83+
var metadataJSON []byte
84+
err := s.db.QueryRowContext(ctx,
85+
`SELECT id, build_id, state, issues, execution_ref, metadata, created_at, updated_at, version FROM fix
86+
WHERE queue = ? AND id = ?`,
87+
s.queue,
88+
id,
89+
).Scan(
90+
&fix.ID,
91+
&fix.BuildID,
92+
&fix.State,
93+
&issuesJSON,
94+
&fix.ExecutionRef,
95+
&metadataJSON,
96+
&fix.CreatedAt,
97+
&fix.UpdatedAt,
98+
&fix.Version,
99+
)
100+
if errors.Is(err, sql.ErrNoRows) {
101+
return entity.Fix{}, storage.WrapNotFound(err)
102+
}
103+
if err != nil {
104+
return entity.Fix{}, fmt.Errorf("failed to get fix queue=%q: %w", s.queue, err)
105+
}
106+
if err := json.Unmarshal(issuesJSON, &fix.Issues); err != nil {
107+
return entity.Fix{}, fmt.Errorf("failed to unmarshal repair issues: %w", err)
108+
}
109+
if err := json.Unmarshal(metadataJSON, &fix.Metadata); err != nil {
110+
return entity.Fix{}, fmt.Errorf("failed to unmarshal repair metadata: %w", err)
111+
}
112+
if fix.Metadata == nil {
113+
fix.Metadata = map[string]string{}
114+
}
115+
return fix, nil
116+
}
117+
118+
func (s *fixStore) Update(ctx context.Context, fix entity.Fix, oldVersion, newVersion int32) (retErr error) {
119+
op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
120+
defer func() { op.Complete(retErr) }()
121+
122+
if fix.Queue != s.queue || fix.ID == "" {
123+
return fmt.Errorf("repair requires a nonempty ID and queue matching %q", s.queue)
124+
}
125+
issuesJSON, metadataJSON, err := marshalFixJSON(fix)
126+
if err != nil {
127+
return fmt.Errorf("failed to marshal repair data: %w", err)
128+
}
129+
130+
result, err := s.db.ExecContext(ctx,
131+
`UPDATE fix SET build_id = ?, state = ?, issues = ?, execution_ref = ?, metadata = ?, created_at = ?, updated_at = ?, version = ?
132+
WHERE queue = ? AND id = ? AND version = ?`,
133+
fix.BuildID,
134+
fix.State,
135+
issuesJSON,
136+
fix.ExecutionRef,
137+
metadataJSON,
138+
fix.CreatedAt,
139+
fix.UpdatedAt,
140+
newVersion,
141+
s.queue,
142+
fix.ID,
143+
oldVersion,
144+
)
145+
if err != nil {
146+
return fmt.Errorf("failed to update fix queue=%q: %w", s.queue, err)
147+
}
148+
rowsAffected, err := result.RowsAffected()
149+
if err != nil {
150+
return fmt.Errorf("failed to get rows affected for fix queue=%q: %w", s.queue, err)
151+
}
152+
if rowsAffected != 1 {
153+
return fmt.Errorf("version mismatch for fix queue=%q expected_version=%d: %w", s.queue, oldVersion, storage.ErrVersionMismatch)
154+
}
155+
return nil
156+
}
157+
158+
func marshalFixJSON(fix entity.Fix) ([]byte, []byte, error) {
159+
issuesJSON, err := json.Marshal(fix.Issues)
160+
if err != nil {
161+
return nil, nil, err
162+
}
163+
metadata := fix.Metadata
164+
if metadata == nil {
165+
metadata = map[string]string{}
166+
}
167+
metadataJSON, err := json.Marshal(metadata)
168+
if err != nil {
169+
return nil, nil, err
170+
}
171+
return issuesJSON, metadataJSON, nil
172+
}

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

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
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+
"database/sql"
20+
"errors"
21+
"testing"
22+
23+
"github.com/DATA-DOG/go-sqlmock"
24+
driver "github.com/go-sql-driver/mysql"
25+
"github.com/stretchr/testify/require"
26+
"github.com/uber/submitqueue/stovepipe/entity"
27+
"github.com/uber/submitqueue/stovepipe/extension/storage"
28+
)
29+
30+
func TestFixStoresErrors(t *testing.T) {
31+
ctx := context.Background()
32+
dbError := errors.New("database unavailable")
33+
for _, scenario := range []string{"create duplicate", "create error", "get missing", "get error", "update error", "update stale", "rows affected error"} {
34+
t.Run(scenario, func(t *testing.T) {
35+
db, mock, err := sqlmock.New()
36+
require.NoError(t, err)
37+
defer db.Close()
38+
backend, err := NewStorage(db, testMetrics())
39+
require.NoError(t, err)
40+
bound, err := backend.For("queue")
41+
require.NoError(t, err)
42+
store := bound.GetFixStore()
43+
fix := entity.Fix{ID: "fix", Queue: "queue"}
44+
var want error
45+
switch scenario {
46+
case "create duplicate":
47+
mock.ExpectExec("INSERT INTO fix").WillReturnError(&driver.MySQLError{Number: mysqlErrDuplicateEntry})
48+
want = storage.ErrAlreadyExists
49+
err = store.Create(ctx, fix)
50+
case "create error":
51+
mock.ExpectExec("INSERT INTO fix").WillReturnError(dbError)
52+
want = dbError
53+
err = store.Create(ctx, fix)
54+
case "get missing":
55+
mock.ExpectQuery("SELECT").WillReturnError(sql.ErrNoRows)
56+
want = storage.ErrNotFound
57+
_, err = store.Get(ctx, fix.ID)
58+
case "get error":
59+
mock.ExpectQuery("SELECT").WillReturnError(dbError)
60+
want = dbError
61+
_, err = store.Get(ctx, fix.ID)
62+
case "update error":
63+
mock.ExpectExec("UPDATE fix").WillReturnError(dbError)
64+
want = dbError
65+
err = store.Update(ctx, fix, 1, 2)
66+
case "update stale":
67+
mock.ExpectExec("UPDATE fix").WillReturnResult(sqlmock.NewResult(0, 0))
68+
want = storage.ErrVersionMismatch
69+
err = store.Update(ctx, fix, 1, 2)
70+
case "rows affected error":
71+
mock.ExpectExec("UPDATE fix").WillReturnResult(sqlmock.NewErrorResult(dbError))
72+
want = dbError
73+
err = store.Update(ctx, fix, 1, 2)
74+
}
75+
require.ErrorIs(t, err, want)
76+
require.NoError(t, mock.ExpectationsWereMet())
77+
})
78+
}
79+
}
80+
81+
func TestFixStoresRejectInvalidWrites(t *testing.T) {
82+
ctx := context.Background()
83+
for _, operation := range []string{"create", "update"} {
84+
t.Run(operation, func(t *testing.T) {
85+
db, mock, err := sqlmock.New()
86+
require.NoError(t, err)
87+
defer db.Close()
88+
fixStore := NewFixStore(db, testMetrics(), "queue")
89+
for _, fix := range []entity.Fix{{Queue: "queue"}, {ID: "fix", Queue: "other"}} {
90+
if operation == "create" {
91+
err = fixStore.Create(ctx, fix)
92+
} else {
93+
err = fixStore.Update(ctx, fix, 1, 2)
94+
}
95+
require.Error(t, err)
96+
}
97+
require.NoError(t, mock.ExpectationsWereMet())
98+
})
99+
}
100+
}
101+
102+
func TestFixStoresRejectCorruptCollections(t *testing.T) {
103+
ctx := context.Background()
104+
for _, tt := range []struct{ name, issues, metadata string }{
105+
{"repair issue openings", `{}`, `{}`},
106+
{"repair metadata", `[]`, `[]`},
107+
} {
108+
t.Run(tt.name, func(t *testing.T) {
109+
db, mock, err := sqlmock.New()
110+
require.NoError(t, err)
111+
defer db.Close()
112+
mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"id", "build_id", "state", "issues", "execution_ref", "metadata", "created_at", "updated_at", "version"}).
113+
AddRow("fix", "build-1", "queued", tt.issues, "", tt.metadata, 1, 1, 1))
114+
_, err = NewFixStore(db, testMetrics(), "queue").Get(ctx, "fix")
115+
require.Error(t, err)
116+
require.NoError(t, mock.ExpectationsWereMet())
117+
})
118+
}
119+
}

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
-- Repairs retain explicit lifecycle fields, typed issue opening membership, and optional metadata.
2+
CREATE TABLE IF NOT EXISTS fix (
3+
queue VARBINARY(1020) NOT NULL,
4+
id VARBINARY(1020) NOT NULL,
5+
build_id TEXT NOT NULL,
6+
state VARCHAR(64) NOT NULL,
7+
issues JSON NOT NULL,
8+
execution_ref TEXT NOT NULL,
9+
metadata JSON NOT NULL,
10+
created_at BIGINT NOT NULL,
11+
updated_at BIGINT NOT NULL,
12+
version INT NOT NULL,
13+
PRIMARY KEY (queue, id)
14+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) {
5858
buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName),
5959
openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName),
6060
validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName),
61+
fixStore: NewFixStore(s.db, queueScope.SubScope("fix_store"), queueName),
6162
}, nil
6263
}
6364

@@ -80,6 +81,7 @@ type mysqlStorage struct {
8081
buildIssueStore storage.BuildIssueStore
8182
openIssueStore storage.OpenIssueStore
8283
validationFactStore storage.ValidationFactStore
84+
fixStore storage.FixStore
8385
}
8486

8587
// Verify mysqlStorage implements the queue-scoped aggregate at compile time.
@@ -125,6 +127,10 @@ func (f *mysqlStorage) GetValidationFactStore() storage.ValidationFactStore {
125127
return f.validationFactStore
126128
}
127129

130+
func (f *mysqlStorage) GetFixStore() storage.FixStore {
131+
return f.fixStore
132+
}
133+
128134
func (f *mysqlStorage) GetQueueSummaryStore() storage.QueueSummaryStore { return f.queueSummaryStore }
129135

130136
func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore {

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -153,6 +153,14 @@ func TestStorageForQueueMetrics(t *testing.T) {
153153
return err
154154
},
155155
},
156+
{
157+
name: "fix_store",
158+
op: "get",
159+
read: func(bound storage.Storage, _ string) error {
160+
_, err := bound.GetFixStore().Get(context.Background(), "fix-id")
161+
return err
162+
},
163+
},
156164
}
157165

158166
for _, tt := range tests {

‎stovepipe/extension/storage/storage.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,4 +102,6 @@ type Storage interface {
102102

103103
// GetValidationFactStore returns the ValidationFactStore instance.
104104
GetValidationFactStore() ValidationFactStore
105+
106+
GetFixStore() FixStore
105107
}

0 commit comments

Comments
 (0)