diff --git a/service/internal/sqldb/BUILD.bazel b/service/internal/sqldb/BUILD.bazel new file mode 100644 index 00000000..92e39c76 --- /dev/null +++ b/service/internal/sqldb/BUILD.bazel @@ -0,0 +1,18 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["pool.go"], + importpath = "github.com/uber/submitqueue/service/internal/sqldb", + visibility = ["//service:__subpackages__"], +) + +go_test( + name = "go_default_test", + srcs = ["pool_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/service/internal/sqldb/pool.go b/service/internal/sqldb/pool.go new file mode 100644 index 00000000..7a1a14a4 --- /dev/null +++ b/service/internal/sqldb/pool.go @@ -0,0 +1,47 @@ +// 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 sqldb contains SQL database wiring helpers shared by services. +package sqldb + +import ( + "database/sql" + "fmt" + "strconv" +) + +// ConfigureMaxOpenConnections applies a positive connection limit encoded as a +// decimal string. It retains up to the same number of idle connections so a +// busy bounded pool reuses sockets instead of repeatedly opening new ones. An +// empty string or zero preserves database/sql defaults. +func ConfigureMaxOpenConnections(db *sql.DB, value string) error { + if db == nil { + return fmt.Errorf("database is required") + } + if value == "" { + return nil + } + maxOpen, err := strconv.Atoi(value) + if err != nil { + return fmt.Errorf("parse maximum open connections %q: %w", value, err) + } + if maxOpen < 0 { + return fmt.Errorf("maximum open connections must be non-negative") + } + if maxOpen > 0 { + db.SetMaxOpenConns(maxOpen) + db.SetMaxIdleConns(maxOpen) + } + return nil +} diff --git a/service/internal/sqldb/pool_test.go b/service/internal/sqldb/pool_test.go new file mode 100644 index 00000000..6234035b --- /dev/null +++ b/service/internal/sqldb/pool_test.go @@ -0,0 +1,50 @@ +// 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 sqldb + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestConfigureMaxOpenConnections(t *testing.T) { + tests := []struct { + name string + value string + want int + wantErr bool + }{ + {name: "empty preserves default", value: "", want: 0}, + {name: "zero preserves default", value: "0", want: 0}, + {name: "positive applies limit", value: "32", want: 32}, + {name: "negative is rejected", value: "-1", wantErr: true}, + {name: "invalid is rejected", value: "many", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := &sql.DB{} + err := ConfigureMaxOpenConnections(db, tt.value) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, db.Stats().MaxOpenConnections) + }) + } +} diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index bafb1cfc..16372b7b 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -19,6 +19,7 @@ go_library( "//runway/controller/mergeconflictcheck:go_default_library", "//runway/extension/merger:go_default_library", "//runway/extension/merger/noop:go_default_library", + "//service/internal/sqldb:go_default_library", "//sqsim/adapter/merger:go_default_library", "//sqsim/model:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", diff --git a/service/runway/server/main.go b/service/runway/server/main.go index 28b41a13..b298f8e2 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -41,6 +41,7 @@ import ( "github.com/uber/submitqueue/runway/controller/mergeconflictcheck" "github.com/uber/submitqueue/runway/extension/merger" "github.com/uber/submitqueue/runway/extension/merger/noop" + "github.com/uber/submitqueue/service/internal/sqldb" simmerger "github.com/uber/submitqueue/sqsim/adapter/merger" "github.com/uber/submitqueue/sqsim/model" "go.uber.org/zap" @@ -131,6 +132,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := sqldb.ConfigureMaxOpenConnections(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("configure queue database connection limit: %w", err) + } mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ DB: queueDB, diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index f0d92d8a..0d562882 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -53,6 +53,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-} # Path to YAML queue configuration baked into the image - QUEUE_CONFIG_PATH=/root/queues.yaml # Stable subscriber name for the request-log consumer @@ -75,6 +76,7 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-} - HOSTNAME=orchestrator-dev - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} volumes: @@ -100,6 +102,7 @@ services: - PORT=:8080 # Queue infrastructure connection (shared with the orchestrator) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-} - HOSTNAME=runway-dev - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} volumes: diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 39241b0a..1d39c98e 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -19,6 +19,7 @@ go_library( "//platform/extension/counter/mysql:go_default_library", "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", + "//service/internal/sqldb:go_default_library", "//service/submitqueue/gateway/server/mapper:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/queueconfig/yaml:go_default_library", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 38f3f9be..c893a19e 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -36,6 +36,7 @@ import ( mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/service/internal/sqldb" "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" @@ -234,6 +235,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := sqldb.ConfigureMaxOpenConnections(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("configure queue database connection limit: %w", err) + } // Initialize queue mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index ff47a216..b7a55754 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -22,6 +22,7 @@ go_library( "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//platform/http:go_default_library", + "//service/internal/sqldb:go_default_library", "//sqsim/adapter/buildrunner:go_default_library", "//sqsim/model:go_default_library", "//submitqueue/core/changeset:go_default_library", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 2f3a583c..c10ea23a 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -42,6 +42,7 @@ import ( extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/http" + "github.com/uber/submitqueue/service/internal/sqldb" simbuildrunner "github.com/uber/submitqueue/sqsim/adapter/buildrunner" "github.com/uber/submitqueue/sqsim/model" "github.com/uber/submitqueue/submitqueue/core/changeset" @@ -196,6 +197,9 @@ func run() error { return fmt.Errorf("failed to open queue database: %w", err) } defer queueDB.Close() + if err := sqldb.ConfigureMaxOpenConnections(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil { + return fmt.Errorf("configure queue database connection limit: %w", err) + } // Initialize queue mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ diff --git a/sqsim/runner/local.go b/sqsim/runner/local.go index fa6ced84..f2063401 100644 --- a/sqsim/runner/local.go +++ b/sqsim/runner/local.go @@ -160,6 +160,7 @@ func newLocalStack(repoRoot, profileDir string, stdout, stderr io.Writer) *local "REPO_ROOT="+repoRoot, "SQSIM_PROFILE_DIR="+profileDir, "SQSIM_SCENARIO_PATH=/sqsim/profile.json", + "QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=32", ), } } diff --git a/sqsim/runner/local_test.go b/sqsim/runner/local_test.go index e3bb495d..f30d7d2d 100644 --- a/sqsim/runner/local_test.go +++ b/sqsim/runner/local_test.go @@ -15,6 +15,7 @@ package runner import ( + "slices" "testing" "github.com/stretchr/testify/assert" @@ -39,3 +40,8 @@ func TestFindRepoRootUsesWorkspaceEnvironment(t *testing.T) { require.NoError(t, err) assert.Equal(t, "/tmp/repo", root) } + +func TestLocalStackBoundsQueueDatabaseConnections(t *testing.T) { + stack := newLocalStack("/repo", "/profile", nil, nil) + assert.True(t, slices.Contains(stack.baseEnv, "QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=32")) +} diff --git a/sqsim/scenarios/BUILD.bazel b/sqsim/scenarios/BUILD.bazel index 160fefc7..72d84e43 100644 --- a/sqsim/scenarios/BUILD.bazel +++ b/sqsim/scenarios/BUILD.bazel @@ -3,7 +3,11 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "failure.go", "happy_path.go", + "load.go", + "mixed.go", + "recovery.go", "registry.go", ], importpath = "github.com/uber/submitqueue/sqsim/scenarios", @@ -14,7 +18,11 @@ go_library( go_test( name = "go_default_test", srcs = [ + "failure_test.go", "happy_path_test.go", + "load_test.go", + "mixed_test.go", + "recovery_test.go", "registry_test.go", ], embed = [":go_default_library"], diff --git a/sqsim/scenarios/failure.go b/sqsim/scenarios/failure.go new file mode 100644 index 00000000..5ab70229 --- /dev/null +++ b/sqsim/scenarios/failure.go @@ -0,0 +1,51 @@ +// 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 scenarios + +import ( + "time" + + "github.com/uber/submitqueue/sqsim" +) + +// BuildFailure returns a request whose Build Runner reports terminal failure. +func BuildFailure() (sqsim.Scenario, error) { + behavior := sqsim.NewBehavior(). + BuildRunner(sqsim.BuildFailedAfter(time.Second)). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + return singleErrorScenario(behavior) +} + +// MergeConflict returns a request rejected by the dry-run mergeability check. +func MergeConflict() (sqsim.Scenario, error) { + behavior := sqsim.NewBehavior(). + BuildRunner(sqsim.SuccessfulBuildRunner()). + MergeConflictCheck(sqsim.ConflictingMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + return singleErrorScenario(behavior) +} + +func singleErrorScenario(behavior *sqsim.BehaviorBuilder) (sqsim.Scenario, error) { + return sqsim.NewScenario(). + Timeout(30 * time.Second). + Land( + sqsim.NewLand("l1"). + Queue("sqsim"). + Behavior(behavior). + Expect(sqsim.RequestError), + ). + Build() +} diff --git a/sqsim/scenarios/failure_test.go b/sqsim/scenarios/failure_test.go new file mode 100644 index 00000000..7893edad --- /dev/null +++ b/sqsim/scenarios/failure_test.go @@ -0,0 +1,38 @@ +// 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 scenarios + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim" +) + +func TestFailureScenariosExpectError(t *testing.T) { + tests := map[string]Builder{ + "build failure": BuildFailure, + "merge conflict": MergeConflict, + } + for name, build := range tests { + t.Run(name, func(t *testing.T) { + scenario, err := build() + require.NoError(t, err) + require.Len(t, scenario.Lands, 1) + assert.Equal(t, sqsim.RequestError, scenario.Lands[0].Expectation.Status) + }) + } +} diff --git a/sqsim/scenarios/load.go b/sqsim/scenarios/load.go new file mode 100644 index 00000000..40e5c9d7 --- /dev/null +++ b/sqsim/scenarios/load.go @@ -0,0 +1,46 @@ +// 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 scenarios + +import ( + "fmt" + "time" + + "github.com/uber/submitqueue/sqsim" +) + +const loadRequestCount = 1000 + +// Load1000 returns an opt-in workload of one thousand successful requests. +func Load1000() (sqsim.Scenario, error) { + lands := make([]*sqsim.LandBuilder, 0, loadRequestCount) + for i := 1; i <= loadRequestCount; i++ { + behavior := sqsim.NewBehavior(). + BuildRunner(sqsim.BuildSucceededAfter(time.Second)). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + lands = append(lands, + sqsim.NewLand(fmt.Sprintf("l%04d", i)). + Queue("sqsim"). + Behavior(behavior). + Expect(sqsim.RequestLanded), + ) + } + + return sqsim.NewScenario(). + Timeout(10 * time.Minute). + Land(lands...). + Build() +} diff --git a/sqsim/scenarios/load_test.go b/sqsim/scenarios/load_test.go new file mode 100644 index 00000000..72ac29db --- /dev/null +++ b/sqsim/scenarios/load_test.go @@ -0,0 +1,30 @@ +// 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 scenarios + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLoad1000BuildsWithoutRunning(t *testing.T) { + scenario, err := Load1000() + require.NoError(t, err) + require.Len(t, scenario.Lands, loadRequestCount) + assert.Equal(t, "l0001", scenario.Lands[0].Name) + assert.Equal(t, "l1000", scenario.Lands[len(scenario.Lands)-1].Name) +} diff --git a/sqsim/scenarios/mixed.go b/sqsim/scenarios/mixed.go new file mode 100644 index 00000000..fd01f40a --- /dev/null +++ b/sqsim/scenarios/mixed.go @@ -0,0 +1,57 @@ +// 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 scenarios + +import ( + "time" + + "github.com/uber/submitqueue/sqsim" +) + +// MixedConcurrent returns staggered requests with successful, conflict, and build-failure outcomes. +func MixedConcurrent() (sqsim.Scenario, error) { + success := sqsim.NewBehavior(). + BuildRunner(sqsim.BuildSucceededAfter(8 * time.Second)). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.NewMergeBehavior().Invoke(sqsim.MergeSucceededAfter(time.Second))) + conflict := sqsim.NewBehavior(). + BuildRunner(sqsim.SuccessfulBuildRunner()). + MergeConflictCheck(sqsim.ConflictingMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + buildFailure := sqsim.NewBehavior(). + BuildRunner(sqsim.BuildFailedAfter(5 * time.Second)). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) + + return sqsim.NewScenario(). + Timeout(90*time.Second). + Land( + sqsim.NewLand("lands"). + Queue("sqsim"). + Behavior(success). + Expect(sqsim.RequestLanded), + sqsim.NewLand("conflicts"). + Queue("sqsim"). + SubmitAfter(2*time.Second). + Behavior(conflict). + Expect(sqsim.RequestError), + sqsim.NewLand("build-fails"). + Queue("sqsim"). + SubmitAfter(4*time.Second). + Behavior(buildFailure). + Expect(sqsim.RequestError), + ). + Build() +} diff --git a/sqsim/scenarios/mixed_test.go b/sqsim/scenarios/mixed_test.go new file mode 100644 index 00000000..dee8bbb8 --- /dev/null +++ b/sqsim/scenarios/mixed_test.go @@ -0,0 +1,38 @@ +// 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 scenarios + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim" +) + +func TestMixedConcurrent(t *testing.T) { + scenario, err := MixedConcurrent() + require.NoError(t, err) + require.Len(t, scenario.Lands, 3) + assert.Equal(t, []sqsim.ExpectedRequestStatus{ + sqsim.RequestLanded, + sqsim.RequestError, + sqsim.RequestError, + }, []sqsim.ExpectedRequestStatus{ + scenario.Lands[0].Expectation.Status, + scenario.Lands[1].Expectation.Status, + scenario.Lands[2].Expectation.Status, + }) +} diff --git a/sqsim/scenarios/recovery.go b/sqsim/scenarios/recovery.go new file mode 100644 index 00000000..66e0fcc8 --- /dev/null +++ b/sqsim/scenarios/recovery.go @@ -0,0 +1,88 @@ +// 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 scenarios + +import ( + "time" + + "github.com/uber/submitqueue/sqsim" +) + +// BuildStatusRecovery returns a request whose first Build Runner Status call fails transiently. +func BuildStatusRecovery() (sqsim.Scenario, error) { + buildRunner := sqsim.NewBuildRunnerBehavior(). + Trigger(sqsim.BuildCreated( + sqsim.StatusAt(0, sqsim.BuildAccepted), + sqsim.StatusAt(500*time.Millisecond, sqsim.BuildRunning), + sqsim.StatusAt(2*time.Second, sqsim.BuildSucceeded), + )). + StatusFaultOnCall(1, sqsim.RetryableErrorBeforeSideEffect()) + + return singleLandedScenario(buildRunner, sqsim.SuccessfulMergeConflictCheck(), sqsim.SuccessfulMerge()) +} + +// BuildTriggerRecovery returns a request whose first Build Runner Trigger call fails transiently. +func BuildTriggerRecovery() (sqsim.Scenario, error) { + buildRunner := sqsim.NewBuildRunnerBehavior().Trigger( + sqsim.BuildTriggerFault(sqsim.RetryableErrorBeforeSideEffect()), + sqsim.BuildCreated( + sqsim.StatusAt(0, sqsim.BuildAccepted), + sqsim.StatusAt(time.Second, sqsim.BuildSucceeded), + ), + ) + + return singleLandedScenario(buildRunner, sqsim.SuccessfulMergeConflictCheck(), sqsim.SuccessfulMerge()) +} + +// MergeConflictCheckRecovery returns a request whose first dry-run mergeability check fails transiently. +func MergeConflictCheckRecovery() (sqsim.Scenario, error) { + check := sqsim.NewMergeConflictCheckBehavior().Invoke( + sqsim.MergeConflictCheckFault(sqsim.RetryableErrorBeforeSideEffect()), + sqsim.MergeConflictCheckSucceeded(), + ) + + return singleLandedScenario(sqsim.BuildSucceededAfter(time.Second), check, sqsim.SuccessfulMerge()) +} + +// MergeResponseLost returns a request whose merge succeeds before its first response is lost. +func MergeResponseLost() (sqsim.Scenario, error) { + merge := sqsim.NewMergeBehavior().Invoke( + sqsim.MergeSucceededAfter(500 * time.Millisecond). + Fault(sqsim.RetryableErrorAfterSideEffect()), + ) + + return singleLandedScenario(sqsim.BuildSucceededAfter(time.Second), sqsim.SuccessfulMergeConflictCheck(), merge) +} + +func singleLandedScenario( + buildRunner *sqsim.BuildRunnerBehaviorBuilder, + check *sqsim.MergeConflictCheckBehaviorBuilder, + merge *sqsim.MergeBehaviorBuilder, +) (sqsim.Scenario, error) { + behavior := sqsim.NewBehavior(). + BuildRunner(buildRunner). + MergeConflictCheck(check). + Merge(merge) + + return sqsim.NewScenario(). + Timeout(45 * time.Second). + Land( + sqsim.NewLand("l1"). + Queue("sqsim"). + Behavior(behavior). + Expect(sqsim.RequestLanded), + ). + Build() +} diff --git a/sqsim/scenarios/recovery_test.go b/sqsim/scenarios/recovery_test.go new file mode 100644 index 00000000..4f1014c8 --- /dev/null +++ b/sqsim/scenarios/recovery_test.go @@ -0,0 +1,40 @@ +// 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 scenarios + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim" +) + +func TestRecoveryScenariosExpectLanding(t *testing.T) { + tests := map[string]Builder{ + "build status": BuildStatusRecovery, + "build trigger": BuildTriggerRecovery, + "merge conflict check": MergeConflictCheckRecovery, + "merge response lost": MergeResponseLost, + } + for name, build := range tests { + t.Run(name, func(t *testing.T) { + scenario, err := build() + require.NoError(t, err) + require.Len(t, scenario.Lands, 1) + assert.Equal(t, sqsim.RequestLanded, scenario.Lands[0].Expectation.Status) + }) + } +} diff --git a/sqsim/scenarios/registry.go b/sqsim/scenarios/registry.go index f00847a0..083f4792 100644 --- a/sqsim/scenarios/registry.go +++ b/sqsim/scenarios/registry.go @@ -26,7 +26,15 @@ import ( type Builder func() (sqsim.Scenario, error) var registry = map[string]Builder{ - "happy-path": HappyPath, + "build-failure": BuildFailure, + "build-status-recovery": BuildStatusRecovery, + "build-trigger-recovery": BuildTriggerRecovery, + "happy-path": HappyPath, + "load-1000": Load1000, + "merge-conflict": MergeConflict, + "merge-conflict-check-recovery": MergeConflictCheckRecovery, + "merge-response-lost": MergeResponseLost, + "mixed-concurrent": MixedConcurrent, } // Names returns registered scenario names in lexical order. diff --git a/sqsim/scenarios/registry_test.go b/sqsim/scenarios/registry_test.go index 67e8fe5d..38fbb397 100644 --- a/sqsim/scenarios/registry_test.go +++ b/sqsim/scenarios/registry_test.go @@ -22,7 +22,17 @@ import ( ) func TestRegistry(t *testing.T) { - assert.Equal(t, []string{"happy-path"}, Names()) + assert.Equal(t, []string{ + "build-failure", + "build-status-recovery", + "build-trigger-recovery", + "happy-path", + "load-1000", + "merge-conflict", + "merge-conflict-check-recovery", + "merge-response-lost", + "mixed-concurrent", + }, Names()) require.NoError(t, ValidateAll()) } diff --git a/sqsim/tui/BUILD.bazel b/sqsim/tui/BUILD.bazel index e1686a7a..4d7c889c 100644 --- a/sqsim/tui/BUILD.bazel +++ b/sqsim/tui/BUILD.bazel @@ -16,7 +16,10 @@ go_library( go_test( name = "go_default_test", - srcs = ["tui_test.go"], + srcs = [ + "stage_test.go", + "tui_test.go", + ], embed = [":go_default_library"], deps = [ "//sqsim/runner:go_default_library", diff --git a/sqsim/tui/stage_test.go b/sqsim/tui/stage_test.go new file mode 100644 index 00000000..3173c3ab --- /dev/null +++ b/sqsim/tui/stage_test.go @@ -0,0 +1,39 @@ +// 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 tui + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim/runner" +) + +func TestStageCellsUseHistory(t *testing.T) { + cells := stageCells(runner.Request{ + Status: "building", + History: []runner.HistoryEvent{ + {Status: "validating"}, + {Status: "validated"}, + {Status: "batched"}, + {Status: "scored"}, + {Status: "speculating"}, + {Status: "building"}, + }, + }, "/") + require.Len(t, cells, 6) + assert.Equal(t, []string{"ok", "ok", "ok", "ok", "/", "."}, cells[:]) +} diff --git a/sqsim/tui/tui_test.go b/sqsim/tui/tui_test.go index b506d234..1c190ee9 100644 --- a/sqsim/tui/tui_test.go +++ b/sqsim/tui/tui_test.go @@ -22,7 +22,6 @@ import ( tea "github.com/charmbracelet/bubbletea" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" "github.com/uber/submitqueue/sqsim/runner" ) @@ -149,22 +148,6 @@ func TestHistoryWindowScrollsFromLatest(t *testing.T) { assert.Equal(t, 6, end) } -func TestStageCellsUseHistory(t *testing.T) { - cells := stageCells(runner.Request{ - Status: "building", - History: []runner.HistoryEvent{ - {Status: "validating"}, - {Status: "validated"}, - {Status: "batched"}, - {Status: "scored"}, - {Status: "speculating"}, - {Status: "building"}, - }, - }, "/") - require.Len(t, cells, 6) - assert.Equal(t, []string{"ok", "ok", "ok", "ok", "/", "."}, cells[:]) -} - func TestTruncate(t *testing.T) { assert.Equal(t, "abcdef", truncate("abcdef", 6)) assert.True(t, strings.HasSuffix(truncate("abcdefgh", 6), "...")) diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index b39a26ef..d23c88af 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "batch.go", + "batch_build.go", "batch_changes.go", "batch_dependent.go", "build.go", diff --git a/submitqueue/entity/batch_build.go b/submitqueue/entity/batch_build.go new file mode 100644 index 00000000..0e135e12 --- /dev/null +++ b/submitqueue/entity/batch_build.go @@ -0,0 +1,23 @@ +// 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 entity + +// BatchBuild maps a batch to the build that verifies it. +type BatchBuild struct { + // BatchID is the globally unique batch identifier and primary key. + BatchID string + // BuildID is the globally unique build identifier associated with the batch. + BuildID string +} diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index d2081a00..df26b8b8 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library") go_library( name = "go_default_library", srcs = [ + "batch_build_store.go", "batch_dependent_store.go", "batch_store.go", "build_store.go", diff --git a/submitqueue/extension/storage/batch_build_store.go b/submitqueue/extension/storage/batch_build_store.go new file mode 100644 index 00000000..943e4c35 --- /dev/null +++ b/submitqueue/extension/storage/batch_build_store.go @@ -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=batch_build_store.go -destination=mock/batch_build_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// BatchBuildStore manages immutable batch-to-build mappings. +type BatchBuildStore interface { + // Create stores a batch-to-build mapping. + // Returns ErrAlreadyExists if a mapping for the batch already exists. + Create(ctx context.Context, batchBuild entity.BatchBuild) error + + // Get retrieves the mapping for a batch. + // Returns ErrNotFound if the batch has no mapping. + Get(ctx context.Context, batchID string) (entity.BatchBuild, error) +} diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel index 4b2f4f51..45add41f 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library") go_library( name = "go_default_library", srcs = [ + "batch_build_store_mock.go", "batch_dependent_store_mock.go", "batch_store_mock.go", "build_store_mock.go", diff --git a/submitqueue/extension/storage/mock/batch_build_store_mock.go b/submitqueue/extension/storage/mock/batch_build_store_mock.go new file mode 100644 index 00000000..f6f7616b --- /dev/null +++ b/submitqueue/extension/storage/mock/batch_build_store_mock.go @@ -0,0 +1,71 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: batch_build_store.go +// +// Generated by this command: +// +// mockgen -source=batch_build_store.go -destination=mock/batch_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" +) + +// MockBatchBuildStore is a mock of BatchBuildStore interface. +type MockBatchBuildStore struct { + ctrl *gomock.Controller + recorder *MockBatchBuildStoreMockRecorder + isgomock struct{} +} + +// MockBatchBuildStoreMockRecorder is the mock recorder for MockBatchBuildStore. +type MockBatchBuildStoreMockRecorder struct { + mock *MockBatchBuildStore +} + +// NewMockBatchBuildStore creates a new mock instance. +func NewMockBatchBuildStore(ctrl *gomock.Controller) *MockBatchBuildStore { + mock := &MockBatchBuildStore{ctrl: ctrl} + mock.recorder = &MockBatchBuildStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockBatchBuildStore) EXPECT() *MockBatchBuildStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockBatchBuildStore) Create(ctx context.Context, batchBuild entity.BatchBuild) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, batchBuild) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockBatchBuildStoreMockRecorder) Create(ctx, batchBuild any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockBatchBuildStore)(nil).Create), ctx, batchBuild) +} + +// Get mocks base method. +func (m *MockBatchBuildStore) Get(ctx context.Context, batchID string) (entity.BatchBuild, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, batchID) + ret0, _ := ret[0].(entity.BatchBuild) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockBatchBuildStoreMockRecorder) Get(ctx, batchID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockBatchBuildStore)(nil).Get), ctx, batchID) +} diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index 6655716f..9ed1e2c1 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -54,6 +54,20 @@ func (mr *MockStorageMockRecorder) Close() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockStorage)(nil).Close)) } +// GetBatchBuildStore mocks base method. +func (m *MockStorage) GetBatchBuildStore() storage.BatchBuildStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetBatchBuildStore") + ret0, _ := ret[0].(storage.BatchBuildStore) + return ret0 +} + +// GetBatchBuildStore indicates an expected call of GetBatchBuildStore. +func (mr *MockStorageMockRecorder) GetBatchBuildStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBatchBuildStore", reflect.TypeOf((*MockStorage)(nil).GetBatchBuildStore)) +} + // GetBatchDependentStore mocks base method. func (m *MockStorage) GetBatchDependentStore() storage.BatchDependentStore { m.ctrl.T.Helper() @@ -152,20 +166,6 @@ func (mr *MockStorageMockRecorder) GetRequestStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestStore", reflect.TypeOf((*MockStorage)(nil).GetRequestStore)) } -// GetSpeculationPathBuildStore mocks base method. -func (m *MockStorage) GetSpeculationPathBuildStore() storage.SpeculationPathBuildStore { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetSpeculationPathBuildStore") - ret0, _ := ret[0].(storage.SpeculationPathBuildStore) - return ret0 -} - -// GetSpeculationPathBuildStore indicates an expected call of GetSpeculationPathBuildStore. -func (mr *MockStorageMockRecorder) GetSpeculationPathBuildStore() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSpeculationPathBuildStore", reflect.TypeOf((*MockStorage)(nil).GetSpeculationPathBuildStore)) -} - // GetRequestSummaryStore mocks base method. func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore { m.ctrl.T.Helper() @@ -194,6 +194,20 @@ func (mr *MockStorageMockRecorder) GetRequestURIStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestURIStore", reflect.TypeOf((*MockStorage)(nil).GetRequestURIStore)) } +// GetSpeculationPathBuildStore mocks base method. +func (m *MockStorage) GetSpeculationPathBuildStore() storage.SpeculationPathBuildStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetSpeculationPathBuildStore") + ret0, _ := ret[0].(storage.SpeculationPathBuildStore) + return ret0 +} + +// GetSpeculationPathBuildStore indicates an expected call of GetSpeculationPathBuildStore. +func (mr *MockStorageMockRecorder) GetSpeculationPathBuildStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSpeculationPathBuildStore", reflect.TypeOf((*MockStorage)(nil).GetSpeculationPathBuildStore)) +} + // GetSpeculationTreeStore mocks base method. func (m *MockStorage) GetSpeculationTreeStore() storage.SpeculationTreeStore { m.ctrl.T.Helper() diff --git a/submitqueue/extension/storage/mysql/BUILD.bazel b/submitqueue/extension/storage/mysql/BUILD.bazel index 1cbbf031..efcfbf83 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "batch_build_store.go", "batch_dependent_store.go", "batch_store.go", "build_store.go", @@ -30,6 +31,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "batch_build_store_test.go", "batch_dependent_store_test.go", "batch_store_test.go", "build_store_test.go", diff --git a/submitqueue/extension/storage/mysql/batch_build_store.go b/submitqueue/extension/storage/mysql/batch_build_store.go new file mode 100644 index 00000000..1189a065 --- /dev/null +++ b/submitqueue/extension/storage/mysql/batch_build_store.go @@ -0,0 +1,79 @@ +// 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/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 batchBuildStore struct { + db *sql.DB + scope tally.Scope +} + +// NewBatchBuildStore creates a MySQL-backed BatchBuildStore. +func NewBatchBuildStore(db *sql.DB, scope tally.Scope) storage.BatchBuildStore { + return &batchBuildStore{db: db, scope: scope} +} + +// Create stores an immutable batch-to-build mapping. +func (s *batchBuildStore) Create(ctx context.Context, batchBuild entity.BatchBuild) (retErr error) { + op := metrics.Begin(s.scope, "create") + defer func() { op.Complete(retErr) }() + + _, err := s.db.ExecContext( + ctx, + "INSERT INTO batch_build (batch_id, build_id) VALUES (?, ?)", + batchBuild.BatchID, + batchBuild.BuildID, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("batch build mapping batch_id=%s: %w", batchBuild.BatchID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert batch build mapping batch_id=%s: %w", batchBuild.BatchID, err) + } + return nil +} + +// Get retrieves the mapping for a batch. +func (s *batchBuildStore) Get(ctx context.Context, batchID string) (ret entity.BatchBuild, retErr error) { + op := metrics.Begin(s.scope, "get") + defer func() { op.Complete(retErr) }() + + err := s.db.QueryRowContext( + ctx, + "SELECT batch_id, build_id FROM batch_build WHERE batch_id = ?", + batchID, + ).Scan(&ret.BatchID, &ret.BuildID) + if errors.Is(err, sql.ErrNoRows) { + return entity.BatchBuild{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.BatchBuild{}, fmt.Errorf("failed to get batch build mapping batch_id=%s: %w", batchID, err) + } + return ret, nil +} diff --git a/submitqueue/extension/storage/mysql/batch_build_store_test.go b/submitqueue/extension/storage/mysql/batch_build_store_test.go new file mode 100644 index 00000000..5872f399 --- /dev/null +++ b/submitqueue/extension/storage/mysql/batch_build_store_test.go @@ -0,0 +1,103 @@ +// 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" + "fmt" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func TestBatchBuildStore_CreateAndGet(t *testing.T) { + mapping := entity.BatchBuild{BatchID: "queue/batch/1", BuildID: "build-1"} + + tests := []struct { + name string + run func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) + }{ + { + name: "create", + run: func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) { + mock.ExpectExec("INSERT INTO batch_build"). + WithArgs(mapping.BatchID, mapping.BuildID). + WillReturnResult(sqlmock.NewResult(0, 1)) + require.NoError(t, store.Create(context.Background(), mapping)) + }, + }, + { + name: "create duplicate", + run: func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) { + mock.ExpectExec("INSERT INTO batch_build"). + WithArgs(mapping.BatchID, mapping.BuildID). + WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) + err := store.Create(context.Background(), mapping) + require.Error(t, err) + assert.ErrorIs(t, err, storage.ErrAlreadyExists) + }, + }, + { + name: "get", + run: func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) { + mock.ExpectQuery("SELECT batch_id, build_id FROM batch_build"). + WithArgs(mapping.BatchID). + WillReturnRows(sqlmock.NewRows([]string{"batch_id", "build_id"}).AddRow(mapping.BatchID, mapping.BuildID)) + got, err := store.Get(context.Background(), mapping.BatchID) + require.NoError(t, err) + assert.Equal(t, mapping, got) + }, + }, + { + name: "get missing", + run: func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) { + mock.ExpectQuery("SELECT batch_id, build_id FROM batch_build"). + WithArgs(mapping.BatchID). + WillReturnError(sql.ErrNoRows) + _, err := store.Get(context.Background(), mapping.BatchID) + require.Error(t, err) + assert.ErrorIs(t, err, storage.ErrNotFound) + }, + }, + { + name: "get failure", + run: func(t *testing.T, mock sqlmock.Sqlmock, store storage.BatchBuildStore) { + mock.ExpectQuery("SELECT batch_id, build_id FROM batch_build"). + WithArgs(mapping.BatchID). + WillReturnError(fmt.Errorf("connection reset")) + _, err := store.Get(context.Background(), mapping.BatchID) + require.Error(t, err) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + + tt.run(t, mock, NewBatchBuildStore(db, testMetrics())) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/submitqueue/extension/storage/mysql/schema/batch_build.sql b/submitqueue/extension/storage/mysql/schema/batch_build.sql new file mode 100644 index 00000000..86b22c81 --- /dev/null +++ b/submitqueue/extension/storage/mysql/schema/batch_build.sql @@ -0,0 +1,5 @@ +CREATE TABLE IF NOT EXISTS batch_build ( + batch_id VARCHAR(255) NOT NULL, + build_id VARCHAR(255) NOT NULL, + PRIMARY KEY (batch_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/submitqueue/extension/storage/mysql/storage.go b/submitqueue/extension/storage/mysql/storage.go index ffcf4d02..42c00933 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -33,6 +33,7 @@ type mysqlStorage struct { changeStore storage.ChangeStore batchStore storage.BatchStore batchDependentStore storage.BatchDependentStore + batchBuildStore storage.BatchBuildStore buildStore storage.BuildStore speculationPathBuildStore storage.SpeculationPathBuildStore speculationTreeStore storage.SpeculationTreeStore @@ -50,6 +51,7 @@ func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) { changeStore: NewChangeStore(db, scope.SubScope("change_store")), batchStore: NewBatchStore(db, scope.SubScope("batch_store")), batchDependentStore: NewBatchDependentStore(db, scope.SubScope("batch_dependent_store")), + batchBuildStore: NewBatchBuildStore(db, scope.SubScope("batch_build_store")), buildStore: NewBuildStore(db, scope.SubScope("build_store")), speculationPathBuildStore: NewSpeculationPathBuildStore(db, scope.SubScope("speculation_path_build_store")), speculationTreeStore: NewSpeculationTreeStore(db, scope.SubScope("speculation_tree_store")), @@ -80,6 +82,11 @@ func (f *mysqlStorage) GetBatchDependentStore() storage.BatchDependentStore { return f.batchDependentStore } +// GetBatchBuildStore returns the MySQL-backed BatchBuildStore. +func (f *mysqlStorage) GetBatchBuildStore() storage.BatchBuildStore { + return f.batchBuildStore +} + // GetBuildStore returns the MySQL-backed BuildStore. func (f *mysqlStorage) GetBuildStore() storage.BuildStore { return f.buildStore diff --git a/submitqueue/extension/storage/mysql/storage_test.go b/submitqueue/extension/storage/mysql/storage_test.go index bf0cb68b..cc0be945 100644 --- a/submitqueue/extension/storage/mysql/storage_test.go +++ b/submitqueue/extension/storage/mysql/storage_test.go @@ -40,6 +40,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, s.GetChangeStore()) assert.NotNil(t, s.GetBatchStore()) assert.NotNil(t, s.GetBatchDependentStore()) + assert.NotNil(t, s.GetBatchBuildStore()) assert.NotNil(t, s.GetBuildStore()) assert.NotNil(t, s.GetSpeculationPathBuildStore()) assert.NotNil(t, s.GetSpeculationTreeStore()) diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index b54736c3..a87894dc 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -56,6 +56,9 @@ type Storage interface { // GetBatchDependentStore returns the BatchDependentStore instance. GetBatchDependentStore() BatchDependentStore + // GetBatchBuildStore returns the BatchBuildStore instance. + GetBatchBuildStore() BatchBuildStore + // GetBuildStore returns the BuildStore instance. GetBuildStore() BuildStore diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index 30493c74..62f154be 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -47,6 +47,8 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +const opName = "process" + // NewController creates a new build controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -72,8 +74,6 @@ func NewController( // Deserializes the batch, triggers a build, and publishes a build entity to the build signal topic. // Returns nil to ack (success), or error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - const opName = "process" - op := metrics.Begin(c.metricsScope, opName) defer func() { op.Complete(retErr) }() @@ -116,51 +116,20 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil } - // Load the dependency batches (base) as identity; the build runner resolves - // each batch's changes itself. head is this batch. - base, err := c.loadBatches(ctx, batch.Dependencies) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to load dependency batches for batch %s: %w", batch.ID, err) - } - - // Trigger the build with the queue's build runner. metadata is nil - // until a caller-supplied source materializes (e.g. requester / ticket - // pulled off the originating LandRequest). - buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue}) + build, err := c.resolveBuild(ctx, batch) if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) - return fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) - } - buildID, err := buildRunner.Trigger(ctx, base, batch, nil) - if err != nil { - metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) - return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err) + return err } if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusBuilding, map[string]string{ "batch_id": batch.ID, - "build_id": buildID.ID, + "build_id": build.ID, "controller": "build", }); err != nil { metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1) return fmt.Errorf("failed to publish building request logs for batch %s: %w", batch.ID, err) } - build := entity.Build{ - ID: buildID.ID, - BatchID: batch.ID, - Status: entity.BuildStatusAccepted, - } - - // Persist the initial Build snapshot so the buildsignal poll loop has a - // row to UpdateStatus against. ErrAlreadyExists is benign — a redelivery - // of this message after a previous successful Create. - if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { - metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to persist build %s: %w", build.ID, err) - } - // Hand off to the buildsignal poll loop; it calls Status, updates the // persisted Build, publishes to speculate, and re-publishes itself via // PublishAfter until terminal. @@ -179,6 +148,72 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil // Success - message will be acked } +// resolveBuild returns the build assigned to a batch, triggering it only when +// no immutable batch-to-build mapping exists. Persisting the mapping before +// the Build row lets redelivery repair a missing Build without triggering the +// external provider again. +func (c *Controller) resolveBuild(ctx context.Context, batch entity.Batch) (entity.Build, error) { + mappingStore := c.store.GetBatchBuildStore() + mapping, err := mappingStore.Get(ctx, batch.ID) + if err == nil { + return c.ensureBuild(ctx, mapping) + } + if !errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.Build{}, fmt.Errorf("failed to get build mapping for batch %s: %w", batch.ID, err) + } + + // Load the dependency batches as the speculative base. The build runner + // resolves each batch's changes itself, and the current batch is the head. + base, err := c.loadBatches(ctx, batch.Dependencies) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.Build{}, fmt.Errorf("failed to load dependency batches for batch %s: %w", batch.ID, err) + } + + buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue}) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) + return entity.Build{}, fmt.Errorf("failed to build runner for batch %s: %w", batch.ID, err) + } + buildID, err := buildRunner.Trigger(ctx, base, batch, nil) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1) + return entity.Build{}, fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err) + } + + mapping = entity.BatchBuild{BatchID: batch.ID, BuildID: buildID.ID} + if err := mappingStore.Create(ctx, mapping); err != nil { + if !errors.Is(err, storage.ErrAlreadyExists) { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.Build{}, fmt.Errorf("failed to persist build mapping for batch %s: %w", batch.ID, err) + } + + // Another delivery won the mapping race. Its mapping is authoritative. + mapping, err = mappingStore.Get(ctx, batch.ID) + if err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.Build{}, fmt.Errorf("failed to reload build mapping for batch %s: %w", batch.ID, err) + } + } + + return c.ensureBuild(ctx, mapping) +} + +// ensureBuild persists the initial Build snapshot if it is missing. +func (c *Controller) ensureBuild(ctx context.Context, mapping entity.BatchBuild) (entity.Build, error) { + build := entity.Build{ + ID: mapping.BuildID, + BatchID: mapping.BatchID, + Status: entity.BuildStatusAccepted, + } + if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return entity.Build{}, fmt.Errorf("failed to persist build %s: %w", build.ID, err) + } + return build, nil +} + // loadBatches loads each batch by ID, preserving order. Used to load the base // (dependency batches) identity handed to BuildRunner.Trigger; the build runner // resolves each batch's changes itself. diff --git a/submitqueue/orchestrator/controller/build/build_test.go b/submitqueue/orchestrator/controller/build/build_test.go index 38a662cb..6ad8ebe2 100644 --- a/submitqueue/orchestrator/controller/build/build_test.go +++ b/submitqueue/orchestrator/controller/build/build_test.go @@ -55,22 +55,24 @@ func testBatch() entity.Batch { } } -// newMockStorage creates a MockStorage with a MockBatchStore that returns the -// given batch on Get, a no-op MockRequestStore, and a MockBuildStore that -// accepts any Create call. Tests that care about Create arguments build their -// own MockBuildStore. +// newMockStorage creates storage for the normal first-delivery path. func newMockStorage(ctrl *gomock.Controller, batch entity.Batch) *storagemock.MockStorage { mockBatchStore := storagemock.NewMockBatchStore(ctrl) mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() mockRequestStore := storagemock.NewMockRequestStore(ctrl) + mockBatchBuildStore := storagemock.NewMockBatchBuildStore(ctrl) + mockBatchBuildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound).AnyTimes() + mockBatchBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + mockBuildStore := storagemock.NewMockBuildStore(ctrl) mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(mockRequestStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mockBatchBuildStore).AnyTimes() store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() return store } @@ -166,6 +168,15 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { mockBatchStore.EXPECT().Get(gomock.Any(), depBatch.ID).Return(depBatch, nil).AnyTimes() var created entity.Build + var createdMapping entity.BatchBuild + mockBatchBuildStore := storagemock.NewMockBatchBuildStore(ctrl) + mockBatchBuildStore.EXPECT().Get(gomock.Any(), headBatch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound) + mockBatchBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, mapping entity.BatchBuild) error { + createdMapping = mapping + return nil + }, + ) mockBuildStore := storagemock.NewMockBuildStore(ctrl) mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, b entity.Build) error { @@ -176,6 +187,7 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mockBatchBuildStore).AnyTimes() store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() br := buildrunnermock.NewMockBuildRunner(ctrl) @@ -233,29 +245,31 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { assert.Equal(t, headBatch.ID, created.BatchID) assert.Equal(t, entity.BuildStatusAccepted, created.Status) assert.Equal(t, published.ID, created.ID) + assert.Equal(t, entity.BatchBuild{BatchID: headBatch.ID, BuildID: "build-xyz"}, createdMapping) } -// TestController_Process_BuildStoreAlreadyExistsIsSwallowed covers the -// redelivery case: Create returns ErrAlreadyExists, the controller proceeds -// to publish to buildsignal anyway. The polling loop will pick up the -// existing row via UpdateStatus. -func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) { +// TestController_Process_RedeliveryReusesMapping verifies that a delivery +// after the external trigger does not trigger a second build. +func TestController_Process_RedeliveryReusesMapping(t *testing.T) { ctrl := gomock.NewController(t) batch := testBatch() + mapping := entity.BatchBuild{BatchID: batch.ID, BuildID: "build-existing"} mockBatchStore := storagemock.NewMockBatchStore(ctrl) mockBatchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + mockBatchBuildStore := storagemock.NewMockBatchBuildStore(ctrl) + mockBatchBuildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(mapping, nil) mockBuildStore := storagemock.NewMockBuildStore(ctrl) mockBuildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mockBatchBuildStore).AnyTimes() store.EXPECT().GetBuildStore().Return(mockBuildStore).AnyTimes() br := buildrunnermock.NewMockBuildRunner(ctrl) - br.EXPECT().Trigger(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.BuildID{ID: "build-dup"}, nil) publishCalled := false mockPub := queuemock.NewMockPublisher(ctrl) @@ -282,7 +296,7 @@ func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) { delivery.EXPECT().Attempt().Return(1).AnyTimes() require.NoError(t, controller.Process(context.Background(), delivery)) - assert.True(t, publishCalled, "publish to buildsignal must run even when Create reports ErrAlreadyExists") + assert.True(t, publishCalled, "redelivery must republish to buildsignal") } // TestController_Process_TriggerFailure verifies a build-runner failure is @@ -296,6 +310,9 @@ func TestController_Process_TriggerFailure(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() + mockBatchBuildStore := storagemock.NewMockBatchBuildStore(ctrl) + mockBatchBuildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound) + store.EXPECT().GetBatchBuildStore().Return(mockBatchBuildStore).AnyTimes() // No build store expectation: Trigger failure must short-circuit before Create. br := buildrunnermock.NewMockBuildRunner(ctrl) diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index dddba2ad..3d847a2b 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -24,6 +24,8 @@ package buildsignal import ( "context" + "crypto/rand" + "encoding/hex" "fmt" "github.com/uber-go/tally" @@ -237,7 +239,14 @@ func (c *Controller) publishBuild(ctx context.Context, key consumer.TopicKey, bu return fmt.Errorf("failed to serialize build ID: %w", err) } - msg := entityqueue.NewMessage(build.ID, payload, build.BatchID, nil) + messageID := build.ID + if delayMs > 0 { + messageID, err = newEventMessageID("build-poll") + if err != nil { + return fmt.Errorf("failed to create build poll message ID: %w", err) + } + } + msg := entityqueue.NewMessage(messageID, payload, build.BatchID, nil) q, ok := c.registry.Queue(key) if !ok { @@ -264,7 +273,11 @@ func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, return fmt.Errorf("failed to serialize batch ID: %w", err) } - msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil) + messageID, err := newEventMessageID("build-status") + if err != nil { + return fmt.Errorf("failed to create build status event ID: %w", err) + } + msg := entityqueue.NewMessage(messageID, payload, partitionKey, nil) q, ok := c.registry.Queue(key) if !ok { @@ -283,6 +296,17 @@ func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, return nil } +// newEventMessageID returns a unique ID for one causal event. Polls and build +// status observations cannot reuse entity IDs because queue message IDs are +// idempotency keys within a partition. +func newEventMessageID(prefix string) (string, error) { + var suffix [16]byte + if _, err := rand.Read(suffix[:]); err != nil { + return "", err + } + return prefix + "-" + hex.EncodeToString(suffix[:]), nil +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "buildsignal" diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index 3e7c0982..aa590efa 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -152,6 +152,7 @@ func TestController_Process_Terminal(t *testing.T) { bid, err := entity.BatchIDFromBytes(msg.Payload) require.NoError(t, err) assert.Equal(t, build.BatchID, bid.ID) + assert.NotEqual(t, build.BatchID, msg.ID) return nil }).Times(1) // No PublishAfter expected on terminal. @@ -229,6 +230,7 @@ func TestController_Process_NonTerminal(t *testing.T) { require.NoError(t, err) // Re-published payload carries only the build ID. assert.Equal(t, build.ID, bid.ID) + assert.NotEqual(t, build.ID, msg.ID) return nil }).Times(1) diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 55f97ac0..8c4fa3a1 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -180,18 +180,53 @@ func (c *Controller) startSpeculation(ctx context.Context, batch entity.Batch) e return nil } -// tryFinalize publishes to merge and transitions to Merging iff every -// dependency batch has reached Succeeded. Cancelled deps are treated as -// out-of-the-way: the cancelled batch will never land, so it can no longer -// conflict — drop it from the chain and proceed. Failed deps still cascade -// via failOnDependency. If some deps are still in flight, the call is a -// no-op and waits for the next event. +// tryFinalize publishes to merge and transitions to Merging only after this +// batch's build and every dependency batch have succeeded. Cancelled +// dependencies are out of the way because they will never land, so they are +// dropped from the chain. Failed dependencies still cascade via +// failOnDependency. If some dependencies are still in flight, the call waits +// for the next event. // // TODO: when a dependency fails we currently fail this batch outright. // We will need to respeculate the failed paths — drop the failed dep // from the chain and re-issue speculation for the surviving ordering(s) // — instead of cascading the failure into requests that could still land. func (c *Controller) tryFinalize(ctx context.Context, batch entity.Batch) error { + mapping, err := c.store.GetBatchBuildStore().Get(ctx, batch.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "waiting_on_build", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get build mapping for batch %s: %w", batch.ID, err) + } + + build, err := c.store.GetBuildStore().Get(ctx, mapping.BuildID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "waiting_on_build", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get build %s for batch %s: %w", mapping.BuildID, batch.ID, err) + } + + switch build.Status { + case entity.BuildStatusSucceeded: + // Continue to dependency evaluation. + case entity.BuildStatusFailed, entity.BuildStatusCancelled: + return c.failOnBuild(ctx, batch, build) + default: + metrics.NamedCounter(c.metricsScope, opName, "waiting_on_build", 1) + c.logger.Debugw("build still in flight; waiting", + "batch_id", batch.ID, + "build_id", build.ID, + "build_status", string(build.Status), + ) + return nil + } + deps, err := c.fetchDependencies(ctx, batch) if err != nil { return err @@ -240,11 +275,20 @@ func (c *Controller) tryFinalize(ctx context.Context, batch entity.Batch) error return nil } +// failOnBuild transitions a batch to Failed after its build reaches a +// non-succeeding terminal status. +func (c *Controller) failOnBuild(ctx context.Context, batch entity.Batch, build entity.Build) error { + metrics.NamedCounter(c.metricsScope, opName, "build_failed", 1) + c.logger.Warnw("build failed; failing batch", + "batch_id", batch.ID, + "build_id", build.ID, + "build_status", string(build.Status), + ) + return c.failBatch(ctx, batch) +} + // failOnDependency transitions a Speculating batch to Failed when one of its -// dependencies has reached a non-succeeding terminal state, then publishes to -// the conclude queue so the request store and request log get reconciled. -// Without this transition the batch would sit in Speculating forever — no -// downstream event ever fires for it again. +// dependencies has reached a non-succeeding terminal state. func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, dep entity.Batch) error { metrics.NamedCounter(c.metricsScope, opName, "dependency_failed", 1) c.logger.Warnw("dependency in non-succeeding terminal state; failing batch", @@ -252,7 +296,12 @@ func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, d "dependency_id", dep.ID, "dependency_state", string(dep.State), ) + return c.failBatch(ctx, batch) +} +// failBatch transitions a Speculating batch to Failed and publishes to +// conclude so request state is reconciled. +func (c *Controller) failBatch(ctx context.Context, batch entity.Batch) error { newVersion := batch.Version + 1 if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateFailed); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) @@ -336,24 +385,33 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error } // cancelBuild flips any in-flight Build entity for the batch to -// BuildStatusCancelled. Builds use build.ID == batch.ID, so a single Get -// covers every build scheduled for the batch. Tolerates ErrNotFound (no -// build was ever scheduled — the batch was cancelled before speculation -// started building) and skips already-terminal builds. +// BuildStatusCancelled. It resolves the runner-minted build ID through the +// immutable batch-to-build mapping. Missing mappings and Build rows are +// tolerated because cancellation can arrive before build scheduling finishes. // // This is the hook point for a future external CI integration: today the // system has no external runner, so the local state flip is the complete // cancellation. Once a runner exists, it must be invoked here before the // local UpdateStatus. func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error { - build, err := c.store.GetBuildStore().Get(ctx, batch.ID) + mapping, err := c.store.GetBatchBuildStore().Get(ctx, batch.ID) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + metrics.NamedCounter(c.metricsScope, opName, "cancel_build_not_found", 1) + return nil + } + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to get build mapping for batch %s: %w", batch.ID, err) + } + + build, err := c.store.GetBuildStore().Get(ctx, mapping.BuildID) if err != nil { if errors.Is(err, storage.ErrNotFound) { metrics.NamedCounter(c.metricsScope, opName, "cancel_build_not_found", 1) return nil } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to get build for batch %s: %w", batch.ID, err) + return fmt.Errorf("failed to get build %s for batch %s: %w", mapping.BuildID, batch.ID, err) } if build.Status.IsTerminal() { @@ -361,9 +419,9 @@ func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error return nil } - if err := c.store.GetBuildStore().UpdateStatus(ctx, batch.ID, entity.BuildStatusCancelled); err != nil { + if err := c.store.GetBuildStore().UpdateStatus(ctx, build.ID, entity.BuildStatusCancelled); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to cancel build for batch %s: %w", batch.ID, err) + return fmt.Errorf("failed to cancel build %s for batch %s: %w", build.ID, batch.ID, err) } metrics.NamedCounter(c.metricsScope, opName, "cancel_build_done", 1) return nil diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 2dbbd68b..0f9018f5 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -52,6 +52,21 @@ func testBatch(state entity.BatchState, deps ...string) entity.Batch { } } +// expectOwnBuildSucceeded satisfies the build gate for dependency-focused tests. +func expectOwnBuildSucceeded(ctrl *gomock.Controller, store *storagemock.MockStorage, batchID string) { + mapping := entity.BatchBuild{BatchID: batchID, BuildID: "build-1"} + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batchID).Return(mapping, nil) + buildStore := storagemock.NewMockBuildStore(ctrl) + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{ + ID: mapping.BuildID, + BatchID: batchID, + Status: entity.BuildStatusSucceeded, + }, nil) + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() + store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() +} + // newTestController wires a controller with a registry covering all topics the // speculate controller may publish to. The publisher returns publishErr (or nil). func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage, publishErr error) *Controller { @@ -188,6 +203,7 @@ func TestController_Process_FinalizeNoDeps(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + expectOwnBuildSucceeded(ctrl, store, batch.ID) controller := newTestController(t, ctrl, store, nil) require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) @@ -208,6 +224,7 @@ func TestController_Process_FinalizeAllDepsSucceeded(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + expectOwnBuildSucceeded(ctrl, store, batch.ID) controller := newTestController(t, ctrl, store, nil) require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) @@ -226,6 +243,7 @@ func TestController_Process_WaitingOnDep(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + expectOwnBuildSucceeded(ctrl, store, batch.ID) controller := newTestController(t, ctrl, store, nil) require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) @@ -246,6 +264,7 @@ func TestController_Process_FailedDepFailsBatch(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + expectOwnBuildSucceeded(ctrl, store, batch.ID) controller := newTestController(t, ctrl, store, nil) require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) @@ -268,11 +287,76 @@ func TestController_Process_CancelledDepSkipped(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + expectOwnBuildSucceeded(ctrl, store, batch.ID) controller := newTestController(t, ctrl, store, nil) require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) } +func TestController_Process_OwnBuildGate(t *testing.T) { + tests := []struct { + name string + status entity.BuildStatus + mappingExists bool + buildExists bool + wantFailed bool + }{ + {name: "mapping not created", mappingExists: false}, + {name: "build not created", mappingExists: true, buildExists: false}, + {name: "accepted", status: entity.BuildStatusAccepted, mappingExists: true, buildExists: true}, + {name: "running", status: entity.BuildStatusRunning, mappingExists: true, buildExists: true}, + {name: "failed", status: entity.BuildStatusFailed, mappingExists: true, buildExists: true, wantFailed: true}, + {name: "cancelled", status: entity.BuildStatusCancelled, mappingExists: true, buildExists: true, wantFailed: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + batch := testBatch(entity.BatchStateSpeculating) + mapping := entity.BatchBuild{BatchID: batch.ID, BuildID: "build-1"} + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + if tt.wantFailed { + batchStore.EXPECT().UpdateState( + gomock.Any(), + batch.ID, + int32(1), + int32(2), + entity.BatchStateFailed, + ).Return(nil) + } + + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + if tt.mappingExists { + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(mapping, nil) + } else { + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound) + } + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() + if tt.mappingExists { + buildStore := storagemock.NewMockBuildStore(ctrl) + if tt.buildExists { + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{ + ID: mapping.BuildID, + BatchID: batch.ID, + Status: tt.status, + }, nil) + } else { + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{}, storage.ErrNotFound) + } + store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + } + + controller := newTestController(t, ctrl, store, nil) + require.NoError(t, runProcess(t, ctrl, controller, batch.ID)) + }) + } +} + // Merging is owned by the merge controller — speculate is a no-op for it. func TestController_Process_MergingNoOp(t *testing.T) { ctrl := gomock.NewController(t) @@ -405,11 +489,14 @@ func TestController_Process_CancellingTerminalFlow(t *testing.T) { batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled).Return(nil) + mapping := entity.BatchBuild{BatchID: batch.ID, BuildID: "build-running"} + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(mapping, nil) buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{ - ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusRunning, + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{ + ID: mapping.BuildID, BatchID: batch.ID, Status: entity.BuildStatusRunning, }, nil) - buildStore.EXPECT().UpdateStatus(gomock.Any(), batch.ID, entity.BuildStatusCancelled).Return(nil) + buildStore.EXPECT().UpdateStatus(gomock.Any(), mapping.BuildID, entity.BuildStatusCancelled).Return(nil) depStore := storagemock.NewMockBatchDependentStore(ctrl) depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{ @@ -420,6 +507,7 @@ func TestController_Process_CancellingTerminalFlow(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() @@ -470,9 +558,12 @@ func TestController_Process_CancellingBuildAlreadyTerminal(t *testing.T) { batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled).Return(nil) + mapping := entity.BatchBuild{BatchID: batch.ID, BuildID: "build-succeeded"} + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(mapping, nil) buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{ - ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusSucceeded, + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{ + ID: mapping.BuildID, BatchID: batch.ID, Status: entity.BuildStatusSucceeded, }, nil) // No UpdateStatus expected — the build is already terminal. @@ -483,6 +574,7 @@ func TestController_Process_CancellingBuildAlreadyTerminal(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() @@ -501,8 +593,11 @@ func TestController_Process_CancellingNoBuildYet(t *testing.T) { batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled).Return(nil) + mapping := entity.BatchBuild{BatchID: batch.ID, BuildID: "build-not-persisted"} + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(mapping, nil) buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) + buildStore.EXPECT().Get(gomock.Any(), mapping.BuildID).Return(entity.Build{}, storage.ErrNotFound) // No UpdateStatus expected. depStore := storagemock.NewMockBatchDependentStore(ctrl) @@ -512,6 +607,7 @@ func TestController_Process_CancellingNoBuildYet(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() @@ -531,15 +627,15 @@ func TestController_Process_CancellingNoDependents(t *testing.T) { batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled).Return(nil) - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound) depStore := storagemock.NewMockBatchDependentStore(ctrl) depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{BatchID: batch.ID, Dependents: []string{}, Version: 1}, nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(depStore).AnyTimes() mockPub := queuemock.NewMockPublisher(ctrl) @@ -575,12 +671,12 @@ func TestController_Process_CancellingTerminalCASVersionMismatch(t *testing.T) { batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled). Return(storage.ErrVersionMismatch) - buildStore := storagemock.NewMockBuildStore(ctrl) - buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound) + mappingStore := storagemock.NewMockBatchBuildStore(ctrl) + mappingStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchBuild{}, storage.ErrNotFound) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() + store.EXPECT().GetBatchBuildStore().Return(mappingStore).AnyTimes() // BatchDependentStore must NOT be touched — terminal CAS failed before fan-out. // No publish expected (terminal CAS failed before fan-out). diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index a07d7cbe..af2f7a5c 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -535,6 +535,29 @@ func (s *StorageContractSuite) TestStorage_BuildCreateAndGet() { assert.NotEqual(t, build.ID, build.SpeculationPathID, "build ID and speculation path ID are distinct identifiers") } +// TestStorage_BatchBuild verifies immutable batch-to-build mapping behavior. +func (s *StorageContractSuite) TestStorage_BatchBuild() { + t := s.T() + ctx := s.ctx + + mapping := entity.BatchBuild{ + BatchID: "q/batch/build-mapping", + BuildID: "runner/build-mapping/1", + } + require.NoError(t, s.storage.GetBatchBuildStore().Create(ctx, mapping)) + + got, err := s.storage.GetBatchBuildStore().Get(ctx, mapping.BatchID) + require.NoError(t, err) + assert.Equal(t, mapping, got) + + other := mapping + other.BuildID = "runner/build-mapping/2" + require.ErrorIs(t, s.storage.GetBatchBuildStore().Create(ctx, other), storage.ErrAlreadyExists) + + _, err = s.storage.GetBatchBuildStore().Get(ctx, "q/batch/build-mapping-missing") + require.ErrorIs(t, err, storage.ErrNotFound) +} + // TestStorage_SpeculationPathBuildCreateAndGet verifies a path->build mapping // round-trips through the store unchanged. func (s *StorageContractSuite) TestStorage_SpeculationPathBuildCreateAndGet() { diff --git a/tool/sqsim/main_test.go b/tool/sqsim/main_test.go index 6d0efe15..9e5acb39 100644 --- a/tool/sqsim/main_test.go +++ b/tool/sqsim/main_test.go @@ -32,7 +32,20 @@ func TestRun(t *testing.T) { wantCode int wantOutput string }{ - {name: "list", args: []string{"list"}, wantCode: 0, wantOutput: "happy-path\n"}, + { + name: "list", + args: []string{"list"}, + wantCode: 0, + wantOutput: "build-failure\n" + + "build-status-recovery\n" + + "build-trigger-recovery\n" + + "happy-path\n" + + "load-1000\n" + + "merge-conflict\n" + + "merge-conflict-check-recovery\n" + + "merge-response-lost\n" + + "mixed-concurrent\n", + }, {name: "validate", args: []string{"validate", "happy-path"}, wantCode: 0, wantOutput: "happy-path is valid\n"}, {name: "unknown scenario", args: []string{"validate", "missing"}, wantCode: 1}, {name: "unknown command", args: []string{"missing"}, wantCode: 2},