Skip to content

refactor(storage): give the orchestrator its own aggregate, keeping store contracts at domain level #743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
behinddwalls merged 1 commit into from
Sep 25, 2026
Merged

refactor(storage): give the orchestrator its own aggregate, keeping store contracts at domain level #743

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,7 +386,7 @@ local-submitqueue-gateway-stop: ## Stop Gateway service

local-init-submitqueue-schemas: ## Manually apply all database schemas
@echo "Applying storage schema to mysql-app..."
@for file in submitqueue/extension/storage/mysql/schema/*.sql; do \
@for file in submitqueue/orchestrator/extension/storage/mysql/schema/*.sql; do \
echo " - Applying $$(basename $$file)..."; \
docker exec -i $(SUBMITQUEUE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \
done
Expand Down
6 changes: 3 additions & 3 deletions service/submitqueue/orchestrator/server/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,11 +60,11 @@ go_library(
"//submitqueue/extension/speculation/scorer/heuristic:go_default_library",
"//submitqueue/extension/speculation/speculator:go_default_library",
"//submitqueue/extension/speculation/speculator/standard:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mysql:go_default_library",
"//submitqueue/extension/validator:go_default_library",
"//submitqueue/extension/validator/fake:go_default_library",
"//submitqueue/orchestrator:go_default_library",
"//submitqueue/orchestrator/extension/storage:go_default_library",
"//submitqueue/orchestrator/extension/storage/mysql:go_default_library",
"@com_github_go_sql_driver_mysql//:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@in_gopkg_yaml_v3//:go_default_library",
Expand Down Expand Up @@ -125,7 +125,7 @@ go_test(
"//submitqueue/extension/conflict:go_default_library",
"//submitqueue/extension/speculation/scorer:go_default_library",
"//submitqueue/extension/speculation/speculator:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/orchestrator/extension/storage:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
Expand Down
9 changes: 5 additions & 4 deletions service/submitqueue/orchestrator/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ import (
"syscall"
"time"

orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"

_ "github.com/go-sql-driver/mysql"

"github.com/uber-go/tally"
Expand All @@ -44,11 +46,10 @@ import (
"github.com/uber/submitqueue/platform/pipeline"
servicemq "github.com/uber/submitqueue/service/messagequeue"
"github.com/uber/submitqueue/submitqueue/core/changeset"
"github.com/uber/submitqueue/submitqueue/extension/storage"
mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
"github.com/uber/submitqueue/submitqueue/extension/validator"
validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
"github.com/uber/submitqueue/submitqueue/orchestrator"
mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
Expand Down Expand Up @@ -438,15 +439,15 @@ func getEnv(key, defaultVal string) string {
}

// storageFactory adapts the MySQL storage backend's queue binding to the
// storage.Factory seam. Routing every queue to the single shared backend is
// orchstorage.Factory seam. Routing every queue to the single shared backend is
// this host's policy; a deployment that splits queues across backends swaps
// this adapter for a routing one.
type storageFactory struct {
backend *mysqlstorage.Storage
}

// For returns the queue-scoped store aggregate bound to the queue named in config.
func (f storageFactory) For(config storage.Config) (storage.Storage, error) {
func (f storageFactory) For(config orchstorage.Config) (orchstorage.Storage, error) {
return f.backend.For(config.QueueName)
}

Expand Down
19 changes: 10 additions & 9 deletions service/submitqueue/orchestrator/server/profiles.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ import (
"fmt"
nethttp "net/http"

orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"

"github.com/uber-go/tally"
"go.uber.org/zap"
"golang.org/x/oauth2"
Expand Down Expand Up @@ -54,7 +56,6 @@ import (
"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/heuristic"
"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
specstandard "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard"
"github.com/uber/submitqueue/submitqueue/extension/storage"
)

// Profile holds the per-queue extension implementations. Grouping them per
Expand All @@ -74,7 +75,7 @@ type Profile struct {
// Storage resolves the queue-scoped store aggregate for this queue. Every
// profile points at the shared backend by default; a deployment that
// splits queues across storage backends overrides this per queue.
Storage storage.Factory
Storage orchstorage.Factory

// Scorer holds this queue's ranking profile. There is no scoring stage: the
// scorer feeds the queue's speculator, which ranks candidate paths by how
Expand Down Expand Up @@ -143,10 +144,10 @@ func (p Profiles) ScorerFactory() scorer.Factory {
})
}

// StorageFactory returns a storage.Factory that routes each queue to its
// StorageFactory returns a orchstorage.Factory that routes each queue to its
// profile's storage backend before binding the queue-scoped store aggregate.
func (p Profiles) StorageFactory() storage.Factory {
return storageFunc(func(c storage.Config) (storage.Storage, error) {
func (p Profiles) StorageFactory() orchstorage.Factory {
return storageFunc(func(c orchstorage.Config) (orchstorage.Storage, error) {
return p.For(c.QueueName).Storage.For(c)
})
}
Expand All @@ -169,9 +170,9 @@ type analyzerFunc func(conflict.Config) (conflict.Analyzer, error)

func (f analyzerFunc) For(c conflict.Config) (conflict.Analyzer, error) { return f(c) }

type storageFunc func(storage.Config) (storage.Storage, error)
type storageFunc func(orchstorage.Config) (orchstorage.Storage, error)

func (f storageFunc) For(c storage.Config) (storage.Storage, error) { return f(c) }
func (f storageFunc) For(c orchstorage.Config) (orchstorage.Storage, error) { return f(c) }

type scorerFunc func(scorer.Config) (scorer.Scorer, error)

Expand All @@ -192,7 +193,7 @@ func newProfiles(
logger *zap.Logger,
scope tally.Scope,
resolver changeset.Resolver,
stores storage.Factory,
stores orchstorage.Factory,
cfg profilesConfig,
) (Profiles, error) {
b := &profileBuilder{
Expand Down Expand Up @@ -245,7 +246,7 @@ type profileBuilder struct {
logger *zap.Logger
scope tally.Scope
resolver changeset.Resolver
stores storage.Factory
stores orchstorage.Factory
built map[string]any
}

Expand Down
7 changes: 4 additions & 3 deletions service/submitqueue/orchestrator/server/profiles_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ import (
"errors"
"testing"

orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

Expand All @@ -28,7 +30,6 @@ import (
"github.com/uber/submitqueue/submitqueue/extension/conflict"
"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer"
"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
"github.com/uber/submitqueue/submitqueue/extension/storage"
)

// recorder captures the queue name each seam's factory was handed, so a test
Expand Down Expand Up @@ -66,7 +67,7 @@ func profileRecording(rec *recorder) Profile {
rec.analyzer = c.QueueName
return nil, nil
}),
Storage: storageFunc(func(c storage.Config) (storage.Storage, error) {
Storage: storageFunc(func(c orchstorage.Config) (orchstorage.Storage, error) {
rec.storage = c.QueueName
return nil, nil
}),
Expand Down Expand Up @@ -109,7 +110,7 @@ func TestProfilesForwardQueueNameToFactories(t *testing.T) {
require.NoError(t, err)
_, err = profiles.AnalyzerFactory().For(conflict.Config{QueueName: tt.queue})
require.NoError(t, err)
_, err = profiles.StorageFactory().For(storage.Config{QueueName: tt.queue})
_, err = profiles.StorageFactory().For(orchstorage.Config{QueueName: tt.queue})
require.NoError(t, err)
_, err = profiles.ScorerFactory().For(scorer.Config{QueueName: tt.queue})
require.NoError(t, err)
Expand Down
2 changes: 2 additions & 0 deletions submitqueue/core/batch/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ go_library(
deps = [
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/orchestrator/extension/storage:go_default_library",
"@org_golang_x_sync//errgroup:go_default_library",
],
)
Expand All @@ -28,6 +29,7 @@ go_test(
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"//submitqueue/orchestrator/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
Expand Down
5 changes: 3 additions & 2 deletions submitqueue/core/batch/find.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ import (
"sort"

"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storage "github.com/uber/submitqueue/submitqueue/extension/storage"
orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
)

// FindByRequestID resolves every batch attempt associated with a request,
Expand All @@ -35,7 +36,7 @@ import (
//
// Unlike ListByStates, which treats a dangling membership record as store
// corruption, a dangling association is an expected retry artifact.
func FindByRequestID(ctx context.Context, store storage.Storage, requestID string) ([]entity.Batch, int, error) {
func FindByRequestID(ctx context.Context, store orchstorage.Storage, requestID string) ([]entity.Batch, int, error) {
associations, err := store.GetRequestBatchStore().GetByRequestID(ctx, requestID)
if err != nil {
return nil, 0, fmt.Errorf("failed to get batch associations for request %s: %w", requestID, err)
Expand Down
5 changes: 3 additions & 2 deletions submitqueue/core/batch/find_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock"
)

const testRequestID = "monorepo/4"
Expand All @@ -36,11 +37,11 @@ func association(batchID string) entity.RequestBatch {
}

// findStores wires a MockStorage over a batch store and a request-batch store.
func findStores(t *testing.T) (*storagemock.MockStorage, *storagemock.MockBatchStore, *storagemock.MockRequestBatchStore) {
func findStores(t *testing.T) (*orchstoragemock.MockStorage, *storagemock.MockBatchStore, *storagemock.MockRequestBatchStore) {
t.Helper()

ctrl := gomock.NewController(t)
mockStorage := storagemock.NewMockStorage(ctrl)
mockStorage := orchstoragemock.NewMockStorage(ctrl)
mockBatchStore := storagemock.NewMockBatchStore(ctrl)
mockAssociationStore := storagemock.NewMockRequestBatchStore(ctrl)
mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes()
Expand Down
4 changes: 2 additions & 2 deletions submitqueue/core/batch/list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import (
"golang.org/x/sync/errgroup"

"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
)

// hydrateConcurrency bounds the parallel per-key batch reads a single
Expand All @@ -39,7 +39,7 @@ const hydrateConcurrency = 16
// A candidate ID whose batch does not exist is returned as an error rather than
// skipped: batch rows are never deleted, so a dangling record means the store is
// inconsistent, not that the batch concluded.
func ListByStates(ctx context.Context, store storage.Storage, states []entity.BatchState) ([]entity.Batch, error) {
func ListByStates(ctx context.Context, store orchstorage.Storage, states []entity.BatchState) ([]entity.Batch, error) {
wanted := make(map[entity.BatchState]bool, len(states))
seen := make(map[string]bool)
var ids []string
Expand Down
6 changes: 3 additions & 3 deletions submitqueue/core/batch/transition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ import (
"fmt"

"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
)

// Transition moves a batch to newState: it performs the optimistic-locking CAS on
Expand All @@ -52,7 +52,7 @@ import (
// applied — the CAS may have committed with the record move incomplete — and the
// caller is expected to let redelivery retry; the retry's already-in-target-state
// branch repairs the record via EnsureRecord.
func Transition(ctx context.Context, store storage.Storage, batch entity.Batch, newState entity.BatchState) (entity.Batch, error) {
func Transition(ctx context.Context, store orchstorage.Storage, batch entity.Batch, newState entity.BatchState) (entity.Batch, error) {
oldState := batch.State
newVersion := batch.Version + 1
updated := batch
Expand All @@ -78,7 +78,7 @@ func Transition(ctx context.Context, store storage.Storage, batch entity.Batch,
// the repair half of the transition protocol: idempotent redelivery branches that
// skip the CAS because the batch is already in the target state call this instead,
// covering a prior attempt that crashed between the CAS and the record move.
func EnsureRecord(ctx context.Context, store storage.Storage, batch entity.Batch) error {
func EnsureRecord(ctx context.Context, store orchstorage.Storage, batch entity.Batch) error {
record := entity.QueueBatchState{Queue: batch.Queue, State: batch.State, BatchID: batch.ID}
if err := store.GetQueueBatchStateStore().Put(ctx, record); err != nil {
return fmt.Errorf("failed to put queue batch state record for batch %s under state %s: %w", batch.ID, batch.State, err)
Expand Down
5 changes: 3 additions & 2 deletions submitqueue/core/batch/transition_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,15 +26,16 @@ import (
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock"
)

// testStores wires a MockStorage whose batch and queue-batch-state accessors
// return the two mocks the tests set expectations on.
func testStores(t *testing.T) (*storagemock.MockStorage, *storagemock.MockBatchStore, *storagemock.MockQueueBatchStateStore) {
func testStores(t *testing.T) (*orchstoragemock.MockStorage, *storagemock.MockBatchStore, *storagemock.MockQueueBatchStateStore) {
t.Helper()

ctrl := gomock.NewController(t)
mockStorage := storagemock.NewMockStorage(ctrl)
mockStorage := orchstoragemock.NewMockStorage(ctrl)
mockBatchStore := storagemock.NewMockBatchStore(ctrl)
mockRecordStore := storagemock.NewMockQueueBatchStateStore(ctrl)
mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes()
Expand Down
3 changes: 2 additions & 1 deletion submitqueue/core/changeset/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ go_library(
deps = [
"//platform/base/change:go_default_library",
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/orchestrator/extension/storage:go_default_library",
],
)

Expand All @@ -24,6 +24,7 @@ go_test(
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"//submitqueue/orchestrator/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
Expand Down
10 changes: 5 additions & 5 deletions submitqueue/core/changeset/resolver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,25 +20,25 @@ import (

"github.com/uber/submitqueue/platform/base/change"
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage"
)

// resolver is the store-backed Resolver. It holds the storage factory and
// resolves the batch's queue-scoped request and change stores per call, since
// every resolution is for exactly one batch and the batch names its queue.
type resolver struct {
stores storage.Factory
stores orchstorage.Factory
}

// New returns a Resolver backed by the given storage factory.
func New(stores storage.Factory) Resolver {
func New(stores orchstorage.Factory) Resolver {
return resolver{stores: stores}
}

// ChangesForBatch resolves a batch's requests to their raw changes, in
// batch.Contains order.
func (r resolver) ChangesForBatch(ctx context.Context, batch entity.Batch) ([]change.Change, error) {
store, err := r.stores.For(storage.Config{QueueName: batch.Queue})
store, err := r.stores.For(orchstorage.Config{QueueName: batch.Queue})
if err != nil {
return nil, fmt.Errorf("failed to resolve storage for queue %q: %w", batch.Queue, err)
}
Expand All @@ -57,7 +57,7 @@ func (r resolver) ChangesForBatch(ctx context.Context, batch entity.Batch) ([]ch
// ChangeInfo per claimed URI, owned by the requesting request, aggregated across
// the whole batch.
func (r resolver) DetailedForBatch(ctx context.Context, batch entity.Batch) (entity.BatchChanges, error) {
store, err := r.stores.For(storage.Config{QueueName: batch.Queue})
store, err := r.stores.For(orchstorage.Config{QueueName: batch.Queue})
if err != nil {
return entity.BatchChanges{}, fmt.Errorf("failed to resolve storage for queue %q: %w", batch.Queue, err)
}
Expand Down
5 changes: 3 additions & 2 deletions submitqueue/core/changeset/resolver_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,16 @@ import (
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
orchstoragemock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock"
)

// newTestResolver builds a Resolver over mock stores exposed through a mock
// storage factory that resolves every queue to the same aggregate.
func newTestResolver(ctrl *gomock.Controller, reqs storage.RequestStore, changes storage.ChangeStore) Resolver {
store := storagemock.NewMockStorage(ctrl)
store := orchstoragemock.NewMockStorage(ctrl)
store.EXPECT().GetRequestStore().Return(reqs).AnyTimes()
store.EXPECT().GetChangeStore().Return(changes).AnyTimes()
f := storagemock.NewMockFactory(ctrl)
f := orchstoragemock.NewMockFactory(ctrl)
f.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes()
return New(f)
}
Expand Down
2 changes: 2 additions & 0 deletions submitqueue/core/request/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ go_library(
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/gateway/extension/storage:go_default_library",
"//submitqueue/orchestrator/extension/storage:go_default_library",
],
)

Expand All @@ -40,6 +41,7 @@ go_test(
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"//submitqueue/gateway/extension/storage/mock:go_default_library",
"//submitqueue/orchestrator/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
"@org_uber_go_mock//gomock:go_default_library",
Expand Down
Loading
Loading