Skip to content

refactor(storage): move the gateway's four stores into submitqueue/gateway/extension/storage #741

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

Open
roychying wants to merge 1 commit into from
Open

refactor(storage): move the gateway's four stores into submitqueue/gateway/extension/storage #741

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
6 changes: 5 additions & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,6 +390,10 @@ local-init-submitqueue-schemas: ## Manually apply all database schemas
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
@for file in submitqueue/gateway/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
@echo "Applying counter schema to mysql-app..."
@for file in platform/extension/counter/mysql/schema/*.sql; do \
echo " - Applying $$(basename $$file)..."; \
Expand Down Expand Up @@ -579,7 +583,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service

mocks: ## Generate mock files using mockgen
@echo "Generating mocks..."
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/gateway/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@echo "Mocks generated successfully!"

proto: ## Generate protobuf files from .proto definitions
Expand Down
4 changes: 2 additions & 2 deletions service/submitqueue/gateway/server/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,10 @@ go_library(
"//submitqueue/core/request:go_default_library",
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/extension/queueconfig/yaml:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mysql:go_default_library",
"//submitqueue/gateway/controller:go_default_library",
"//submitqueue/gateway/controller/log:go_default_library",
"//submitqueue/gateway/extension/storage:go_default_library",
"//submitqueue/gateway/extension/storage/mysql: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",
Expand Down
6 changes: 3 additions & 3 deletions service/submitqueue/gateway/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,10 @@ import (
requestcore "github.com/uber/submitqueue/submitqueue/core/request"
"github.com/uber/submitqueue/submitqueue/core/topickey"
yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml"
"github.com/uber/submitqueue/submitqueue/extension/storage"
mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
"github.com/uber/submitqueue/submitqueue/gateway/controller"
logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log"
storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
mysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
Expand Down Expand Up @@ -375,7 +375,7 @@ func run() error {
// directly so reconciliation does not depend on another asynchronous path.
logConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
errs.NewClassifierProcessor(
// Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql)
// Storage (submitqueue/gateway/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.
genericerrs.Classifier,
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 @@ -17,6 +17,7 @@ go_library(
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/gateway/extension/storage:go_default_library",
],
)

Expand All @@ -38,6 +39,7 @@ go_test(
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"//submitqueue/gateway/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
13 changes: 7 additions & 6 deletions submitqueue/core/request/materializer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@ import (
"slices"

"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
basestorage "github.com/uber/submitqueue/submitqueue/extension/storage"
storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"

@behinddwalls behinddwalls Sep 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agent-written.

This is the dependency inversion caused by moving the interfaces: domain core/ now imports a service package. Keeping the interfaces in the domain storage contract avoids it.

@roychying roychying Sep 23, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But even if we keep the store interfaces in domain level, Materializer still takes Factory and resolves per queue (m.stores.For(...{QueueName: log.Queue})), so it imports gateway/extension/storage either way. Same for terminate.go, batch/{find,list,transition}.go and changeset/resolver.go, which also take aggregates. I don't think we can actually avoid importing a service package here

@behinddwalls behinddwalls Sep 23, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i guess the question is? should this also be moved to gateway now?

@roychying roychying Sep 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Technically yes. This is the last section of the RFC (#740). I'd raised it there as a follow-on for a later stage and keep this stack focusing on the schema refactor. Happy to fold it in here if you like.

)

// Materializer appends request logs and projects the winning public request state.
Expand Down Expand Up @@ -70,7 +71,7 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
updated.Metadata = cloneMetadata(log.Metadata)

if err := summaries.Update(ctx, updated, oldVersion, newVersion); err != nil {
if errors.Is(err, storage.ErrVersionMismatch) {
if errors.Is(err, basestorage.ErrVersionMismatch) {
continue
}
return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err)
Expand All @@ -93,12 +94,12 @@ func (m *Materializer) repairPublicProjections(ctx context.Context, stores stora
queueSummaries := stores.GetRequestQueueSummaryStore()
for {
current, err := queueSummaries.Get(ctx, desired.ReceivedAtMs, desired.RequestID)
if errors.Is(err, storage.ErrNotFound) {
if errors.Is(err, basestorage.ErrNotFound) {
if err := m.createURIMappings(ctx, stores, authoritative); err != nil {
return err
}
if err := queueSummaries.Create(ctx, desired); err != nil {
if errors.Is(err, storage.ErrAlreadyExists) {
if errors.Is(err, basestorage.ErrAlreadyExists) {
continue
}
return fmt.Errorf("failed to recreate queue summary request_id=%s: %w", desired.RequestID, err)
Expand All @@ -116,7 +117,7 @@ func (m *Materializer) repairPublicProjections(ctx context.Context, stores stora
return nil
}
if err := queueSummaries.Update(ctx, desired, current.Version, desired.Version); err != nil {
if errors.Is(err, storage.ErrVersionMismatch) {
if errors.Is(err, basestorage.ErrVersionMismatch) {
continue
}
return fmt.Errorf("failed to update queue summary request_id=%s: %w", desired.RequestID, err)
Expand All @@ -134,7 +135,7 @@ func (m *Materializer) createURIMappings(ctx context.Context, stores storage.Sto
ReceivedAtMs: summary.ReceivedAtMs,
RequestID: summary.RequestID,
}
if err := uris.Create(ctx, mapping); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
if err := uris.Create(ctx, mapping); err != nil && !errors.Is(err, basestorage.ErrAlreadyExists) {
return fmt.Errorf("failed to create request URI mapping request_id=%s change_uri=%s: %w", summary.RequestID, changeURI, err)
}
}
Expand Down
5 changes: 3 additions & 2 deletions submitqueue/core/request/materializer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock"
"go.uber.org/mock/gomock"
)

Expand Down Expand Up @@ -316,12 +317,12 @@ func materializerStores(ctrl *gomock.Controller) (*Materializer, *storagemock.Mo
queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl)
uriStore := storagemock.NewMockRequestURIStore(ctrl)
logStore := storagemock.NewMockRequestLogStore(ctrl)
queueScoped := storagemock.NewMockStorage(ctrl)
queueScoped := gwstoragemock.NewMockStorage(ctrl)
queueScoped.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes()
queueScoped.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
queueScoped.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes()
queueScoped.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes()
factory := storagemock.NewMockFactory(ctrl)
factory := gwstoragemock.NewMockFactory(ctrl)
factory.EXPECT().For(gomock.Any()).Return(queueScoped, nil).AnyTimes()
return NewMaterializer(factory), summaryStore, queueStore, uriStore, logStore
}
Expand Down
2 changes: 1 addition & 1 deletion submitqueue/core/request/request.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import (
"fmt"

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

// CurrentState holds the current request status obtained from the request log. It is eventually consistent with the request status in the request store. It might take some time to converge, typically no more than a few seconds.
Expand Down
13 changes: 13 additions & 0 deletions submitqueue/extension/storage/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,3 +35,16 @@ go_test(
"@com_github_stretchr_testify//assert:go_default_library",
],
)

# The whole SubmitQueue storage schema. Each service owns the tables it reads,
# so a deployment that separates them depends on the service package it runs;
# one that colocates them — the end-to-end tests, the local compose stack —
# takes this and does not have to track which services exist.
filegroup(
name = "schema",
srcs = [
"//submitqueue/extension/storage/mysql/schema",
"//submitqueue/gateway/extension/storage/mysql/schema",
],
visibility = ["//visibility:public"],
)
56 changes: 0 additions & 56 deletions submitqueue/extension/storage/mock/storage_mock.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 0 additions & 8 deletions submitqueue/extension/storage/mysql/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,7 @@ go_library(
"path_build_store.go",
"queue_batch_state_store.go",
"request_batch_store.go",
"request_log_store.go",
"request_queue_summary_store.go",
"request_store.go",
"request_summary_store.go",
"request_uri_store.go",
"speculation_path_set_store.go",
"storage.go",
],
Expand All @@ -38,11 +34,7 @@ go_test(
"change_store_test.go",
"queue_batch_state_store_test.go",
"request_batch_store_test.go",
"request_log_store_test.go",
"request_queue_summary_store_test.go",
"request_store_test.go",
"request_summary_store_test.go",
"request_uri_store_test.go",
"speculation_path_set_store_test.go",
"storage_test.go",
],
Expand Down
28 changes: 0 additions & 28 deletions submitqueue/extension/storage/mysql/storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,10 +59,6 @@ func (s *Storage) For(queueName string) (storage.Storage, error) {
buildStore: NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName),
speculationPathSetStore: NewSpeculationPathSetStore(s.db, s.scope.SubScope("speculation_path_set_store"), queueName),
pathBuildStore: NewPathBuildStore(s.db, s.scope.SubScope("path_build_store"), queueName),
requestQueueStore: NewRequestQueueSummaryStore(s.db, s.scope.SubScope("request_queue_summary_store"), queueName),
requestSummaryStore: NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName),
requestLogStore: NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName),
requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName),
}, nil
}

Expand All @@ -82,10 +78,6 @@ type boundStorage struct {
buildStore storage.BuildStore
speculationPathSetStore storage.SpeculationPathSetStore
pathBuildStore storage.PathBuildStore
requestQueueStore storage.RequestQueueSummaryStore
requestSummaryStore storage.RequestSummaryStore
requestLogStore storage.RequestLogStore
requestURIStore storage.RequestURIStore
}

// Verify boundStorage implements the queue-scoped aggregate at compile time.
Expand Down Expand Up @@ -135,23 +127,3 @@ func (f *boundStorage) GetSpeculationPathSetStore() storage.SpeculationPathSetSt
func (f *boundStorage) GetPathBuildStore() storage.PathBuildStore {
return f.pathBuildStore
}

// GetRequestQueueSummaryStore returns the bound MySQL-backed RequestQueueSummaryStore.
func (f *boundStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore {
return f.requestQueueStore
}

// GetRequestSummaryStore returns the bound MySQL-backed RequestSummaryStore.
func (f *boundStorage) GetRequestSummaryStore() storage.RequestSummaryStore {
return f.requestSummaryStore
}

// GetRequestLogStore returns the bound MySQL-backed RequestLogStore.
func (f *boundStorage) GetRequestLogStore() storage.RequestLogStore {
return f.requestLogStore
}

// GetRequestURIStore returns the bound MySQL-backed RequestURIStore.
func (f *boundStorage) GetRequestURIStore() storage.RequestURIStore {
return f.requestURIStore
}
4 changes: 0 additions & 4 deletions submitqueue/extension/storage/mysql/storage_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,6 @@ func TestNewStorage(t *testing.T) {
assert.NotNil(t, bound.GetQueueBatchStateStore())
assert.NotNil(t, bound.GetBuildStore())
assert.NotNil(t, bound.GetSpeculationPathSetStore())
assert.NotNil(t, bound.GetRequestQueueSummaryStore())
assert.NotNil(t, bound.GetRequestSummaryStore())
assert.NotNil(t, bound.GetRequestLogStore())
assert.NotNil(t, bound.GetRequestURIStore())

_, err = s.For("")
assert.Error(t, err, "resolving an empty queue name must fail")
Expand Down
12 changes: 0 additions & 12 deletions submitqueue/extension/storage/storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,16 +94,4 @@ type Storage interface {

// GetPathBuildStore returns the PathBuildStore instance.
GetPathBuildStore() PathBuildStore

// GetRequestQueueSummaryStore returns the RequestQueueSummaryStore instance.
GetRequestQueueSummaryStore() RequestQueueSummaryStore

// GetRequestSummaryStore returns the RequestSummaryStore instance.
GetRequestSummaryStore() RequestSummaryStore

// GetRequestLogStore returns the RequestLogStore instance.
GetRequestLogStore() RequestLogStore

// GetRequestURIStore returns the RequestURIStore instance.
GetRequestURIStore() RequestURIStore
}
3 changes: 3 additions & 0 deletions submitqueue/gateway/controller/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ go_library(
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/queueconfig:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/gateway/extension/storage:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
Expand Down Expand Up @@ -61,6 +62,8 @@ go_test(
"//submitqueue/extension/queueconfig/mock:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"//submitqueue/gateway/extension/storage:go_default_library",
"//submitqueue/gateway/extension/storage/mock: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
5 changes: 3 additions & 2 deletions submitqueue/gateway/controller/cancel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ import (
requestcore "github.com/uber/submitqueue/submitqueue/core/request"
"github.com/uber/submitqueue/submitqueue/core/topickey"
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/storage"
basestorage "github.com/uber/submitqueue/submitqueue/extension/storage"
storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
"go.uber.org/zap"
)

Expand Down Expand Up @@ -97,7 +98,7 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest)
return fmt.Errorf("failed to resolve storage for queue %q: %w", req.Queue, err)
}
if _, err := stores.GetRequestSummaryStore().Get(ctx, req.ID); err != nil {
if storage.IsNotFound(err) {
if basestorage.IsNotFound(err) {
metrics.NamedCounter(c.metricsScope, opName, "not_found", 1)
return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID})
}
Expand Down
2 changes: 1 addition & 1 deletion submitqueue/gateway/controller/land.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import (
"github.com/uber/submitqueue/submitqueue/core/topickey"
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/queueconfig"
"github.com/uber/submitqueue/submitqueue/extension/storage"
storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
"go.uber.org/zap"
)

Expand Down
Loading
Loading