From 529ce0c12fa7d0217f3142d86085eed2e8bee5f0 Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Wed, 15 Jul 2026 23:22:14 -0700 Subject: [PATCH] feat(sqsim): wire modeled adapters into local stack --- service/runway/server/BUILD.bazel | 17 +++++- service/runway/server/docker-compose.yml | 3 ++ service/runway/server/main.go | 53 ++++++++++++++---- service/runway/server/main_test.go | 51 ++++++++++++++++++ service/submitqueue/docker-compose.yml | 6 +++ .../submitqueue/gateway/server/queues.yaml | 1 + .../orchestrator/server/BUILD.bazel | 20 +++++-- .../orchestrator/server/docker-compose.yml | 3 ++ .../submitqueue/orchestrator/server/main.go | 54 +++++++++++++++---- .../orchestrator/server/main_test.go | 42 +++++++++++++++ 10 files changed, 228 insertions(+), 22 deletions(-) create mode 100644 service/runway/server/main_test.go create mode 100644 service/submitqueue/orchestrator/server/main_test.go diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index 49067b0f..bafb1cfc 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_binary", "go_library") +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") go_library( name = "go_default_library", @@ -19,6 +19,8 @@ go_library( "//runway/controller/mergeconflictcheck:go_default_library", "//runway/extension/merger:go_default_library", "//runway/extension/merger/noop:go_default_library", + "//sqsim/adapter/merger:go_default_library", + "//sqsim/model:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", @@ -32,3 +34,16 @@ go_binary( embed = [":go_default_library"], visibility = ["//visibility:public"], ) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":go_default_library"], + deps = [ + "//runway/extension/merger:go_default_library", + "//sqsim/model:go_default_library", + "//sqsim/scenarios:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/service/runway/server/docker-compose.yml b/service/runway/server/docker-compose.yml index d39026ab..627995de 100644 --- a/service/runway/server/docker-compose.yml +++ b/service/runway/server/docker-compose.yml @@ -41,6 +41,9 @@ services: # Queue infrastructure connection - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true - HOSTNAME=runway-dev + - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} + volumes: + - ${SQSIM_PROFILE_DIR:-../../../sqsim}:/sqsim:ro depends_on: mysql-queue: condition: service_healthy diff --git a/service/runway/server/main.go b/service/runway/server/main.go index ee6a0936..28b41a13 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -41,6 +41,8 @@ import ( "github.com/uber/submitqueue/runway/controller/mergeconflictcheck" "github.com/uber/submitqueue/runway/extension/merger" "github.com/uber/submitqueue/runway/extension/merger/noop" + simmerger "github.com/uber/submitqueue/sqsim/adapter/merger" + "github.com/uber/submitqueue/sqsim/model" "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/reflection" @@ -115,6 +117,11 @@ func run() error { metricsWgDone.Wait() }() + sqsimRuntime, err := loadSQSimRuntime(os.Getenv("SQSIM_SCENARIO_PATH")) + if err != nil { + return err + } + queueDSN := os.Getenv("QUEUE_MYSQL_DSN") if queueDSN == "" { return fmt.Errorf("QUEUE_MYSQL_DSN environment variable is required") @@ -147,14 +154,18 @@ func run() error { return fmt.Errorf("failed to create topic registry: %w", err) } + classifiers := []errs.Classifier{ + genericerrs.Classifier, + mysqlerrs.Classifier, + } + if sqsimRuntime != nil { + classifiers = append(classifiers, model.Classifier) + } primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, - errs.NewClassifierProcessor( - genericerrs.Classifier, - mysqlerrs.Classifier, - ), + errs.NewClassifierProcessor(classifiers...), ) - mergerFactory := newMergerFactory() + mergerFactory := newMergerFactory(sqsimRuntime) mergeConflictCheckController := mergeconflictcheck.NewController(mergeconflictcheck.Params{ Logger: logger.Sugar(), @@ -245,16 +256,40 @@ func run() error { // newMergerFactory returns a merger.Factory for the example server. The noop // implementation always succeeds; a real deployment wires a VCS-backed factory. -func newMergerFactory() merger.Factory { - return &noopMergerFactory{} +func newMergerFactory(sqsimRuntime *model.Runtime) merger.Factory { + factory := &routingMergerFactory{} + if sqsimRuntime != nil { + factory.sqsim = simmerger.New(sqsimRuntime, "sqsim") + } + return factory } -type noopMergerFactory struct{} +type routingMergerFactory struct { + sqsim merger.Merger +} -func (f *noopMergerFactory) For(_ merger.Config) (merger.Merger, error) { +func (f *routingMergerFactory) For(cfg merger.Config) (merger.Merger, error) { + if cfg.QueueName == "sqsim" && f.sqsim != nil { + return f.sqsim, nil + } return noop.New(), nil } +func loadSQSimRuntime(path string) (*model.Runtime, error) { + if path == "" { + return nil, nil + } + profile, err := model.Load(path) + if err != nil { + return nil, fmt.Errorf("load SQSim profile: %w", err) + } + runtime, err := model.NewRuntime(profile, model.RealClock{}) + if err != nil { + return nil, fmt.Errorf("initialize SQSim runtime: %w", err) + } + return runtime, nil +} + // newTopicRegistry builds the TopicRegistry for Runway's merge queues. Inbound // topics (merge-conflict-check, merge) have subscriptions; outbound signal topics // are publish-only. diff --git a/service/runway/server/main_test.go b/service/runway/server/main_test.go new file mode 100644 index 00000000..3d8e6c6b --- /dev/null +++ b/service/runway/server/main_test.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 main + +import ( + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/runway/extension/merger" + "github.com/uber/submitqueue/sqsim/model" + "github.com/uber/submitqueue/sqsim/scenarios" +) + +func TestSQSimMergerFactoryIsOptIn(t *testing.T) { + defaultFactory := newMergerFactory(nil) + first, err := defaultFactory.For(merger.Config{QueueName: "sqsim"}) + require.NoError(t, err) + second, err := defaultFactory.For(merger.Config{QueueName: "sqsim"}) + require.NoError(t, err) + assert.NotSame(t, first, second) + + scenario, err := scenarios.HappyPath() + require.NoError(t, err) + profile, err := model.Compile("happy-path", scenario) + require.NoError(t, err) + path := filepath.Join(t.TempDir(), "profile.json") + require.NoError(t, model.Write(path, profile)) + runtime, err := loadSQSimRuntime(path) + require.NoError(t, err) + + sqsimFactory := newMergerFactory(runtime) + first, err = sqsimFactory.For(merger.Config{QueueName: "sqsim"}) + require.NoError(t, err) + second, err = sqsimFactory.For(merger.Config{QueueName: "sqsim"}) + require.NoError(t, err) + assert.Same(t, first, second) +} diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index e8d2c4bd..f0d92d8a 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -76,6 +76,9 @@ services: # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true - HOSTNAME=orchestrator-dev + - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} + volumes: + - ${SQSIM_PROFILE_DIR:-../../sqsim}:/sqsim:ro depends_on: mysql-app: condition: service_healthy @@ -98,6 +101,9 @@ services: # Queue infrastructure connection (shared with the orchestrator) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true - HOSTNAME=runway-dev + - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} + volumes: + - ${SQSIM_PROFILE_DIR:-../../sqsim}:/sqsim:ro depends_on: mysql-queue: condition: service_healthy diff --git a/service/submitqueue/gateway/server/queues.yaml b/service/submitqueue/gateway/server/queues.yaml index df7571fe..0b3764b8 100644 --- a/service/submitqueue/gateway/server/queues.yaml +++ b/service/submitqueue/gateway/server/queues.yaml @@ -4,6 +4,7 @@ # VCS configuration lives in the extension factory implementations wired in the # server, not here. queues: + - name: sqsim - name: test-queue - name: e2e-test-queue - name: e2e-cancel-queue diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 2ccb3493..ff47a216 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_binary", "go_library") +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") exports_files( ["docker-compose.yml"], @@ -6,7 +6,7 @@ exports_files( ) go_library( - name = "orchestrator_lib", + name = "go_default_library", srcs = ["main.go"], importpath = "github.com/uber/submitqueue/service/submitqueue/orchestrator/server", visibility = ["//visibility:private"], @@ -22,6 +22,8 @@ go_library( "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//platform/http:go_default_library", + "//sqsim/adapter/buildrunner:go_default_library", + "//sqsim/model:go_default_library", "//submitqueue/core/changeset:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", @@ -73,6 +75,18 @@ go_library( go_binary( name = "orchestrator", - embed = [":orchestrator_lib"], + embed = [":go_default_library"], visibility = ["//visibility:public"], ) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":go_default_library"], + deps = [ + "//sqsim/model:go_default_library", + "//sqsim/scenarios:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/service/submitqueue/orchestrator/server/docker-compose.yml b/service/submitqueue/orchestrator/server/docker-compose.yml index 90db979f..f67a4359 100644 --- a/service/submitqueue/orchestrator/server/docker-compose.yml +++ b/service/submitqueue/orchestrator/server/docker-compose.yml @@ -54,6 +54,9 @@ services: # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true - HOSTNAME=orchestrator-dev + - SQSIM_SCENARIO_PATH=${SQSIM_SCENARIO_PATH:-} + volumes: + - ${SQSIM_PROFILE_DIR:-../../../../sqsim}:/sqsim:ro depends_on: mysql-app: condition: service_healthy diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index eb1ce9ab..2f3a583c 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -42,6 +42,8 @@ import ( extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/http" + simbuildrunner "github.com/uber/submitqueue/sqsim/adapter/buildrunner" + "github.com/uber/submitqueue/sqsim/model" "github.com/uber/submitqueue/submitqueue/core/changeset" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -159,6 +161,11 @@ func run() error { metricsWgDone.Wait() }() + sqsimRuntime, err := loadSQSimRuntime(os.Getenv("SQSIM_SCENARIO_PATH")) + if err != nil { + return err + } + // Open app database connection for counter // Docker Compose healthchecks ensure MySQL is ready before service starts appDSN := os.Getenv("MYSQL_DSN") @@ -220,14 +227,18 @@ func run() error { // so every non-nil error from a DLQ controller is forced retryable — // reconciliation must redeliver on any failure because the DLQ // subscriptions are final destinations (there is no further DLQ). + classifiers := []errs.Classifier{ + genericerrs.Classifier, + // Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) + // both run on the same MySQL driver, so a single classifier covers + // errors surfaced from either backend. + mysqlerrs.Classifier, + } + if sqsimRuntime != nil { + classifiers = append(classifiers, model.Classifier) + } primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, - errs.NewClassifierProcessor( - genericerrs.Classifier, - // Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) - // both run on the same MySQL driver, so a single classifier covers - // errors surfaced from either backend. - mysqlerrs.Classifier, - ), + errs.NewClassifierProcessor(classifiers...), ) dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry, errs.AlwaysRetryableProcessor, @@ -238,7 +249,8 @@ func run() error { // back to a baseline profile for queues without an explicit entry. This is // the single place queue topology is known; the extension packages stay // queue-agnostic. - queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore())) + resolver := changeset.New(store.GetRequestStore(), store.GetChangeStore()) + queues, err := newQueueRegistry(logger, scope, resolver, sqsimRuntime) if err != nil { return fmt.Errorf("failed to build queue registry: %w", err) } @@ -893,7 +905,7 @@ const defaultPrioritizationLimit = 1000 // conflict analyzer. Queues without an explicit profile fall back to the // baseline. This is the one place queue topology lives; extension packages stay // queue-agnostic. -func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (queueRegistry, error) { +func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver, sqsimRuntime *model.Runtime) (queueRegistry, error) { cp, err := newChangeProvider(logger, scope) if err != nil { return queueRegistry{}, fmt.Errorf("failed to create change provider: %w", err) @@ -969,6 +981,14 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset. fileOverlapQueue := base fileOverlapQueue.analyzer = fileoverlap.New(resolver) + // sqsim: maximum request-level parallelism for scenario and load runs. The + // modeled Build Runner is selected only when an explicit profile is loaded. + sqsimQueue := base + sqsimQueue.analyzer = none.New() + if sqsimRuntime != nil { + sqsimQueue.buildRunner = simbuildrunner.New(sqsimRuntime, resolver, "sqsim") + } + return queueRegistry{ def: base, byQueue: map[string]queueExtensions{ @@ -976,6 +996,22 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset. "e2e-test-queue": e2eQueue, "e2e-conflict-error-queue": conflictErrQueue, "file-overlap-queue": fileOverlapQueue, + "sqsim": sqsimQueue, }, }, nil } + +func loadSQSimRuntime(path string) (*model.Runtime, error) { + if path == "" { + return nil, nil + } + profile, err := model.Load(path) + if err != nil { + return nil, fmt.Errorf("load SQSim profile: %w", err) + } + runtime, err := model.NewRuntime(profile, model.RealClock{}) + if err != nil { + return nil, fmt.Errorf("initialize SQSim runtime: %w", err) + } + return runtime, nil +} diff --git a/service/submitqueue/orchestrator/server/main_test.go b/service/submitqueue/orchestrator/server/main_test.go new file mode 100644 index 00000000..b90d52b3 --- /dev/null +++ b/service/submitqueue/orchestrator/server/main_test.go @@ -0,0 +1,42 @@ +// 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 main + +import ( + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim/model" + "github.com/uber/submitqueue/sqsim/scenarios" +) + +func TestLoadSQSimRuntimeIsOptIn(t *testing.T) { + runtime, err := loadSQSimRuntime("") + require.NoError(t, err) + assert.Nil(t, runtime) + + scenario, err := scenarios.HappyPath() + require.NoError(t, err) + profile, err := model.Compile("happy-path", scenario) + require.NoError(t, err) + path := filepath.Join(t.TempDir(), "profile.json") + require.NoError(t, model.Write(path, profile)) + + runtime, err = loadSQSimRuntime(path) + require.NoError(t, err) + assert.NotNil(t, runtime) +}