From edc7d3a4983398dcdc42daefad85a0db85985719 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 3 Aug 2026 10:02:27 -0700 Subject: [PATCH] feat(storage): path-build link store for per-path builds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Speculation needs one build per *path*, and something must record which build belongs to which path attempt. The runner chooses the build ID, so a caller holding a path cannot derive it; a keyed record is the key-value contract's mechanism for a reverse lookup. ### What? `entity.PathBuild` is new: the `(pathID, attempt) -> buildID` link, with `storage.PathBuildStore`, a MySQL implementation, schema, and mocks. The record is write-once: created only once the runner has named the build, never updated. Absent means no build is recorded for the attempt; present names its build permanently, and a retried path is a new attempt under a different key. Because creation is the only write, the first insert also decides concurrent dispatches for the same attempt — `ErrAlreadyExists` tells the loser the attempt's build is someone else's. There is deliberately no reservation state and no version column: the stage that watches builds (later in this stack) stops any build whose path no longer wants it, so no reader needs to distinguish "idle" from "mid-dispatch". `entity.Build` gains `PathID` and `Attempt`, and `entity.SpeculationPath` gains `Base()` — the dependencies the path assumes will succeed — so every consumer derives the base one way. ## Test Plan ✅ `bazel test //submitqueue/entity/... //submitqueue/extension/storage/...` ✅ `bazel test //test/integration/submitqueue/extension/storage/mysql:go_default_test` — an attempt resolving to its build, `ErrNotFound` when undispatched, and a duplicate `Create` refused so the first link stands. ✅ `make fmt`, `make gazelle`, `make mocks` --- submitqueue/entity/BUILD.bazel | 1 + submitqueue/entity/build.go | 13 +++ submitqueue/entity/path_build.go | 35 ++++++++ submitqueue/entity/speculation.go | 25 +++++- submitqueue/extension/storage/BUILD.bazel | 1 + .../extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/path_build_store_mock.go | 71 ++++++++++++++++ .../storage/mock/request_batch_store_mock.go | 4 +- .../extension/storage/mock/storage_mock.go | 14 ++++ .../extension/storage/mysql/BUILD.bazel | 1 + .../extension/storage/mysql/build_store.go | 12 +-- .../storage/mysql/build_store_test.go | 28 ++++--- .../storage/mysql/path_build_store.go | 83 +++++++++++++++++++ .../extension/storage/mysql/schema/build.sql | 2 + .../storage/mysql/schema/path_build.sql | 14 ++++ .../extension/storage/mysql/storage.go | 7 ++ .../extension/storage/path_build_store.go | 51 ++++++++++++ submitqueue/extension/storage/storage.go | 3 + .../submitqueue/extension/storage/suite.go | 46 ++++++++++ 19 files changed, 389 insertions(+), 23 deletions(-) create mode 100644 submitqueue/entity/path_build.go create mode 100644 submitqueue/extension/storage/mock/path_build_store_mock.go create mode 100644 submitqueue/extension/storage/mysql/path_build_store.go create mode 100644 submitqueue/extension/storage/mysql/schema/path_build.sql create mode 100644 submitqueue/extension/storage/path_build_store.go diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index c31bf21f..bc3d5f2a 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -14,6 +14,7 @@ go_library( "land.go", "list.go", "merge_result.go", + "path_build.go", "push_result.go", "queue_config.go", "request.go", diff --git a/submitqueue/entity/build.go b/submitqueue/entity/build.go index 11a62c18..ffb96d7b 100644 --- a/submitqueue/entity/build.go +++ b/submitqueue/entity/build.go @@ -55,12 +55,25 @@ func (s BuildStatus) IsTerminal() bool { // Build represents a build scheduled for a batch along a specific speculation path. // All fields except the Status are immutable after creation. +// +// It is keyed by the runner's build ID, which is the identifier every stage +// downstream of the trigger already holds: a poll, a webhook, and a runner-side +// log line all name a build, none of them names a speculation path. The path +// coordinates ride along on the record so those stages never have to +// understand speculation to do their job. type Build struct { // ID is the identifier minted by the queue's build runner when the build // is triggered; this is the primary storage key. ID string // BatchID is the batch for which this build is scheduled. BatchID string + // PathID is the speculation path this build verifies, as carried by + // SpeculationPathEntry.ID. + PathID string + // Attempt is which build attempt for that path this is, starting at 1. + // A path may be built more than once, so ID names the run while + // (PathID, Attempt) names the slot it occupies. + Attempt int // Status represents the state of the build lifecycle this build is in. Status BuildStatus } diff --git a/submitqueue/entity/path_build.go b/submitqueue/entity/path_build.go new file mode 100644 index 00000000..2bfd1641 --- /dev/null +++ b/submitqueue/entity/path_build.go @@ -0,0 +1,35 @@ +// Copyright (c) 2025 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 entity + +// PathBuild names the build started for one attempt of one speculation path. +// +// It is the reverse of Build's key. A Build is keyed by the identifier the +// runner minted, which is what every stage watching a build already holds; a +// caller starting from a path has no way to derive that identifier, so the link +// is recorded under the coordinates it does hold. +// +// A record is write-once: it is created already naming its build and never +// changes, so an attempt maps to one build for good — a retried path is a new +// attempt under a different key. An absent record means no build is recorded +// for the attempt; it does not promise that none is starting. +type PathBuild struct { + // PathID is the speculation path, as carried by SpeculationPathEntry.ID. + PathID string + // Attempt is which build attempt for that path this is, starting at 1. + Attempt int + // BuildID is the build started for that attempt. Never empty. + BuildID string +} diff --git a/submitqueue/entity/speculation.go b/submitqueue/entity/speculation.go index bb4cb4f4..d4631b61 100644 --- a/submitqueue/entity/speculation.go +++ b/submitqueue/entity/speculation.go @@ -79,6 +79,24 @@ func (p SpeculationPath) ID() string { return hex.EncodeToString(sum[:]) } +// Base returns the path's base — the batches its head is stacked on top of: +// the IDs of the dependencies the path assumes will succeed, in the path's +// dependency order. +// +// It is a projection of the path rather than a decision about it — a +// dependency the path assumes will fail is by definition built without, and +// an ignored one is not built on either — so every caller that needs the base +// derives it here rather than re-reading the assumptions itself. +func (p SpeculationPath) Base() []string { + var deps []string + for _, dep := range p.Dependencies { + if dep.Assumption == DependencyAssumptionSucceeds { + deps = append(deps, dep.Batch) + } + } + return deps +} + // SpeculationPathStatus is the lifecycle status of one speculation path's // current build attempt. type SpeculationPathStatus string @@ -122,9 +140,10 @@ func (s SpeculationPathStatus) IsTerminal() bool { } // SpeculationPathEntry is the stored record of one chosen speculation path, -// keyed by the hash of its content. It holds no build reference (that lives on -// the separate execution record, keyed by (ID, Attempt)) and no score (a score -// is meaningful only within a single speculation run). +// keyed by the hash of its content. It holds no build reference — a build is +// linked to an attempt by PathBuild, so the path stays what the speculation run +// decided rather than a mirror of what the build system is doing — and no score +// (a score is meaningful only within a single speculation run). type SpeculationPathEntry struct { // ID is the primary key: the hash of the path's content (head plus its // assumptions). It always equals Path.ID() — it is materialized here, rather diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index 25b945be..e1865e89 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "path_build_store.go", "request_batch_store.go", "request_log_store.go", "request_queue_summary_store.go", diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel index 986a3d4d..6ac3b88e 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store_mock.go", "build_store_mock.go", "change_store_mock.go", + "path_build_store_mock.go", "request_batch_store_mock.go", "request_log_store_mock.go", "request_queue_summary_store_mock.go", diff --git a/submitqueue/extension/storage/mock/path_build_store_mock.go b/submitqueue/extension/storage/mock/path_build_store_mock.go new file mode 100644 index 00000000..a17098ac --- /dev/null +++ b/submitqueue/extension/storage/mock/path_build_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: path_build_store.go +// +// Generated by this command: +// +// mockgen -source=path_build_store.go -destination=mock/path_build_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/submitqueue/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockPathBuildStore is a mock of PathBuildStore interface. +type MockPathBuildStore struct { + ctrl *gomock.Controller + recorder *MockPathBuildStoreMockRecorder + isgomock struct{} +} + +// MockPathBuildStoreMockRecorder is the mock recorder for MockPathBuildStore. +type MockPathBuildStoreMockRecorder struct { + mock *MockPathBuildStore +} + +// NewMockPathBuildStore creates a new mock instance. +func NewMockPathBuildStore(ctrl *gomock.Controller) *MockPathBuildStore { + mock := &MockPathBuildStore{ctrl: ctrl} + mock.recorder = &MockPathBuildStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockPathBuildStore) EXPECT() *MockPathBuildStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockPathBuildStore) Create(ctx context.Context, pathBuild entity.PathBuild) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, pathBuild) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockPathBuildStoreMockRecorder) Create(ctx, pathBuild any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockPathBuildStore)(nil).Create), ctx, pathBuild) +} + +// Get mocks base method. +func (m *MockPathBuildStore) Get(ctx context.Context, pathID string, attempt int) (entity.PathBuild, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, pathID, attempt) + ret0, _ := ret[0].(entity.PathBuild) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockPathBuildStoreMockRecorder) Get(ctx, pathID, attempt any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockPathBuildStore)(nil).Get), ctx, pathID, attempt) +} diff --git a/submitqueue/extension/storage/mock/request_batch_store_mock.go b/submitqueue/extension/storage/mock/request_batch_store_mock.go index 2c4c15f0..24e87e12 100644 --- a/submitqueue/extension/storage/mock/request_batch_store_mock.go +++ b/submitqueue/extension/storage/mock/request_batch_store_mock.go @@ -1,9 +1,9 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: submitqueue/extension/storage/request_batch_store.go +// Source: request_batch_store.go // // Generated by this command: // -// mockgen -source=submitqueue/extension/storage/request_batch_store.go -destination=submitqueue/extension/storage/mock/request_batch_store_mock.go -package=mock +// mockgen -source=request_batch_store.go -destination=mock/request_batch_store_mock.go -package=mock // // Package mock is a generated GoMock package. diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index 3f73c35e..2162372e 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -110,6 +110,20 @@ func (mr *MockStorageMockRecorder) GetChangeStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeStore", reflect.TypeOf((*MockStorage)(nil).GetChangeStore)) } +// GetPathBuildStore mocks base method. +func (m *MockStorage) GetPathBuildStore() storage.PathBuildStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetPathBuildStore") + ret0, _ := ret[0].(storage.PathBuildStore) + return ret0 +} + +// GetPathBuildStore indicates an expected call of GetPathBuildStore. +func (mr *MockStorageMockRecorder) GetPathBuildStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPathBuildStore", reflect.TypeOf((*MockStorage)(nil).GetPathBuildStore)) +} + // GetRequestBatchStore mocks base method. func (m *MockStorage) GetRequestBatchStore() storage.RequestBatchStore { m.ctrl.T.Helper() diff --git a/submitqueue/extension/storage/mysql/BUILD.bazel b/submitqueue/extension/storage/mysql/BUILD.bazel index bf1c93c3..c6c00d9a 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "path_build_store.go", "request_batch_store.go", "request_log_store.go", "request_queue_summary_store.go", diff --git a/submitqueue/extension/storage/mysql/build_store.go b/submitqueue/extension/storage/mysql/build_store.go index f1795ffb..c79718b7 100644 --- a/submitqueue/extension/storage/mysql/build_store.go +++ b/submitqueue/extension/storage/mysql/build_store.go @@ -46,9 +46,9 @@ func (s *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE var build entity.Build err := s.db.QueryRowContext(ctx, - "SELECT id, batch_id, status FROM build WHERE id = ?", + "SELECT id, batch_id, path_id, attempt, status FROM build WHERE id = ?", id, - ).Scan(&build.ID, &build.BatchID, &build.Status) + ).Scan(&build.ID, &build.BatchID, &build.PathID, &build.Attempt, &build.Status) if errors.Is(err, sql.ErrNoRows) { return entity.Build{}, storage.WrapNotFound(err) @@ -66,8 +66,8 @@ func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr err defer func() { op.Complete(retErr) }() _, err := s.db.ExecContext(ctx, - "INSERT INTO build (id, batch_id, status) VALUES (?, ?, ?)", - build.ID, build.BatchID, build.Status, + "INSERT INTO build (id, batch_id, path_id, attempt, status) VALUES (?, ?, ?, ?, ?)", + build.ID, build.BatchID, build.PathID, build.Attempt, build.Status, ) if err != nil { var mysqlErr *mysql.MySQLError @@ -86,8 +86,8 @@ func (s *buildStore) Update(ctx context.Context, build entity.Build) (retErr err defer func() { op.Complete(retErr) }() result, err := s.db.ExecContext(ctx, - "UPDATE build SET batch_id = ?, status = ? WHERE id = ?", - build.BatchID, build.Status, build.ID, + "UPDATE build SET batch_id = ?, path_id = ?, attempt = ?, status = ? WHERE id = ?", + build.BatchID, build.PathID, build.Attempt, build.Status, build.ID, ) if err != nil { return fmt.Errorf("failed to update build entity id=%q: %w", build.ID, err) diff --git a/submitqueue/extension/storage/mysql/build_store_test.go b/submitqueue/extension/storage/mysql/build_store_test.go index 9d350db8..1d279bb7 100644 --- a/submitqueue/extension/storage/mysql/build_store_test.go +++ b/submitqueue/extension/storage/mysql/build_store_test.go @@ -44,6 +44,8 @@ func TestBuildStore_Get(t *testing.T) { want := entity.Build{ ID: "bk-1001", BatchID: "monorepo/batch/1", + PathID: "path-1", + Attempt: 1, Status: entity.BuildStatusRunning, } @@ -59,9 +61,9 @@ func TestBuildStore_Get(t *testing.T) { name: "found", id: want.ID, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"id", "batch_id", "status"}). - AddRow(want.ID, want.BatchID, string(want.Status)) - mock.ExpectQuery("SELECT id, batch_id, status"). + rows := sqlmock.NewRows([]string{"id", "batch_id", "path_id", "attempt", "status"}). + AddRow(want.ID, want.BatchID, want.PathID, want.Attempt, string(want.Status)) + mock.ExpectQuery("SELECT id, batch_id, path_id, attempt, status"). WithArgs(want.ID). WillReturnRows(rows) }, @@ -71,7 +73,7 @@ func TestBuildStore_Get(t *testing.T) { name: "not found", id: "missing", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, batch_id, status"). + mock.ExpectQuery("SELECT id, batch_id, path_id, attempt, status"). WithArgs("missing"). WillReturnError(sql.ErrNoRows) }, @@ -82,7 +84,7 @@ func TestBuildStore_Get(t *testing.T) { name: "query error", id: "bad", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, batch_id, status"). + mock.ExpectQuery("SELECT id, batch_id, path_id, attempt, status"). WithArgs("bad"). WillReturnError(fmt.Errorf("connection reset")) }, @@ -116,6 +118,8 @@ func TestBuildStore_Create(t *testing.T) { build := entity.Build{ ID: "bk-1001", BatchID: "monorepo/batch/1", + PathID: "path-1", + Attempt: 1, Status: entity.BuildStatusAccepted, } @@ -129,7 +133,7 @@ func TestBuildStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs(build.ID, build.BatchID, build.PathID, build.Attempt, build.Status). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -137,7 +141,7 @@ func TestBuildStore_Create(t *testing.T) { name: "duplicate id returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs(build.ID, build.BatchID, build.PathID, build.Attempt, build.Status). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -147,7 +151,7 @@ func TestBuildStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs(build.ID, build.BatchID, build.Status). + WithArgs(build.ID, build.BatchID, build.PathID, build.Attempt, build.Status). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -192,7 +196,7 @@ func TestBuildStore_Update(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.PathID, build.Attempt, build.Status, build.ID). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -200,7 +204,7 @@ func TestBuildStore_Update(t *testing.T) { name: "not found", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.PathID, build.Attempt, build.Status, build.ID). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -210,7 +214,7 @@ func TestBuildStore_Update(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.PathID, build.Attempt, build.Status, build.ID). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -219,7 +223,7 @@ func TestBuildStore_Update(t *testing.T) { name: "rows affected error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE build"). - WithArgs(build.BatchID, build.Status, build.ID). + WithArgs(build.BatchID, build.PathID, build.Attempt, build.Status, build.ID). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/submitqueue/extension/storage/mysql/path_build_store.go b/submitqueue/extension/storage/mysql/path_build_store.go new file mode 100644 index 00000000..ab699d48 --- /dev/null +++ b/submitqueue/extension/storage/mysql/path_build_store.go @@ -0,0 +1,83 @@ +// Copyright (c) 2025 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/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type pathBuildStore struct { + db *sql.DB + scope tally.Scope +} + +// NewPathBuildStore creates a new MySQL-backed PathBuildStore. +func NewPathBuildStore(db *sql.DB, scope tally.Scope) storage.PathBuildStore { + return &pathBuildStore{db: db, scope: scope} +} + +// Get resolves an attempt to its build. Returns ErrNotFound if no build has +// been recorded for it. +func (s *pathBuildStore) Get(ctx context.Context, pathID string, attempt int) (ret entity.PathBuild, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + var pathBuild entity.PathBuild + + err := s.db.QueryRowContext(ctx, + "SELECT path_id, attempt, build_id FROM path_build WHERE path_id = ? AND attempt = ?", + pathID, attempt, + ).Scan(&pathBuild.PathID, &pathBuild.Attempt, &pathBuild.BuildID) + + if errors.Is(err, sql.ErrNoRows) { + return entity.PathBuild{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.PathBuild{}, fmt.Errorf("failed to get path build entity pathID=%s attempt=%d from the database: %w", pathID, attempt, err) + } + + return pathBuild, nil +} + +// Create records the build for an attempt, permanently. Returns +// ErrAlreadyExists if the attempt already has a build. +func (s *pathBuildStore) Create(ctx context.Context, pathBuild entity.PathBuild) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + _, err := s.db.ExecContext(ctx, + "INSERT INTO path_build (path_id, attempt, build_id) VALUES (?, ?, ?)", + pathBuild.PathID, pathBuild.Attempt, pathBuild.BuildID, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("path build entity pathID=%s attempt=%d: %w", pathBuild.PathID, pathBuild.Attempt, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert path build entity pathID=%s attempt=%d: %w", pathBuild.PathID, pathBuild.Attempt, err) + } + + return nil +} diff --git a/submitqueue/extension/storage/mysql/schema/build.sql b/submitqueue/extension/storage/mysql/schema/build.sql index f720397f..62c2105e 100644 --- a/submitqueue/extension/storage/mysql/schema/build.sql +++ b/submitqueue/extension/storage/mysql/schema/build.sql @@ -1,6 +1,8 @@ CREATE TABLE IF NOT EXISTS build ( id VARCHAR(255) NOT NULL, batch_id VARCHAR(255) NOT NULL, + path_id VARCHAR(255) NOT NULL, + attempt INT NOT NULL, status VARCHAR(64) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/schema/path_build.sql b/submitqueue/extension/storage/mysql/schema/path_build.sql new file mode 100644 index 00000000..0d48936d --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/path_build.sql @@ -0,0 +1,14 @@ +-- Resolves one attempt of one speculation path to the build started for it. +-- The runner chooses the build ID, so a caller holding a path cannot derive it; +-- this is the reverse lookup, made a keyed record rather than an index because +-- the store contract has no lookup by attribute. +-- +-- A row is write-once: inserted only once the runner has named the build, and +-- never updated. The primary key decides concurrent dispatches for the same +-- attempt — the first insert wins, the duplicate is refused. +CREATE TABLE IF NOT EXISTS path_build ( + path_id VARCHAR(255) NOT NULL, + attempt INT NOT NULL, + build_id VARCHAR(255) NOT NULL, + PRIMARY KEY (path_id, attempt) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/storage.go b/submitqueue/extension/storage/mysql/storage.go index 2a9b1953..77fd5eba 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -36,6 +36,7 @@ type mysqlStorage struct { batchDependentStore storage.BatchDependentStore buildStore storage.BuildStore speculationPathSetStore storage.SpeculationPathSetStore + pathBuildStore storage.PathBuildStore requestLogStore storage.RequestLogStore requestSummaryStore storage.RequestSummaryStore requestQueueStore storage.RequestQueueSummaryStore @@ -53,6 +54,7 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) { batchDependentStore: NewBatchDependentStore(db, scope.SubScope("batch_dependent_store")), buildStore: NewBuildStore(db, scope.SubScope("build_store")), speculationPathSetStore: NewSpeculationPathSetStore(db, scope.SubScope("speculation_path_set_store")), + pathBuildStore: NewPathBuildStore(db, scope.SubScope("path_build_store")), requestLogStore: NewRequestLogStore(db, scope.SubScope("request_log_store")), requestSummaryStore: NewRequestSummaryStore(db, scope.SubScope("request_summary_store")), requestQueueStore: NewRequestQueueSummaryStore(db, scope.SubScope("request_queue_summary_store")), @@ -95,6 +97,11 @@ func (f *mysqlStorage) GetSpeculationPathSetStore() storage.SpeculationPathSetSt return f.speculationPathSetStore } +// GetPathBuildStore returns the MySQL-backed PathBuildStore. +func (f *mysqlStorage) GetPathBuildStore() storage.PathBuildStore { + return f.pathBuildStore +} + // GetRequestLogStore returns the MySQL-backed RequestLogStore. func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore { return f.requestLogStore diff --git a/submitqueue/extension/storage/path_build_store.go b/submitqueue/extension/storage/path_build_store.go new file mode 100644 index 00000000..1664d97a --- /dev/null +++ b/submitqueue/extension/storage/path_build_store.go @@ -0,0 +1,51 @@ +// Copyright (c) 2025 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=path_build_store.go -destination=mock/path_build_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// PathBuildStore resolves one attempt of one speculation path to the build +// started for it, keyed by (path ID, attempt). +// +// It exists because the runner chooses the build ID: a caller holding a path +// and an attempt cannot derive it, and there is no lookup by attribute in this +// contract. Promoting the relationship to a keyed record is the mechanism for a +// reverse lookup here, not a workaround. +// +// A record is write-once and complete from the start: it is created only once +// the runner has named the build, and never updated afterwards. It answers two +// questions — absent means no build is recorded for the attempt, present names +// the attempt's build permanently. A retried path is a new attempt under a +// different key. +// +// Because creation is the only write, a duplicate Create is how concurrent +// dispatches for the same attempt are decided: the first insert wins, and +// ErrAlreadyExists tells the loser the attempt's build is someone else's. +type PathBuildStore interface { + // Get resolves an attempt to its build. + // Returns ErrNotFound if no build has been recorded for the attempt. + Get(ctx context.Context, pathID string, attempt int) (entity.PathBuild, error) + + // Create records the build for an attempt, permanently. + // Returns ErrAlreadyExists if the attempt already has a build, which means + // a concurrent dispatch recorded its own first. + Create(ctx context.Context, pathBuild entity.PathBuild) error +} diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index fbdaee24..ec620593 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -67,6 +67,9 @@ type Storage interface { // GetSpeculationPathSetStore returns the SpeculationPathSetStore instance. GetSpeculationPathSetStore() SpeculationPathSetStore + // GetPathBuildStore returns the PathBuildStore instance. + GetPathBuildStore() PathBuildStore + // GetRequestLogStore returns the RequestLogStore instance. GetRequestLogStore() RequestLogStore diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index 9b8c1eb2..ff9110a1 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -787,6 +787,52 @@ func speculationPathSet(head, dep string) entity.SpeculationPathSet { } } +// TestStorage_PathBuildResolvesAnAttemptToItsBuild tests the reverse lookup the +// speculate run depends on: it holds a path and an attempt, and the runner +// chose the build ID, so this record is the only way back. +func (s *StorageContractSuite) TestStorage_PathBuildResolvesAnAttemptToItsBuild() { + t := s.T() + ctx := s.ctx + store := s.storage.GetPathBuildStore() + + want := entity.PathBuild{PathID: "pb/path/1", Attempt: 1, BuildID: "pb/build/1"} + require.NoError(t, store.Create(ctx, want)) + + got, err := store.Get(ctx, want.PathID, want.Attempt) + require.NoError(t, err) + assert.Equal(t, want, got) + + // An attempt that was never dispatched has no build. + _, err = store.Get(ctx, "pb/path/undispatched", 1) + assert.ErrorIs(t, err, storage.ErrNotFound) +} + +// TestStorage_PathBuildIsImmutable tests that an attempt keeps the build it was +// first linked to. A redelivered dispatch must be refused rather than silently +// repointing the attempt at a second build, which would orphan the first. +func (s *StorageContractSuite) TestStorage_PathBuildIsImmutable() { + t := s.T() + ctx := s.ctx + store := s.storage.GetPathBuildStore() + + first := entity.PathBuild{PathID: "pb/path/immutable", Attempt: 1, BuildID: "pb/build/first"} + require.NoError(t, store.Create(ctx, first)) + + second := first + second.BuildID = "pb/build/second" + assert.ErrorIs(t, store.Create(ctx, second), storage.ErrAlreadyExists) + + got, err := store.Get(ctx, first.PathID, first.Attempt) + require.NoError(t, err) + assert.Equal(t, "pb/build/first", got.BuildID) + + // A retried path is a different attempt, so it is a different key. + retry := first + retry.Attempt = 2 + retry.BuildID = "pb/build/retry" + require.NoError(t, store.Create(ctx, retry)) +} + // TestStorage_SpeculationPathSetCreateAndGet tests that a set round-trips whole, // including each path's assumptions — a path's identity hashes them, so an // encoding that dropped or reordered them would silently change every path ID.