Skip to content

Commit f82d947

Browse files
committed
feat(stovepipe): persist repairs with queue-scoped discovery
1 parent 7a82ea5 commit f82d947

19 files changed

Lines changed: 720 additions & 24 deletions

File tree

‎doc/rfc/stovepipe/fixer.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ Stovepipe accepts a broken check, decides whether to start a repair or attach it
44

55
`buildanalysis.Controller.Process` persists observations and manages failure episodes and recovery evidence. Each episode also stores its authoritative `FixID` assignment, updated by `fix`. The `fix` queue stage owns repair decisions, associations, and submission; `fixsignal` observes execution and resolves repairs. Both stages implement [`consumer.Controller`](../../../platform/consumer/controller.go) through `Process(context.Context, consumer.Delivery) error`. [`fixer.RepairEvaluator`](../../../stovepipe/extension/fixer/fixer.go) recommends Start, Attach, or Defer; `fixer.RepairExecutor` submits and observes work through a host-supplied implementation.
66

7-
**Implementation status:** the [queue messages and topic keys](../../../stovepipe/core/messagequeue/messagequeue.go), [extension interfaces](../../../stovepipe/extension/fixer/fixer.go), [repair entity](../../../stovepipe/entity/fix.go), and [FixStore contract](../../../stovepipe/extension/storage/fix_store.go) are defined. [`Queue.ActiveFixIDs`](../../../stovepipe/entity/queue.go) is part of the existing queue snapshot. Persistence implementations follow in the storage layer. `buildanalysis` and both fixer runtime controllers remain to be implemented. The episode requirements below define the integration needed from build analysis.
7+
**Implementation status:** the [queue messages and topic keys](../../../stovepipe/core/messagequeue/messagequeue.go), [extension interfaces](../../../stovepipe/extension/fixer/fixer.go), [repair entity](../../../stovepipe/entity/fix.go), and [MySQL FixStore](../../../stovepipe/extension/storage/mysql/fix_store.go) are implemented. [`Queue.ActiveFixIDs`](../../../stovepipe/entity/queue.go) is part of the existing queue snapshot. `buildanalysis` and both fixer runtime controllers remain to be implemented. The episode requirements below define the integration needed from build analysis.
88

99
## 1. Persist the failure episode in buildanalysis
1010

@@ -125,11 +125,11 @@ All access uses primary keys. Candidate discovery reads `Queue.ActiveFixIDs` and
125125

126126
`fix.metadata` follows the existing [request-log metadata pattern](../../../stovepipe/entity/request_log.go). Nil and empty metadata maps are equivalent and persist as `{}`. Nil and empty `Queue.ActiveFixIDs` both persist as `[]` and read as nil. Conditional updates replace complete snapshots. Every queue writer preserves the active IDs alongside existing coordination fields. Queue updates can conflict with ingestion or validation updates and retry using the latest full snapshot.
127127

128-
For an existing database, add and backfill `queue.active_fix_ids` with an empty JSON array before running the new queue-store code. The storage implementation supplies schemas for fresh databases.
128+
For an existing database, add and backfill `queue.active_fix_ids` with an empty JSON array before running the new queue-store code. The checked-in schemas initialize fresh databases.
129129

130130
## Implementation
131131

132-
1. **Implement persistence and episode tracking.** Implement `FixStore` and persist `Queue.ActiveFixIDs`. In buildanalysis, supply the episode data and primary-key operations listed above, preserve historical generations and provenance, and publish/reconcile `FixRequest` handoffs.
132+
1. **Implement episode tracking in buildanalysis.** Supply the episode data and primary-key operations listed above, preserve historical generations and provenance, and publish/reconcile `FixRequest` handoffs.
133133
2. **Implement submission transitions.** Persist the submitting state before dispatch and the execution ID after success; use snapshot versions to reject stale results. Define retry policy and replayable reassociation before enabling execution.
134134
3. **Implement the queue stages.** Add `fix.Controller.Process` and `fixsignal.Controller.Process`, queue membership updates, episode assignment, bounded deferral, and reconciliation; register both consumers.
135135
4. **Implement extensions.** Supply `RepairEvaluator` and `RepairExecutor`, including evidence refresh, observation correlation, and overlap handling. Keep transport adapters in host integration.

‎stovepipe/extension/storage/README.md‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,4 +32,10 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
3232

3333
Build analysis manages failure episodes, failure and recovery evidence, and the authoritative `FixID` assignment. Fixer updates preserve episode evidence, and build-analysis updates preserve assignments. The [required failure-episode contract](../../../doc/rfc/stovepipe/fixer.md#required-failure-episode-contract) defines the integration; its entity and persistence implementation belong to buildanalysis.
3434

35-
The [dispatch sequence](../../../doc/rfc/stovepipe/fixer.md#4-trigger-the-repair) uses `Fix.State`, `Fix.ExecutionRef`, and `Fix.Version` to persist intent before submission and reject stale results. Persistence implementations follow in the storage layer.
35+
The [dispatch sequence](../../../doc/rfc/stovepipe/fixer.md#4-trigger-the-repair) uses `Fix.State`, `Fix.ExecutionRef`, and `Fix.Version` to persist intent before submission and reject stale results.
36+
37+
### MySQL implementation
38+
39+
The MySQL `FixStore` preserves bytewise identity for queue and repair IDs, up to 1020 bytes each. Nil and empty metadata maps persist as `{}`. Nil and empty active-ID lists persist as `[]` and read as nil. Updates replace the complete snapshot under the caller-supplied version guard.
40+
41+
Existing databases need `queue.active_fix_ids` added and backfilled with an empty JSON array before deploying the new queue-store code. The checked-in `fix` and `queue` schemas initialize fresh databases.

‎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
@@ -4,6 +4,7 @@ go_library(
44
name = "go_default_library",
55
srcs = [
66
"build_store.go",
7+
"fix_store.go",
78
"queue_log_store.go",
89
"queue_store.go",
910
"queue_summary_store.go",
@@ -30,6 +31,7 @@ go_test(
3031
name = "go_default_test",
3132
srcs = [
3233
"build_store_test.go",
34+
"fix_store_test.go",
3335
"queue_store_test.go",
3436
"queue_summary_store_test.go",
3537
"request_acceptance_store_test.go",
Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
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+
breakagesJSON, 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, state, breakages, execution_ref, metadata, created_at, updated_at, version)
55+
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
56+
s.queue,
57+
fix.ID,
58+
fix.State,
59+
breakagesJSON,
60+
fix.ExecutionRef,
61+
metadataJSON,
62+
fix.CreatedAt,
63+
fix.UpdatedAt,
64+
fix.Version,
65+
)
66+
if err != nil {
67+
if isDuplicateEntry(err) {
68+
return fmt.Errorf("fix queue=%q: %w", s.queue, storage.ErrAlreadyExists)
69+
}
70+
return fmt.Errorf("failed to insert fix queue=%q: %w", s.queue, err)
71+
}
72+
return nil
73+
}
74+
75+
func (s *fixStore) Get(ctx context.Context, id string) (ret entity.Fix, retErr error) {
76+
op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
77+
defer func() { op.Complete(retErr) }()
78+
79+
var fix entity.Fix
80+
fix.Queue = s.queue
81+
var breakagesJSON []byte
82+
var metadataJSON []byte
83+
err := s.db.QueryRowContext(ctx,
84+
`SELECT id, state, breakages, execution_ref, metadata, created_at, updated_at, version FROM fix
85+
WHERE queue = ? AND id = ?`,
86+
s.queue,
87+
id,
88+
).Scan(
89+
&fix.ID,
90+
&fix.State,
91+
&breakagesJSON,
92+
&fix.ExecutionRef,
93+
&metadataJSON,
94+
&fix.CreatedAt,
95+
&fix.UpdatedAt,
96+
&fix.Version,
97+
)
98+
if errors.Is(err, sql.ErrNoRows) {
99+
return entity.Fix{}, storage.WrapNotFound(err)
100+
}
101+
if err != nil {
102+
return entity.Fix{}, fmt.Errorf("failed to get fix queue=%q: %w", s.queue, err)
103+
}
104+
if err := json.Unmarshal(breakagesJSON, &fix.Breakages); err != nil {
105+
return entity.Fix{}, fmt.Errorf("failed to unmarshal repair breakages: %w", err)
106+
}
107+
if err := json.Unmarshal(metadataJSON, &fix.Metadata); err != nil {
108+
return entity.Fix{}, fmt.Errorf("failed to unmarshal repair metadata: %w", err)
109+
}
110+
if fix.Metadata == nil {
111+
fix.Metadata = map[string]string{}
112+
}
113+
return fix, nil
114+
}
115+
116+
func (s *fixStore) Update(ctx context.Context, fix entity.Fix, oldVersion, newVersion int32) (retErr error) {
117+
op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
118+
defer func() { op.Complete(retErr) }()
119+
120+
if fix.Queue != s.queue || fix.ID == "" {
121+
return fmt.Errorf("repair requires a nonempty ID and queue matching %q", s.queue)
122+
}
123+
breakagesJSON, metadataJSON, err := marshalFixJSON(fix)
124+
if err != nil {
125+
return fmt.Errorf("failed to marshal repair data: %w", err)
126+
}
127+
128+
result, err := s.db.ExecContext(ctx,
129+
`UPDATE fix SET state = ?, breakages = ?, execution_ref = ?, metadata = ?, created_at = ?, updated_at = ?, version = ?
130+
WHERE queue = ? AND id = ? AND version = ?`,
131+
fix.State,
132+
breakagesJSON,
133+
fix.ExecutionRef,
134+
metadataJSON,
135+
fix.CreatedAt,
136+
fix.UpdatedAt,
137+
newVersion,
138+
s.queue,
139+
fix.ID,
140+
oldVersion,
141+
)
142+
if err != nil {
143+
return fmt.Errorf("failed to update fix queue=%q: %w", s.queue, err)
144+
}
145+
rowsAffected, err := result.RowsAffected()
146+
if err != nil {
147+
return fmt.Errorf("failed to get rows affected for fix queue=%q: %w", s.queue, err)
148+
}
149+
if rowsAffected != 1 {
150+
return fmt.Errorf("version mismatch for fix queue=%q expected_version=%d: %w", s.queue, oldVersion, storage.ErrVersionMismatch)
151+
}
152+
return nil
153+
}
154+
155+
func marshalFixJSON(fix entity.Fix) ([]byte, []byte, error) {
156+
breakagesJSON, err := json.Marshal(fix.Breakages)
157+
if err != nil {
158+
return nil, nil, err
159+
}
160+
metadata := fix.Metadata
161+
if metadata == nil {
162+
metadata = map[string]string{}
163+
}
164+
metadataJSON, err := json.Marshal(metadata)
165+
if err != nil {
166+
return nil, nil, err
167+
}
168+
return breakagesJSON, metadataJSON, nil
169+
}
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, breakages, metadata string }{
105+
{"repair episodes", `{}`, `{}`},
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", "state", "breakages", "execution_ref", "metadata", "created_at", "updated_at", "version"}).
113+
AddRow("fix", "queued", tt.breakages, "", 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+
}

0 commit comments

Comments
 (0)